From ec8bfb27e248ac15729622d8cfbe0e0de32ac6df Mon Sep 17 00:00:00 2001 From: Justin Drew <2396364+jdrew82@users.noreply.github.com> Date: Wed, 21 Aug 2024 10:53:55 -0500 Subject: [PATCH] Migrate IPFabric SSoT Command (#318) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: ✨ Replicate IPFabric SSoT Job command from SSoT project. * docs: 📝 Add changelog fragment --------- Co-authored-by: Stephen Kiely --- changes/317.added | 1 + .../integrations/ipfabric/worker.py | 115 ++++++++++++++++++ 2 files changed, 116 insertions(+) create mode 100644 changes/317.added diff --git a/changes/317.added b/changes/317.added new file mode 100644 index 00000000..648cca4c --- /dev/null +++ b/changes/317.added @@ -0,0 +1 @@ +Migrated IPFabric SSoT Job command from SSoT project. \ No newline at end of file diff --git a/nautobot_chatops/integrations/ipfabric/worker.py b/nautobot_chatops/integrations/ipfabric/worker.py index 59f3645a..9897d692 100644 --- a/nautobot_chatops/integrations/ipfabric/worker.py +++ b/nautobot_chatops/integrations/ipfabric/worker.py @@ -3,15 +3,20 @@ import logging import os import tempfile +import uuid from datetime import datetime from django.conf import settings +from django.contrib.contenttypes.models import ContentType from ipfabric_diagrams import Unicast, icmp +from nautobot.core.settings_funcs import is_truthy +from nautobot.extras.models import JobResult from netutils.ip import is_ip from netutils.mac import is_valid_mac from pkg_resources import parse_version from nautobot_chatops.choices import CommandStatusChoices +from nautobot_chatops.dispatchers import Dispatcher from nautobot_chatops.workers import handle_subcommands, subcommand_of from .context import get_context, set_context @@ -25,6 +30,14 @@ logger = logging.getLogger("nautobot") +try: + from nautobot_ssot.integrations.ipfabric.jobs import IpFabricDataSource + + IPFABRIC_SSOT_JOB_FOUND = True +except ImportError: + logger.error(msg="Unable to find IPFabric SSoT Job.") + IPFABRIC_SSOT_JOB_FOUND = False + inventory_field_mapping = { "site": "siteName", "model": "model", @@ -94,6 +107,13 @@ def prompt_find_host_filter_keys(action_id, help_text, dispatcher, choices=None) return False +def prompt_for_bool(dispatcher: Dispatcher, action_id: str, help_text: str): + """Prompt the user to select a True or False choice.""" + choices = [("Yes", "True"), ("No", "False")] + dispatcher.prompt_from_menu(action_id, help_text, choices, default=("Yes", "True")) + return False + + def ipf_api_client(): """Get the IP Fabric API Client handle. @@ -1068,3 +1088,98 @@ def table_diff(dispatcher, category, table, view, snapshot): # pylint: disable= else: dispatcher.send_markdown(f"{key.title()}: None") return CommandStatusChoices.STATUS_SUCCEEDED + + +@subcommand_of("ipfabric") +def ssot_sync_to_nautobot( + dispatcher, + dry_run=None, + safe_delete_mode=None, + sync_ipfabric_tagged_only=None, +): + """Start an SSoT sync from IPFabric to Nautobot.""" + if not IPFABRIC_SSOT_JOB_FOUND: + dispatcher.send_error( + "Unable to find IPFabric SSoT Job installed. Please confirm the IPFabric SSoT integration is installed and enabled." + ) + return CommandStatusChoices.STATUS_FAILED + + if dry_run is None: + prompt_for_bool(dispatcher, f"{BASE_CMD} ssot-sync-to-nautobot", "Do you want to run a `Dry Run`?") + return (CommandStatusChoices.STATUS_SUCCEEDED, "Success") + + if safe_delete_mode is None: + prompt_for_bool( + dispatcher, f"{BASE_CMD} ssot-sync-to-nautobot {dry_run}", "Do you want to run in `Safe Delete Mode`?" + ) + return (CommandStatusChoices.STATUS_SUCCEEDED, "Success") + + if sync_ipfabric_tagged_only is None: + prompt_for_bool( + dispatcher, + f"{BASE_CMD} ssot-sync-to-nautobot {dry_run} {safe_delete_mode}", + "Do you want to sync against `ssot-tagged-from-ipfabric` tagged objects only?", + ) + return (CommandStatusChoices.STATUS_SUCCEEDED, "Success") + + # if location_filter is None: + # prompt_for_site( + # dispatcher, + # f"{BASE_CMD} ssot-sync-to-nautobot {dry_run} {safe_delete_mode} {sync_ipfabric_tagged_only}", + # "Select a Site to use as an optional filter?", + # ) + # return (CommandStatusChoices.STATUS_SUCCEEDED, "Success") + + # Implement filter in future release + location_filter = False + + sync_job = IpFabricDataSource() + + sync_job.job_result = JobResult( + name=sync_job.class_path, + obj_type=ContentType.objects.get( + app_label="extras", + model="job", + ), + job_id=uuid.uuid4(), + ) + sync_job.job_result.validated_save() + + dispatcher.send_markdown( + f"Stand by {dispatcher.user_mention()}, I'm running your sync with options set to `Dry Run`: {dry_run}, `Safe Delete Mode`: {safe_delete_mode}. `Sync Tagged Only`: {sync_ipfabric_tagged_only}", + ephemeral=True, + ) + + sync_job.run( + dryrun=is_truthy(dry_run), + memory_profiling=False, + safe_delete_mode=is_truthy(safe_delete_mode), + sync_ipfabric_tagged_only=is_truthy(sync_ipfabric_tagged_only), + location_filter=location_filter, + debug=False, + ) + sync_job.job_result.validated_save() + + blocks = [ + *dispatcher.command_response_header( + "ipfabric", + "ssot-sync-to-nautobot", + [ + ("Dry Run", str(dry_run)), + ("Safe Delete Mode", str(safe_delete_mode)), + ("Sync IPFabric Tagged Only", str(sync_ipfabric_tagged_only)), + ], + "sync job", + ipfabric_logo(dispatcher), + ), + ] + dispatcher.send_blocks(blocks) + if sync_job.job_result.status == "completed": + dispatcher.send_markdown( + f"Sync completed succesfully. Here is the link to your job: {os.getenv('NAUTOBOT_HOST')}{sync_job.sync.get_absolute_url()}." + ) + else: + dispatcher.send_warning( + f"Sync failed. Here is the link to your job: {os.getenv('NAUTOBOT_HOST')}{sync_job.sync.get_absolute_url()}" + ) + return CommandStatusChoices.STATUS_SUCCEEDED