chore: update CI workflows, add pytest-timeout, and switch to ThreadPoolExecutor

.github/dependabot.yml

@@ -1,13 +1,11 @@
 ---
 version: 2
 updates:
-  - package-ecosystem: pip
-    directory: /
+  - package-ecosystem: "pip"
+    directory: "/"
     schedule:
-      interval: monthly
-      time: '02:00'
-      timezone: America/New_York
-    labels:
-      - dependencies
-    assignees:
-        - "danner26"
+      interval: "weekly"
+  - package-ecosystem: "github-actions"
+    directory: "/"
+    schedule:
+      interval: "weekly"

.github/workflows/docker.yml

@@ -1,6 +1,10 @@
 ---
 name: ci
 
+permissions:
+  contents: read
+  packages: write
+
 on:
   push:
     branches:
@@ -21,13 +25,16 @@ jobs:
       -
         name: Checkout
         uses: actions/checkout@v4
+      -
+        name: Set lowercase repo name
+        run: echo "REPO_LOWER=$(echo '${{ github.event.repository.name }}' | tr '[:upper:]' '[:lower:]')" >> $GITHUB_ENV
       -
         name: Docker meta
         id: meta
         uses: docker/metadata-action@v5
         with:
           images: |
-            ghcr.io/minitriga/Netbox-Device-Type-Library-Import
+            ghcr.io/${{ github.repository_owner }}/${{ env.REPO_LOWER }}
           tags: |
             type=raw,value=latest,enable=${{ endsWith(github.ref, github.event.repository.default_branch) }}
             type=ref,event=branch
@@ -57,4 +64,4 @@ jobs:
           push: ${{ github.event_name != 'pull_request' }}
           platforms: linux/amd64
           tags: ${{ steps.meta.outputs.tags }}
-          labels: ${{ steps.meta.outputs.labels }}
+          labels: ${{ steps.meta.outputs.labels }}

Dockerfile

@@ -27,5 +27,8 @@ ENV PATH="/app/.venv/bin:$PATH"
 # Copy source code
 COPY *.py ./
 
+RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app
+USER appuser
+
 # Set the command
 CMD ["uv", "run", "nb-dt-import.py"]

env.example

@@ -0,0 +1,17 @@
+# NetBox instance URL (without trailing slash)
+NETBOX_URL=https://netbox.example.org
+
+# NetBox API token with write permissions
+# Should have permissions: dcim.add_manufacturer, dcim.add_devicetype, dcim.add_moduletype
+NETBOX_TOKEN=XXXXXXXXX-xxxxxxxxx
+
+# Device Type Library repository
+REPO_URL=https://github.com/netbox-community/devicetype-library.git
+
+# Branch of devicetype-library to use (default: master)
+REPO_BRANCH=master
+
+# Optional: Disable SSL certificate verification (INSECURE - development/testing only)
+# WARNING: Setting this to True bypasses certificate validation and exposes connections
+# to man-in-the-middle attacks. Never enable in production environments.
+# IGNORE_SSL_ERRORS=True

log_handler.py

@@ -3,18 +3,41 @@
 
 
 class LogHandler:
-    def __new__(cls, *args, **kwargs):
-        return super().__new__(cls)
+    """
+    Handles logging and exception reporting for the device type import process.
+
+    Provides timestamped logging methods, verbose mode support, and formatted
+    error messages that terminate the program on critical failures.
+    """
 
     def __init__(self, args):
+        """
+        Initialize the LogHandler with parsed arguments or a configuration object.
+
+        Parameters:
+            args: Parsed command-line arguments or a configuration object with at least a `verbose` attribute (bool);
+                  stored on the instance as `self.args`.
+        """
         self.args = args
 
     def exception(self, exception_type, exception, stack_trace=None):
+        """
+        Handle an error by formatting a user-facing message and terminating the program.
+
+        Parameters:
+            exception_type (str): Key identifying the error category (expected keys include "EnvironmentError", "SSLError", "GitCommandError", "GitInvalidRepositoryError", "InvalidGitURL", "Exception").
+            exception (str): Value used to populate the chosen error message (e.g., environment variable name, repo name, or raw error text).
+            stack_trace (str | None): Optional stack trace or additional context. If provided and the instance was constructed with verbose enabled, the stack trace is printed.
+
+        Raises:
+            SystemExit: Exits the process with a formatted message corresponding to `exception_type`.
+        """
         exception_dict = {
             "EnvironmentError": f'Environment variable "{exception}" is not set.',
             "SSLError": f"SSL verification failed. IGNORE_SSL_ERRORS is {exception}. Set IGNORE_SSL_ERRORS to True if you want to ignore this error. EXITING.",
             "GitCommandError": f'The repo "{exception}" is not a valid git repo.',
             "GitInvalidRepositoryError": f'The repo "{exception}" is not a valid git repo.',
+            "InvalidGitURL": f"Invalid Git URL: {exception}. URL must use HTTPS or SSH protocol.",
             "Exception": f'An unknown error occurred: "{exception}"',
         }
 

