diff --git a/gso/services/librenms_client.py b/gso/services/librenms_client.py
index 48bc1dc4bdbfef7f564032576e35c749f8514756..e4564b832482346569e29bae3fa3d1dc6f06d686 100644
--- a/gso/services/librenms_client.py
+++ b/gso/services/librenms_client.py
@@ -98,12 +98,15 @@ class LibreNMSClient:
:return list[str]: A list of errors, if empty the device is successfully validated.
"""
errors = []
- device = self.get_device(fqdn)
-
- if device["status"] != "ok":
- errors += ["Device does not exist in LibreNMS."]
+ try:
+ device = self.get_device(fqdn)
- if device["hostname"] != fqdn:
- errors += ["Device hostname in LibreNMS does not match FQDN."]
+ if device["devices"][0]["hostname"] != fqdn:
+ errors += ["Device hostname in LibreNMS does not match FQDN."]
+ except HTTPError as e:
+ if e.response.status_code == HTTPStatus.NOT_FOUND:
+ errors += ["Device does not exist in LibreNMS."]
+ else:
+ raise
return errors
diff --git a/test/conftest.py b/test/conftest.py
index 0ce79f9d8f3f89c169ba9e614c5edf5457cdb832..edb1dffba401f2f57b5c140d171570468c7ee8e7 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -176,7 +176,7 @@ def configuration_data() -> dict:
},
"MONITORING": {
"LIBRENMS": {
- "base_url": "http://fake.url.local",
+ "base_url": "http://librenms",
"token": "secret-token",
},
"SNMP": {
diff --git a/test/services/test_librenms_client.py b/test/services/test_librenms_client.py
new file mode 100644
index 0000000000000000000000000000000000000000..55df5ce176329a66587b53c366988b325ecf49e0
--- /dev/null
+++ b/test/services/test_librenms_client.py
@@ -0,0 +1,458 @@
+from http import HTTPStatus
+from unittest.mock import patch
+
+import pytest
+from requests import HTTPError
+
+from gso.services.librenms_client import LibreNMSClient
+from gso.utils.helpers import SNMPVersion
+
+
+@pytest.fixture()
+def mock_get_device_success(faker):
+ with patch("gso.services.librenms_client.requests.get") as mock_get_device:
+ mock_get_device().status_code = HTTPStatus.OK
+ mock_get_device().json.return_value = {
+ "status": "ok",
+ "devices": [
+ {
+ "device_id": 1,
+ "inserted": faker.date("%Y-%m-%dT%H:%M:%S.%fZ"),
+ "hostname": "localhost",
+ "sysName": "librenms",
+ "display": None,
+ "ip": faker.ipv4(),
+ "overwrite_ip": None,
+ "community": "librenms-community",
+ "authlevel": None,
+ "authname": None,
+ "authpass": None,
+ "authalgo": None,
+ "cryptopass": None,
+ "cryptoalgo": None,
+ "snmpver": "v2c",
+ "port": faker.port_number(),
+ "transport": "udp",
+ "timeout": None,
+ "retries": None,
+ "snmp_disable": 0,
+ "bgpLocalAs": None,
+ "sysObjectID": ".1.3.6.1.4.1.8072.3.2.10",
+ "sysDescr": "Linux librenms 5.15.0-79-generic #86-Ubuntu SMP Mon Jul 10 16:07:21 UTC 2023 x86_64",
+ "sysContact": "Your Name <your@email.address>",
+ "version": "5.15.0-79-generic",
+ "hardware": "Generic x86 64-bit",
+ "features": "Ubuntu 22.04",
+ "location_id": 1,
+ "os": "linux",
+ "status": True,
+ "status_reason": "",
+ "ignore": 0,
+ "disabled": 0,
+ "uptime": faker.pyint(),
+ "agent_uptime": 0,
+ "last_polled": faker.date("%Y-%m-%dT%H:%M:%S.%fZ"),
+ "last_poll_attempted": None,
+ "last_polled_timetaken": faker.pyfloat(left_digits=1, positive=True),
+ "last_discovered_timetaken": faker.pyfloat(left_digits=1, positive=True),
+ "last_discovered": faker.date("%Y-%m-%dT%H:%M:%S.%fZ"),
+ "last_ping": faker.date("%Y-%m-%dT%H:%M:%S.%fZ"),
+ "last_ping_timetaken": faker.pyfloat(left_digits=1, positive=True),
+ "purpose": None,
+ "type": "server",
+ "serial": None,
+ "icon": "images/os/ubuntu.svg",
+ "poller_group": 0,
+ "override_sysLocation": 0,
+ "notes": None,
+ "port_association_mode": 1,
+ "max_depth": 0,
+ "disable_notify": 0,
+ "location": "Rack, Room, Building, City, Country [Lat, Lon]",
+ "lat": None,
+ "lng": None,
+ },
+ ],
+ "count": 1,
+ }
+
+ yield mock_get_device
+
+
+@pytest.fixture()
+def mock_get_device_not_found():
+ with patch("gso.services.librenms_client.requests.get") as mock_get_not_found:
+ mock_get_not_found().status_code = HTTPStatus.NOT_FOUND
+ mock_get_not_found().json.return_value = {
+ "status": "error",
+ "message": "Device non-existent-url does not exist",
+ }
+ mock_get_not_found().raise_for_status.side_effect = HTTPError(
+ "404 Client Error: Not Found for url: http://librenms/devices/non-existent-url",
+ response=mock_get_not_found(),
+ )
+
+ yield mock_get_not_found
+
+
+@pytest.fixture()
+def mock_get_device_misconfigured(faker):
+ with patch("gso.services.librenms_client.requests.get") as mock_get_device:
+ mock_get_device().status_code = HTTPStatus.OK
+ mock_get_device().json.return_value = {
+ "status": "ok",
+ "devices": [
+ {
+ "device_id": 1,
+ "inserted": faker.date("%Y-%m-%dT%H:%M:%S.%fZ"),
+ "hostname": "127.0.0.1",
+ "sysName": "librenms",
+ "display": None,
+ "ip": faker.ipv4(),
+ "overwrite_ip": None,
+ "community": "librenms-community",
+ "authlevel": None,
+ "authname": None,
+ "authpass": None,
+ "authalgo": None,
+ "cryptopass": None,
+ "cryptoalgo": None,
+ "snmpver": "v2c",
+ "port": faker.port_number(),
+ "transport": "udp",
+ "timeout": None,
+ "retries": None,
+ "snmp_disable": 0,
+ "bgpLocalAs": None,
+ "sysObjectID": ".1.3.6.1.4.1.8072.3.2.10",
+ "sysDescr": "Linux librenms 5.15.0-79-generic #86-Ubuntu SMP Mon Jul 10 16:07:21 UTC 2023 x86_64",
+ "sysContact": "Your Name <your@email.address>",
+ "version": "5.15.0-79-generic",
+ "hardware": "Generic x86 64-bit",
+ "features": "Ubuntu 22.04",
+ "location_id": 1,
+ "os": "linux",
+ "status": True,
+ "status_reason": "",
+ "ignore": 0,
+ "disabled": 0,
+ "uptime": faker.pyint(),
+ "agent_uptime": 0,
+ "last_polled": faker.date("%Y-%m-%dT%H:%M:%S.%fZ"),
+ "last_poll_attempted": None,
+ "last_polled_timetaken": faker.pyfloat(left_digits=1, positive=True),
+ "last_discovered_timetaken": faker.pyfloat(left_digits=1, positive=True),
+ "last_discovered": faker.date("%Y-%m-%dT%H:%M:%S.%fZ"),
+ "last_ping": faker.date("%Y-%m-%dT%H:%M:%S.%fZ"),
+ "last_ping_timetaken": faker.pyfloat(left_digits=1, positive=True),
+ "purpose": None,
+ "type": "server",
+ "serial": None,
+ "icon": "images/os/ubuntu.svg",
+ "poller_group": 0,
+ "override_sysLocation": 0,
+ "notes": None,
+ "port_association_mode": 1,
+ "max_depth": 0,
+ "disable_notify": 0,
+ "location": "Rack, Room, Building, City, Country [Lat, Lon]",
+ "lat": None,
+ "lng": None,
+ },
+ ],
+ "count": 1,
+ }
+
+ yield mock_get_device
+
+
+@pytest.fixture()
+def mock_get_device_unauthenticated():
+ with patch("gso.services.librenms_client.requests.get") as mock_get_unauthorized, patch(
+ "gso.services.librenms_client.LibreNMSClient.get_device",
+ ) as mock_get_device:
+ mock_get_unauthorized().status_code = HTTPStatus.UNAUTHORIZED
+ mock_get_unauthorized().json.return_value = {"message": "Unauthenticated."}
+ mock_get_device.side_effect = HTTPError(
+ "401 Client Error: Unauthorized for url: http://librenms/devices/naughty-url",
+ response=mock_get_unauthorized(),
+ )
+
+ yield mock_get_unauthorized
+
+
+@pytest.fixture()
+def mock_add_device_success():
+ with patch("gso.services.librenms_client.requests.post") as mock_post_device:
+ mock_post_device().status_code = HTTPStatus.OK
+ mock_post_device().json.return_value = {
+ "status": "ok",
+ "devices": [
+ {
+ "community": "secret-community",
+ "display": "localhost",
+ "hostname": "localhost",
+ "snmpver": "v2c",
+ "port": 161,
+ "transport": "udp",
+ "poller_group": 0,
+ "os": "linux",
+ "status_reason": "",
+ "sysName": "librenms",
+ "port_association_mode": 1,
+ "authlevel": None,
+ "authname": None,
+ "authalgo": None,
+ "cryptopass": None,
+ "cryptoalgo": None,
+ "sysDescr": "Linux librenms 5.15.0-79-generic #86-Ubuntu SMP Mon Jul 10 16:07:21 UTC 2023 x86_64",
+ "sysObjectID": ".1.3.6.1.4.1.8072.3.2.10",
+ "device_id": 2,
+ },
+ ],
+ "message": "Device localhost has been added successfully",
+ "count": 1,
+ }
+
+ yield mock_post_device
+
+
+@pytest.fixture()
+def mock_add_device_bad_url():
+ with patch("gso.services.librenms_client.requests.post") as mock_post_device:
+ mock_post_device().status_code = HTTPStatus.INTERNAL_SERVER_ERROR
+ mock_post_device().json.return_value = {
+ "status": "error",
+ "message": "Could not ping non-existent-url (Hostname did not resolve to IP)",
+ }
+ mock_post_device().raise_for_status.side_effect = HTTPError(
+ "500 Server Error: Internal server error for url: http://librenms/devices",
+ response=mock_post_device(),
+ )
+
+ yield mock_post_device
+
+
+@pytest.fixture()
+def mock_add_device_unreachable():
+ with patch("gso.services.librenms_client.requests.post") as mock_post_device:
+ mock_post_device().status_code = HTTPStatus.INTERNAL_SERVER_ERROR
+ mock_post_device().json.return_value = {
+ "status": "error",
+ "message": "Could not connect to non-existent-url, please check the snmp details and snmp reachability",
+ }
+ mock_post_device().raise_for_status.side_effect = HTTPError(
+ "500 Server Error: Internal server error for url: http://librenms/devices",
+ response=mock_post_device(),
+ )
+
+ yield mock_post_device
+
+
+@pytest.fixture()
+def mock_remove_device_success(faker):
+ with patch("gso.services.librenms_client.requests.delete") as mock_remove_device:
+ mock_remove_device().status_code = HTTPStatus.OK
+ mock_remove_device().json.return_value = {
+ "status": "ok",
+ "devices": [
+ {
+ "device_id": 2,
+ "inserted": faker.date("%Y-%m-%dT%H:%M:%S.%fZ"),
+ "hostname": "localhost",
+ "sysName": "librenms",
+ "display": "localhost",
+ "ip": faker.ipv4(),
+ "overwrite_ip": None,
+ "community": "snmp-community",
+ "authlevel": None,
+ "authname": None,
+ "authpass": None,
+ "authalgo": None,
+ "cryptopass": None,
+ "cryptoalgo": None,
+ "snmpver": "v2c",
+ "port": 161,
+ "transport": "udp",
+ "timeout": None,
+ "retries": None,
+ "snmp_disable": 0,
+ "bgpLocalAs": None,
+ "sysObjectID": ".1.3.6.1.4.1.8072.3.2.10",
+ "sysDescr": "Linux librenms 5.15.0-79-generic #86-Ubuntu SMP Mon Jul 10 16:07:21 UTC 2023 x86_64",
+ "sysContact": "Your Name <your@email.address>",
+ "version": "5.15.0-79-generic",
+ "hardware": "Generic x86 64-bit",
+ "features": "Ubuntu 22.04",
+ "location_id": 1,
+ "os": "linux",
+ "status": True,
+ "status_reason": "",
+ "ignore": 0,
+ "disabled": 0,
+ "uptime": 8057430,
+ "agent_uptime": 0,
+ "last_polled": faker.date("%Y-%m-%dT%H:%M:%S.%fZ"),
+ "last_poll_attempted": None,
+ "last_polled_timetaken": faker.pyfloat(left_digits=1, positive=True),
+ "last_discovered_timetaken": faker.pyfloat(left_digits=1, positive=True),
+ "last_discovered": faker.date("%Y-%m-%dT%H:%M:%S.%fZ"),
+ "last_ping": faker.date("%Y-%m-%dT%H:%M:%S.%fZ"),
+ "last_ping_timetaken": faker.pyfloat(left_digits=1, positive=True),
+ "purpose": None,
+ "type": "server",
+ "serial": None,
+ "icon": "images/os/ubuntu.svg",
+ "poller_group": 0,
+ "override_sysLocation": 0,
+ "notes": None,
+ "port_association_mode": 1,
+ "max_depth": 0,
+ "disable_notify": 0,
+ "location": "Rack, Room, Building, City, Country [Lat, Lon]",
+ "lat": None,
+ "lng": None,
+ },
+ ],
+ "message": "Removed device localhost\n",
+ "count": 1,
+ }
+
+ yield mock_remove_device
+
+
+@pytest.fixture()
+def mock_remove_device_non_existent(faker):
+ with patch("gso.services.librenms_client.requests.delete") as mock_remove_device:
+ mock_remove_device().status_code = HTTPStatus.NOT_FOUND
+ mock_remove_device().json.return_value = {"status": "error", "message": "Device non-existent-url not found"}
+ mock_remove_device().raise_for_status.side_effect = HTTPError(
+ "404 Client Error: Not Found for url: http://librenms/devices/non-existent-url",
+ response=mock_remove_device(),
+ )
+
+ yield mock_remove_device
+
+
+def test_get_device_success(mock_get_device_success):
+ client = LibreNMSClient()
+ device = client.get_device("localhost")
+
+ assert device["status"] == "ok"
+ assert device["devices"][0]["hostname"] == "localhost"
+
+
+def test_get_device_not_found(mock_get_device_not_found):
+ client = LibreNMSClient()
+
+ with pytest.raises(HTTPError) as e:
+ client.get_device("non-existent-url")
+
+ assert e.value.response.status_code == HTTPStatus.NOT_FOUND
+ assert e.value.response.json() == {"status": "error", "message": "Device non-existent-url does not exist"}
+ assert e.value.args[0] == "404 Client Error: Not Found for url: http://librenms/devices/non-existent-url"
+
+
+def test_device_exists_true(mock_get_device_success):
+ client = LibreNMSClient()
+
+ assert client.device_exists("localhost")
+
+
+def test_device_exists_false(mock_get_device_not_found):
+ client = LibreNMSClient()
+
+ assert not client.device_exists("non-existent-url")
+
+
+def test_device_exists_bad_request(mock_get_device_unauthenticated):
+ client = LibreNMSClient()
+
+ with pytest.raises(HTTPError) as e:
+ client.device_exists("naughty-url")
+
+ assert e.value.response.status_code == HTTPStatus.UNAUTHORIZED
+ assert e.value.response.json() == {"message": "Unauthenticated."}
+ assert e.value.args[0] == "401 Client Error: Unauthorized for url: http://librenms/devices/naughty-url"
+
+
+def test_add_device_success(mock_add_device_success):
+ fqdn = "localhost"
+ client = LibreNMSClient()
+ new_device = client.add_device(fqdn, SNMPVersion.V2C)
+
+ assert new_device["status"] == "ok"
+ assert new_device["devices"][0]["hostname"] == fqdn
+ assert new_device["devices"][0]["snmpver"] == SNMPVersion.V2C.value
+
+
+def test_add_device_bad_fqdn(mock_add_device_bad_url):
+ fqdn = "non-existent-url"
+ client = LibreNMSClient()
+
+ with pytest.raises(HTTPError) as e:
+ client.add_device(fqdn, SNMPVersion.V2C)
+
+ assert e.value.response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
+ assert e.value.response.json() == {
+ "status": "error",
+ "message": "Could not ping non-existent-url (Hostname did not resolve to IP)",
+ }
+ assert e.value.args[0] == "500 Server Error: Internal server error for url: http://librenms/devices"
+
+
+def test_add_device_no_ping(mock_add_device_unreachable):
+ fqdn = "non-existent-url"
+ client = LibreNMSClient()
+
+ with pytest.raises(HTTPError) as e:
+ client.add_device(fqdn, SNMPVersion.V2C)
+
+ assert e.value.response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
+ assert e.value.response.json() == {
+ "status": "error",
+ "message": "Could not connect to non-existent-url, please check the snmp details and snmp reachability",
+ }
+ assert e.value.args[0] == "500 Server Error: Internal server error for url: http://librenms/devices"
+
+
+def test_remove_device_success(mock_remove_device_success):
+ client = LibreNMSClient()
+ device = client.remove_device("localhost")
+
+ assert device["status"] == "ok"
+ assert device["devices"][0]["hostname"] == "localhost"
+
+
+def test_remove_non_existent_device(mock_remove_device_non_existent):
+ client = LibreNMSClient()
+
+ with pytest.raises(HTTPError) as e:
+ client.remove_device("non-existent-url")
+
+ assert e.value.response.status_code == HTTPStatus.NOT_FOUND
+ assert e.value.response.json() == {"status": "error", "message": "Device non-existent-url not found"}
+ assert e.value.args[0] == "404 Client Error: Not Found for url: http://librenms/devices/non-existent-url"
+
+
+def test_validate_device_success(mock_get_device_success):
+ client = LibreNMSClient()
+ errors = client.validate_device("localhost")
+
+ assert not errors
+
+
+def test_validate_device_non_existing(mock_get_device_not_found):
+ client = LibreNMSClient()
+ errors = client.validate_device("localhost")
+
+ assert len(errors) == 1
+ assert errors[0] == "Device does not exist in LibreNMS."
+
+
+def test_validate_device_misconfigured(mock_get_device_misconfigured):
+ client = LibreNMSClient()
+ errors = client.validate_device("localhost")
+
+ assert len(errors) == 1
+ assert errors[0] == "Device hostname in LibreNMS does not match FQDN."