Fix tests for running OnboardingTask Job

.github/workflows/ci.yml

@@ -106,7 +106,7 @@ jobs:
       fail-fast: true
       matrix:
         python-version: ["3.11"]
-        nautobot-version: ["2.0.3"]
+        nautobot-version: ["2.2.2"]
     env:
       INVOKE_NAUTOBOT_DEVICE_ONBOARDING_PYTHON_VER: "${{ matrix.python-version }}"
       INVOKE_NAUTOBOT_DEVICE_ONBOARDING_NAUTOBOT_VER: "${{ matrix.nautobot-version }}"
@@ -152,7 +152,7 @@ jobs:
         include:
           - python-version: "3.11"
             db-backend: "postgresql"
-            nautobot-version: "2.0.3"
+            nautobot-version: "2.2.2"
           # - python-version: "3.11"
           #   db-backend: "mysql"
           #   nautobot-version: "stable"

changes/176.fixed

@@ -0,0 +1 @@
+Updated tests for OnboardingTask Job.

development/app_config_schema.py

@@ -1,4 +1,5 @@
 """App Config Schema Generator and Validator."""
+
 import json
 from importlib import import_module
 from os import getenv

development/nautobot_config.py

@@ -1,4 +1,5 @@
 """Nautobot development configuration file."""
+
 import os
 import sys
 

nautobot_device_onboarding/__init__.py

@@ -1,4 +1,5 @@
 """App declaration for nautobot_device_onboarding."""
+
 # Metadata is inherited from Nautobot. If not including Nautobot in the environment, this should be added
 from importlib import metadata
 

nautobot_device_onboarding/jobs.py

@@ -1,4 +1,5 @@
 """Device Onboarding Jobs."""
+
 from django.conf import settings
 from nautobot.apps.jobs import Job, ObjectVar, IntegerVar, StringVar, BooleanVar
 from nautobot.core.celery import register_jobs
@@ -113,9 +114,9 @@ def _onboard(self, address):
             password=self.password,
             secret=self.secret,
             napalm_driver=self.platform.napalm_driver if self.platform and self.platform.napalm_driver else None,
-            optional_args=self.platform.napalm_args
-            if self.platform and self.platform.napalm_args
-            else settings.NAPALM_ARGS,
+            optional_args=(
+                self.platform.napalm_args if self.platform and self.platform.napalm_args else settings.NAPALM_ARGS
+            ),
         )
         netdev.get_onboarding_facts()
         netdev_dict = netdev.get_netdev_dict()

nautobot_device_onboarding/tests/test_basic.py

@@ -1,4 +1,5 @@
 """Basic tests that do not require Django."""
+
 import unittest
 import os
 import toml

nautobot_device_onboarding/tests/test_nautobot_keeper.py

@@ -1,4 +1,5 @@
 """Unit tests for nautobot_device_onboarding.onboard module and its classes."""
+
 from django.conf import settings
 from django.contrib.contenttypes.models import ContentType
 from django.test import TestCase

nautobot_device_onboarding/tests/test_onboarding.py

@@ -1,15 +1,15 @@
 """Unit tests for nautobot_device_onboarding.netdev_keeper module and its classes."""
-# from unittest import mock
 
-from django.conf import settings
-
-# from django.test import TestCase
-# from django.contrib.contenttypes.models import ContentType
+from unittest import mock
 
+from nautobot.apps.testing import TransactionTestCase, create_job_result_and_run_job
+from nautobot.extras.models import JobLogEntry, Status
+from nautobot.dcim.models import Device, Location, LocationType, Platform
+from nautobot.ipam.models import Namespace
 
-# from nautobot.dcim.models import Device, Location, LocationType, Platform
-# from nautobot.extras.models import Status
+from django.conf import settings
 
+from django.contrib.contenttypes.models import ContentType
 
 PLUGIN_SETTINGS = settings.PLUGINS_CONFIG["nautobot_device_onboarding"]
 
@@ -23,26 +23,8 @@ def __init__(self, *args, **kwargs):
     def open(self):
         """Mock test open ssh connection."""
 
-
-class NapalmMockNxos(NapalmMock):
-    """Mock napalm for nxos tests."""
-
-    def get_facts(self):
-        """Mock test get napalm facts."""
-        return {
-            "uptime": 4066631,
-            "vendor": "Cisco",
-            "hostname": "nxos-spine1",
-            "fqdn": "nxos-spine1.domain.net",
-            "os_version": "7.3(1)D1(1) [build 7.3(1)D1(0.10)]",
-            "serial_number": "TM6017D760B",
-            "model": "NX-OSv Chassis",
-            "interface_list": ["mgmt0"],
-        }
-
-    def get_interfaces_ip(self):
-        """Mock test get interface ip facts."""
-        return {"mgmnt0": {"ipv4": {"1.1.1.1": {"prefix_length": 32}}}}
+    def close(self):
+        """Mock test close ssh connection."""
 
 
 class NapalmMockEos(NapalmMock):