nb-dt-import.py

@@ -20,6 +20,13 @@ def get_progress_wrapper(iterable, desc=None, **kwargs):
 
 
 def main():
+    """
+    Orchestrate importing device- and module-types from a Git repository into NetBox.
+
+    Parses CLI arguments, validates environment variables, clones/pulls the DTL repo,
+    parses YAML files, and creates manufacturers, device types, and module types in NetBox.
+    Reports progress and summary counters.
+    """
     startTime = datetime.now()
 
     parser = ArgumentParser(description="Import Netbox Device Types")
@@ -53,7 +60,7 @@ def main():
             handle.exception(
                 "EnvironmentError",
                 var,
-                f'Environment variable "{var}" is not set.\n\nMANDATORY_ENV_VARS: {str(settings.MANDATORY_ENV_VARS)}.\n\nCURRENT_ENV_VARS: {str(os.environ)}',
+                f'Environment variable "{var}" is not set.\n\nMANDATORY_ENV_VARS: {str(settings.MANDATORY_ENV_VARS)}\n',
             )
 
     dtl_repo = DTLRepo(args, settings.REPO_PATH, handle)

netbox_api.py

@@ -541,6 +541,17 @@ def create_rear_ports(self, rear_ports, device_type, context=None):
         )
 
     def create_front_ports(self, front_ports, device_type, context=None):
+        """
+        Create front port templates for a device type, resolving rear-port references before creation.
+
+        For each front-port entry, attempts to resolve its `rear_port` name to the corresponding rear-port ID; front ports whose `rear_port` cannot be resolved are removed and a log entry is emitted (including the optional context). After resolving and pruning entries, the function creates the remaining front-port templates in NetBox and records created items via the shared counters/handlers.
+
+        Parameters:
+            front_ports (list[dict]): List of front-port template definitions. Each item is expected to include a "name" and a "rear_port" (the rear-port name to resolve).
+            device_type (int): ID of the device type (device_type) to which the front ports belong.
+            context (str | None): Optional context string appended to log messages for disambiguation.
+        """
+
         def link_rear_ports(items, pid):
             # Use cached rear ports if available, otherwise fetch from API
             cache_key = ("device", pid)
@@ -569,6 +580,13 @@ def link_rear_ports(items, pid):
             for port in ports_to_remove:
                 items.remove(port)
 
