Implement api methods for global SNMP configuration management (CRUD operations), SNMP config Pydantic Model, uuid validation and test

PaulWinterstein · PaulWinterstein · commit 89492560bba1 · 2025-06-26T11:34:40.000+02:00
diff --git a/src/videoipath_automation_tool/apps/inventory/inventory_api.py b/src/videoipath_automation_tool/apps/inventory/inventory_api.py
@@ -13,6 +13,8 @@
 )
 from videoipath_automation_tool.apps.inventory.model.device_status import DeviceStatus
 from videoipath_automation_tool.apps.inventory.model.drivers import CustomSettingsType, DriverLiteral
+from videoipath_automation_tool.apps.inventory.model.global_snmp_config import SnmpConfiguration
+from videoipath_automation_tool.apps.inventory.model.global_snmp_request_rpc import SnmpRequestRpc
 from videoipath_automation_tool.apps.inventory.model.inventory_device import InventoryDevice
 from videoipath_automation_tool.apps.inventory.model.inventory_discovered_device import DiscoveredInventoryDevice
 from videoipath_automation_tool.apps.inventory.model.inventory_request_rpc import InventoryRequestRpc
@@ -664,16 +666,16 @@ def get_discovered_device(self, discovered_device_id: str) -> DiscoveredInventor
             response.data["status"]["devman"]["discoveredDevices"]["_items"][0]
         )
 
-    # --- Global Configuration Helpers ---
-    def get_global_snmp_config_id_by_label(self, label: str) -> Optional[str]:
+    # --- Global SNMP Configuration Helpers ---
+    def get_global_snmp_config_id_by_label(self, label: str) -> Optional[str | List[str]]:
         """Method to get the global SNMP configuration id by label.
-        Note: If multiple SNMP configurations with the same label exist, the first one is returned.
+        Note: If multiple SNMP configurations with the same label exist, a list of ids is returned.
 
         Args:
             label (str): Label of the SNMP configuration
 
         Returns:
-            Optional[str]: SNMP configuration id, None if not found
+            Optional[str | List[str]]: SNMP configuration id, None if not found, List of ids if multiple configurations with the same label exist
         """
         if not label:
             raise ValueError("Label must not be empty.")
@@ -687,9 +689,9 @@ def get_global_snmp_config_id_by_label(self, label: str) -> Optional[str]:
                 return list(matches.keys())[0]
             elif len(matches) > 1:
                 self._logger.warning(
-                    f"Multiple SNMP configurations found with label '{label}'. Returning the first one."
+                    f"Multiple SNMP configurations found with label '{label}''. Returning all matching ids."
                 )
-                return list(matches.keys())[0]
+                return list(matches.keys())
         return None
 
     def get_global_snmp_config_label_by_id(self, snmp_config_id: str) -> Optional[str]:
@@ -705,9 +707,10 @@ def get_global_snmp_config_label_by_id(self, snmp_config_id: str) -> Optional[st
             raise ValueError("SNMP configuration id must not be empty.")
 
         url = f"/rest/v2/data/config/system/snmp/session/{snmp_config_id}/descriptor/label"
-        response = self.vip_connector.rest.get(url)
-        if response.data and response.data["config"]["system"]["snmp"]["session"][snmp_config_id]:
-            return response.data["config"]["system"]["snmp"]["session"][snmp_config_id]["descriptor"]["label"]
+        response = self.vip_connector.rest.get(url, node_check=False)
+        if response.data and response.data["config"]["system"]["snmp"]["session"]:
+            if snmp_config_id in response.data["config"]["system"]["snmp"]["session"]:
+                return response.data["config"]["system"]["snmp"]["session"][snmp_config_id]["descriptor"]["label"]
         return None
 
     def get_all_global_snmp_config_ids(self) -> dict[str, str]:
@@ -726,6 +729,106 @@ def get_all_global_snmp_config_ids(self) -> dict[str, str]:
             snmp_config_id: snmp_config["descriptor"]["label"] for snmp_config_id, snmp_config in snmp_configs.items()
         }
 
