add SNMP API method wrappers in InventoryApp, engine ID validation in global SNMP model, and getter/setter in InventoryDevice for global SNMP config

Author: PaulWinterstein

src/videoipath_automation_tool/apps/inventory/app/app.py

@@ -10,11 +10,13 @@
 from videoipath_automation_tool.apps.inventory.app.get_device import InventoryGetDeviceMixin
 from videoipath_automation_tool.apps.inventory.inventory_api import InventoryAPI
 from videoipath_automation_tool.apps.inventory.model.drivers import CustomSettings, CustomSettingsType, DriverLiteral
+from videoipath_automation_tool.apps.inventory.model.global_snmp_config import SnmpConfiguration
 from videoipath_automation_tool.apps.inventory.model.inventory_device import InventoryDevice
 from videoipath_automation_tool.apps.inventory.model.inventory_device_configuration_compare import (
     InventoryDeviceComparison,
 )
 from videoipath_automation_tool.apps.inventory.model.inventory_discovered_device import DiscoveredInventoryDevice
+from videoipath_automation_tool.connector.models.response_rpc import ResponseRPC
 from videoipath_automation_tool.connector.vip_connector import VideoIPathConnector
 from videoipath_automation_tool.validators.device_id import validate_device_id
 
@@ -349,6 +351,83 @@ def parse_configuration(config: dict) -> InventoryDevice:
         """
         return InventoryDevice.parse_configuration(config)
 
+    # --- 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, a list of ids is returned.
+
+        Args:
+            label (str): Label of the SNMP configuration
+
+        Returns:
+            Optional[str | List[str]]: SNMP configuration id, None if not found, List of ids if multiple configurations with the same label exist
+        """
+        return self._inventory_api.get_global_snmp_config_id_by_label(label=label)
+
+    def get_global_snmp_config_label_by_id(self, snmp_config_id: str) -> Optional[str]:
+        """Method to get the global SNMP configuration label by id.
+
+        Args:
+            snmp_config_id (str): SNMP configuration id
+
+        Returns:
+            Optional[str]: SNMP configuration label, None if not found
+        """
+        return self._inventory_api.get_global_snmp_config_label_by_id(snmp_config_id=snmp_config_id)
+
+    def get_all_global_snmp_config_ids(self) -> dict[str, str]:
+        """Method to list all global SNMP configuration ids with their labels.
+
+        Returns:
+            dict: {snmp_config_id: snmp_config_label}
+        """
+        return self._inventory_api.get_all_global_snmp_config_ids()
+
+    # --- 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
+        """
+        return self._inventory_api.get_global_snmp_config(snmp_config_id=snmp_config_id)
+
+    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
+        """
+        return self._inventory_api.add_global_snmp_config(snmp_config=snmp_config)
+
+    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
+        """
+        return self._inventory_api.update_global_snmp_config(snmp_config=snmp_config)
+
+    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
+        """
+        return self._inventory_api.remove_global_snmp_config(snmp_config_id=snmp_config_id)
+
     # --- Deprecated Methods ---
     @deprecated(
         "This method is deprecated and will be removed in future versions.",

src/videoipath_automation_tool/apps/inventory/model/global_snmp_config.py

@@ -2,7 +2,34 @@
 from typing import Literal, cast
 from uuid import uuid4
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, field_validator
+
+
+def _validate_engine_id_format(engine_id: str) -> str:
+    """
+    Basic format check for SNMPv3 Engine ID.
+    - Allows empty string
+    - Must be valid hex string (after cleaning)
+    - Length: 5–32 bytes (10–64 hex digits)
+    - No RFC compliance guaranteed
+    """
+    if not engine_id:
+        return engine_id  # allowed for v1/v2c
+
+    cleaned = engine_id.replace(":", "").replace(" ", "").lower()
+
+    if len(cleaned) % 2 != 0:
+        raise ValueError("Engine ID must contain full bytes (even number of hex digits)")
+
+    try:
+        raw = bytes.fromhex(cleaned)
+    except ValueError:
+        raise ValueError("Engine ID must be a valid hex string")
+
+    if not (5 <= len(raw) <= 32):
+        raise ValueError("Engine ID must be 5–32 bytes long")
+
+    return cleaned
 
 
 # --- User Enum Classes ---
@@ -44,7 +71,7 @@ class SnmpVersion(int, Enum):
 
 
 # --- Data Model Classes ---
-class SnmpUser(BaseModel):
+class SnmpUser(BaseModel, validate_assignment=True):
     level: SecurityLevel = SecurityLevel.NO_AUTH_NO_PRIV
     name: str = "New User"
     authProtocol: AuthProtocol = AuthProtocol.MD5
@@ -53,32 +80,47 @@ class SnmpUser(BaseModel):
     privPassword: str = ""
     authPassword: str = ""
 
+    @field_validator("engineId")
+    def validate_engine_id(cls, value: str) -> str:
+        """
+        Validates the SNMPv3 Engine ID format.
+        - Allows empty string
+        - Must be valid hex string (after cleaning)
+        - Length: 5–32 bytes (10–64 hex digits)
+        """
+        return _validate_engine_id_format(value)
+
 
-class SnmpDescriptor(BaseModel):
+class SnmpDescriptor(BaseModel, validate_assignment=True):
     label: str = ""
     desc: str = ""
 
 
-class SnmpSecurityEntry(BaseModel):
+class SnmpSecurityEntry(BaseModel, validate_assignment=True):
     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):
