Add unit tests

diode-napalm-agent/requirements.txt

@@ -1,4 +1,5 @@
 napalm==5.0.0
 netboxlabs-diode-sdk==0.0.3
 pydantic==2.7.1
-python-dotenv==1.0.1
+python-dotenv==1.0.1
+pytest==8.2.1

diode-napalm-agent/tests/test_parser.py

@@ -0,0 +1,98 @@
+#!/usr/bin/env python
+# Copyright 2024 NetBox Labs Inc
+"""NetBox Labs - Tests."""
+import pytest
+import os
+from pathlib import Path
+from unittest.mock import patch, mock_open
+from diode_napalm.parser import parse_config, parse_config_file, ParseException, resolve_env_vars, Config
+
+
+@pytest.fixture
+def valid_yaml():
+    return """
+    diode:
+      config:
+        target: "target_value"
+        api_key: "api_key_value"
+        tls_verify: true
+      policies:
+        policy1:
+          config:
+            netbox:
+              site: "New York"
+          data:
+            - driver: "ios"
+              hostname: "router1"
+              username: "admin"
+              password: "password"
+    """
+
+
+@pytest.fixture
+def invalid_yaml():
+    return """
+    diode:
+      config:
+        target: "target_value"
+        api_key: "api_key_value"
+        tls_verify: true
+      policies:
+        policy1:
+          config:
+            netbox:
+              site: "New York"
+          data:
+            - driver: "ios"
+              hostname: "router1"
+              username: "admin"
+              # Missing password field
+    """
+
+
+def test_parse_valid_config(valid_yaml):
+    """Ensure we can parse a valid configuration."""
+    config = parse_config(valid_yaml)
+    assert isinstance(config, Config)
+    assert config.diode.config.target == "target_value"
+    assert config.diode.policies["policy1"].data[0].hostname == "router1"
+
+
+def test_parse_invalid_config(invalid_yaml):
+    """Ensure an invalid configuration raises a ParseException."""
+    with pytest.raises(ParseException):
+        parse_config(invalid_yaml)
+
+
+@patch("builtins.open", new_callable=mock_open, read_data="valid_yaml")
+def test_parse_config_file(mock_file, valid_yaml):
+    """Ensure we can parse a configuration file."""
+    with patch("parser.parse_config", return_value=parse_config(valid_yaml)):
+        config = parse_config_file(Path("fake_path.yaml"))
+        assert config.config.target == "target_value"
+        mock_file.assert_called_once_with("fake_path.yaml", "r")
+
+
+@patch.dict(os.environ, {"API_KEY": "env_api_key"})
+def test_resolve_env_vars():
+    """Ensure environment variables are resolved correctly."""
+    config_with_env_var = {
+        "api_key": "${API_KEY}"
+    }
+    resolved_config = resolve_env_vars(config_with_env_var)
+    assert resolved_config["api_key"] == "env_api_key"
+
+
+def test_resolve_env_vars_no_env():
+    """Ensure missing environment variables are handled correctly."""
+    config_with_no_env_var = {
+        "api_key": "${MISSING_KEY}"
+    }
+    resolved_config = resolve_env_vars(config_with_no_env_var)
+    assert resolved_config["api_key"] == "${MISSING_KEY}"
+
+
+def test_parse_config_file_exception():
+    """Ensure file parsing errors are handled correctly."""
+    with pytest.raises(Exception):
+        parse_config_file(Path("non_existent_file.yaml"))

diode-napalm-agent/tests/test_translate.py

