Skip to content
Snippets Groups Projects
Verified Commit 7fbf4dae authored by Karel van Klink's avatar Karel van Klink :smiley_cat:
Browse files

add unit tests to the LibreNMS client, improve the validation method

parent 00950520
No related branches found
No related tags found
1 merge request!126Add iBGP workflow and LibreNMS client
Hide whitespace changes
Inline Side-by-side
gso/services/librenms_client.py
+ 9
− 6
View file @ 7fbf4dae
......@@ -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
test/conftest.py
+ 1
− 1
View file @ 7fbf4dae
......@@ -176,7 +176,7 @@ def configuration_data() -> dict:
},
"MONITORING": {
"LIBRENMS": {
"base_url": "http://fake.url.local",
"base_url": "http://librenms",
"token": "secret-token",
},
"SNMP": {
......
test/services/test_librenms_client.py 0 → 100644
+ 458
− 0
View file @ 7fbf4dae
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."
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment