ft-orchestration add docstrings and type hints

Author: pascal260303

tools/ft-orchestration/src/probe/cento.py

@@ -57,15 +57,13 @@
 @dataclass
 class CentoSettings(ABC):
     """
-    These settings can be set in the probes.yml under `connector:`\\
-    For example with:
-    ```
-    connector:
-        sample_rate: "1:1"
-    ```
-    For information on possible values see `cento --help`\\
-    If it's not clear which setting affects which arg see `SETTINGS_TO_ARGS` dict above\\
-    Settings not in that dict are ignored
+    These settings can be set in probes.yml under `connector:`.
+    Example:
+        connector:
+            sample_rate: "1:1"
+    For possible values see `cento --help`.
+    If unclear which setting affects which arg, see `SETTINGS_TO_ARGS` above.
+    Settings not in that dict are ignored.
     """
 
     # general options
@@ -251,7 +249,9 @@ def _before_start(self):
             self._set_rss_queues(name)
 
     def start(self):
-        """Start the probe."""
+        """
+        Start the probe.
+        """
         logging.getLogger().info(
             "Starting cento exporter on %s.", ",".join(self._interfaces)
         )
@@ -326,7 +326,9 @@ def _stop_process(self, pid):
         ).run()
 
     def supported_fields(self):
-        """Get list of IPFIX fields the probe may export in its current configuration."""
+        """
+        Get list of IPFIX fields the probe may export in its current configuration.
+        """
         output, _ = Tool(
             r"cento --help | awk '/Supported template elements \(\-\-template\):/ {p=1} p' | grep '^ %'",
             executor=self._executor,
@@ -336,14 +338,18 @@ def supported_fields(self):
         return fields
 
     def get_special_fields(self):
-        """Return dictionary of exported fields that need special evaluation."""
+        """
+        Return dictionary of exported fields that need special evaluation.
+        """
         basic_fields = set(BASIC_TEMPLATE.replace('"', "").split(","))
         used_fields = set(self._settings.flow_template.replace('"', "").split(","))
         special_fields = {field: None for field in (used_fields - basic_fields)}
         return special_fields
 
     def stop(self):
-        """Stop the probe."""
+        """
+        Stop the probe.
+        """
         # if process not running, method has no effect
         if self._process is None:
             return
@@ -380,17 +386,18 @@ def stop(self):
         self._after_stop()
 
     def cleanup(self):
-        """Clean any artifacts which were created by the connector or the active probe itself."""
+        """
+        Clean any artifacts created by the connector or the active probe itself.
+        """
         Tool(f"rm -rf {self._local_workdir}").run()
         self.host_statistics.cleanup()
 
     def download_logs(self, directory: str):
-        """Download logs to given directory.
+        """
+        Download logs to the given directory.
 
-        Parameters
-        ----------
-        directory : str
-            Path to a local directory where logs should be stored.
+        Args:
+            directory (str): Path to a local directory where logs should be stored.
         """
         try:
             shutil.move(self._log_file, directory)
@@ -399,14 +406,16 @@ def download_logs(self, directory: str):
             logging.getLogger().warning("Cannot download ipfixprobe log, %s", err)
 
     def get_timeouts(self) -> tuple[int, int]:
-        """Get active and inactive timeouts of the probe (in seconds).
+        """
+        Get active and inactive timeouts of the probe (in seconds).
 
-        Returns
-        -------
-        tuple
-            active_timeout, inactive_timeout
+        Returns:
+            tuple: active_timeout, inactive_timeout
         """
         return self._timeouts
 
     def set_prefilter(self, ip_ranges: list[str]) -> None:
+        """
+        Set probe input filter. Probe will drop all traffic except specified IP ranges.
+        """
         raise NotImplementedError

tools/ft-orchestration/src/probe/interface.py

@@ -14,67 +14,87 @@
 
 
 class HostStats(ABC):
-    """Abstract class defining an Interface for a programm that reads statistics from a host"""
+    """
+    Abstract class defining an interface for a program that reads statistics from a host.
+    """
 
     @abstractmethod
     def __init__(self, executor: Executor, watch_cmd: str):
-        """Constructor
+        """
+        Constructor.
 
         Args:
-            executor (Executor): _description_
-            watch_cmd (str): the program name to watch statistics for
+            executor (Executor): Executor object.
+            watch_cmd (str): Program name to watch statistics for.
         """
         pass
 
     @property
     @abstractmethod
     def cpus(self) -> int:
-        """number of cpus"""
+        """
+        Number of CPUs.
+        """
         pass
 
     @property
     @abstractmethod
     def total_ram(self) -> int:
-        """size of ram in kB"""
+        """
+        Size of RAM in kB.
+        """
         pass
 
     @property
     @abstractmethod
     def local_file(self) -> PathLike:
-        """Path to locally stored csv file"""
+        """
+        Path to locally stored CSV file.
+        """
         pass
 
     @abstractmethod
     def start(self):
-        """Start collecting statistics"""
+        """
+        Start collecting statistics.
+        """
         pass
 
     @abstractmethod
     def stop(self):
-        """Stop collecting statistics"""
+        """
+        Stop collecting statistics.
+        """
         pass
 
     @abstractmethod
     def cleanup(self):
-        """Remove temporary files"""
+        """
+        Remove temporary files.
+        """
         pass
 
     @abstractmethod
     def get_csv(self, output_dir: PathLike):
-        """Download the csv file containing statistics to the output_dir
+        """
+        Download the CSV file containing statistics to the output directory.
 
         Args:
-            output_dir (PathLike): path where to copy csv to
+            output_dir (PathLike): Path where to copy CSV to.
         """
         pass
 
 
 class ProbeException(Exception):
-    """Basic exception raised by the probe implementations"""
+    """
+    Basic exception raised by probe implementations.
+    """
 
 
 class ProbeInterface(ABC):
-    """Abstract class defining common interface for all probes"""
+    """
+    Abstract class defining common interface for all probes.
+    """
 
     @abstractmethod
     def __init__(
@@ -91,106 +111,96 @@ def __init__(
         cache_size,
         **kwargs,
     ):
-        """Initialize the local or remote probe interface as object
-
-        Parameters
-        ----------
-
-        executor : lbr_testsuite.executable.Executor
-            Initialized executor object with the deployed probe.
-
-        target : src.probe.probe_target
-            Target object for the exporter/probe
-
-        protocols : list
-            List of the networking protocols which the probe should parse and export.
-
-        interfaces : list(InterfaceCfg)
-            Network interfaces where the exporting process should be initiated.
-
-        verbose : bool, optional
-            Increase verbosity of probe logs.
-
-        mtu : int, optional
-            The maximum transmission unit to be set at the probe input.
-
-        active_timeout : int, optional
-            Maximum duration of an ongoing flow before the probe exports it (in seconds).
-
-        inactive_timeout : int, optional
-            Maximum duration for which a flow is kept in the probe if no new data updates it (in seconds).
-
-        kwargs : dict
-            Additional startup arguments for specific probe variants.
+        """
+        Initialize the local or remote probe interface as an object.
 
-        Raises
-        ------
-        ProbeException
-            Unable to initialize the probe.
+        Args:
+            executor (Executor): Initialized executor object with the deployed probe.
+            target: Target object for the exporter/probe.
+            protocols (list): List of networking protocols to parse and export.
+            interfaces (list[InterfaceCfg]): Network interfaces for the exporting process.
+            verbose (bool, optional): Increase verbosity of probe logs.
+            mtu (int, optional): Maximum transmission unit for the probe input.
+            active_timeout (int, optional): Maximum duration of an ongoing flow before export (seconds).
+            inactive_timeout (int, optional): Maximum duration for which a flow is kept if no new data updates it (seconds).
+            cache_size: Additional cache size argument.
+            **kwargs: Additional startup arguments for specific probe variants.
+
+        Raises:
+            ProbeException: Unable to initialize the probe.
         """
         raise NotImplementedError
 
     @property
     @abstractmethod
     def host_statistics(self) -> HostStats:
-        """Class with which to read statistics from the host"""
+        """
+        Class to read statistics from the host.
+        """
         pass
 
     @abstractmethod
     def start(self):
-        """Start the probe."""
+        """
+        Start the probe.
+        """
         raise NotImplementedError
 
     @abstractmethod
     def supported_fields(self):
-        """Get list of IPFIX fields the probe may export in its current configuration."""
+        """
+        Get list of IPFIX fields the probe may export in its current configuration.
+        """
         raise NotImplementedError
 
     @abstractmethod
     def get_special_fields(self):
-        """Return dictionary of exported fields that need special evaluation."""
+        """
+        Return dictionary of exported fields that need special evaluation.
+        """
         raise NotImplementedError
 
     @abstractmethod
     def stop(self):
-        """Stop the probe."""
+        """
+        Stop the probe.
+        """
         raise NotImplementedError
 
     @abstractmethod
     def cleanup(self):
-        """Clean any artifacts which were created by the connector or the active probe itself."""
+        """
+        Clean any artifacts created by the connector or the active probe itself.
+        """
         raise NotImplementedError
 
     @abstractmethod
     def download_logs(self, directory: str):
-        """Download logs to given directory.
+        """
+        Download logs to the given directory.
 
-        Parameters
-        ----------
-        directory : str
-            Path to a local directory where logs should be stored.
+        Args:
+            directory (str): Path to a local directory where logs should be stored.
         """
         raise NotImplementedError
 
     @abstractmethod
     def get_timeouts(self) -> tuple[int, int]:
-        """Get active and inactive timeouts of the probe (in seconds).
-
-        Returns
-        -------
-        tuple
-            active_timeout, inactive_timeout
         """
+        Get active and inactive timeouts of the probe (in seconds).
+
+        Returns:
+            tuple: active_timeout, inactive_timeout
 
+        """
+        raise NotImplementedError
         raise NotImplementedError
 
     def set_prefilter(self, ip_ranges: list[str]) -> None:
-        """Set probe input filter. Probe will drop all the traffic except specified IP ranges.
-
-        Parameters
-        ----------
-        ip_ranges : list[str]
-            IP ranges passed by the filter.
         """
+        Set probe input filter. Probe will drop all traffic except specified IP ranges.
 
+        Args:
+            ip_ranges (list[str]): IP ranges passed by the filter.
+        """
         raise NotImplementedError