Expose onboarding details in device view

netbox_onboarding/choices.py

@@ -22,12 +22,14 @@ class OnboardingStatusChoices(ChoiceSet):
     STATUS_PENDING = "pending"
     STATUS_RUNNING = "running"
     STATUS_SUCCEEDED = "succeeded"
+    STATUS_SKIPPED = "skipped"
 
     CHOICES = (
         (STATUS_FAILED, "failed"),
         (STATUS_PENDING, "pending"),
         (STATUS_RUNNING, "running"),
         (STATUS_SUCCEEDED, "succeeded"),
+        (STATUS_SKIPPED, "skipped"),
     )
 
 

netbox_onboarding/migrations/0002_onboardingdevice.py

@@ -0,0 +1,22 @@
+# Generated by Django 2.2.10 on 2020-08-21 11:05
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("netbox_onboarding", "0001_initial"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="OnboardingDevice",
+            fields=[
+                ("id", models.AutoField(auto_created=True, primary_key=True, serialize=False)),
+                ("enabled", models.BooleanField(default=True)),
+                ("device", models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, to="dcim.Device")),
+            ],
+        ),
+    ]

netbox_onboarding/models.py

@@ -11,8 +11,11 @@
 See the License for the specific language governing permissions and
 limitations under the License.
 """
+from django.db.models.signals import post_save
+from django.dispatch import receiver
 from django.db import models
 from django.urls import reverse
+from dcim.models import Device
 from .choices import OnboardingStatusChoices, OnboardingFailChoices
 from .release import NETBOX_RELEASE_CURRENT, NETBOX_RELEASE_29
 
@@ -64,3 +67,58 @@ def get_absolute_url(self):
         from utilities.querysets import RestrictedQuerySet  # pylint: disable=no-name-in-module, import-outside-toplevel
 
         objects = RestrictedQuerySet.as_manager()
+
+
+class OnboardingDevice(models.Model):
+    """The status of each Onboarded Device is tracked in the OnboardingDevice table."""
+
+    device = models.OneToOneField(to="dcim.Device", on_delete=models.CASCADE)
+    enabled = models.BooleanField(default=True, help_text="Whether (re)onboarding of this device is permitted")
+
+    @property
+    def last_check_attempt_date(self):
+        """Date of last onboarding attempt for a device."""
+        try:
+            return OnboardingTask.objects.filter(created_device=self.device).latest("created_on").created_on
+        except ValueError:
+            return "unknown"
+
+    @property
+    def last_check_successful_date(self):
+        """Date of last successful onboarding for a device."""
+        try:
+            return (
+                OnboardingTask.objects.filter(
+                    created_device=self.device, status=OnboardingStatusChoices.STATUS_SUCCEEDED
+                )
+                .latest("created_on")
+                .created_on
+            )
+        except ValueError:
+            return "unknown"
+
+    @property
+    def status(self):
+        """Last onboarding status."""
+        try:
+            return OnboardingTask.objects.filter(created_device=self.device).latest("created_on").status
+        except ValueError:
+            return "unknown"
+
+    @property
+    def last_ot(self):
+        """Last onboarding task."""
+        try:
+            return OnboardingTask.objects.filter(created_device=self.device).latest("created_on")
+        except ValueError:
+            return None
+
+
+@receiver(post_save, sender=Device)
+def init_onboarding_for_new_device(sender, instance, created, **kwargs):  # pylint: disable=unused-argument
+    """Register to create a OnboardingDevice object for each new Device Object using Django Signal.
+
+    https://docs.djangoproject.com/en/3.0/ref/signals/#post-save
+    """
+    if created:
+        OnboardingDevice.objects.create(device=instance)

netbox_onboarding/template_content.py

@@ -0,0 +1,47 @@
+"""Onboarding template content.
+
+(c) 2020 Network To Code
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+  http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+from extras.plugins import PluginTemplateExtension
+from .models import OnboardingDevice
+
+
+class DeviceContent(PluginTemplateExtension):  # pylint: disable=abstract-method
+    """Table to show onboarding details on Device objects."""
+
+    model = "dcim.device"
+
+    def right_page(self):
+        """Show table on right side of view."""
+        onboarding = OnboardingDevice.objects.filter(device=self.context["object"]).first()
+
+        if not onboarding.enabled:
+            return None
+
+        status = onboarding.status
+        last_check_attempt_date = onboarding.last_check_attempt_date
+        last_check_successful_date = onboarding.last_check_successful_date
+        last_ot = onboarding.last_ot
+
+        return self.render(
+            "netbox_onboarding/device_onboarding_table.html",
+            extra_context={
+                "status": status,
+                "last_check_attempt_date": last_check_attempt_date,
+                "last_check_successful_date": last_check_successful_date,
+                "last_ot": last_ot,
+            },
+        )
+
+
+template_extensions = [DeviceContent]

netbox_onboarding/templates/netbox_onboarding/device_onboarding_table.html

@@ -0,0 +1,31 @@
+{% block content %}
+<div class="panel panel-default">
+    <div class="panel-heading">
+        <strong>Device Onboarding</strong>
+    </div>
+    <table class="table table-hover panel-body">
+        <tbody>
+            <tr>
+                <th>Date</th>
+                <th>Status</th>
+                <th>Date of last success</th>
+                <th>Latest Task</th>
+            </tr>
+            <tr>
+                <td>
+                    {{ last_check_attempt_date }}
+                </td>
+                <td>
+                    {{ status }}
+                </td>
+                <td>
+                    {{ last_check_successful_date }}
+                </td>
+                <td>
+                    {{ last_ot.pk }}
+                </td>
+            </tr>
+        </tbody>
+    </table>
+</div>
+{% endblock %}