@@ -63,7 +45,28 @@ def get_facts(self):
 
     def get_interfaces_ip(self):
         """Mock test get napalm interfaces on eos."""
-        return {"Vlan100": {"ipv4": {"2.2.2.2": {"prefix_length": 32}}}}
+        return {"Vlan100": {"ipv4": {"1.1.1.1": {"prefix_length": 32}}}}
+
+
+class NapalmMockNxos(NapalmMock):
+    """Mock napalm for nxos tests."""
+
+    def get_facts(self):
+        """Mock test get napalm facts."""
+        return {
+            "uptime": 4066631,
+            "vendor": "Cisco",
+            "hostname": "nxos-spine1",
+            "fqdn": "nxos-spine1.domain.net",
+            "os_version": "7.3(1)D1(1) [build 7.3(1)D1(0.10)]",
+            "serial_number": "TM6017D760B",
+            "model": "NX-OSv Chassis",
+            "interface_list": ["mgmt0"],
+        }
+
+    def get_interfaces_ip(self):
+        """Mock test get interface ip facts."""
+        return {"mgmnt0": {"ipv4": {"2.2.2.2": {"prefix_length": 32}}}}
 
 
 class SSHDetectMock:  # pylint: disable=too-few-public-methods
@@ -77,59 +80,103 @@ def autodetect(self):
         return self.driver
 
 
-# class OnboardingTestCase(TestCase):
-#     """Test the OnboardingManager Class."""
-
-#     def setUp(self):
-#         """Prepare test objects."""
-#         PLUGIN_SETTINGS["platform_map"] = {}  # Reset platform map to default
-#         status = Status.objects.get(name="Active")
-#         location_type = LocationType.objects.create(name="site")
-#         location_type.content_types.set([ContentType.objects.get_for_model(Device)])
-#         self.site = Location.objects.create(name="TEST_SITE", location_type=location_type, status=status)
-#         self.eos_platform = Platform.objects.create(name="arista_eos", napalm_driver="eos")
-
-#         self.onboarding_task1 = OnboardingTask.objects.create(ip_address="1.1.1.1", location=self.site)
-#         self.onboarding_task2 = OnboardingTask.objects.create(
-#             ip_address="2.2.2.2", location=self.site, platform=self.eos_platform, port=443
-#         )
-
-#         # Patch socket as it would be able to verify connectivity
-#         self.patcher = mock.patch("nautobot_device_onboarding.netdev_keeper.socket")
-#         self.patcher.start()
-
-#     def tearDown(self):
-#         """Disable patch on socket."""
-#         self.patcher.stop()
-
-#     @mock.patch("nautobot_device_onboarding.netdev_keeper.SSHDetect")
-#     @mock.patch("nautobot_device_onboarding.netdev_keeper.get_network_driver")
-#     def test_onboarding_nxos(self, mock_napalm, mock_ssh_detect):
-#         """Test device onboarding nxos."""
-
-#         mock_napalm.return_value = NapalmMockNxos
-#         mock_ssh_detect.return_value = SSHDetectMock("cisco_nxos")
-
-#         # Run onboarding
-#         on_manager = OnboardingManager(self.onboarding_task1, "user", "pass", "secret")
-
-#         self.assertEqual(on_manager.created_device.name, "nxos-spine1")
-#         self.assertEqual(on_manager.created_device.platform.name, "cisco_nxos")
-#         self.assertEqual(on_manager.created_device.platform.napalm_driver, "nxos_ssh")
-#         self.assertEqual(str(on_manager.created_device.primary_ip4), "1.1.1.1/32")
-
-#     @mock.patch("nautobot_device_onboarding.netdev_keeper.SSHDetect")
-#     @mock.patch("nautobot_device_onboarding.netdev_keeper.get_network_driver")
-#     def test_onboarding_eos(self, mock_napalm, mock_ssh_detect):
-#         """Test device onboarding eos."""
-
-#         mock_napalm.return_value = NapalmMockEos
-#         mock_ssh_detect.return_value = SSHDetectMock("arista_eos")
-
-#         # Run onboarding
-#         on_manager = OnboardingManager(self.onboarding_task2, "user", "pass", "secret")
+class OnboardingTestCase(TransactionTestCase):
+    """Test the OnboardingManager Class."""
+
+    def setUp(self):
+        """Prepare test objects."""
+        super().setUp()
+        # Default objects are removed after each test. Recreate Namespace if needed
+        if not Namespace.objects.filter(name="Global").exists():
+            Namespace.objects.create(name="Global")
+
+        PLUGIN_SETTINGS["platform_map"] = {}  # Reset platform map to default
+        status = Status.objects.get(name="Active")
+        location_type = LocationType.objects.create(name="site")
+        location_type.content_types.set([ContentType.objects.get_for_model(Device)])
+        self.site = Location.objects.create(name="TEST_SITE", location_type=location_type, status=status)
+        self.eos_platform = Platform.objects.create(name="arista_eos", napalm_driver="eos")
+
+        self.onboarding_task1 = {
+            "ip_address": "1.1.1.1",
+            "location": self.site.id,
+            "platform": self.eos_platform.id,
+            "port": 443,
+            "credentials": None,
+            "timeout": 10,
+            "device_type": None,
+            "role": None,
+            "continue_on_failure": False,
+        }
+        self.onboarding_task2 = {
+            "ip_address": "2.2.2.2",
+            "location": self.site.id,
+            "platform": None,
+            "port": 22,
+            "credentials": None,
+            "timeout": 10,
+            "device_type": None,
+            "role": None,
+            "continue_on_failure": False,
+        }
 