+class SnmpSecurity(BaseModel, validate_assignment=True):
     read: SnmpSecurityEntry = SnmpSecurityEntry(community="public")
     write: SnmpSecurityEntry = SnmpSecurityEntry(community="private")
 
 
-class SnmpProtocolSettings(BaseModel):
+class SnmpProtocolSettings(BaseModel, validate_assignment=True):
     preferredVersion: SnmpVersion = SnmpVersion.V2C
-    retries: int = 1
-    maxRepetitions: int = 10
+    retries: int = Field(default=1, ge=0)
+    maxRepetitions: int = Field(default=10, ge=0)
     useGetBulk: bool = True
-    timeout: int = 5000
+    timeout: int = Field(default=5000, ge=0)
     localEngineId: str = ""
 
+    @field_validator("localEngineId")
+    @classmethod
+    def validate_engine_id(cls, value: str) -> str:
+        return _validate_engine_id_format(value)
+
 
-class SnmpConfiguration(BaseModel):
+class SnmpConfiguration(BaseModel, validate_assignment=True):
     id: str = Field(alias="_id")
     descriptor: SnmpDescriptor = Field(default_factory=SnmpDescriptor)
     users: dict[str, SnmpUser] = Field(default_factory=dict)
@@ -161,8 +203,6 @@ def retries(self) -> int:
     @retries.setter
     def retries(self, value: int):
         """Sets the number of retries for SNMP requests."""
-        if not isinstance(value, int) or value < 0:
-            raise ValueError("Retries must be a non-negative integer.")
         self.protocol.retries = value
 
     @property
@@ -173,8 +213,6 @@ def timeout(self) -> int:
     @timeout.setter
     def timeout(self, value: int):
         """Sets the timeout for SNMP requests in milliseconds."""
-        if not isinstance(value, int) or value < 0:
-            raise ValueError("Timeout must be a non-negative integer.")
         self.protocol.timeout = value
 
     @property
@@ -185,8 +223,6 @@ def local_engine_id(self) -> str:
     @local_engine_id.setter
     def local_engine_id(self, value: str):
         """Sets the local engine ID for SNMP."""
-        if not isinstance(value, str) or not value:
-            raise ValueError("Local Engine ID must be a non-empty string.")
         self.protocol.localEngineId = value
 
     @property
