from fastapi.testclient import TestClient from app.core.secrets import encrypt_secret from app.models import Credential from app.services.snmp import ( ENT_PHYSICAL_NAME, ENT_PHY_SENSOR_TYPE, ENT_PHY_SENSOR_VALUE, HR_PROCESSOR_LOAD, HR_STORAGE_ALLOCATION_UNITS, HR_STORAGE_DESCR, HR_STORAGE_FIXED_DISK, HR_STORAGE_RAM, HR_STORAGE_SIZE, HR_STORAGE_TYPE, HR_STORAGE_USED, SYS_DESCR, SYS_NAME, SYS_UPTIME, UCD_DSK_PATH, UCD_DSK_PERCENT, UCD_LA_LOAD_INT, UCD_MEM_AVAIL_REAL, UCD_MEM_TOTAL_REAL, DiscoveredSnmpDevice, DiscoveredSnmpInterface, IF_DESCR, IF_NAME, SnmpCredential, SnmpDiscoveryError, discover_snmp_device, ) def test_snmp_discovery_uses_profile_and_returns_friendly_results(client: TestClient, db_session, monkeypatch) -> None: profile = Credential( name="Core Switch", credential_type="snmp", encrypted_secret=encrypt_secret("private-community"), extra={"version": "2c", "port": 1161, "timeout_seconds": 4, "retries": 2}, ) db_session.add(profile) db_session.commit() calls: list[tuple[str, SnmpCredential, str | None]] = [] def fake_discover(host: str, credential: SnmpCredential, asset_type: str | None = None) -> DiscoveredSnmpDevice: calls.append((host, credential, asset_type)) return DiscoveredSnmpDevice( host=host, device_name="core-sw-1", description="Core switch", uptime_seconds=12345, interfaces=[ DiscoveredSnmpInterface( index=1, name="Gi1/0/1", label="GigabitEthernet 1/0/1", description="Uplink", admin_status="up", oper_status="up", speed_bps=1_000_000_000, ) ], ) monkeypatch.setattr("app.api.discovery.discover_snmp_device", fake_discover) response = client.post("/discovery/snmp", json={"host": "192.0.2.10", "credential_profile_id": profile.id}) assert response.status_code == 200 assert calls == [ ( "192.0.2.10", SnmpCredential(community="private-community", port=1161, timeout_seconds=4, retries=2), None, ) ] body = response.json() assert body["host"] == "192.0.2.10" assert body["credential_profile_id"] == profile.id assert body["profile_key"] == "generic_snmp" assert body["profile_name"] == "Generic SNMP" assert body["capabilities"] == {} assert body["device_name"] == "core-sw-1" assert body["description"] == "Core switch" assert body["uptime_seconds"] == 12345 assert body["interfaces"] == [ { "index": 1, "name": "Gi1/0/1", "label": "GigabitEthernet 1/0/1", "description": "Uplink", "admin_status": "up", "oper_status": "up", "speed_bps": 1_000_000_000, } ] assert body["monitorable_items"] == [ { "item_id": "device.uptime", "item_type": "device_uptime", "group": "Device Health", "label": "Device uptime", "unit": "seconds", }, { "item_id": "interface.1.status", "item_type": "interface_status", "group": "Interface GigabitEthernet 1/0/1", "label": "GigabitEthernet 1/0/1 status", "unit": None, }, { "item_id": "interface.1.traffic", "item_type": "interface_traffic", "group": "Interface GigabitEthernet 1/0/1", "label": "GigabitEthernet 1/0/1 traffic", "unit": "bps", }, { "item_id": "interface.1.errors", "item_type": "interface_errors", "group": "Interface GigabitEthernet 1/0/1", "label": "GigabitEthernet 1/0/1 errors and discards", "unit": "count", }, ] assert "private-community" not in response.text assert "1.3.6" not in response.text def test_snmp_interface_discovery_normalizes_network_interface_labels(monkeypatch) -> None: class FakeClient: def __init__(self, host: str, credential: SnmpCredential) -> None: self.host = host self.credential = credential def get_many(self, _oids): return { SYS_NAME: "lan-party-router", SYS_DESCR: "MikroTik RouterOS", SYS_UPTIME: 10_000, } def walk(self, base_oid, max_items=128): values = { IF_NAME: { (*IF_NAME, 1): "ether1", (*IF_NAME, 2): "sfp-sfpplus1", (*IF_NAME, 3): "bridge1", (*IF_NAME, 4): "vlan20", (*IF_NAME, 5): "lo", (*IF_NAME, 6): "Gi1/0/1", (*IF_NAME, 7): "Te1/1/1", (*IF_NAME, 8): "xe-0/0/1", }, IF_DESCR: { (*IF_DESCR, 1): "WAN", (*IF_DESCR, 2): "Fiber uplink", (*IF_DESCR, 3): "LAN bridge", (*IF_DESCR, 4): "Guest VLAN", }, } return values.get(base_oid, {}) monkeypatch.setattr("app.services.snmp.SnmpV2Client", FakeClient) discovered = discover_snmp_device("10.10.10.254", SnmpCredential(community="private-community")) assert [(interface.name, interface.label, interface.description) for interface in discovered.interfaces] == [ ("ether1", "Port 1", "WAN"), ("sfp-sfpplus1", "SFP+ 1", "Fiber uplink"), ("bridge1", "Bridge 1", "LAN bridge"), ("vlan20", "VLAN 20", "Guest VLAN"), ("lo", "Loopback", None), ("Gi1/0/1", "GigabitEthernet 1/0/1", None), ("Te1/1/1", "TenGigabitEthernet 1/1/1", None), ("xe-0/0/1", "TenGigabitEthernet 0/0/1", None), ] def test_snmp_server_asset_type_uses_linux_server_mibs_and_keeps_interfaces(monkeypatch) -> None: class FakeClient: def __init__(self, host: str, credential: SnmpCredential) -> None: self.host = host self.credential = credential def get_many(self, oids): values = { SYS_NAME: "app-1", SYS_DESCR: "Linux app-1 6.1.0 net-snmp", SYS_UPTIME: 10_000, UCD_MEM_TOTAL_REAL: 1_000_000, UCD_MEM_AVAIL_REAL: 250_000, } return {oid: values.get(oid) for oid in oids} def walk(self, base_oid, max_items=128): values = { IF_NAME: {(*IF_NAME, 1): "eth0"}, UCD_LA_LOAD_INT: { (*UCD_LA_LOAD_INT, 1): 123, (*UCD_LA_LOAD_INT, 2): 97, (*UCD_LA_LOAD_INT, 3): 88, }, UCD_DSK_PATH: {(*UCD_DSK_PATH, 1): "/"}, UCD_DSK_PERCENT: {(*UCD_DSK_PERCENT, 1): 42}, } return values.get(base_oid, {}) monkeypatch.setattr("app.services.snmp.SnmpV2Client", FakeClient) discovered = discover_snmp_device("192.0.2.21", SnmpCredential(community="private-community"), asset_type="server") assert discovered.profile_key == "linux_server" assert discovered.profile_name == "Linux Server" assert [(interface.name, interface.label) for interface in discovered.interfaces] == [("eth0", "eth0")] assert discovered.capabilities["interfaces"] is True assert discovered.capabilities["cpu"] is True assert discovered.capabilities["memory"] is True assert discovered.capabilities["storage"] is True assert [(item.item_id, item.item_type, item.group, item.label, item.unit) for item in discovered.health_items] == [ ("linux.load.1", "linux_load_average", "Server Health", "Load average 1 minute", None), ("linux.load.2", "linux_load_average", "Server Health", "Load average 5 minutes", None), ("linux.load.3", "linux_load_average", "Server Health", "Load average 15 minutes", None), ("linux.memory.real", "linux_memory_usage", "Server Health", "Memory used", "%"), ("linux.disk.1", "linux_disk_usage", "Storage", "Disk / usage", "%"), ] def test_snmp_server_asset_type_falls_back_to_host_resources(monkeypatch) -> None: class FakeClient: def __init__(self, host: str, credential: SnmpCredential) -> None: self.host = host self.credential = credential def get_many(self, oids): values = { SYS_NAME: "app-2", SYS_DESCR: "Linux app-2", SYS_UPTIME: 10_000, UCD_MEM_TOTAL_REAL: None, UCD_MEM_AVAIL_REAL: None, } return {oid: values.get(oid) for oid in oids} def walk(self, base_oid, max_items=128): values = { IF_NAME: {(*IF_NAME, 1): "eth0"}, HR_PROCESSOR_LOAD: {(*HR_PROCESSOR_LOAD, 196608): 17}, HR_STORAGE_TYPE: { (*HR_STORAGE_TYPE, 1): HR_STORAGE_RAM, (*HR_STORAGE_TYPE, 31): HR_STORAGE_FIXED_DISK, (*HR_STORAGE_TYPE, 35): HR_STORAGE_FIXED_DISK, }, HR_STORAGE_DESCR: { (*HR_STORAGE_DESCR, 1): "Physical memory", (*HR_STORAGE_DESCR, 31): "/", (*HR_STORAGE_DESCR, 35): "/run/credentials/systemd-journald.service", }, HR_STORAGE_ALLOCATION_UNITS: { (*HR_STORAGE_ALLOCATION_UNITS, 1): 1024, (*HR_STORAGE_ALLOCATION_UNITS, 31): 4096, (*HR_STORAGE_ALLOCATION_UNITS, 35): 4096, }, HR_STORAGE_SIZE: { (*HR_STORAGE_SIZE, 1): 2048, (*HR_STORAGE_SIZE, 31): 4096, (*HR_STORAGE_SIZE, 35): 4096, }, HR_STORAGE_USED: { (*HR_STORAGE_USED, 1): 1024, (*HR_STORAGE_USED, 31): 1024, (*HR_STORAGE_USED, 35): 1024, }, } return values.get(base_oid, {}) monkeypatch.setattr("app.services.snmp.SnmpV2Client", FakeClient) discovered = discover_snmp_device("192.0.2.22", SnmpCredential(community="private-community"), asset_type="server") assert discovered.profile_key == "linux_server" assert [(interface.name, interface.label) for interface in discovered.interfaces] == [("eth0", "eth0")] assert [(item.item_id, item.item_type, item.group, item.label, item.unit) for item in discovered.health_items] == [ ("cpu.196608.load", "cpu_load", "Device Health", "CPU load", "%"), ("storage.1.memory", "memory_usage", "Device Health", "Memory used", "%"), ("storage.31.usage", "storage_usage", "Storage", "Disk / usage", "%"), ] def test_snmp_profile_mapping_discovers_standard_health_items(monkeypatch) -> None: class FakeClient: def __init__(self, host: str, credential: SnmpCredential) -> None: self.host = host self.credential = credential def get_many(self, _oids): return { SYS_NAME: "edge-router", SYS_DESCR: "Generic SNMP appliance", SYS_UPTIME: 10_000, } def walk(self, base_oid, max_items=128): values = { HR_PROCESSOR_LOAD: {(*HR_PROCESSOR_LOAD, 196608): 17}, HR_STORAGE_TYPE: { (*HR_STORAGE_TYPE, 1): HR_STORAGE_RAM, (*HR_STORAGE_TYPE, 31): HR_STORAGE_FIXED_DISK, }, HR_STORAGE_DESCR: { (*HR_STORAGE_DESCR, 1): "Physical memory", (*HR_STORAGE_DESCR, 31): "/", }, HR_STORAGE_ALLOCATION_UNITS: { (*HR_STORAGE_ALLOCATION_UNITS, 1): 1024, (*HR_STORAGE_ALLOCATION_UNITS, 31): 4096, }, HR_STORAGE_SIZE: { (*HR_STORAGE_SIZE, 1): 2048, (*HR_STORAGE_SIZE, 31): 4096, }, HR_STORAGE_USED: { (*HR_STORAGE_USED, 1): 1024, (*HR_STORAGE_USED, 31): 1024, }, ENT_PHY_SENSOR_TYPE: {(*ENT_PHY_SENSOR_TYPE, 10): 8}, ENT_PHY_SENSOR_VALUE: {(*ENT_PHY_SENSOR_VALUE, 10): 310}, ENT_PHYSICAL_NAME: {(*ENT_PHYSICAL_NAME, 10): "Inlet"}, } return values.get(base_oid, {}) monkeypatch.setattr("app.services.snmp.SnmpV2Client", FakeClient) discovered = discover_snmp_device("192.0.2.20", SnmpCredential(community="private-community")) assert discovered.profile_key == "generic_snmp" assert discovered.profile_name == "Generic SNMP" assert discovered.capabilities["cpu"] is True assert discovered.capabilities["memory"] is True assert discovered.capabilities["storage"] is True assert discovered.capabilities["sensors"] is True assert [(item.item_id, item.item_type, item.group, item.label, item.unit) for item in discovered.health_items] == [ ("cpu.196608.load", "cpu_load", "Device Health", "CPU load", "%"), ("storage.1.memory", "memory_usage", "Device Health", "Memory used", "%"), ("storage.31.usage", "storage_usage", "Storage", "Disk / usage", "%"), ("sensor.10.value", "sensor_value", "Environmental", "Temperature Inlet", "C"), ] def test_snmp_profile_mapping_identifies_proxmox_before_linux(monkeypatch) -> None: class FakeClient: def __init__(self, host: str, credential: SnmpCredential) -> None: self.host = host self.credential = credential def get_many(self, _oids): return { SYS_NAME: "pve-1", SYS_DESCR: "Linux pve-1 6.8.12-9-pve #1 SMP PREEMPT_DYNAMIC PMX", SYS_UPTIME: 10_000, } def walk(self, base_oid, max_items=128): return {} monkeypatch.setattr("app.services.snmp.SnmpV2Client", FakeClient) discovered = discover_snmp_device("192.0.2.30", SnmpCredential(community="private-community")) assert discovered.profile_key == "proxmox_ve" assert discovered.profile_name == "Proxmox VE Server" def test_snmp_discovery_rejects_missing_profile(client: TestClient) -> None: response = client.post("/discovery/snmp", json={"host": "192.0.2.10", "credential_profile_id": 999}) assert response.status_code == 404 def test_snmp_discovery_rejects_profile_without_secret(client: TestClient, db_session) -> None: profile = Credential( name="No Secret", credential_type="snmp", encrypted_secret=None, extra={"version": "2c"}, ) db_session.add(profile) db_session.commit() response = client.post("/discovery/snmp", json={"host": "192.0.2.10", "credential_profile_id": profile.id}) assert response.status_code == 400 def test_snmp_discovery_reports_probe_failure(client: TestClient, db_session, monkeypatch) -> None: profile = Credential( name="Core Switch", credential_type="snmp", encrypted_secret=encrypt_secret("private-community"), extra={"version": "2c"}, ) db_session.add(profile) db_session.commit() def fake_discover(host: str, credential: SnmpCredential, asset_type: str | None = None) -> DiscoveredSnmpDevice: raise SnmpDiscoveryError("timeout") monkeypatch.setattr("app.api.discovery.discover_snmp_device", fake_discover) response = client.post("/discovery/snmp", json={"host": "192.0.2.10", "credential_profile_id": profile.id}) assert response.status_code == 502 assert "private-community" not in response.text