refactor: Minor

Author: snaselj

nautobot_netbox_importer/diffsync/__init__.py

@@ -1 +1,4 @@
-"""DiffSync adapter and model implementation for nautobot-netbox-importer."""
+"""DiffSync adapter and model implementation for nautobot-netbox-importer.
+
+This folder is an importer implementation specific to NetBox, in opposite to `generator` folder, that is a generic Source => Nautobot importer.
+"""

nautobot_netbox_importer/diffsync/adapters/netbox.py

@@ -94,8 +94,10 @@ def __init__(self, input_ref: _FileRef, options: NetBoxImporterOptions, job=None
     def load(self) -> None:
         """Load data from NetBox."""
         self.import_data()
+
         if self.options.fix_powerfeed_locations:
             fix_power_feed_locations(self)
+
         if self.options.unrack_zero_uheight_devices:
             unrack_zero_uheight_devices(self)
 

nautobot_netbox_importer/diffsync/models/base.py

@@ -1,20 +1,17 @@
 """NetBox to Nautobot Base Models Mapping."""
 
-from diffsync.enum import DiffSyncModelFlags
-
-from nautobot_netbox_importer.base import RecordData
-from nautobot_netbox_importer.generator import DiffSyncBaseModel, SourceAdapter, SourceField, fields
+from nautobot_netbox_importer.generator import SourceAdapter, fields
 
 from .locations import define_locations
 
 
 def setup(adapter: SourceAdapter) -> None:
     """Map NetBox base models to Nautobot."""
-    adapter.disable_model("sessions.session", "Nautobot has own sessions, sessions should never cross apps.")
     adapter.disable_model("admin.logentry", "Not directly used in Nautobot.")
-    adapter.disable_model("users.userconfig", "May not have a 1 to 1 translation to Nautobot.")
     adapter.disable_model("auth.permission", "Handled via a Nautobot model and may not be a 1 to 1.")
-
+    adapter.disable_model("extras.imageattachment", "Images are not imported yet.")
+    adapter.disable_model("sessions.session", "Nautobot has own sessions, sessions should never cross apps.")
+    adapter.disable_model("users.userconfig", "May not have a 1 to 1 translation to Nautobot.")
 
     adapter.configure_model(
         "extras.Status",

nautobot_netbox_importer/diffsync/models/dcim.py

@@ -4,13 +4,7 @@
 from uuid import UUID
 
 from nautobot_netbox_importer.base import RecordData
-from nautobot_netbox_importer.generator import (
-    DiffSyncBaseModel,
-    PreImportRecordResult,
-    SourceAdapter,
-    SourceField,
-    fields,
-)
+from nautobot_netbox_importer.generator import DiffSyncBaseModel, SourceAdapter, SourceField, fields
 
 from .locations import define_location
 
@@ -175,7 +169,7 @@ def fix_power_feed_locations(adapter: SourceAdapter) -> None:
         if not isinstance(location_uid, UUID):
             raise TypeError(f"Location UID must be UUID, got {type(location_uid)}")
 
-        target.location_id = location_uid
+        target.location_id = location_uid  # type: ignore[assignment]
         adapter.update(target)
 
         # Need to update references, to properly update `content_types` fields

tasks.py

@@ -627,12 +627,16 @@ def import_db(context, db_name="", input_file="dump.sql"):
 @task(
     help={
         "db-name": "Database name to backup (default: Nautobot database)",
+        "format": "Database dump format (default: `sql`)",
         "output-file": "Ouput file, overwrite if exists (default: `dump.sql`)",
         "readable": "Flag to dump database data in more readable format (default: `True`)",
     }
 )
-def backup_db(context, db_name="", output_file="dump.sql", readable=True):
+def backup_db(context, db_name="", format="sql", output_file="", readable=True):
     """Dump database into `output_file` file from `db` container."""
+    if not output_file:
+        output_file = f"dump.{format}"
+
     start(context, "db")
     _await_healthy_service(context, "db")
 
@@ -651,6 +655,7 @@ def backup_db(context, db_name="", output_file="dump.sql", readable=True):
             "pg_dump",
             "--username=$POSTGRES_USER",
             f"--dbname={db_name or '$POSTGRES_DB'}",
+            f"--format={format}",
             "--inserts" if readable else "",
         ]
     else:
@@ -1037,8 +1042,9 @@ def validate_app_config(context):
 @task(
     help={
         "file": "URL or path to the JSON file to import.",
-        "bypass-data-validation": "Bypass as much of Nautobot's internal data validation logic as possible, allowing the import of data from NetBox that would be rejected as invalid if entered as-is through the GUI or REST API. USE WITH CAUTION: it is generally more desirable to *take note* of any data validation errors, *correct* the invalid data in NetBox, and *re-import* with the corrected data! (default: False)",
         "demo-version": "Version of the demo data to import from `https://github.com/netbox-community/netbox-demo-data/json` instead of using the `--file` option (default: empty).",
+        "test-input": "Version of the test data to import from `nautobot_netbox_importer/tests/fixtures/nautobot-v<value>` instead of using the `--file` option (default: empty).",
+        "bypass-data-validation": "Bypass as much of Nautobot's internal data validation logic as possible, allowing the import of data from NetBox that would be rejected as invalid if entered as-is through the GUI or REST API. USE WITH CAUTION: it is generally more desirable to *take note* of any data validation errors, *correct* the invalid data in NetBox, and *re-import* with the corrected data! (default: False)",
         "create-missing-cable-terminations": "Create missing cable terminations as Nautobot requires both cable terminations to be defined to save cable instances.",
         "customizations": "Path to a Python module containing customizations to apply during the import. (default: empty)",
         "deduplicate-prefixes": "Deduplicate `ipam.prefix` and `ipam.aggregate` from NetBox. `prefix` value will be unique. (default: False)",
@@ -1048,61 +1054,77 @@ def validate_app_config(context):
         "save-json-summary-path": "File path to write the JSON mapping to. (default: generated-mappings.json)",
         "save-text-summary-path": "File path to write the text mapping to. (default: generated-mappings.txt)",
         "sitegroup-parent-always-region": "When importing `dcim.sitegroup` to `dcim.locationtype`, always set the parent of a site group, to be a `Region` location type. This is a workaround to fix validation errors `'A Location of type Location may only have a Location of the same type as its parent.'`. (default: False)",
-        "update-paths": "Call management command `trace_paths` to update paths after the import. (default: False)",
-        "unrack-zero-uheight-devices": "Cleans the `position` field in `dcim.device` instances with `u_height == 0`. (default: True)",
         "tag-issues": "If specified, tag records with any importer issues. (default: '')",
         "trace-issues": "Show a detailed trace of issues originated from any `Exception` found during the import.",
+        "unrack-zero-uheight-devices": "Cleans the `position` field in `dcim.device` instances with `u_height == 0`. (default: True)",
+        "update-paths": "Call management command `trace_paths` to update paths after the import. (default: False)",
     }
 )
 def import_netbox(  # noqa: PLR0913
     context,
     file="",
     demo_version="",
-    save_json_summary_path="",
-    save_text_summary_path="",
+    test_input="",
     bypass_data_validation=False,
     create_missing_cable_terminations=False,
     customizations="",
     deduplicate_prefixes=False,
     dry_run=True,
     fix_powerfeed_locations=False,
-    sitegroup_parent_always_region=False,
     print_summary=True,
-    update_paths=False,
-    unrack_zero_uheight_devices=True,
+    save_json_summary_path="",
+    save_text_summary_path="",
+    sitegroup_parent_always_region=False,
     tag_issues="",
     trace_issues=False,
+    unrack_zero_uheight_devices=True,
+    update_paths=False,
 ):
     """Import NetBox data into Nautobot."""
-    if demo_version:
-        if file:
-            raise ValueError("Cannot specify both, `file` and `demo` arguments")
+    if sum(bool(x) for x in [file, demo_version, test_input]) > 1:
+        raise ValueError("Cannot specify more than one of `file`, `demo`, or `test_input` arguments")
 
+    if demo_version:
         file = (
             "https://raw.githubusercontent.com/netbox-community/netbox-demo-data/master/json/netbox-demo-v"
             + demo_version
             + ".json"
         )
 
+    if test_input:
+        if is_truthy(context.nautobot_netbox_importer.local):
+            path = Path(__file__).parent
+        else:
+            path = Path('/source')
+
+        path = path / f"nautobot_netbox_importer/tests/fixtures/nautobot-v{test_input}"
+
+        file = path / "input.json"
+
+        if not save_json_summary_path:
+            save_json_summary_path = path / 'summary.json'
+        if not save_text_summary_path:
+            save_text_summary_path = path / 'summary.json'
+
     command = [
         "nautobot-server",
         "import_netbox",
-        f"--save-json-summary-path={save_json_summary_path}" if save_json_summary_path else "",
-        f"--save-text-summary-path={save_text_summary_path}" if save_text_summary_path else "",
         "--bypass-data-validation" if bypass_data_validation else "",
         "--create-missing-cable-terminations" if create_missing_cable_terminations else "",
         f"--customizations={customizations}" if customizations else "",
         "--deduplicate-prefixes" if deduplicate_prefixes else "",
         "--dry-run" if dry_run else "",
         "--fix-powerfeed-locations" if fix_powerfeed_locations else "",
-        "--sitegroup-parent-always-region" if sitegroup_parent_always_region else "",
-        "--print-summary" if print_summary else "",
-        "--update-paths" if update_paths else "",
         "--no-color",
-        "" if unrack_zero_uheight_devices else "--no-unrack-zero-uheight-devices",
+        "--print-summary" if print_summary else "",
+        "--sitegroup-parent-always-region" if sitegroup_parent_always_region else "",
+        f"--save-json-summary-path={save_json_summary_path}" if save_json_summary_path else "",
+        f"--save-text-summary-path={save_text_summary_path}" if save_text_summary_path else "",
         f"--tag-issues={tag_issues}" if tag_issues else "",
         "--trace-issues" if trace_issues else "",
-        file,
+        "" if unrack_zero_uheight_devices else "--no-unrack-zero-uheight-devices",
+        "--update-paths" if update_paths else "",
+        f"{file}",
     ]
 
     run_command(context, " ".join(command))