Initiating Nautobot jobs

changes/224.added

@@ -0,0 +1,3 @@
+Add init_job Nautobot subcommand, which starts a Nautobot job by job name.
+Add get_jobs Nautobot subcommand, which gets all Nautobot jobs.
+Add filter_jobs Nautobot subcommand, which gets filtered set of Nautobot jobs.

docs/admin/release_notes/version_2.0.md

@@ -2,6 +2,19 @@
 # v2.0 Release Notes
 
 <!-- towncrier release notes start -->
+## [v2.0.3 (2023-09-22)](https://github.com/nautobot/nautobot-plugin-chatops/releases/tag/v2.0.3)
+
+### Added
+
+- [#227](https://github.com/nautobot/nautobot-plugin-chatops/issues/227) - Added some tests for VLAN chatops.
+
+### Fixed
+
+- [#227](https://github.com/nautobot/nautobot-plugin-chatops/issues/227) - Fixed parameters that should be set to None if they have not been defined yet by default.
+- [#239](https://github.com/nautobot/nautobot-plugin-chatops/issues/239) - Updated IP Fabric Logo.
+- [#241](https://github.com/nautobot/nautobot-plugin-chatops/issues/241) - Remove Grafana Navigation and urls if disabled.
+- [#253](https://github.com/nautobot/nautobot-plugin-chatops/issues/253) - Sorted the ICMP types from IP Fabric diagrams package.
+
 ## [v2.0.2 (2023-08-11)](https://github.com/nautobot/nautobot-plugin-chatops/releases/tag/v2.0.2)
 
 ### Changed

nautobot_chatops/integrations/grafana/api/urls.py

@@ -1,16 +1,16 @@
 """Django urlpatterns declaration for nautobot_chatops.integrations.grafana plugin."""
 
-# from django.urls import path
+from django.conf import settings
 
 from nautobot.core.api import OrderedDefaultRouter
 from nautobot_chatops.integrations.grafana.api.views.generic import NautobotPluginChatopsGrafanaRootView
 
 
 urlpatterns = []
+if settings.PLUGINS_CONFIG["nautobot_chatops"]["enable_grafana"]:
+    router = OrderedDefaultRouter()
+    router.APIRootView = NautobotPluginChatopsGrafanaRootView
 
-router = OrderedDefaultRouter()
-router.APIRootView = NautobotPluginChatopsGrafanaRootView
+    app_name = "nautobot_chatops.grafana-api"
 
-app_name = "nautobot_chatops.grafana-api"
-
-urlpatterns += router.urls
+    urlpatterns += router.urls

nautobot_chatops/integrations/ipfabric/worker.py

@@ -564,7 +564,7 @@ def pathlookup_icmp(dispatcher, src_ip, dst_ip, icmp_type):  # pylint: disable=t
     """Path simulation diagram lookup between source and target IP address."""
     sub_cmd = "pathlookup-icmp"
     icmp_type = icmp_type.upper() if isinstance(icmp_type, str) else icmp_type
-    icmp_types = [(icmp_type_name.upper(), icmp_type_name) for icmp_type_name in icmp.__all__]
+    icmp_types = sorted([(icmp_type_name.upper(), icmp_type_name) for icmp_type_name in icmp.__all__])
 
     # identical to dialog_list in end-to-end-path; consolidate dialog_list if maintaining both cmds
     dialog_list = [

nautobot_chatops/navigation.py

@@ -1,9 +1,13 @@
 """Plugin additions to the Nautobot navigation menu."""
 
+from django.conf import settings
 from nautobot.extras.plugins import PluginMenuItem, PluginMenuButton
 from nautobot.utilities.choices import ButtonColorChoices
 
-from .integrations.grafana.navigation import menu_items as grafana_menu_items
+if settings.PLUGINS_CONFIG["nautobot_chatops"]["enable_grafana"]:
+    from .integrations.grafana.navigation import menu_items as grafana_menu_items
+else:
+    grafana_menu_items = ()
 
 menu_items = (
     PluginMenuItem(

nautobot_chatops/tests/workers/test_nautobot.py

@@ -0,0 +1,56 @@
+"""Tests for the /nautobot chatops commands."""
+from unittest.mock import MagicMock
+
+from django.test import TestCase
+from nautobot.dcim.models import Site
+from nautobot.ipam.models import VLAN
+from nautobot.extras.models import Status
+from nautobot_chatops.choices import CommandStatusChoices
+
+from nautobot_chatops.dispatchers import Dispatcher
+from nautobot_chatops.workers.nautobot import get_vlans
+
+
+class IpamTestCase(TestCase):
+    """Tests related to IPAM ChatOps commands."""
+
+    def setUp(self):
+        """Per-test-case setup function."""
+        self.active_status = Status.objects.get(name="Active")
+        self.site = Site.objects.create(name="site-1", status=self.active_status)
+        self.vlans, _ = VLAN.objects.get_or_create(vid=1, name="vlan-1", status=self.active_status, site=self.site)
+
+        # Mock the dispatcher
+        self.dispatcher = MagicMock(Dispatcher)
+
+    def test_get_vlans_initial_prompt(self):
+        """Test get VLANs initial command."""
+        self.assertFalse(get_vlans(self.dispatcher))
+        self.dispatcher.send_error.assert_not_called()
+        self.dispatcher.prompt_from_menu.assert_called_with(
+            "nautobot get-vlans",
+            "select a vlan filter",
+            [
+                ("VLAN ID", "id"),
+                ("Group", "group"),
+                ("Name", "name"),
+                ("Role", "role"),
+                ("Site", "site"),
+                ("Status", "status"),
+                ("Tenant", "tenant"),
+                ("All (no filter)", "all"),
+            ],
+        )
+
+    def test_get_vlans_filter_type_sent_filter_name(self):
+        """Test get VLANs with filter type Name selected."""
+        self.assertFalse(get_vlans(self.dispatcher, "name"))
+        self.dispatcher.send_error.assert_not_called()
+        self.dispatcher.prompt_from_menu.assert_called_with(
+            "nautobot get-vlans name", "select a vlan name", [("vlan-1", "vlan-1")], offset=0
+        )
+
+    def test_get_vlans_filter_type_sent_filter_all(self):
+        """Test get VLANs with filter type All selected."""
+        self.assertEqual(get_vlans(self.dispatcher, "all"), CommandStatusChoices.STATUS_SUCCEEDED)
+        self.dispatcher.send_error.assert_not_called()

nautobot_chatops/urls.py

@@ -1,6 +1,7 @@
 """Django urlpatterns declaration for nautobot_chatops plugin."""
 import logging
 
+from django.conf import settings
 from django.urls import path
 
 from nautobot.extras.views import ObjectChangeLogView
@@ -18,15 +19,19 @@
     AccessGrantBulkDeleteView,
 )
 
-try:
-    from nautobot_chatops.integrations.grafana.urls import urlpatterns as grafana_urlpatterns
-# pylint: disable-next=broad-except
-except Exception:
+if settings.PLUGINS_CONFIG["nautobot_chatops"]["enable_grafana"]:
+    try:
+        from nautobot_chatops.integrations.grafana.urls import urlpatterns as grafana_urlpatterns
+    # pylint: disable-next=broad-except
+    except Exception:
+        grafana_urlpatterns = []
+        logger = logging.getLogger(__name__)
+        logger.warning("Grafana ChatOps integration is not available.", exc_info=True)
+else:
     grafana_urlpatterns = []
     logger = logging.getLogger(__name__)
     logger.warning("Grafana ChatOps integration is not available.", exc_info=True)
 
-
 urlpatterns = [
     path("", NautobotHomeView.as_view(), name="home"),
     path("access/", AccessGrantListView.as_view(), name="accessgrant_list"),

nautobot_chatops/workers/nautobot.py

@@ -1,5 +1,9 @@
 """Worker functions for interacting with Nautobot."""
 
+import uuid
+
+
+from django.contrib.auth import get_user_model
 from django.core.exceptions import ValidationError
 from django.db.models import Count
 from django.contrib.contenttypes.models import ContentType
@@ -11,7 +15,10 @@
 from nautobot.dcim.models import Device, Site, DeviceRole, DeviceType, Manufacturer, Rack, Region, Cable
 from nautobot.ipam.models import VLAN, Prefix, VLANGroup, Role
 from nautobot.tenancy.models import Tenant
-from nautobot.extras.models import Status
+from nautobot.extras.context_managers import web_request_context
+from nautobot.extras.jobs import run_job
+from nautobot.extras.models import Job, JobResult, Status
+from nautobot.extras.utils import get_job_content_type
 
 from nautobot_chatops.choices import CommandStatusChoices
 from nautobot_chatops.workers import subcommand_of, handle_subcommands
@@ -160,7 +167,7 @@ def examine_termination_endpoints(circuit):
 
 # pylint: disable=too-many-statements
 @subcommand_of("nautobot")
-def get_vlans(dispatcher, filter_type, filter_value_1):
+def get_vlans(dispatcher, filter_type=None, filter_value_1=None):
     """Return a filtered list of VLANs based on filter type and/or `filter_value_1`."""
     # pylint: disable=no-else-return
     if not filter_type:
@@ -1016,6 +1023,103 @@ def get_circuit_providers(dispatcher, *args):
     return CommandStatusChoices.STATUS_SUCCEEDED
 
 
+@subcommand_of("nautobot")
+def filter_jobs(
+    dispatcher, job_filters: str = ""
+):  # We can use a Literal["enabled", "installed", "runnable"] here instead
+    """Get a filtered list of jobs from Nautobot.
+
+    Args:
+        job_filters (str): Filter job results by literals in a comma-separated string.
+                           Available filters are: enabled, installed or runnable.
+    """
+    # Check for filters in user supplied input
+    job_filters_list = [item.strip() for item in job_filters.split(",")] if isinstance(job_filters, str) else ""
+    filters = ["enabled", "installed", "runnable"]
+    if any([key in job_filters for key in filters]):
+        filter_args = {key: True for key in filters if key in job_filters_list}
+        jobs = Job.objects.restrict(dispatch.user, "view").filter(
+            **filter_args
+        )  # enabled=True, installed=True, runnable=True
+    else:
+        jobs = Job.objects.restrict(dispatch.user, "view").all()
+
+    header = ["Name", "ID"]
+    rows = [
+        (
+            str(job.name),
+            str(job.id),
+        )
+        for job in jobs
+    ]
+
+    dispatcher.send_large_table(header, rows)
+
+    return CommandStatusChoices.STATUS_SUCCEEDED
+
+
+@subcommand_of("nautobot")
+def get_jobs(dispatcher):
+    """Get all jobs from Nautobot."""
+    jobs = Job.objects.restrict(dispatch.user, "view").all()
+
+    header = ["Name", "ID"]
+    rows = [
+        (
+            str(job.name),
+            str(job.id),
+        )
+        for job in jobs
+    ]
+
+    dispatcher.send_large_table(header, rows)
+
+    return CommandStatusChoices.STATUS_SUCCEEDED
+
+
+@subcommand_of("nautobot")
+def init_job(dispatcher, job_name):
+    """Initiate a job in Nautobot by job name."""
+    # Get instance of the user who will run the job
+    user = get_user_model()
+    try:
+        user_instance = user.objects.get(username=dispatch.user)
+    except user.DoesNotExist:  # Unsure if we need to check this case?
+        dispatcher.send_error(f"User {dispatch.user} not found")
+        return (CommandStatusChoices.STATUS_FAILED, f'User "{dispatch.user}" not found')
+
+    # Get the job model instance using job name
+    try:
+        job_model = Job.objects.restrict(dispatch.user, "view").get(name=job_name)
+    except Job.DoesNotExist:
+        dispatcher.send_error(f"Job {job_name} not found")
+        return (CommandStatusChoices.STATUS_FAILED, f'Job "{job_name}" not found')
+
+    job_class_path = job_model.class_path
+
+    # Create an instance of job result
+    job_result = JobResult.objects.create(
+        name=job_model.class_path,
+        job_kwargs={"data": {}, "commit": True, "profile": False},
+        obj_type=get_job_content_type(),
+        user=user_instance,
+        job_model=job_model,
+        job_id=uuid.uuid4(),
+    )
+
+    # Emulate HTTP context for the request as the user
+    with web_request_context(user=user_instance) as request:
+        run_job(data={}, request=request, commit=True, job_result_pk=job_result.pk)
+
+    blocks = [
+        dispatcher.markdown_block(f"The requested job {job_class_path} was initiated!"),
+    ]
+
+    dispatcher.send_blocks(blocks)
+
+    return CommandStatusChoices.STATUS_SUCCEEDED
+
+
 @subcommand_of("nautobot")
 def about(dispatcher, *args):
     """Provide link for more information on Nautobot Apps."""

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "nautobot-chatops"
-version = "2.0.2"
+version = "2.0.3"
 description = "A plugin providing chatbot capabilities for Nautobot"
 authors = ["Network to Code, LLC <[email protected]>"]
 readme = "README.md"
@@ -44,7 +44,7 @@ ipfabric-diagrams = { version = "~6.0.2", optional = true }
 isodate =  { version = "^0.6.1", optional = true }
 meraki = { version = "^1.7.2", optional = true }
 nautobot = "^1.5.4"
-nautobot-capacity-metrics = "*"
+nautobot-capacity-metrics = "^2.0.0"
 netmiko = { version = "^3.4.0", optional = true }
 netutils = { version = "^1.1.0", optional = true }
 pan-os-python = { version = "^1.3.0", optional = true }