@@ -0,0 +1,156 @@
+#!/usr/bin/env python
+# Copyright 2024 NetBox Labs Inc
+"""NetBox Labs - Translate Tests."""
+
+import pytest
+from netboxlabs.diode.sdk.ingester import (
+    Device,
+    DeviceType,
+    Platform,
+    Interface,
+    IPAddress,
+    Entity
+)
+from diode_napalm.translate import (
+    translate_device,
+    translate_interface,
+    translate_interface_ips,
+    translate_data
+)
+
+
+def test_translate_device():
+    """Ensure device information is correctly translated."""
+    device_info = {
+        "hostname": "device1",
+        "model": "modelX",
+        "vendor": "vendorY",
+        "driver": "driverZ",
+        "serial_number": "12345",
+        "site": "siteA"
+    }
+
+    device = translate_device(device_info)
+    assert device.name == "device1"
+    assert device.device_type.model == "modelX"
+    assert device.device_type.manufacturer == "vendorY"
+    assert device.platform.name == "driverZ"
+    assert device.platform.manufacturer == "vendorY"
+    assert device.serial == "12345"
+    assert device.status == "active"
+    assert device.site == "siteA"
+
+
+def test_translate_interface():
+    """Ensure interface information is correctly translated."""
+    device = Device(
+        name="device1",
+        device_type=DeviceType(model="modelX", manufacturer="vendorY"),
+        platform=Platform(name="driverZ", manufacturer="vendorY"),
+        serial="12345",
+        status="active",
+        site="siteA"
+    )
+    interface_info = {
+        "is_enabled": True,
+        "mtu": 1500,
+        "mac_address": "00:11:22:33:44:55",
+        "speed": 1000,
+        "description": "uplink"
+    }
+
+    interface = translate_interface(device, "eth0", interface_info)
+    assert interface.device == device
+    assert interface.name == "eth0"
+    assert interface.enabled is True
+    assert interface.mtu == 1500
+    assert interface.mac_address == "00:11:22:33:44:55"
+    assert interface.speed == 1000
+    assert interface.description == "uplink"
+
+
+def test_translate_interface_ips():
+    """Ensure interface IP addresses are correctly translated."""
+    device = Device(
+        name="device1",
+        device_type=DeviceType(model="modelX", manufacturer="vendorY"),
+        platform=Platform(name="driverZ", manufacturer="vendorY"),
+        serial="12345",
+        status="active",
+        site="siteA"
+    )
+    interface = Interface(
+        device=device,
+        name="eth0",
+        type="other",
+        enabled=True,
+        mtu=1500,
+        mac_address="00:11:22:33:44:55",
+        speed=1000,
+        description="uplink"
+    )
+    interfaces_ip = {
+        "eth0": {
+            "ipv4": {
+                "192.168.1.1": {"prefix_length": 24}
+            },
+            "ipv6": {
+                "fe80::1": {"prefix_length": 64}
+            }
+        }
+    }
+
+    ip_entities = list(translate_interface_ips(interface, interfaces_ip))
+    assert len(ip_entities) == 2
+    assert ip_entities[0].ip_address.address == "192.168.1.1/24"
+    assert ip_entities[1].ip_address.address == "fe80::1/64"
+    assert ip_entities[0].ip_address.interface == interface
+    assert ip_entities[1].ip_address.interface == interface
+
+
+def test_translate_data():
+    """Ensure full data translation works correctly."""
+    data = {
+        "device": {
+            "hostname": "device1",
+            "model": "modelX",
+            "vendor": "vendorY",
+            "serial_number": "12345",
+            "interface_list": ["eth0"],
+            "site": "siteA"
+        },
+        "driver": "driverZ",
+        "site": "siteA",
+        "interface": {
+            "eth0": {
+                "is_enabled": True,
+                "mtu": 1500,
+                "mac_address": "00:11:22:33:44:55",
+                "speed": 1000,
+                "description": "uplink"
+            }
+        },
+        "interface_ip": {
+            "eth0": {
+                "ipv4": {
+                    "192.168.1.1": {"prefix_length": 24}
+                },
+                "ipv6": {
+                    "fe80::1": {"prefix_length": 64}
+                }
+            }
+        }
+    }
+
+    entities = list(translate_data(data))
+    # 1 device, 1 interface, 2 IPs (but IPs are in one entity)
+    assert len(entities) == 3
+    device_entity = entities[0].device
+    assert device_entity.name == "device1"
+    assert device_entity.device_type.model == "modelX"
+    assert device_entity.device_type.manufacturer == "vendorY"
+    assert device_entity.platform.name == "driverZ"
+    assert device_entity.platform.manufacturer == "vendorY"
+    assert device_entity.serial == "12345"
+    assert device_entity.status == "active"
+    assert device_entity.site == "siteA"