@@ -197,8 +233,6 @@ def use_get_bulk(self) -> bool:
     @use_get_bulk.setter
     def use_get_bulk(self, value: bool):
         """Sets whether to use GetBulk for SNMP requests."""
-        if not isinstance(value, bool):
-            raise ValueError("Use GetBulk must be a boolean value.")
         self.protocol.useGetBulk = value
 
     @property
@@ -209,8 +243,6 @@ def max_repetitions(self) -> int:
     @max_repetitions.setter
     def max_repetitions(self, value: int):
         """Sets the maximum number of repetitions for SNMP requests."""
-        if not isinstance(value, int) or value < 0:
-            raise ValueError("Max Repetitions must be a non-negative integer.")
         self.protocol.maxRepetitions = value
 
     @property
@@ -221,8 +253,6 @@ def read_community(self) -> str:
     @read_community.setter
     def read_community(self, value: str):
         """Sets the read community for SNMP."""
-        if not isinstance(value, str) or not value:
-            raise ValueError("Read Community must be a non-empty string.")
         self.security.read.community = value
 
     @property
@@ -233,8 +263,6 @@ def write_community(self) -> str:
     @write_community.setter
     def write_community(self, value: str):
         """Sets the write community for SNMP."""
-        if not isinstance(value, str) or not value:
-            raise ValueError("Write Community must be a non-empty string.")
         self.security.write.community = value
 
     def list_usernames(self) -> list[str]:

src/videoipath_automation_tool/apps/inventory/model/inventory_device_configuration.py

@@ -4,6 +4,7 @@
 from typing_extensions import deprecated
 
 from videoipath_automation_tool.apps.inventory.model.drivers import CustomSettingsType
+from videoipath_automation_tool.validators.uuid_4 import validate_uuid_4
 
 
 class CinfoOverridesSNMP(BaseModel, validate_assignment=True):
@@ -180,6 +181,55 @@ def metadata(self):
     def metadata(self, value):
         self.meta = value
 
+    @property
+    def use_global_snmp_settings(self) -> bool:
+        """Use (activate) global SNMP settings."""
+        return self.cinfoOverrides.snmp.useDefault
+
+    @use_global_snmp_settings.setter
+    def use_global_snmp_settings(self, value: bool):
+        """Use (activate) global SNMP settings."""
+        self.cinfoOverrides.snmp.useDefault = value
+
+    def get_global_snmp_setting_id(self) -> str:
+        """
+        Returns the ID of the global SNMP setting currently in use.
+
+        Returns:
+            str: The ID of the global SNMP setting.
+                - "default" if the system's default configuration is used.
+                - Otherwise, returns the UUID of the configured global SNMP setting.
+                Note: Retrieve the Label of the global SNMP setting using:
+                `app.inventory.get_global_snmp_config_label_by_id(id)`
+        """
+        return self.cinfoOverrides.snmp.id
+
+    def set_global_snmp_setting(self, setting_id: str, activate: bool = True):
+        """
+        Sets the global SNMP setting by ID.
+
+        Args:
+            setting_id (str): The ID of the global SNMP setting to use.
+                - Use "default" to apply the system's default configuration.
+                - For other configurations, retrieve the ID by label using:
+                    `app.inventory.get_global_snmp_config_id_by_label(label)`
+
+            activate (bool): Whether to activate the global SNMP settings (`use_global_snmp_settings = True`).
+                Defaults to True. Set to False if you only want to store the ID without activating it yet.
+
+        Raises:
+            ValueError: If the setting ID is empty or not a valid UUID (unless "default").
+        """
+        if not setting_id:
+            raise ValueError("Setting ID cannot be empty.")
+        if setting_id != "default":
+            setting_id = validate_uuid_4(setting_id)
+
+        self.cinfoOverrides.snmp.id = setting_id
+
+        if activate:
+            self.cinfoOverrides.snmp.useDefault = True
+
     # --- Deprecated properties ---
     @property
     @deprecated("The property `custom` is deprecated, use `custom_settings` instead.")