Skip to content

Commit

Permalink
refactor: ♻️ refactor tests per review suggestions
Browse files Browse the repository at this point in the history
  • Loading branch information
bile0026 committed Jan 11, 2025
1 parent 43e9349 commit 6e0a901
Show file tree
Hide file tree
Showing 3 changed files with 49 additions and 38 deletions.
22 changes: 3 additions & 19 deletions nautobot_ssot/integrations/librenms/diffsync/adapters/librenms.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
from diffsync import DiffSync
from diffsync.exceptions import ObjectNotFound
from django.contrib.contenttypes.models import ContentType
from nautobot.dcim.models import Device, Location, LocationType
from nautobot.dcim.models import Location, LocationType
from nautobot.extras.models import Status

from nautobot_ssot.integrations.librenms.constants import (
Expand All @@ -17,7 +17,6 @@
LibrenmsLocation,
)
from nautobot_ssot.integrations.librenms.utils import (
is_running_tests,
normalize_gps_coordinates,
)
from nautobot_ssot.integrations.librenms.utils.librenms import LibreNMSApi
Expand Down Expand Up @@ -110,12 +109,7 @@ def load(self):
else self.job.hostname_field or "sysName"
)

if is_running_tests():
load_source = "file"
self.job.sync_locations = True
self.hostname_field = "sysName"
else:
load_source = self.job.load_type
load_source = self.job.load_type

if load_source != "file":
all_devices = self.lnms_api.get_librenms_devices()
Expand All @@ -124,8 +118,6 @@ def load(self):

self.job.logger.info(f'Loading {all_devices["count"]} Devices from LibreNMS.')

if is_running_tests():
print(f"All devices fetched: {all_devices}")

for _device in all_devices["devices"]:
self.load_device(device=_device)
Expand All @@ -134,13 +126,7 @@ def load(self):
_site, _created = LocationType.objects.get_or_create(name="Site")
if _created:
_site.content_types.add(ContentType.objects.get(app_label="dcim", model="device"))
if is_running_tests():
_status = Status.objects.get_or_create(name="Active")[0]
_status.content_types.add(ContentType.objects.get_for_model(Device))
_status.content_types.add(ContentType.objects.get_for_model(Location))
_status.validated_save()
else:
_status = Status.objects.get(name="Active")
_status = Status.objects.get(name="Active")
Location.objects.get_or_create(
name="Unknown",
location_type=_site,
Expand All @@ -154,8 +140,6 @@ def load(self):

self.job.logger.info(f'Loading {all_locations["count"]} Locations from LibreNMS.')

if is_running_tests():
print(f"All locations fetched: {all_locations}")

for _location in all_locations["locations"]:
self.load_location(location=_location)
Expand Down
13 changes: 13 additions & 0 deletions nautobot_ssot/tests/librenms/fixtures/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
"""Fixtures for tests."""

import json


def load_json(path):
"""Load a json file."""
with open(path, encoding="utf-8") as file:
return json.loads(file.read())


DEVICE_FIXTURE_RECV = load_json("./nautobot_ssot/tests/librenms/fixtures/get_librenms_devices.json")["devices"]
LOCATION_FIXURE_RECV = load_json("./nautobot_ssot/tests/librenms/fixtures/get_librenms_locations.json")["locations"]
52 changes: 33 additions & 19 deletions nautobot_ssot/tests/librenms/test_librenms_adapter.py
Original file line number Diff line number Diff line change
@@ -1,46 +1,59 @@
"""Unit test for LibreNMS object models."""

import json
from unittest.mock import MagicMock

from django.contrib.contenttypes.models import ContentType
from nautobot.core.testing import TransactionTestCase
from nautobot.extras.models import JobResult
from nautobot.dcim.models import Device, Location
from nautobot.extras.models import JobResult, Status

from nautobot_ssot.integrations.librenms.diffsync.adapters.librenms import LibrenmsAdapter
from nautobot_ssot.integrations.librenms.jobs import LibrenmsDataSource


def load_json(path):
"""Load a JSON file."""
with open(path, encoding="utf-8") as file:
return json.load(file)


DEVICE_FIXTURE = load_json("./nautobot_ssot/tests/librenms/fixtures/get_librenms_devices.json")["devices"]
LOCATION_FIXTURE = load_json("./nautobot_ssot/tests/librenms/fixtures/get_librenms_locations.json")["locations"]
from nautobot_ssot.tests.librenms.fixtures import DEVICE_FIXTURE_RECV, LOCATION_FIXURE_RECV


class TestLibreNMSAdapterTestCase(TransactionTestCase):
"""Test NautobotSsotLibreNMSAdapter class."""

databases = ("default", "job_logs")

def setUp(self):
def __init__(self, *args, **kwargs):
"""Initialize test case."""
super().__init__(*args, **kwargs)

def setUp(self):
"""Setup shared objects for tests."""
# Create Active status first
self.active_status, _ = Status.objects.get_or_create(
name="Active",
defaults={
"color": "4caf50",
}
)
self.active_status.content_types.add(ContentType.objects.get_for_model(Device))
self.active_status.content_types.add(ContentType.objects.get_for_model(Location))

self.librenms_client = MagicMock()
self.librenms_client.name = "Test"
self.librenms_client.remote_url = "https://test.com"
self.librenms_client.verify_ssl = True

# Mock device and location data
self.librenms_client.get_librenms_devices_from_file.return_value = {
"devices": DEVICE_FIXTURE,
"count": len(DEVICE_FIXTURE),
"count": len(DEVICE_FIXTURE_RECV),
"devices": DEVICE_FIXTURE_RECV
}
self.librenms_client.get_librenms_locations_from_file.return_value = {
"locations": LOCATION_FIXTURE,
"count": len(LOCATION_FIXTURE),
"count": len(LOCATION_FIXURE_RECV),
"locations": LOCATION_FIXURE_RECV
}

self.job = LibrenmsDataSource()
self.job.load_type = "file"
self.job.hostname_field = "sysName"
self.job.sync_locations = True
self.job.logger.warning = MagicMock()
self.job.sync_locations = True
self.job.job_result = JobResult.objects.create(
name=self.job.class_path, task_name="fake task", worker="default"
)
Expand All @@ -52,11 +65,12 @@ def test_data_loading(self):

# Debugging outputs
print("Adapter Devices:", list(self.librenms_adapter.get_all("device")))
print("Adapter Locations:", list(self.librenms_adapter.get_all("location")))

expected_locations = {loc["location"].strip() for loc in LOCATION_FIXTURE}
expected_locations = {loc["location"].strip() for loc in LOCATION_FIXURE_RECV}
loaded_locations = {loc.get_unique_id() for loc in self.librenms_adapter.get_all("location")}
self.assertEqual(expected_locations, loaded_locations, "Locations are not loaded correctly.")

expected_devices = {dev["sysName"].strip() for dev in DEVICE_FIXTURE}
expected_devices = {dev["sysName"].strip() for dev in DEVICE_FIXTURE_RECV}
loaded_devices = {dev.get_unique_id() for dev in self.librenms_adapter.get_all("device")}
self.assertEqual(expected_devices, loaded_devices, "Devices are not loaded correctly.")

0 comments on commit 6e0a901

Please sign in to comment.