+            if ports_to_remove:
+                skipped_names = [p["name"] for p in ports_to_remove]
+                ctx = f" (Context: {context})" if context else ""
+                self.handle.log(
+                    f"Skipped {len(ports_to_remove)} front port(s) with invalid rear port refs: {skipped_names}{ctx}"
+                )
+
         self._create_generic(
             front_ports,
             device_type,
@@ -674,6 +692,15 @@ def create_module_console_server_ports(self, console_server_ports, module_type,
         )
 
     def create_module_rear_ports(self, rear_ports, module_type, context=None):
+        """
+        Create rear-port templates for a module type in NetBox.
+
+        Adds any rear port templates from `rear_ports` that do not already exist for the specified `module_type`.
+        Parameters:
+            rear_ports (list[dict]): List of rear-port template definitions to create; each item must include a `name` and any other template fields required by NetBox.
+            module_type (int|object): The module type identifier or object used to associate created templates with the parent module type.
+            context (str, optional): Optional context string used for logging to identify the source of these templates.
+        """
         self._create_generic(
             rear_ports,
             module_type,
@@ -685,8 +712,28 @@ def create_module_rear_ports(self, rear_ports, module_type, context=None):
         )
 
     def create_module_front_ports(self, front_ports, module_type, context=None):
+        """
+        Create front-port templates for a module-type and link them to their rear ports.
+
+        Creates any missing module front-port templates under the given module_type. If a front port references a rear port by name, the rear port name is resolved to the rear-port ID; front ports with unresolved rear-port names are removed from creation and a log message is emitted (includes `context` if provided).
+
+        Parameters:
+            front_ports (list[dict]): List of front-port template definitions. Each dict must include at least "name"; items may reference a rear port by the "rear_port" key (name).
+            module_type (int | object): Module type identifier or object to associate the created front ports with.
+            context (str | None): Optional context string appended to log messages for easier debugging.
+        """
+
         def link_rear_ports(items, pid):
             # Use cached rear ports if available, otherwise fetch from API
+            """
+            Resolve each front-port's `rear_port` name to the corresponding rear-port ID for a module and remove any front-ports whose `rear_port` cannot be resolved.
+
+            This function mutates `items` in place: for each port dict it replaces the string `rear_port` value with the matching rear-port `.id` when found, logs a message for each missing rear-port (including the available rear-port names), and removes ports with unresolved `rear_port` references. After processing it logs a summary of how many module front ports were skipped. The logs may include the outer-scope `context` value if present.
+
+            Parameters:
+                items (list[dict]): List of front-port dictionaries; each must contain at least `"name"` and `"rear_port"` (the latter initially a name string).
+                pid (int): The module type ID used to scope the rear-port lookup.
+            """
             cache_key = ("module", pid)
             if "rear_port_templates" in self.cached_components:
                 existing_rp = self.cached_components["rear_port_templates"].get(cache_key, {})
@@ -713,6 +760,13 @@ def link_rear_ports(items, pid):
             for port in ports_to_remove:
                 items.remove(port)
 
+            if ports_to_remove:
+                skipped_names = [p["name"] for p in ports_to_remove]
+                ctx = f" (Context: {context})" if context else ""
+                self.handle.log(
+                    f"Skipped {len(ports_to_remove)} module front port(s) with invalid rear port refs: {skipped_names}{ctx}"
+                )
+
         self._create_generic(
             front_ports,
             module_type,
@@ -725,22 +779,34 @@ def link_rear_ports(items, pid):
         )
 
     def upload_images(self, baseurl, token, images, device_type):
-        """Upload front_image and/or rear_image for the given device type
+        """
+        Upload front and/or rear image files to the specified NetBox device type.
 
-        Args:
-        baseurl: URL for Netbox instance
-        token: Token to access Netbox instance
-        images: map of front_image and/or rear_image filename
-        device_type: id for the device-type to update
+        Sends a PATCH request to the device-type endpoint attaching the provided image files, increments self.counter["images"] by the number of files sent, and ensures all opened file handles are closed. Respects self.ignore_ssl to determine SSL verification behavior.
 
-        Returns:
-        None
+        Parameters:
+            baseurl (str): Base URL of the NetBox instance (e.g. "https://netbox.example.com").
+            token (str): API token used for the Authorization header.
+            images (dict): Mapping of form field name to local file path (e.g. {"front_image": "/path/front.jpg", "rear_image": "/path/rear.jpg"}).
+            device_type (int | str): Identifier of the device type to update in NetBox (used in the endpoint URL).
         """
         url = f"{baseurl}/api/dcim/device-types/{device_type}/"
         headers = {"Authorization": f"Token {token}"}
 
-        files = {i: (os.path.basename(f), open(f, "rb")) for i, f in images.items()}
-        response = requests.patch(url, headers=headers, files=files, verify=(not self.ignore_ssl))
-
-        self.handle.log(f"Images {images} updated at {url}: {response}")
-        self.counter["images"] += len(images)
+        # Open files with proper cleanup to avoid resource leaks
+        file_handles = {}
+        try:
+            for field, path in images.items():
+                file_handles[field] = (os.path.basename(path), open(path, "rb"))
+            response = requests.patch(
+                url, headers=headers, files=file_handles, verify=(not self.ignore_ssl), timeout=60
+            )
+            response.raise_for_status()
+            self.handle.log(f"Images {images} updated at {url}: {response.status_code}")
+            self.counter["images"] += len(images)
+        finally:
+            for _, (_, fh) in file_handles.items():
+                try:
+                    fh.close()
+                except Exception:
+                    pass

pyproject.toml

@@ -18,6 +18,7 @@ dev = [
     "pre-commit>=4.5.0",
     "pytest>=9.0.2",
     "pytest-mock>=3.15.1",
+    "pytest-timeout>=2.4.0",
     "ruff>=0.14.9",
 ]
 
@@ -26,3 +27,7 @@ line-length = 120
 
 [tool.ruff]
 line-length = 120
+
+[tool.pytest.ini_options]
+timeout = 20
+timeout_method = "thread"