+    # --- Global SNMP Configuration CRUD Methods ---
+    def get_global_snmp_config(self, snmp_config_id: str) -> SnmpConfiguration:
+        """Method to get a global SNMP configuration by id from VideoIPath-Inventory
+
+        Args:
+            snmp_config_id (str): SNMP configuration id
+
+        Returns:
+            GlobalSnmpConfig: Global SNMP configuration object
+        """
+        if not snmp_config_id:
+            raise ValueError("SNMP configuration id must not be empty.")
+
+        url = f"/rest/v2/data/config/system/snmp/session/{snmp_config_id}/**"
+        response = self.vip_connector.rest.get(url)
+        if not response.data:
+            raise ValueError("Response data is empty.")
+
+        return SnmpConfiguration.parse_from_dict(response.data["config"]["system"]["snmp"]["session"])
+
+    def add_global_snmp_config(self, snmp_config: SnmpConfiguration) -> SnmpConfiguration:
+        """Method to add a new global SNMP configuration
+
+        Args:
+            snmp_config (SnmpConfiguration): SNMP configuration object to add
+
+        Returns:
+            SnmpConfiguration: Added SNMP configuration object
+        """
+        if not snmp_config.id:
+            raise ValueError("SNMP configuration id must be set.")
+
+        self._logger.debug(f"Adding new global SNMP configuration with id '{snmp_config.id}'.")
+
+        existing_configs_label = self.get_global_snmp_config_label_by_id(snmp_config.id)
+        if existing_configs_label is not None:
+            raise ValueError(f"SNMP configuration with id '{snmp_config.id}' already exists. Please update it instead.")
+
+        body = SnmpRequestRpc()
+        body.add(snmp_config)
+
+        response = self.vip_connector.rpc.post("/api/updateSnmpConfig", body=body)
+
+        if response.header.status != "OK":
+            raise ValueError(f"Failed to add global SNMP configuration. Error: {response}")
+
+        return self.get_global_snmp_config(snmp_config_id=snmp_config.id)
+
+    def update_global_snmp_config(self, snmp_config: SnmpConfiguration) -> SnmpConfiguration:
+        """Method to update a global SNMP configuration
+
+        Args:
+            snmp_config (SnmpConfiguration): SNMP configuration object to update
+
+        Returns:
+            SnmpConfiguration: Updated SNMP configuration object
+        """
+        if not snmp_config.id:
+            raise ValueError("SNMP configuration id must be set.")
+
+        self._logger.debug(f"Updating global SNMP configuration with id '{snmp_config.id}'.")
+
+        existing_configs_label = self.get_global_snmp_config_label_by_id(snmp_config.id)
+        if existing_configs_label is None:
+            raise ValueError(f"SNMP configuration with id '{snmp_config.id}' does not exist. Please add it first.")
+
+        body = SnmpRequestRpc()
+        body.update(snmp_config)
+
+        response = self.vip_connector.rpc.post("/api/updateSnmpConfig", body=body)
+
+        if response.header.status != "OK":
+            raise ValueError(f"Failed to update global SNMP configuration. Error: {response}")
+
+        return self.get_global_snmp_config(snmp_config_id=snmp_config.id)
+
+    def remove_global_snmp_config(self, snmp_config_id: str) -> ResponseRPC:
+        """Method to remove a global SNMP configuration by id from VideoIPath-Inventory
+
+        Args:
+            snmp_config_id (str): SNMP configuration id
+
+        Returns:
+            ResponseRPC: Response object
+        """
+        if not snmp_config_id:
+            raise ValueError("SNMP configuration id must be set.")
+
+        self._logger.debug(f"Removing global SNMP configuration with id '{snmp_config_id}'.")
+
+        body = SnmpRequestRpc()
+        body.remove(snmp_config_id)
+
+        response = self.vip_connector.rpc.post("/api/updateSnmpConfig", body=body)
+
+        if response.header.status != "OK":
+            raise ValueError(f"Failed to remove global SNMP configuration. Error: {response}")
+
+        return response
+
     # --- Deprecated Methods ---
     @deprecated(
         "The method `fetch_device_ids_list` is deprecated and will be removed in a future release. ",
diff --git a/src/videoipath_automation_tool/apps/inventory/model/global_snmp_config.py b/src/videoipath_automation_tool/apps/inventory/model/global_snmp_config.py
@@ -0,0 +1,134 @@
+from enum import Enum
+from uuid import uuid4
+
+from pydantic import BaseModel, Field
+
+
+# --- User Enum Classes ---
+class SecurityLevel(int, Enum):
+    UNDEFINED = 0
+    """Undefined security level."""
+    NO_AUTH_NO_PRIV = 1
+    """NoAuthNoPriv | Without authentication and without privacy."""
+    AUTH_NO_PRIV = 2
+    """AuthNoPriv | With authentication but without privacy."""
+    AUTH_PRIV = 3
+    """AuthPriv | With authentication and with privacy."""
+
+
+class AuthProtocol(int, Enum):
+    MD5 = 2
+    """MD5 | HMAC-MD5-96 digest authentication protocol."""
+    SHA = 3
+    """SHA | HMAC-SHA-96 digest authentication protocol."""
+
+
+class PrivProtocol(int, Enum):
+    DES = 2
+    """DES | CBC-DES symmetric encryption protocol."""
+    THREE_DES = 3
+    """3DES | 3DES-EDE symmetric encryption protocol."""
+    AES128 = 4
+    """AES127 | CFB128-AES-128 privacy protocol."""
+
+
+# --- Version Enum Class ---
+class SnmpVersion(int, Enum):
+    V1 = 0
+    """SNMP version 1."""
+    V2C = 1
+    """SNMP version 2 with community security."""
+    V3 = 3
+    """SNMP version 3."""
+
+
+# --- Data Model Classes ---
+class SnmpUser(BaseModel):
+    level: SecurityLevel = SecurityLevel.NO_AUTH_NO_PRIV
+    name: str = "New User"
+    authProtocol: AuthProtocol = AuthProtocol.MD5
+    privProtocol: PrivProtocol = PrivProtocol.DES
+    engineId: str = ""
+    privPassword: str = ""
+    authPassword: str = ""
+
+
+class SnmpDescriptor(BaseModel):
+    label: str = ""
+    desc: str = ""
+
+
+class SnmpSecurityEntry(BaseModel):
+    user: str = ""  # User ID from "Users" section. Must be a valid UUID of an existing user.
+    community: str = ""  # Value from "Protocol Settings => SNMP v1/v2c Security => Write / Read community"
+
+
+class SnmpSecurity(BaseModel):
+    read: SnmpSecurityEntry = SnmpSecurityEntry(community="public")
+    write: SnmpSecurityEntry = SnmpSecurityEntry(community="private")
+
+
+class SnmpProtocolSettings(BaseModel):
+    preferredVersion: SnmpVersion = SnmpVersion.V2C
+    retries: int = 1
+    maxRepetitions: int = 10
+    useGetBulk: bool = True
+    timeout: int = 5000
+    localEngineId: str = ""
+
+
+class SnmpConfiguration(BaseModel):
+    id: str = Field(alias="_id")
+    descriptor: SnmpDescriptor = Field(default_factory=SnmpDescriptor)
+    users: dict[str, SnmpUser] = Field(default_factory=dict)
+    security: SnmpSecurity = Field(default_factory=SnmpSecurity)
+    protocol: SnmpProtocolSettings = Field(default_factory=SnmpProtocolSettings)
+
+    @classmethod
+    def create(cls):
+        """
+        Creates a new instance of SnmpConfiguration with default values.
+
+        Returns:
+            SnmpConfiguration: A new instance of SnmpConfiguration.
+        """
+        config_id = str(uuid4())
+        return cls(
+            _id=config_id,
+            descriptor=SnmpDescriptor(label="New SNMP Configuration", desc=""),
+        )
+
+    @classmethod
+    def parse_from_dict(cls, data: dict) -> "SnmpConfiguration":
+        """
+        Parses a dictionary into a SnmpConfiguration instance.
+
+        Args:
+            data (dict): The dictionary to parse.
+
+        Returns:
+            SnmpConfiguration: An instance of SnmpConfiguration.
+        """
+        if len(data.keys()) == 1:
+            config_id = list(data.keys())[0]
+            data = data[config_id]
+            data["_id"] = config_id
+        else:
+            raise ValueError("Data dictionary must contain exactly one key/value pair: <id>: <configuration>")
+        return cls(**data)
+
+    # def dump_to_dict(self) -> dict:
+    #     """
+    #     Dumps the SnmpConfiguration instance to a dictionary.
+
+    #     Returns:
+    #         dict: A dictionary representation of the SnmpConfiguration instance.
+    #     """
+    #     config_id = self.id
+    #     data = self.model_dump(mode="json", exclude={"id"})
+    #     return {config_id: data}
+
+
+# class SnmpConfig(BaseModel):
+#     id: str
+#     configuration: SnmpConfiguration = Field(default_factory=SnmpConfiguration)
diff --git a/src/videoipath_automation_tool/apps/inventory/model/global_snmp_request_rpc.py b/src/videoipath_automation_tool/apps/inventory/model/global_snmp_request_rpc.py
@@ -0,0 +1,51 @@
+from videoipath_automation_tool.apps.inventory.model.global_snmp_config import SnmpConfiguration
+from videoipath_automation_tool.connector.models.request_rpc import RequestRPC
+from videoipath_automation_tool.validators.uuid_4 import validate_uuid_4
+
+
+class SnmpRequestRpc(RequestRPC):
+    # Wrapper class for RequestRpc
+
+    def add(self, config: SnmpConfiguration):
+        """Method to add a new global SNMP configuration
+
+        Args:
+            config (SnmpConfiguration): SNMP configuration to add
+        """
+        try:
+            validate_uuid_4(config.id)
+        except ValueError as e:
+            raise ValueError(
+                f"To add a new global SNMP configuration, a valid 'id' (UUID 4 format) must be set in the configuration. Error: {e}"
+            )
+        return super().add(config.id, config)
+
+    def update(self, config: SnmpConfiguration):
+        """Method to update a global SNMP configuration
+
+        Args:
+            config (SnmpConfiguration): SNMP configuration to update
+        """
+        try:
+            validate_uuid_4(config.id)
+        except ValueError as e:
+            raise ValueError(
+                f"To update a global SNMP configuration, a valid 'id' (UUID 4 format) must be set in the configuration. Error: {e}"
+            )
+        return super().update(config.id, config)
+
+    def remove(self, config_id: str | list[str]):
+        """Method to remove a global SNMP configuration
+
+        Args:
+            config_id (str | list[str]): Id or List of Ids of the configuration
+        """
+        try:
+            if isinstance(config_id, list):
+                for c_id in config_id:
+                    validate_uuid_4(c_id)
+            else:
+                validate_uuid_4(config_id)
+        except ValueError as e:
+            raise ValueError(f"Invalid 'config_id' format. Must be a valid UUID 4 format. Error: {e}")
+        return super().remove(config_id)
diff --git a/src/videoipath_automation_tool/connector/vip_rpc_connector.py b/src/videoipath_automation_tool/connector/vip_rpc_connector.py
@@ -15,6 +15,7 @@ class VideoIPathRPCConnector(VideoIPathBaseConnector):
         "/api/uploadLicense",
         "/api/activateLicense",
         "/api/deactivateLicense",
+        "/api/updateSnmpConfig",
     }
 
     def post(self, url_path: str, body: RequestRPC, url_validation: bool = True) -> ResponseRPC:
