skip onboarding

Author: mzbroch

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/models.py

@@ -14,8 +14,8 @@
 from django.db.models.signals import post_save
 from django.dispatch import receiver
 from django.db import models
-from dcim.models import Device
 from django.urls import reverse
+from dcim.models import Device
 from .choices import OnboardingStatusChoices, OnboardingFailChoices
 from .release import NETBOX_RELEASE_CURRENT, NETBOX_RELEASE_29
 
@@ -113,3 +113,12 @@ def last_ot(self):
         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/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