-#         self.assertEqual(on_manager.created_device.name, "arista-device")
-#         self.assertEqual(on_manager.created_device.platform.name, "arista_eos")
-#         self.assertEqual(on_manager.created_device.platform.napalm_driver, "eos")
-#         self.assertEqual(str(on_manager.created_device.primary_ip4), "2.2.2.2/32")
+        # Patch socket as it would be able to verify connectivity
+        self.patcher = mock.patch("nautobot_device_onboarding.netdev_keeper.socket")
+        self.patcher.start()
+
+    def tearDown(self):
+        """Disable patch on socket."""
+        super().tearDown()
+        self.patcher.stop()
+
+    @mock.patch("nautobot_device_onboarding.netdev_keeper.get_network_driver")
+    def test_onboarding_eos(self, mock_napalm):
+        """Test device onboarding eos."""
+
+        mock_napalm.return_value = NapalmMockEos
+
+        job_result = create_job_result_and_run_job(
+            module="nautobot_device_onboarding.jobs",
+            name="OnboardingTask",
+            **self.onboarding_task1,
+        )
+        # Get Job log messages
+        job_logs = list(JobLogEntry.objects.filter(job_result=job_result))
+        job_logs = [log.message for log in job_logs]
+
+        created_device = Device.objects.get(name="arista-device")
+
+        self.assertEqual(job_result.status, "SUCCESS", (job_result.status, job_logs))
+        self.assertIn("Successfully onboarded arista-device with a management IP of 1.1.1.1", job_logs, job_logs)
+        self.assertEqual(created_device.name, "arista-device")
+        self.assertEqual(created_device.platform.name, "arista_eos")
+        self.assertEqual(created_device.platform.napalm_driver, "eos")
+        self.assertEqual(str(created_device.primary_ip4), "1.1.1.1/32")
+
+    @mock.patch("nautobot_device_onboarding.netdev_keeper.SSHDetect")
+    @mock.patch("nautobot_device_onboarding.netdev_keeper.get_network_driver")
+    def test_onboarding_nxos(self, mock_napalm, mock_ssh_detect):
+        """Test device onboarding nxos."""
+
+        mock_napalm.return_value = NapalmMockNxos
+        mock_ssh_detect.return_value = SSHDetectMock("cisco_nxos")
+
+        # Run onboarding
+        job_result = create_job_result_and_run_job(
+            module="nautobot_device_onboarding.jobs",
+            name="OnboardingTask",
+            **self.onboarding_task2,
+        )
+
+        # Get Job log messages
+        job_logs = list(JobLogEntry.objects.filter(job_result=job_result))
+        job_logs = [log.message for log in job_logs]
+
+        created_device = Device.objects.get(name="nxos-spine1")
+
+        self.assertEqual(job_result.status, "SUCCESS", job_result.status)
+        self.assertIn("Successfully onboarded nxos-spine1 with a management IP of 2.2.2.2", job_logs, job_logs)
+        self.assertEqual(created_device.name, "nxos-spine1")
+        self.assertEqual(created_device.platform.name, "cisco_nxos")
+        self.assertEqual(created_device.platform.napalm_driver, "nxos_ssh")
+        self.assertEqual(str(created_device.primary_ip4), "2.2.2.2/32")