netbox_onboarding/tests/test_models.py

@@ -0,0 +1,88 @@
+"""Unit tests for netbox_onboarding OnboardingDevice model.
+
+(c) 2020 Network To Code
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+  http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+from django.test import TestCase
+
+from dcim.models import Site, DeviceRole, DeviceType, Manufacturer, Device
+
+from netbox_onboarding.models import OnboardingTask
+from netbox_onboarding.models import OnboardingDevice
+from netbox_onboarding.choices import OnboardingStatusChoices
+
+
+class OnboardingDeviceModelTestCase(TestCase):
+    """Test the Onboarding models."""
+
+    def setUp(self):
+        """Setup objects for Onboarding Model tests."""
+        self.site = Site.objects.create(name="USWEST", slug="uswest")
+        manufacturer = Manufacturer.objects.create(name="Juniper", slug="juniper")
+        device_role = DeviceRole.objects.create(name="Firewall", slug="firewall")
+        device_type = DeviceType.objects.create(slug="srx3600", model="SRX3600", manufacturer=manufacturer)
+
+        self.device = Device.objects.create(
+            device_type=device_type, name="device1", device_role=device_role, site=self.site
+        )
+
+        self.succeeded_task1 = OnboardingTask.objects.create(
+            ip_address="10.10.10.10",
+            site=self.site,
+            status=OnboardingStatusChoices.STATUS_SUCCEEDED,
+            created_device=self.device,
+        )
+
+        self.succeeded_task2 = OnboardingTask.objects.create(
+            ip_address="10.10.10.10",
+            site=self.site,
+            status=OnboardingStatusChoices.STATUS_SUCCEEDED,
+            created_device=self.device,
+        )
+
+        self.failed_task1 = OnboardingTask.objects.create(
+            ip_address="10.10.10.10",
+            site=self.site,
+            status=OnboardingStatusChoices.STATUS_FAILED,
+            created_device=self.device,
+        )
+
+        self.failed_task2 = OnboardingTask.objects.create(
+            ip_address="10.10.10.10",
+            site=self.site,
+            status=OnboardingStatusChoices.STATUS_FAILED,
+            created_device=self.device,
+        )
+
+    def test_onboardingdevice_autocreated(self):
+        """Verify that OnboardingDevice is auto-created."""
+        onboarding_device = OnboardingDevice.objects.get(device=self.device)
+        self.assertEqual(self.device, onboarding_device.device)
+
+    def test_last_check_attempt_date(self):
+        """Verify OnboardingDevice last attempt."""
+        onboarding_device = OnboardingDevice.objects.get(device=self.device)
+        self.assertEqual(onboarding_device.last_check_attempt_date, self.failed_task2.created_on)
+
+    def test_last_check_successful_date(self):
+        """Verify OnboardingDevice last success."""
+        onboarding_device = OnboardingDevice.objects.get(device=self.device)
+        self.assertEqual(onboarding_device.last_check_successful_date, self.succeeded_task2.created_on)
+
+    def test_status(self):
+        """Verify OnboardingDevice status."""
+        onboarding_device = OnboardingDevice.objects.get(device=self.device)
+        self.assertEqual(onboarding_device.status, self.failed_task2.status)
+
+    def test_last_ot(self):
+        """Verify OnboardingDevice last ot."""
+        onboarding_device = OnboardingDevice.objects.get(device=self.device)
+        self.assertEqual(onboarding_device.last_ot, self.failed_task2)

netbox_onboarding/worker.py

@@ -21,6 +21,7 @@
 from .choices import OnboardingFailChoices
 from .choices import OnboardingStatusChoices
 from .exceptions import OnboardException
+from .models import OnboardingDevice
 from .models import OnboardingTask
 from .onboard import OnboardingManager
 
@@ -29,7 +30,7 @@
 
 
 @job("default")
-def onboard_device(task_id, credentials):  # pylint: disable=R0915
+def onboard_device(task_id, credentials):  # pylint: disable=too-many-statements
     """Process a single OnboardingTask instance."""
     username = credentials.username
     password = credentials.password
@@ -44,6 +45,12 @@ def onboard_device(task_id, credentials):  # pylint: disable=R0915
         try:
             if ot.ip_address:
                 onboarded_device = Device.objects.get(primary_ip4__address__net_host=ot.ip_address)
+
+            if OnboardingDevice.objects.filter(device=onboarded_device, enabled=False):
+                ot.status = OnboardingStatusChoices.STATUS_SKIPPED
+
+                return dict(ok=True)
+
         except Device.DoesNotExist as exc:
             logger.info("Getting device with IP lookup failed: %s", str(exc))
         except Device.MultipleObjectsReturned as exc:
@@ -80,7 +87,7 @@ def onboard_device(task_id, credentials):  # pylint: disable=R0915
         ot.save()
         onboarding_status = False
 
-    except Exception as exc:  # pylint: disable=W0703
+    except Exception as exc:  # pylint: disable=broad-except
         if onboarded_device:
             ot.created_device = onboarded_device