diff --git a/src/videoipath_automation_tool/validators/uuid_4.py b/src/videoipath_automation_tool/validators/uuid_4.py
@@ -0,0 +1,28 @@
+import uuid
+
+
+def validate_uuid_4(uuid_4: str) -> str:
+    """
+    Validates and normalizes a UUIDv4 string.
+
+    Args:
+        uuid_4: The input string to validate.
+
+    Returns:
+        The normalized UUIDv4 string (lowercase, with dashes).
+
+    Raises:
+        ValueError: If input is not a valid UUIDv4.
+    """
+    if not isinstance(uuid_4, str):
+        raise ValueError(f"UUID must be a string, got {type(uuid_4).__name__}")
+
+    try:
+        u = uuid.UUID(uuid_4)
+    except (ValueError, AttributeError, TypeError):
+        raise ValueError(f"Invalid UUID format: '{uuid_4}'")
+
+    if u.version != 4:
+        raise ValueError(f"UUID is not version 4: '{uuid_4}'")
+
+    return str(u)
diff --git a/tests/validators/test_uuid_4_validator.py b/tests/validators/test_uuid_4_validator.py

Original file line number	Diff line number	Diff line change
`@@ -15,6 +15,7 @@ class VideoIPathRPCConnector(VideoIPathBaseConnector):`
`15`	`15`	`"/api/uploadLicense",`
`16`	`16`	`"/api/activateLicense",`
`17`	`17`	`"/api/deactivateLicense",`
	`18`	`+ "/api/updateSnmpConfig",`
`18`	`19`	`}`
`19`	`20`
`20`	`21`	`def post(self, url_path: str, body: RequestRPC, url_validation: bool = True) -> ResponseRPC:`