Keith Smith
139 lines
4.6 KiB
Python
139 lines
4.6 KiB
Python
from fastapi.testclient import TestClient
|
|
|
|
from app.core.secrets import encrypt_secret
|
|
from app.models import Credential
|
|
from app.services.snmp import DiscoveredSnmpDevice, DiscoveredSnmpInterface, SnmpCredential, SnmpDiscoveryError
|
|
|
|
|
|
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]] = []
|
|
|
|
def fake_discover(host: str, credential: SnmpCredential) -> DiscoveredSnmpDevice:
|
|
calls.append((host, credential))
|
|
return DiscoveredSnmpDevice(
|
|
host=host,
|
|
device_name="core-sw-1",
|
|
description="Core switch",
|
|
uptime_seconds=12345,
|
|
interfaces=[
|
|
DiscoveredSnmpInterface(
|
|
index=1,
|
|
name="Gi1/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),
|
|
)
|
|
]
|
|
body = response.json()
|
|
assert body["host"] == "192.0.2.10"
|
|
assert body["credential_profile_id"] == profile.id
|
|
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",
|
|
"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 Gi1/0/1",
|
|
"label": "Gi1/0/1 status",
|
|
"unit": None,
|
|
},
|
|
{
|
|
"item_id": "interface.1.traffic",
|
|
"item_type": "interface_traffic",
|
|
"group": "Interface Gi1/0/1",
|
|
"label": "Gi1/0/1 traffic",
|
|
"unit": "bps",
|
|
},
|
|
{
|
|
"item_id": "interface.1.errors",
|
|
"item_type": "interface_errors",
|
|
"group": "Interface Gi1/0/1",
|
|
"label": "Gi1/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_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) -> 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
|