dts: code adjustments for doc generation

The standard Python tool for generating API documentation, Sphinx,
imports modules one-by-one when generating the documentation. This
requires code changes:
* properly guarding argument parsing in the if __name__ == '__main__'
  block,
* the logger used by DTS runner underwent the same treatment so that it
  doesn't create log files outside of a DTS run,
* however, DTS uses the arguments to construct an object holding global
  variables. The defaults for the global variables needed to be moved
  from argument parsing elsewhere,
* importing the remote_session module from framework resulted in
  circular imports because of one module trying to import another
  module. This is fixed by reorganizing the code,
* some code reorganization was done because the resulting structure
  makes more sense, improving documentation clarity.

The are some other changes which are documentation related:
* added missing type annotation so they appear in the generated docs,
* reordered arguments in some methods,
* removed superfluous arguments and attributes,
* change private functions/methods/attributes to private and vice-versa.

The above all appear in the generated documentation and the with them,
the documentation is improved.

Signed-off-by: Juraj Linkeš <juraj.linkes@pantheon.tech>
Signed-off-by: 0-day Robot <robot@bytheb.org>

dts/framework/config/__init__.py

@@ -17,6 +17,7 @@
 import warlock  # type: ignore[import]
 import yaml
 
+from framework.exception import ConfigurationError
 from framework.settings import SETTINGS
 from framework.utils import StrEnum
 
@@ -89,14 +90,18 @@ class TrafficGeneratorConfig:
     traffic_generator_type: TrafficGeneratorType
 
     @staticmethod
-    def from_dict(d: dict):
+    def from_dict(d: dict) -> "ScapyTrafficGeneratorConfig":
         # This looks useless now, but is designed to allow expansion to traffic
         # generators that require more configuration later.
         match TrafficGeneratorType(d["type"]):
             case TrafficGeneratorType.SCAPY:
                 return ScapyTrafficGeneratorConfig(
                     traffic_generator_type=TrafficGeneratorType.SCAPY
                 )
+            case _:
+                raise ConfigurationError(
+                    f'Unknown traffic generator type "{d["type"]}".'
+                )
 
 
 @dataclass(slots=True, frozen=True)
@@ -324,6 +329,3 @@ def load_config() -> Configuration:
     config: dict[str, Any] = warlock.model_factory(schema, name="_Config")(config_data)
     config_obj: Configuration = Configuration.from_dict(dict(config))
     return config_obj
-
-
-CONFIGURATION = load_config()

dts/framework/dts.py

@@ -6,19 +6,19 @@
 import sys
 
 from .config import (
-    CONFIGURATION,
     BuildTargetConfiguration,
     ExecutionConfiguration,
     TestSuiteConfig,
+    load_config,
 )
 from .exception import BlockingTestSuiteError
 from .logger import DTSLOG, getLogger
 from .test_result import BuildTargetResult, DTSResult, ExecutionResult, Result
 from .test_suite import get_test_suites
 from .testbed_model import SutNode, TGNode
-from .utils import check_dts_python_version
 
-dts_logger: DTSLOG = getLogger("DTSRunner")
+# dummy defaults to satisfy linters
+dts_logger: DTSLOG = None  # type: ignore[assignment]
 result: DTSResult = DTSResult(dts_logger)
 
 
@@ -30,14 +30,18 @@ def run_all() -> None:
     global dts_logger
     global result
 
+    # create a regular DTS logger and create a new result with it
+    dts_logger = getLogger("DTSRunner")
+    result = DTSResult(dts_logger)
+
     # check the python version of the server that run dts
-    check_dts_python_version()
+    _check_dts_python_version()
 
     sut_nodes: dict[str, SutNode] = {}
     tg_nodes: dict[str, TGNode] = {}
     try:
         # for all Execution sections
-        for execution in CONFIGURATION.executions:
+        for execution in load_config().executions:
             sut_node = sut_nodes.get(execution.system_under_test_node.name)
             tg_node = tg_nodes.get(execution.traffic_generator_node.name)
 
@@ -82,6 +86,25 @@ def run_all() -> None:
     _exit_dts()
 
 
