Implement legacy field check & tests

hyperglass/models/config/devices.py

@@ -8,23 +8,29 @@ from pathlib import Path
 from ipaddress import IPv4Address, IPv6Address
 
 # Third Party
-from pydantic import StrictInt, StrictStr, StrictBool, validator, root_validator, Field
+from pydantic import StrictInt, StrictStr, StrictBool, validator, root_validator
 
 # Project
 from hyperglass.log import log
-from hyperglass.util import get_driver, get_fmt_keys, validate_device_type, resolve_hostname
+from hyperglass.util import (
+    get_driver,
+    get_fmt_keys,
+    resolve_hostname,
+    validate_device_type,
+)
 from hyperglass.constants import SCRAPE_HELPERS, SUPPORTED_STRUCTURED_OUTPUT
 from hyperglass.exceptions.private import ConfigError, UnsupportedDevice
-from hyperglass.models.commands.generic import Directive
 
 # Local
 from .ssl import Ssl
 from ..main import HyperglassModel, HyperglassModelWithId
+from ..util import check_legacy_fields
 from .proxy import Proxy
 from .params import Params
 from ..fields import SupportedDriver
 from .network import Network
 from .credential import Credential
+from ..commands.generic import Directive
 
 
 class Device(HyperglassModelWithId, extra="allow"):
@@ -39,7 +45,7 @@ class Device(HyperglassModelWithId, extra="allow"):
     display_name: Optional[StrictStr]
     port: StrictInt = 22
     ssl: Optional[Ssl]
-    type: StrictStr = Field(..., alias="nos")
+    type: StrictStr
     commands: List[Directive]
     structured_output: Optional[StrictBool]
     driver: Optional[SupportedDriver]
@@ -47,6 +53,7 @@ class Device(HyperglassModelWithId, extra="allow"):
 
     def __init__(self, **kwargs) -> None:
         """Set the device ID."""
+        kwargs = check_legacy_fields("Device", **kwargs)
         _id, values = self._generate_id(kwargs)
         super().__init__(id=_id, **values)
         self._validate_directive_attrs()
@@ -98,6 +105,18 @@ class Device(HyperglassModelWithId, extra="allow"):
             for command in rule.commands
         ]
 
+    @property
+    def directive_ids(self) -> List[str]:
+        """Get all directive IDs associated with the device."""
+        return [directive.id for directive in self.commands]
+
+    def has_directives(self, *directive_ids: str) -> bool:
+        """Determine if a directive is used on this device."""
+        for directive_id in directive_ids:
+            if directive_id in self.directive_ids:
+                return True
+        return False
+
     def _validate_directive_attrs(self) -> None:
 
         # Set of all keys except for built-in key `target`.
@@ -132,20 +151,6 @@ class Device(HyperglassModelWithId, extra="allow"):
                 )
         return value
 
-    @validator("type", pre=True, always=True)
-    def validate_type(cls: "Device", value: Any, values: Dict[str, Any]) -> str:
-        """Validate device type."""
-        legacy = values.pop("nos", None)
-        if legacy is not None and value is None:
-            log.warning(
-                "The 'nos' field on device '{}' has been deprecated and will be removed in a future release. Use the 'type' field moving forward.",
-                values.get("name", values.get("display_name", "Unknown")),
-            )
-            return legacy
-        if value is not None:
-            return value
-        raise ValueError("type is missing")
-
     @validator("structured_output", pre=True, always=True)
     def validate_structured_output(cls, value: bool, values: Dict) -> bool:
         """Validate structured output is supported on the device & set a default."""
@@ -183,8 +188,8 @@ class Device(HyperglassModelWithId, extra="allow"):
     def validate_device_commands(cls, values: Dict) -> Dict:
         """Validate & rewrite device type, set default commands."""
 
