Strongly define the connection arg

dev-setup.sh

@@ -14,13 +14,13 @@ export UV_SYSTEM_CERTS=1
 
 # Create venv if not already present or if Python version is below 3.9
 if [ ! -d "venv" ] || ! venv/bin/python -c 'import sys; exit(0 if sys.version_info >= (3, 9) else 1)' 2>/dev/null; then
-    uv venv venv --python python3 --no-python-downloads
+    uv venv venv --python 3.9  --no-python-downloads
 fi
 
 # Activate the desired venv
 source venv/bin/activate
 
-uv pip install -e ".[dev]"
+uv pip install -e . --group dev
 
 # Only install pre-commit hooks if not in CI environment
 if [ -z "${CI:-}" ]; then

nodescraper/connection/inband/inbandmanager.py

@@ -26,7 +26,7 @@
 from __future__ import annotations
 
 from logging import Logger
-from typing import Any, Optional, Union
+from typing import Any, ClassVar, Optional, Union
 
 from nodescraper.enums import (
     EventCategory,
@@ -46,6 +46,8 @@
 
 
 class InBandConnectionManager(ConnectionManager[InBandConnection, SSHConnectionParams]):
+    CONNECTION_ARGS_MODEL: ClassVar[type[SSHConnectionParams]] = SSHConnectionParams
+
     def __init__(
         self,
         system_info: SystemInfo,

nodescraper/connection/oob_ssh/oob_ssh_connection_manager.py

@@ -26,7 +26,7 @@
 from __future__ import annotations
 
 from logging import Logger
-from typing import Optional, Union
+from typing import ClassVar, Optional, Union
 
 from nodescraper.enums import EventCategory, EventPriority, ExecutionStatus
 from nodescraper.interfaces.connectionmanager import ConnectionManager
@@ -42,6 +42,8 @@
 class OobSshConnectionManager(ConnectionManager[InBandConnection, RedfishConnectionParams]):
     """SSH to the BMC using the same host and credentials as Redfish (OOB shell)."""
 
+    CONNECTION_ARGS_MODEL: ClassVar[type[RedfishConnectionParams]] = RedfishConnectionParams
+
     def __init__(
         self,
         system_info: SystemInfo,

nodescraper/connection/redfish/redfish_manager.py

@@ -26,7 +26,7 @@
 from __future__ import annotations
 
 from logging import Logger
-from typing import Any, Mapping, Optional, Union
+from typing import Any, ClassVar, Mapping, Optional, Union
 
 from nodescraper.enums import EventCategory, EventPriority, ExecutionStatus
 from nodescraper.interfaces.connectionmanager import ConnectionManager
@@ -83,6 +83,8 @@ def collected_multi_target_data(connection_manager: Any) -> dict[str, Any]:
 class RedfishConnectionManager(ConnectionManager[RedfishConnection, RedfishConnectionParams]):
     """Connection manager for Redfish (BMC) API."""
 
+    CONNECTION_ARGS_MODEL: ClassVar[type[RedfishConnectionParams]] = RedfishConnectionParams
+
     def __init__(
         self,
         system_info: SystemInfo,

nodescraper/connection/redfish/ssh_proxy_manager.py

@@ -26,7 +26,7 @@
 from __future__ import annotations
 
 from logging import Logger
-from typing import Optional, Union
+from typing import Any, ClassVar, Optional, Union
 
 from nodescraper.enums import EventCategory, EventPriority, ExecutionStatus
 from nodescraper.interfaces.connectionmanager import ConnectionManager
@@ -46,14 +46,18 @@ class RedfishSshProxyConnectionManager(
 ):
     """SSH to a BMC and curl Redfish on an address reachable only from that host (AMC)."""
 
+    CONNECTION_ARGS_MODEL: ClassVar[type[RedfishSshProxyConnectionParams]] = (
+        RedfishSshProxyConnectionParams
+    )
+
     def __init__(
         self,
         system_info: SystemInfo,
         logger: Optional[Logger] = None,
         max_event_priority_level: Union[EventPriority, str] = EventPriority.CRITICAL,
         parent: Optional[str] = None,
         task_result_hooks: Optional[list[TaskResultHook]] = None,
-        connection_args: Optional[RedfishSshProxyConnectionParams] = None,
+        connection_args: Optional[Union[RedfishSshProxyConnectionParams, dict[str, Any]]] = None,
         **kwargs,
     ):
         super().__init__(

nodescraper/interfaces/connectionmanager.py

@@ -29,14 +29,13 @@
 import logging
 import types
 from functools import wraps
-from typing import Any, Callable, Generic, Optional, TypeVar, Union
+from typing import Any, Callable, ClassVar, Generic, Optional, TypeVar, Union
 
 from pydantic import BaseModel
 
 from nodescraper.constants import DEFAULT_EVENT_REPORTER
 from nodescraper.enums import EventCategory, EventPriority, ExecutionStatus
 from nodescraper.models import SystemInfo, TaskResult
-from nodescraper.typeutils import TypeUtils
 from nodescraper.utils import get_exception_traceback
 
 from .task import Task
@@ -89,6 +88,8 @@ class ConnectionManager(Task, Generic[TConnection, TConnectArg]):
 
     TASK_TYPE = "CONNECTION_MANAGER"
 
+    CONNECTION_ARGS_MODEL: ClassVar[Optional[type[BaseModel]]]
+
     def __init__(
         self,
         system_info: SystemInfo,
@@ -130,19 +131,19 @@ def __init__(
         )
 
         if isinstance(connection_args, dict):
-            generic_map = TypeUtils.get_generic_map(self.__class__)
-            connection_arg_model = generic_map.get(TConnectArg)
-            if not connection_arg_model:
-                raise ValueError("No model defined for connection args")
-
-            connection_args = connection_arg_model(**connection_args)
+            connection_arg_model = self.CONNECTION_ARGS_MODEL
+            if connection_arg_model is not None:
+                connection_args = connection_arg_model(**connection_args)
+            elif connection_args:
+                self.logger.warning("Connection args provided but no model defined")
 
         self.connection_args = connection_args
         self.connection: Optional[TConnection] = None
 
     def __init_subclass__(cls, **kwargs) -> None:
         super().__init_subclass__(**kwargs)
-
+        if not hasattr(cls, "CONNECTION_ARGS_MODEL"):
+            raise ValueError("Subclasses of ConnectionManager must define CONNECTION_ARGS_MODEL")
         if "connect" in cls.__dict__:
             cls.connect = connect_decorator(cls.connect)
 

pyproject.toml

@@ -29,23 +29,6 @@ dependencies = [
     "urllib3>=2.6.0"
 ]
 
-[project.optional-dependencies]
-dev = [
-    "build",
-    "black",
-    "pylint",
-    "coverage",
-    "twine",
-    "ruff",
-    "pre-commit",
-    "pytest",
-    "pytest-cov",
-    "mypy",
-    "types-paramiko",
-    "types-requests",
-    "types-setuptools",
-]
-
 [project.urls]
 homepage = "https://github.com/amd/node-scraper"
 documentation = "https://github.com/amd/node-scraper/blob/main/docs/PLUGIN_DOC.md"

test/unit/connection/test_connectionmanager.py

@@ -28,8 +28,19 @@
 
 from nodescraper.connection.inband.inbandmanager import InBandConnectionManager
 from nodescraper.connection.inband.inbandremote import RemoteShell
+from nodescraper.connection.inband.sshparams import SSHConnectionParams
+from nodescraper.connection.oob_ssh.oob_ssh_connection_manager import (
+    OobSshConnectionManager,
+)
 from nodescraper.connection.redfish.redfish_connection import RedfishConnection
 from nodescraper.connection.redfish.redfish_manager import RedfishConnectionManager
+from nodescraper.connection.redfish.redfish_params import RedfishConnectionParams
+from nodescraper.connection.redfish.ssh_proxy_manager import (
+    RedfishSshProxyConnectionManager,
+)
+from nodescraper.connection.redfish.ssh_proxy_params import (
+    RedfishSshProxyConnectionParams,
+)
 from nodescraper.enums import ExecutionStatus, SystemLocation
 from nodescraper.interfaces.connectionmanager import ConnectionManager
 from nodescraper.models import SystemInfo, TaskResult
@@ -44,6 +55,8 @@ class DummyConnection:
 class DummyConnectionManager(ConnectionManager[DummyConnection, None]):
     """Concrete implementation of ConnectionManager for testing base class behavior"""
 
+    CONNECTION_ARGS_MODEL = None
+
     def connect(self) -> TaskResult:
         """Minimal connect implementation for testing"""
         self.connection = DummyConnection()
@@ -182,6 +195,86 @@ def test_disconnect_closes_redfish_connection(self):
         self.assertIsNone(manager.connection)
 
 
+class TestConnectionArgsModelClassVar(unittest.TestCase):
+    """CONNECTION_ARGS_MODEL classvar drives dict -> model conversion of connection_args."""
+
+    def setUp(self):
+        self.system_info = SystemInfo(name="test_system", location=SystemLocation.REMOTE)
+
+    def test_subclasses_declare_expected_models(self):
+        expected = {
+            InBandConnectionManager: SSHConnectionParams,
+            RedfishConnectionManager: RedfishConnectionParams,
+            RedfishSshProxyConnectionManager: RedfishSshProxyConnectionParams,
+            OobSshConnectionManager: RedfishConnectionParams,
+        }
+        for manager_cls, model in expected.items():
+            with self.subTest(manager=manager_cls.__name__):
+                self.assertIs(manager_cls.CONNECTION_ARGS_MODEL, model)
+
+    def test_classvar_not_set_on_instance(self):
+        manager = InBandConnectionManager(system_info=self.system_info)
+        self.assertNotIn("CONNECTION_ARGS_MODEL", vars(manager))
+
+    def test_dict_args_converted_to_model(self):
+        manager = InBandConnectionManager(
+            system_info=self.system_info,
+            connection_args={"hostname": "host1", "username": "user1"},
+        )
+        self.assertIsInstance(manager.connection_args, SSHConnectionParams)
+        self.assertEqual(manager.connection_args.hostname, "host1")
+        self.assertEqual(manager.connection_args.username, "user1")
+
+    def test_redfish_dict_args_converted_to_model(self):
+        manager = RedfishConnectionManager(
+            system_info=self.system_info,
+            connection_args={"host": "bmc1", "username": "admin"},
+        )
+        self.assertIsInstance(manager.connection_args, RedfishConnectionParams)
+        self.assertEqual(manager.connection_args.host, "bmc1")
+
+    def test_oob_ssh_dict_args_converted_to_model(self):
+        manager = OobSshConnectionManager(
+            system_info=self.system_info,
+            connection_args={"host": "bmc1", "username": "admin"},
+        )
+        self.assertIsInstance(manager.connection_args, RedfishConnectionParams)
+
+    def test_ssh_proxy_dict_args_converted_to_model(self):
+        manager = RedfishSshProxyConnectionManager(
+            system_info=self.system_info,
+            connection_args={"host": "", "ssh": {"hostname": "", "username": ""}, "max_workers": 2},
+        )
+        self.assertIsInstance(manager.connection_args, RedfishSshProxyConnectionParams)
+        self.assertEqual(manager.connection_args.max_workers, 2)
+
+    def test_model_instance_passed_through_unchanged(self):
+        params = SSHConnectionParams(hostname="host1", username="user1")
+        manager = InBandConnectionManager(system_info=self.system_info, connection_args=params)
+        self.assertIs(manager.connection_args, params)
+
+    def test_none_args_stay_none(self):
+        manager = InBandConnectionManager(system_info=self.system_info)
+        self.assertIsNone(manager.connection_args)
+
+    def test_custom_subclass_classvar_is_used(self):
+        class CustomManager(ConnectionManager[DummyConnection, SSHConnectionParams]):
+            CONNECTION_ARGS_MODEL = SSHConnectionParams
+
+            def connect(self) -> TaskResult:
+                return self.result
+
+        manager = CustomManager(
+            system_info=self.system_info,
+            connection_args={"hostname": "h", "username": "u"},
+        )
+        self.assertIsInstance(manager.connection_args, SSHConnectionParams)
+
+    def test_invalid_dict_args_raise_validation_error(self):
+        with self.assertRaises(ValueError):
+            InBandConnectionManager(system_info=self.system_info, connection_args={"port": 22})
+
+
 if __name__ == "__main__":
     unittest.main()
 

test/unit/framework/common/shared_utils.py

@@ -36,6 +36,7 @@
 class MockConnectionManager(ConnectionManager):
     # Class variable to store the mock connector
     mock_connector = None
+    CONNECTION_ARGS_MODEL = None
 
     def __init__(
         self,