+def _check_dts_python_version() -> None:
+    def RED(text: str) -> str:
+        return f"\u001B[31;1m{str(text)}\u001B[0m"
+
+    if sys.version_info.major < 3 or (
+        sys.version_info.major == 3 and sys.version_info.minor < 10
+    ):
+        print(
+            RED(
+                (
+                    "WARNING: DTS execution node's python version is lower than"
+                    "python 3.10, is deprecated and will not work in future releases."
+                )
+            ),
+            file=sys.stderr,
+        )
+        print(RED("Please use Python >= 3.10 instead"), file=sys.stderr)
+
+
 def _run_execution(
     sut_node: SutNode,
     tg_node: TGNode,

dts/framework/exception.py

@@ -42,38 +42,33 @@ class SSHTimeoutError(DTSError):
     Command execution timeout.
     """
 
-    command: str
-    output: str
     severity: ClassVar[ErrorSeverity] = ErrorSeverity.SSH_ERR
+    _command: str
 
-    def __init__(self, command: str, output: str):
-        self.command = command
-        self.output = output
+    def __init__(self, command: str):
+        self._command = command
 
     def __str__(self) -> str:
-        return f"TIMEOUT on {self.command}"
-
-    def get_output(self) -> str:
-        return self.output
+        return f"TIMEOUT on {self._command}"
 
 
 class SSHConnectionError(DTSError):
     """
     SSH connection error.
     """
 
-    host: str
-    errors: list[str]
     severity: ClassVar[ErrorSeverity] = ErrorSeverity.SSH_ERR
+    _host: str
+    _errors: list[str]
 
     def __init__(self, host: str, errors: list[str] | None = None):
-        self.host = host
-        self.errors = [] if errors is None else errors
+        self._host = host
+        self._errors = [] if errors is None else errors
 
     def __str__(self) -> str:
-        message = f"Error trying to connect with {self.host}."
-        if self.errors:
-            message += f" Errors encountered while retrying: {', '.join(self.errors)}"
+        message = f"Error trying to connect with {self._host}."
+        if self._errors:
+            message += f" Errors encountered while retrying: {', '.join(self._errors)}"
 
         return message
 
@@ -84,14 +79,14 @@ class SSHSessionDeadError(DTSError):
     It can no longer be used.
     """
 
-    host: str
     severity: ClassVar[ErrorSeverity] = ErrorSeverity.SSH_ERR
+    _host: str
 
     def __init__(self, host: str):
-        self.host = host
+        self._host = host
 
     def __str__(self) -> str:
-        return f"SSH session with {self.host} has died"
+        return f"SSH session with {self._host} has died"
 
 
 class ConfigurationError(DTSError):
@@ -107,18 +102,18 @@ class RemoteCommandExecutionError(DTSError):
     Raised when a command executed on a Node returns a non-zero exit status.
     """
 
-    command: str
-    command_return_code: int
     severity: ClassVar[ErrorSeverity] = ErrorSeverity.REMOTE_CMD_EXEC_ERR
+    command: str
+    _command_return_code: int
 
     def __init__(self, command: str, command_return_code: int):
         self.command = command
-        self.command_return_code = command_return_code
+        self._command_return_code = command_return_code
 
     def __str__(self) -> str:
         return (
             f"Command {self.command} returned a non-zero exit code: "
-            f"{self.command_return_code}"
+            f"{self._command_return_code}"
         )
 
 
@@ -143,22 +138,15 @@ class TestCaseVerifyError(DTSError):
     Used in test cases to verify the expected behavior.
     """
 
-    value: str
     severity: ClassVar[ErrorSeverity] = ErrorSeverity.TESTCASE_VERIFY_ERR
 
-    def __init__(self, value: str):
-        self.value = value
-
-    def __str__(self) -> str:
-        return repr(self.value)
-
 
 class BlockingTestSuiteError(DTSError):
-    suite_name: str
     severity: ClassVar[ErrorSeverity] = ErrorSeverity.BLOCKING_TESTSUITE_ERR
+    _suite_name: str
 
     def __init__(self, suite_name: str) -> None:
-        self.suite_name = suite_name
+        self._suite_name = suite_name
 
     def __str__(self) -> str:
-        return f"Blocking suite {self.suite_name} failed."
+        return f"Blocking suite {self._suite_name} failed."

dts/framework/remote_session/__init__.py

@@ -12,29 +12,24 @@
 
 # pylama:ignore=W0611
 
-from framework.config import OS, NodeConfiguration
-from framework.exception import ConfigurationError
+from framework.config import NodeConfiguration
 from framework.logger import DTSLOG
 
-from .linux_session import LinuxSession
-from .os_session import InteractiveShellType, OSSession
-from .remote import (
-    CommandResult,
-    InteractiveRemoteSession,
-    InteractiveShell,
-    PythonShell,
-    RemoteSession,
-    SSHSession,
-    TestPmdDevice,
-    TestPmdShell,
-)
-
-
-def create_session(
+from .interactive_remote_session import InteractiveRemoteSession
+from .interactive_shell import InteractiveShell
+from .python_shell import PythonShell
+from .remote_session import CommandResult, RemoteSession
+from .ssh_session import SSHSession
+from .testpmd_shell import TestPmdShell
+
+
+def create_remote_session(
     node_config: NodeConfiguration, name: str, logger: DTSLOG
-) -> OSSession:
-    match node_config.os:
-        case OS.linux:
-            return LinuxSession(node_config, name, logger)
-        case _:
-            raise ConfigurationError(f"Unsupported OS {node_config.os}")
+) -> RemoteSession:
+    return SSHSession(node_config, name, logger)
+
+
+def create_interactive_session(
+    node_config: NodeConfiguration, logger: DTSLOG
+) -> InteractiveRemoteSession:
+    return InteractiveRemoteSession(node_config, logger)

dts/framework/remote_session/remote/ssh_session.py → dts/framework/remote_session/ssh_session.py

@@ -18,9 +18,7 @@
     SSHException,
 )
 
-from framework.config import NodeConfiguration
 from framework.exception import SSHConnectionError, SSHSessionDeadError, SSHTimeoutError
-from framework.logger import DTSLOG
 
 from .remote_session import CommandResult, RemoteSession
 
@@ -45,14 +43,6 @@ class SSHSession(RemoteSession):
 
     session: Connection
 
-    def __init__(
-        self,
-        node_config: NodeConfiguration,
-        session_name: str,
-        logger: DTSLOG,
-    ):
-        super(SSHSession, self).__init__(node_config, session_name, logger)
-
     def _connect(self) -> None:
         errors = []
         retry_attempts = 10
@@ -117,7 +107,7 @@ def _send_command(
 
         except CommandTimedOut as e:
             self._logger.exception(e)
-            raise SSHTimeoutError(command, e.result.stderr) from e
+            raise SSHTimeoutError(command) from e
 
         return CommandResult(
             self.name, command, output.stdout, output.stderr, output.return_code