-        _type = values.get("type", values.get("nos"))
-        if not _type:
+        _type = values.get("type")
+        if _type is None:
             # Ensure device type is defined.
             raise ValueError(
                 f"Device {values['name']} is missing a 'type' (Network Operating System) property."

hyperglass/models/config/proxy.py

@@ -1,19 +1,19 @@
 """Validate SSH proxy configuration variables."""
 
 # Standard Library
-from typing import Union, Any, Dict
+from typing import Any, Dict, Union
 from ipaddress import IPv4Address, IPv6Address
 
 # Third Party
-from pydantic import StrictInt, StrictStr, validator, Field
+from pydantic import StrictInt, StrictStr, validator
 
 # Project
-from hyperglass.log import log
 from hyperglass.util import resolve_hostname
 from hyperglass.exceptions.private import ConfigError, UnsupportedDevice
 
 # Local
 from ..main import HyperglassModel
+from ..util import check_legacy_fields
 from .credential import Credential
 
 
@@ -24,7 +24,12 @@ class Proxy(HyperglassModel):
     address: Union[IPv4Address, IPv6Address, StrictStr]
     port: StrictInt = 22
     credential: Credential
-    type: StrictStr = Field("linux_ssh", alias="nos")
+    type: StrictStr = "linux_ssh"
+
+    def __init__(self: "Proxy", **kwargs: Any) -> None:
+        """Check for legacy fields."""
+        kwargs = check_legacy_fields("Proxy", **kwargs)
+        super().__init__(**kwargs)
 
     @property
     def _target(self):
@@ -46,19 +51,11 @@ class Proxy(HyperglassModel):
     @validator("type", pre=True, always=True)
     def validate_type(cls: "Proxy", value: Any, values: Dict[str, Any]) -> str:
         """Validate device type."""
-        legacy = values.pop("nos", None)
-        if legacy is not None and value is None:
-            log.warning(
-                "The 'nos' field on proxy '{}' has been deprecated and will be removed in a future release. Use the 'type' field moving forward.",
-                values.get("name", "Unknown"),
+
+        if value != "linux_ssh":
+            raise UnsupportedDevice(
+                "Proxy '{p}' uses type '{t}', which is currently unsupported.",
+                p=values["name"],
+                t=value,
             )
-            return legacy
-        if value is not None:
-            if value != "linux_ssh":
-                raise UnsupportedDevice(
-                    "Proxy '{p}' uses type '{t}', which is currently unsupported.",
-                    p=values["name"],
-                    t=value,
-                )
-            return value
-        raise ValueError("type is missing")
+        return value

hyperglass/models/tests/__init__.py

@@ -0,0 +1 @@
+"""Model tests."""

hyperglass/models/tests/test_util.py

@@ -0,0 +1,22 @@
+"""Test model utilities."""
+
+# Third Party
+import pytest
+
+# Local
+from ..util import check_legacy_fields
+
+
+def test_check_legacy_fields():
+    test1 = {"name": "Device A", "nos": "juniper"}
+    test1_expected = {"name": "Device A", "type": "juniper"}
+    test2 = {"name": "Device B", "type": "juniper"}
+    test3 = {"name": "Device C"}
+    assert set(check_legacy_fields("Device", **test1).keys()) == set(
+        test1_expected.keys()
+    ), "legacy field not replaced"
+    assert set(check_legacy_fields("Device", **test2).keys()) == set(
+        test2.keys()
+    ), "new field not left unmodified"
+    with pytest.raises(ValueError):
+        check_legacy_fields("Device", **test3)

hyperglass/models/util.py

@@ -0,0 +1,30 @@
+"""Model utilities."""
+
+# Standard Library
+from typing import Any, Dict, Tuple
+
+# Project
+from hyperglass.log import log
+
+LEGACY_FIELDS: Dict[str, Tuple[Tuple[str, str], ...]] = {
+    "Device": (("nos", "type"),),
+    "Proxy": (("nos", "type"),),
+}
+
+
+def check_legacy_fields(model: str, **kwargs: Dict[str, Any]) -> Dict[str, Any]:
+    """Check for legacy fields prior to model initialization."""
+    if model in LEGACY_FIELDS:
+        for legacy_key, new_key in LEGACY_FIELDS[model]:
+            legacy_value = kwargs.pop(legacy_key, None)
+            new_value = kwargs.get(new_key)
+            if legacy_value is not None and new_value is None:
+                log.warning(
+                    "The {} field has been deprecated and will be removed in a future release. Use the '{}' field moving forward.",
+                    f"{model}.{legacy_key}",
+                    new_key,
+                )
+                kwargs[new_key] = legacy_value
+            elif legacy_value is None and new_value is None:
+                raise ValueError(f"'{new_key}' is missing")
+    return kwargs