From f5d2133246a1aa855f4a53604addcddb5729260d Mon Sep 17 00:00:00 2001 From: Marcin Zieba Date: Wed, 9 Sep 2026 14:52:59 +0200 Subject: [PATCH 1/4] style: apply the ruff autofix sweep Machine-applied only, no behaviour change: PEP 585/604 builtin generics (List -> list, Optional[X] -> X | None), import ordering, redundant intermediate assignments, composite test assertions split into one assert each, and unused tuple elements renamed to _. The rules that produce these land in the follow-up commit, so this one stays a pure diff to skim and the config arrives with the changes that need judgment. Excluded UP042 (str, Enum -> StrEnum) from the sweep: StrEnum.__str__ returns the bare value where (str, Enum) returns ClassName.MEMBER, which would change logged and serialized output. --- core/change_detector.py | 32 +++---- core/component_registry.py | 3 +- core/export.py | 38 ++++---- core/graphql_client.py | 1 + core/import_run.py | 5 +- core/log_handler.py | 4 +- core/nb_serializer.py | 3 +- core/netbox_api.py | 42 +++++---- core/outcomes.py | 29 +++---- core/repo.py | 19 ++-- core/update_failure_resolver.py | 15 ++-- nb-dt-import.py | 4 +- tests/conftest.py | 3 +- tests/integration/test_import.py | 4 +- tests/test_change_detector.py | 16 ++-- tests/test_component_cache.py | 4 +- tests/test_docker_workflow.py | 1 - tests/test_errors.py | 6 +- tests/test_export_manifest.py | 3 +- tests/test_exporter.py | 8 +- tests/test_graphql_client.py | 39 +++++---- tests/test_import_run.py | 7 +- tests/test_log_handler.py | 4 +- tests/test_module_bay_type_sync.py | 5 +- tests/test_module_bay_types.py | 17 ++-- tests/test_nb_dt_import.py | 42 ++++----- tests/test_netbox_api.py | 120 ++++++++++++++++---------- tests/test_normalization.py | 16 ++-- tests/test_repo.py | 63 +++++++------- tests/test_update_failure_resolver.py | 3 +- 30 files changed, 295 insertions(+), 261 deletions(-) diff --git a/core/change_detector.py b/core/change_detector.py index 5e2ecf787..06700f12d 100644 --- a/core/change_detector.py +++ b/core/change_detector.py @@ -6,13 +6,13 @@ import os from dataclasses import dataclass, field -from functools import lru_cache -from typing import Any, List, Optional from enum import Enum +from functools import lru_cache +from typing import Any from core.component_registry import BY_YAML_KEY, COMPONENT_TYPES -from core.normalization import is_explicit_list, normalize_values from core.formatting import log_property_diffs +from core.normalization import is_explicit_list, normalize_values from core.schema_reader import load_properties_for_type @@ -150,7 +150,7 @@ class ComponentChange: component_type: str # e.g., "interfaces", "power-ports" component_name: str change_type: ChangeType - property_changes: List[PropertyChange] = field(default_factory=list) + property_changes: list[PropertyChange] = field(default_factory=list) @dataclass @@ -161,9 +161,9 @@ class DeviceTypeChange: model: str slug: str is_new: bool = False - property_changes: List[PropertyChange] = field(default_factory=list) - component_changes: List[ComponentChange] = field(default_factory=list) - netbox_id: Optional[int] = None + property_changes: list[PropertyChange] = field(default_factory=list) + component_changes: list[ComponentChange] = field(default_factory=list) + netbox_id: int | None = None @property def has_changes(self) -> bool: @@ -180,8 +180,8 @@ def has_updates(self) -> bool: class ChangeReport: """Aggregated change report for all device types.""" - new_device_types: List[DeviceTypeChange] = field(default_factory=list) - modified_device_types: List[DeviceTypeChange] = field(default_factory=list) + new_device_types: list[DeviceTypeChange] = field(default_factory=list) + modified_device_types: list[DeviceTypeChange] = field(default_factory=list) unchanged_count: int = 0 @@ -257,7 +257,7 @@ def __init__(self, device_types_instance, handle, remove_unmanaged_types: bool = self.verbose = verbose self.remove_unmanaged_types = remove_unmanaged_types - def detect_changes(self, device_types: List[dict], progress=None) -> ChangeReport: + def detect_changes(self, device_types: list[dict], progress=None) -> ChangeReport: """Analyze all device types and generate a change report. Args: @@ -308,7 +308,7 @@ def detect_changes(self, device_types: List[dict], progress=None) -> ChangeRepor return report - def _compare_device_type_properties(self, yaml_data: dict, netbox_dt) -> List[PropertyChange]: + def _compare_device_type_properties(self, yaml_data: dict, netbox_dt) -> list[PropertyChange]: """Compare YAML device type properties against NetBox device type. Args: @@ -349,7 +349,7 @@ def _compare_device_type_properties(self, yaml_data: dict, netbox_dt) -> List[Pr return changes @staticmethod - def _compare_image_properties(yaml_data: dict, netbox_dt) -> List[PropertyChange]: + def _compare_image_properties(yaml_data: dict, netbox_dt) -> list[PropertyChange]: """Compare image properties between YAML and NetBox device type. YAML uses boolean flags (front_image: true) meaning "an image should exist", @@ -388,7 +388,7 @@ def _compare_components( yaml_data: dict, device_type_id: int, parent_type: str = "device", - ) -> List[ComponentChange]: + ) -> list[ComponentChange]: """Compare all components between YAML and cached NetBox data. Args: @@ -481,10 +481,10 @@ def _compare_component_properties( self, yaml_comp: dict, netbox_comp, - properties: List[str], + properties: list[str], comp_type: str = "", manufacturer: str = "", - ) -> List[PropertyChange]: + ) -> list[PropertyChange]: """Compare properties between YAML and NetBox component. *manufacturer* is the owning manufacturer's slug, used to resolve a relation @@ -631,7 +631,7 @@ def _log_modified_summary(self, report: ChangeReport) -> None: if parts: self.handle.log(f" Breakdown: {', '.join(parts)}") - def _log_property_diffs(self, prop_changes: List[PropertyChange], indent: str) -> None: + def _log_property_diffs(self, prop_changes: list[PropertyChange], indent: str) -> None: """Emit diff-u style lines for *prop_changes* at the given *indent*.""" log_property_diffs( [(pc.property_name, pc.old_value, pc.new_value) for pc in prop_changes], diff --git a/core/component_registry.py b/core/component_registry.py index 8fb4e474a..10609fb69 100644 --- a/core/component_registry.py +++ b/core/component_registry.py @@ -10,7 +10,6 @@ """ from dataclasses import dataclass, field -from typing import Optional # What a create call must resolve from a name to a NetBox id before it can POST. LINK_BRIDGE = "bridge" @@ -41,7 +40,7 @@ class ComponentType: relations: tuple[str, ...] = field(default_factory=tuple) graphql_extra: tuple[str, ...] = field(default_factory=tuple) compare_extra: tuple[str, ...] = field(default_factory=tuple) - link: Optional[str] = None + link: str | None = None @property def graphql_fields(self): diff --git a/core/export.py b/core/export.py index 9dd485074..9c3dd853c 100644 --- a/core/export.py +++ b/core/export.py @@ -7,9 +7,10 @@ import os import re import threading +from collections.abc import Sequence from dataclasses import dataclass from pathlib import Path -from typing import Any, List, Optional, Sequence +from typing import Any import requests import yaml @@ -31,6 +32,7 @@ ) from core.netbox_api import IMAGE_EXTENSIONS, _build_auth_header from core.repo import LIBRARY_TYPE_DIRS, library_dirs_present +import contextlib class _SkipSentinel: @@ -108,7 +110,7 @@ class ExportItem: kind: str # "device-type" | "module-type" | "rack-type" nb_record: Any - repo_yaml: Optional[dict] # None when absent from repo + repo_yaml: dict | None # None when absent from repo serialized: dict # What we will write reason: str # "absent" | "differs" | "images-missing" mfr_name: str @@ -248,7 +250,7 @@ def _is_subset(sub: Any, sup: Any) -> bool: class Exporter: """Exports NetBox device/module/rack types to a local directory in DTL format.""" - def __init__(self, config, handle, export_dir: str, force_overwrite: bool, vendor_slugs: Optional[Sequence[str]]): + def __init__(self, config, handle, export_dir: str, force_overwrite: bool, vendor_slugs: Sequence[str] | None): """Initialize the Exporter from the resolved run configuration.""" self.config = config self.handle = handle @@ -270,7 +272,7 @@ def __init__(self, config, handle, export_dir: str, force_overwrite: bool, vendo handle=handle, page_size=config.graphql_page_size, ) - self._module_image_details: Optional[dict] = None + self._module_image_details: dict | None = None def _get_module_image_details(self) -> dict: """Return module type image details, fetching from NetBox at most once per run.""" @@ -297,7 +299,7 @@ def run(self, progress=None) -> None: self.graphql.detect_module_bay_type_support() # ── Fetch all types from NetBox ────────────────────────────────────── - by_model, by_slug = self.graphql.get_device_types(manufacturer_slugs=self.vendor_slugs) + by_model, _by_slug = self.graphql.get_device_types(manufacturer_slugs=self.vendor_slugs) all_mt = self.graphql.get_module_types(manufacturer_slugs=self.vendor_slugs) all_rt = self.graphql.get_rack_types(manufacturer_slugs=self.vendor_slugs) @@ -371,9 +373,9 @@ def _compare_vendors_to_items( repo_dt_by_slug, repo_mt_by_key, progress, - ) -> tuple[List[ExportItem], int]: + ) -> tuple[list[ExportItem], int]: """Compare stale device/module types per vendor and return export items.""" - items: List[ExportItem] = [] + items: list[ExportItem] = [] skipped_fresh = 0 compare_task = ( progress.add_task("Comparing vendors", total=len(all_vendor_slugs)) @@ -441,9 +443,9 @@ def _compare_vendors_to_items( return items, skipped_fresh - def _compare_racks_to_items(self, all_rt, manifest, repo_rt_by_key, progress) -> tuple[List[ExportItem], int]: + def _compare_racks_to_items(self, all_rt, manifest, repo_rt_by_key, progress) -> tuple[list[ExportItem], int]: """Compare stale rack types and return export items.""" - items: List[ExportItem] = [] + items: list[ExportItem] = [] skipped_fresh = 0 rack_records = [record for models in all_rt.values() for record in models.values()] rack_task = ( @@ -698,10 +700,8 @@ def _fetch_one(endpoint_name): results = list(pool.map(_fetch_one, COMPONENT_ENDPOINT_NAMES)) finally: for client in _clients: - try: + with contextlib.suppress(Exception): client.close() - except Exception: - pass for endpoint_name, records in results: for rec in records: @@ -717,7 +717,7 @@ def _fetch_one(endpoint_name): def _determine_export_set_for_device_types( self, nb_records: list, repo_dt_by_slug: dict, components_by_dt_id: dict - ) -> List[ExportItem]: + ) -> list[ExportItem]: """Build the list of device types that need exporting to the repo. Includes records absent from the repo, those whose serialized form @@ -733,7 +733,7 @@ def _determine_export_set_for_device_types( manifest_key = f"{mfr_name}/{rec.slug}" repo_yaml = repo_dt_by_slug.get((mfr_slug, rec.slug)) - reason: Optional[str] + reason: str | None if repo_yaml is None: reason = "absent" elif _repo_supersedes(repo_yaml, serialized): @@ -759,7 +759,7 @@ def _determine_export_set_for_device_types( def _determine_export_set_for_module_types( self, nb_records: list, repo_mt_by_key: dict, components_by_mt_id: dict - ) -> List[ExportItem]: + ) -> list[ExportItem]: """Build the list of module types that need exporting to the repo. Includes records absent from the repo and those whose serialized form @@ -796,7 +796,7 @@ def _determine_export_set_for_module_types( ) return items - def _determine_export_set_for_rack_types(self, nb_records: list, repo_rt_by_key: dict) -> List[ExportItem]: + def _determine_export_set_for_rack_types(self, nb_records: list, repo_rt_by_key: dict) -> list[ExportItem]: """Build the list of rack types that need exporting to the repo. Includes records absent from the repo and those whose serialized form @@ -833,7 +833,7 @@ def _determine_export_set_for_rack_types(self, nb_records: list, repo_rt_by_key: ) return items - def _check_missing_images(self, front_url, rear_url, mfr_name: str, slug: str) -> Optional[str]: + def _check_missing_images(self, front_url, rear_url, mfr_name: str, slug: str) -> str | None: """Return ``'images-missing'`` if any expected local image is absent; else None. DTL stores images under ``elevation-images//.{front,rear}.{png,jpg,jpeg,gif}`` @@ -874,7 +874,7 @@ def _download_type_images(self, item: ExportItem) -> bool: """Download images for *item*. Returns True if all downloads succeeded.""" if item.kind == "device-type": return self._download_device_type_images(item) - elif item.kind == "module-type": + if item.kind == "module-type": return self._download_module_type_images(item) return True # rack types have no images @@ -975,7 +975,7 @@ def _download_module_type_images(self, item: ExportItem) -> bool: return ok def _download_image( - self, url_path: str, dest: Path, content_type_out: "Optional[list]" = None + self, url_path: str, dest: Path, content_type_out: "list | None" = None ) -> "str | _SkipSentinel | None": """Download an image from NetBox and write to *dest*. diff --git a/core/graphql_client.py b/core/graphql_client.py index a188ea6ef..98b71d3cd 100644 --- a/core/graphql_client.py +++ b/core/graphql_client.py @@ -299,6 +299,7 @@ def query(self, graphql_query, variables=None, _retries=3): raise GraphQLSchemaError(messages) return body.get("data", {}) + return None def query_all(self, graphql_query, list_key, page_size=None, variables=None, on_page=None): """Auto-paginate a GraphQL list query using offset/limit. diff --git a/core/import_run.py b/core/import_run.py index cdac353b0..6bc35f528 100644 --- a/core/import_run.py +++ b/core/import_run.py @@ -1,19 +1,18 @@ """Import pipeline planning and execution.""" +import os from collections import Counter from contextlib import contextmanager from dataclasses import dataclass from datetime import datetime, timedelta -import os from typing import Any -from core.change_detector import ChangeDetector, ChangeType, IMAGE_PROPERTIES +from core.change_detector import IMAGE_PROPERTIES, ChangeDetector, ChangeType from core.component_cache import NullTaskDisplay, RichTaskDisplay from core.config import RunConfig from core.errors import VendorSelectionError from core.outcomes import EntityKind, Outcome - _PROGRESS_DESC_WIDTH = 28 diff --git a/core/log_handler.py b/core/log_handler.py index 2a53eab90..6c543abac 100644 --- a/core/log_handler.py +++ b/core/log_handler.py @@ -74,6 +74,6 @@ def log_ports_created(self, created_ports, parent_type: str, port_type: str = "p for port in created_ports: self.verbose_log( f"{port_type} Template Created: {port.name} - " - + f"{port.type if hasattr(port, 'type') else ''} - {getattr(port, parent_attribute).id} - " - + f"{port.id}" + f"{port.type if hasattr(port, 'type') else ''} - {getattr(port, parent_attribute).id} - " + f"{port.id}" ) diff --git a/core/nb_serializer.py b/core/nb_serializer.py index 4977553b4..0e10446b7 100644 --- a/core/nb_serializer.py +++ b/core/nb_serializer.py @@ -4,7 +4,8 @@ comparison against existing repo YAML files. """ -from typing import Any, Sequence +from collections.abc import Sequence +from typing import Any from core.component_registry import BY_ENDPOINT, COMPONENT_TYPES, MODULE_TYPE_RELATIONS diff --git a/core/netbox_api.py b/core/netbox_api.py index 3321d45d5..237359088 100644 --- a/core/netbox_api.py +++ b/core/netbox_api.py @@ -1,21 +1,23 @@ """NetBox REST and GraphQL API client for importing device and module type libraries.""" -from collections import Counter -from dataclasses import replace -from contextlib import contextmanager -from functools import lru_cache +import glob import hashlib import json +import os import tempfile import time +from collections import Counter +from contextlib import contextmanager, suppress +from dataclasses import replace +from functools import lru_cache +from pathlib import Path +from typing import Any + import pynetbox import requests -import os -import glob -from pathlib import Path -from typing import Any, Optional from core.change_detector import ChangeDetector, ChangeType +from core.compat import parse_netbox_version, supports_module_bay_types from core.component_cache import ComponentCache from core.component_registry import ( BY_YAML_KEY, @@ -26,9 +28,8 @@ MODULE_TYPE_COMPONENTS, MODULE_TYPE_RELATIONS, ) -from core.compat import parse_netbox_version, supports_module_bay_types -from core.formatting import log_property_diffs from core.errors import FatalError, UnknownError +from core.formatting import log_property_diffs from core.graphql_client import GraphQLError, NetBoxGraphQLClient from core.normalization import values_equal from core.outcomes import EntityKind, Outcome, OutcomeRegistry @@ -247,7 +248,7 @@ def _is_image_hash_changed(local_path: str, hash_cache: dict, log_fn=None) -> bo return current != cached -def _load_image_hash_cache(path: Optional[str], log_fn=None) -> dict: +def _load_image_hash_cache(path: str | None, log_fn=None) -> dict: """Load the image-hash cache from *path* (JSON), returning an empty dict when it cannot be read. An absent file is the normal first run and stays quiet. Anything else means the @@ -298,10 +299,8 @@ def _save_image_hash_cache(path: str, cache: dict) -> bool: return True except OSError: if tmp_path is not None: - try: + with suppress(OSError): os.unlink(tmp_path) - except OSError: - pass return False @@ -360,6 +359,7 @@ def _retry_on_connection_error(func, *args, **kwargs): raise wait = _RETRY_BACKOFF[attempt] if attempt < len(_RETRY_BACKOFF) else _RETRY_BACKOFF[-1] time.sleep(wait) + return None # Module type scalar properties that can be compared and updated. @@ -542,7 +542,7 @@ def __init__(self, config, handle): _cache_dir = Path(os.environ.get("XDG_CACHE_HOME", Path.home() / ".cache")) / "nb-dt-import" try: _cache_dir.mkdir(parents=True, exist_ok=True) - self._image_hash_cache_path: Optional[str] = str(_cache_dir / "image-hashes.json") + self._image_hash_cache_path: str | None = str(_cache_dir / "image-hashes.json") except OSError as exc: self.handle.log( "[yellow]Warning: could not create image hash cache directory " @@ -1900,7 +1900,7 @@ def _process_single_module_type( if module_type_res is not None: self.handle.verbose_log( f"Module Type Cached: {module_type_res.manufacturer.name} - " - + f"{module_type_res.model} - {module_type_res.id}" + f"{module_type_res.model} - {module_type_res.id}" ) # Upload images before the scalar PATCH so attachments are created # even if the property update later fails (module already exists in @@ -1940,7 +1940,7 @@ def _process_single_module_type( all_module_types.setdefault(manufacturer_slug, {})[curr_mt["model"]] = module_type_res self.handle.verbose_log( f"Module Type Created: {module_type_res.manufacturer.name} - " - + f"{module_type_res.model} - {module_type_res.id}" + f"{module_type_res.model} - {module_type_res.id}" ) except pynetbox.RequestError as excep: self.handle.log(f"Error creating Module Type: {excep.error} (Context: {src_file})") @@ -2283,7 +2283,7 @@ class _FrontPortRecordWithMappings: All other attribute accesses are forwarded to the underlying record. """ - __slots__ = ("_record", "_mappings_canonical", "_mappings_m2m") + __slots__ = ("_mappings_canonical", "_mappings_m2m", "_record") def __init__(self, record): """Wrap *record* and pre-compute a canonical mappings list for ChangeDetector compatibility. @@ -2294,7 +2294,7 @@ def __init__(self, record): """ object.__setattr__(self, "_record", record) mappings_raw = getattr(record, "mappings", None) - canonical: Optional[list] + canonical: list | None if mappings_raw is not None: # NetBox >= 4.5: mappings is a list of PortTemplateMapping objects canonical = [] @@ -3173,10 +3173,8 @@ def upload_images(self, baseurl, token, images, device_type): self.handle.log(f"Error reading image file for device type {device_type}: {e}") finally: for _, (_, fh) in file_handles.items(): - try: + with suppress(Exception): fh.close() - except Exception: - pass def upload_image_attachment(self, baseurl, token, image_path, object_type, object_id): """Upload an image as an Image Attachment to a NetBox object. diff --git a/core/outcomes.py b/core/outcomes.py index cf0b75e80..a9a9c7983 100644 --- a/core/outcomes.py +++ b/core/outcomes.py @@ -16,7 +16,6 @@ from dataclasses import dataclass, field from enum import Enum -from typing import Dict, List, Optional class EntityKind(str, Enum): @@ -48,9 +47,9 @@ class OutcomeRecord: kind: EntityKind identity: str # human-readable identifier (e.g. "Supermicro/SuperServer-6028TR-HTR") outcome: Outcome - reason: Optional[str] = None - blocking_objects: List[str] = field(default_factory=list) - hint: Optional[str] = None + reason: str | None = None + blocking_objects: list[str] = field(default_factory=list) + hint: str | None = None class OutcomeRegistry: @@ -62,7 +61,7 @@ class OutcomeRegistry: def __init__(self) -> None: """Initialise an empty registry.""" - self._records: List[OutcomeRecord] = [] + self._records: list[OutcomeRecord] = [] def record( self, @@ -70,9 +69,9 @@ def record( identity: str, outcome: Outcome, *, - reason: Optional[str] = None, - blocking_objects: Optional[List[str]] = None, - hint: Optional[str] = None, + reason: str | None = None, + blocking_objects: list[str] | None = None, + hint: str | None = None, ) -> None: """Append a new outcome record.""" self._records.append( @@ -87,27 +86,27 @@ def record( ) @property - def records(self) -> List[OutcomeRecord]: + def records(self) -> list[OutcomeRecord]: """All recorded outcomes (read-only view).""" return list(self._records) - def failures(self) -> List[OutcomeRecord]: + def failures(self) -> list[OutcomeRecord]: """Return only the FAILED records.""" return [r for r in self._records if r.outcome == Outcome.FAILED] - def partials(self) -> List[OutcomeRecord]: + def partials(self) -> list[OutcomeRecord]: """Return only the PARTIAL records (some but not all changes applied).""" return [r for r in self._records if r.outcome == Outcome.PARTIAL] - def summary_by_kind(self) -> Dict[EntityKind, Dict[Outcome, int]]: + def summary_by_kind(self) -> dict[EntityKind, dict[Outcome, int]]: """Aggregate counts grouped by ``(kind, outcome)``.""" - agg: Dict[EntityKind, Dict[Outcome, int]] = {} + agg: dict[EntityKind, dict[Outcome, int]] = {} for r in self._records: agg.setdefault(r.kind, {}).setdefault(r.outcome, 0) agg[r.kind][r.outcome] += 1 return agg - def render_failure_report(self) -> List[str]: + def render_failure_report(self) -> list[str]: """Render a multi-line operator-facing failure report. Returns an empty list when no failures or partials were recorded. @@ -118,7 +117,7 @@ def render_failure_report(self) -> List[str]: if not failures and not partials: return [] - lines: List[str] = [] + lines: list[str] = [] lines.append("=" * 60) lines.append("FAILED / PARTIAL UPDATE REPORT") lines.append("=" * 60) diff --git a/core/repo.py b/core/repo.py index 62472f7ac..11c47ceb8 100644 --- a/core/repo.py +++ b/core/repo.py @@ -4,12 +4,13 @@ import json import os import pickle +from collections.abc import Sequence from glob import glob from re import sub as re_sub -from typing import Optional, Sequence from urllib.parse import urlparse -from git import Repo, exc + import yaml +from git import Repo, exc from core.config import LOCAL_REPO_URL, is_local_repo_url from core.errors import FatalError, UnknownError @@ -79,7 +80,7 @@ def find_class(self, module, name): _INDEX_MAX_BYTES = 10 * 1024 * 1024 # 10 MiB — DTL index files are typically <1 MiB -def _resolve_index_path(base_dir: str, stem: str) -> "Optional[str]": +def _resolve_index_path(base_dir: str, stem: str) -> "str | None": """Return the path to a DTL index file, preferring the JSON form over the legacy pickle. Upstream DTL replaced ``tests/known-*.pickle`` with ``tests/known-*.json`` (GHSA-492p-5wp7-2w7c). @@ -96,8 +97,8 @@ def _resolve_index_path(base_dir: str, stem: str) -> "Optional[str]": def _vendor_slugs_from_index( - index_path: "Optional[str]", slugs_lower: list, slug_format, subdir_filter: "Optional[str]" = None -) -> "Optional[set]": + index_path: "str | None", slugs_lower: list, slug_format, subdir_filter: "str | None" = None +) -> "set | None": """Load a (model_name, vendor_dir) index and return the set of vendor slugs matching *slugs_lower*. *index_path* may point at a ``.json`` (current) or ``.pickle`` (legacy) file; the loader is @@ -122,7 +123,7 @@ def _vendor_slugs_from_index( return result -def _safe_abs_path(repo_root: str, relpath: str) -> "Optional[str]": +def _safe_abs_path(repo_root: str, relpath: str) -> "str | None": """Return the absolute path for *relpath* inside *repo_root*, or None if it escapes the root.""" abs_path = os.path.normpath(os.path.join(repo_root, *relpath.replace("\\", "/").split("/"))) return abs_path if abs_path.startswith(os.path.normpath(repo_root) + os.sep) else None @@ -434,7 +435,7 @@ def parse_single_file(file): `src` set to the file path. str: Error string beginning with "Error:" describing YAML parsing or other failure. """ - with open(file, "r") as stream: + with open(file) as stream: try: data = yaml.safe_load(stream) manufacturer = data["manufacturer"] @@ -625,7 +626,7 @@ def clone_repo(self): except Exception as git_error: raise UnknownError("Git Repository Error", cause=git_error) from git_error - def get_devices(self, base_path, vendors: Optional[Sequence[str]] = None): + def get_devices(self, base_path, vendors: Sequence[str] | None = None): """Discover device YAML files and vendor directories under a base path. Args: @@ -769,7 +770,7 @@ def discover_vendors(self, devices_path, modules_path, racks_path): # Return sorted list by slug return sorted(vendors_dict.values(), key=lambda v: v["slug"]) - def parse_files(self, files: list, slugs: Optional[Sequence[str]] = None, progress=None): + def parse_files(self, files: list, slugs: Sequence[str] | None = None, progress=None): """Parse YAML device files into device type dicts, optionally filtering and tracking progress. Args: diff --git a/core/update_failure_resolver.py b/core/update_failure_resolver.py index 82543a5dc..54bb6f104 100644 --- a/core/update_failure_resolver.py +++ b/core/update_failure_resolver.py @@ -20,9 +20,10 @@ from __future__ import annotations import json +from collections.abc import Callable from dataclasses import dataclass, field from enum import Enum -from typing import Any, Callable, List, Optional +from typing import Any class FailureKind(str, Enum): @@ -57,10 +58,10 @@ class FailureResolution: kind: FailureKind description: str = "" - blocking_objects: List[str] = field(default_factory=list) - dependent_devices_count: Optional[int] = None - dependent_devices_sample: List[str] = field(default_factory=list) - remediation_steps: List[Callable[[], None]] = field(default_factory=list) + blocking_objects: list[str] = field(default_factory=list) + dependent_devices_count: int | None = None + dependent_devices_sample: list[str] = field(default_factory=list) + remediation_steps: list[Callable[[], None]] = field(default_factory=list) hint: str = "" @property @@ -120,7 +121,7 @@ def _matches_subdevice_role_constraint(payload: Any) -> bool: return False -def _count_dependent_devices(netbox: Any, device_type_id: int) -> tuple[int, List[str]]: +def _count_dependent_devices(netbox: Any, device_type_id: int) -> tuple[int, list[str]]: """Query NetBox for devices using *device_type_id*. Returns ``(count, sample_names)`` where ``sample_names`` is up to 5 names @@ -148,7 +149,7 @@ def _count_dependent_devices(netbox: Any, device_type_id: int) -> tuple[int, Lis return total, sample -def _list_device_bay_templates(netbox: Any, device_type_id: int) -> Optional[List[Any]]: +def _list_device_bay_templates(netbox: Any, device_type_id: int) -> list[Any] | None: """Return all ``DeviceBayTemplate`` records attached to *device_type_id*. Returns ``None`` when the NetBox query itself fails (network error, 5xx, etc.) diff --git a/nb-dt-import.py b/nb-dt-import.py index 6d53f2da9..61946e167 100644 --- a/nb-dt-import.py +++ b/nb-dt-import.py @@ -1,12 +1,12 @@ #!/usr/bin/env python3 """Import NetBox device and module types from the community library.""" +import sys from contextlib import contextmanager from datetime import datetime -import sys -from pynetbox.core.query import RequestError as NetBoxRequestError import requests +from pynetbox.core.query import RequestError as NetBoxRequestError from rich.panel import Panel from rich.progress import ( BarColumn, diff --git a/tests/conftest.py b/tests/conftest.py index a2d55c4e3..32544ae5e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,7 +1,8 @@ import os -import pytest from unittest.mock import MagicMock, patch +import pytest + from core.config import RunConfig diff --git a/tests/integration/test_import.py b/tests/integration/test_import.py index acb110f70..429717d81 100644 --- a/tests/integration/test_import.py +++ b/tests/integration/test_import.py @@ -67,8 +67,8 @@ import urllib3 from core.change_detector import get_device_type_properties -from core.config import resolve_run_config from core.component_registry import COMPONENT_TYPES +from core.config import resolve_run_config from core.graphql_client import NetBoxGraphQLClient urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) @@ -399,7 +399,7 @@ def test_graphql_schema() -> None: ok("get_manufacturers() returned TestVendor") # ── Device types: every schema property present ── - dt_by_model, dt_by_slug = client.get_device_types() + _dt_by_model, dt_by_slug = client.get_device_types() fd = dt_by_slug.get(("testvendor", "testvendor-full-device")) if fd is None: fail("get_device_types() did not return full-device") diff --git a/tests/test_change_detector.py b/tests/test_change_detector.py index f56450d37..bb79c6751 100644 --- a/tests/test_change_detector.py +++ b/tests/test_change_detector.py @@ -1,8 +1,6 @@ from types import SimpleNamespace from unittest.mock import MagicMock -from core.component_cache import ComponentCache -from core.graphql_client import DotDict from core.change_detector import ( ChangeDetector, ChangeReport, @@ -11,6 +9,8 @@ DeviceTypeChange, PropertyChange, ) +from core.component_cache import ComponentCache +from core.graphql_client import DotDict def _cache(**records): @@ -651,12 +651,14 @@ def test_a_reader_failure_propagates(self): from core.change_detector import _load_device_type_properties - with patch( - "core.change_detector.load_properties_for_type", - side_effect=RuntimeError("schema unavailable"), + with ( + patch( + "core.change_detector.load_properties_for_type", + side_effect=RuntimeError("schema unavailable"), + ), + pytest.raises(RuntimeError, match="schema unavailable"), ): - with pytest.raises(RuntimeError, match="schema unavailable"): - _load_device_type_properties("/nonexistent") + _load_device_type_properties("/nonexistent") # --------------------------------------------------------------------------- diff --git a/tests/test_component_cache.py b/tests/test_component_cache.py index a915936f5..181d77df1 100644 --- a/tests/test_component_cache.py +++ b/tests/test_component_cache.py @@ -19,7 +19,6 @@ from core.component_registry import COMPONENT_TYPES from core.graphql_client import GraphQLCountMismatchError, GraphQLSchemaError - # ── Fakes ───────────────────────────────────────────────────────────────────── @@ -267,7 +266,8 @@ def test_a_miss_filters_by_device_type_and_caches_the_answer(self): first = cache.get("interface_templates", "device", 1, endpoint) second = cache.get("interface_templates", "device", 1, endpoint) - assert set(first) == {"eth0"} and first == second + assert set(first) == {"eth0"} + assert first == second assert endpoint.filter_calls == [{"device_type_id": 1}] def test_a_miss_filters_by_module_type_for_a_module_parent(self): diff --git a/tests/test_docker_workflow.py b/tests/test_docker_workflow.py index 8b8ce6024..19fb5d4fd 100644 --- a/tests/test_docker_workflow.py +++ b/tests/test_docker_workflow.py @@ -4,7 +4,6 @@ import yaml - _WORKFLOW = Path(__file__).resolve().parents[1] / ".github" / "workflows" / "docker.yml" diff --git a/tests/test_errors.py b/tests/test_errors.py index 16a91d646..cbd7e0d60 100644 --- a/tests/test_errors.py +++ b/tests/test_errors.py @@ -23,8 +23,10 @@ ), ( SSLVerificationError(False), - "SSL verification failed. IGNORE_SSL_ERRORS is False. " - "Set IGNORE_SSL_ERRORS to True if you want to ignore this error. EXITING.", + ( + "SSL verification failed. IGNORE_SSL_ERRORS is False. " + "Set IGNORE_SSL_ERRORS to True if you want to ignore this error. EXITING." + ), ), (GitCommandError("my-repo"), 'Git error for repo "my-repo".'), (GitInvalidRepositoryError("my-repo"), 'The repo "my-repo" is not a valid git repo.'), diff --git a/tests/test_export_manifest.py b/tests/test_export_manifest.py index c646a2419..0cfddd016 100644 --- a/tests/test_export_manifest.py +++ b/tests/test_export_manifest.py @@ -1,7 +1,8 @@ """Tests for core/export_manifest.py.""" import json -from core.export_manifest import load_manifest, save_manifest, is_entry_fresh, update_entry + +from core.export_manifest import is_entry_fresh, load_manifest, save_manifest, update_entry class TestLoadManifest: diff --git a/tests/test_exporter.py b/tests/test_exporter.py index 8e0638849..7bfa5bd1e 100644 --- a/tests/test_exporter.py +++ b/tests/test_exporter.py @@ -1,15 +1,16 @@ """Tests for core/export.py — Exporter class.""" from dataclasses import replace +from unittest.mock import MagicMock, patch import pytest -from unittest.mock import MagicMock, patch import yaml from core.config import RunConfig from core.export import ( - ExportItem, + _SKIP, Exporter, + ExportItem, _canon_mfr_slug, _is_subset, _make_filename, @@ -17,11 +18,9 @@ _repo_supersedes, _sanitize_attachment_filename, _yaml_equal, - _SKIP, ) from core.log_handler import LogHandler - # ── Fixtures ────────────────────────────────────────────────────────────────── @@ -286,6 +285,7 @@ def test_a_default_positions_does_not_make_every_front_port_differ(self): against an entry that omits it would report every such definition as differing. """ from types import SimpleNamespace + from core.nb_serializer import _serialize_front_port legacy = SimpleNamespace(name="FP1", type="8p8c", label="", description="", color="") diff --git a/tests/test_graphql_client.py b/tests/test_graphql_client.py index 378a83e16..9eab667eb 100644 --- a/tests/test_graphql_client.py +++ b/tests/test_graphql_client.py @@ -3,9 +3,9 @@ import ast import pathlib import textwrap +from unittest.mock import MagicMock, patch import pytest -from unittest.mock import MagicMock, patch import requests from core.graphql_client import DotDict, NetBoxGraphQLClient @@ -161,9 +161,10 @@ def test_a_v1_token_clone_keeps_the_legacy_auth_scheme(self): @pytest.mark.real_http def test_an_unreachable_server_fails_the_probe_as_a_graphql_error(self): """It is the export's first request, so a raw transport error escapes as a traceback.""" - from core.graphql_client import GraphQLError from helpers import FakeNetBox + from core.graphql_client import GraphQLError + server = FakeNetBox() url = server.url server.close() # nothing is listening on that port any more @@ -606,8 +607,7 @@ class TestGetManufacturers: def _make_client(self): from core.graphql_client import NetBoxGraphQLClient - client = NetBoxGraphQLClient("http://netbox.local", "tok") - return client + return NetBoxGraphQLClient("http://netbox.local", "tok") def test_returns_dict_keyed_by_name(self, mock_post): data = { @@ -1216,7 +1216,8 @@ def test_a_clone_still_selects_the_module_bay_type_relation(self, mock_post): client.clone().get_component_templates("module_bay_templates") queries = [call.kwargs["json"]["query"] for call in mock_post.call_args_list] - assert queries and all("module_bay_types" in q for q in queries) + assert queries + assert all("module_bay_types" in q for q in queries) def test_the_module_type_query_selects_the_relation_only_when_supported(self, mock_post): """The selection has to track the server, in both directions. @@ -1772,11 +1773,11 @@ def test_primary_fails_no_mappings_field_reraises(self, mock_post): selection — meaning there is no older tier left to fall back to. """ from dataclasses import replace + from unittest.mock import patch + import core.graphql_client as gc_module from core.component_registry import BY_ENDPOINT from core.graphql_client import GraphQLError - from unittest.mock import patch - import core.graphql_client as gc_module # A NetBox whose schema dropped the mappings block entirely. stripped = {"front_port_templates": replace(BY_ENDPOINT["front_port_templates"], graphql_extra=())} @@ -1889,9 +1890,8 @@ def test_request_exception_retries_then_raises_graphql_error(self, mock_post): mock_post.side_effect = requests.exceptions.ConnectionError("connection refused") client = self._make_client() - with patch("core.graphql_client.time.sleep"): - with pytest.raises(GraphQLError, match="connection refused"): - client.query("{ test }", _retries=1) + with patch("core.graphql_client.time.sleep"), pytest.raises(GraphQLError, match="connection refused"): + client.query("{ test }", _retries=1) def test_request_exception_retries_before_exhausting(self, mock_post): """A RequestException on the first attempt stores last_exc and retries.""" @@ -1903,9 +1903,8 @@ def test_request_exception_retries_before_exhausting(self, mock_post): mock_post.side_effect = requests.exceptions.Timeout("timed out") client = self._make_client() - with patch("core.graphql_client.time.sleep"): - with pytest.raises(GraphQLError, match="timed out"): - client.query("{ test }", _retries=1) + with patch("core.graphql_client.time.sleep"), pytest.raises(GraphQLError, match="timed out"): + client.query("{ test }", _retries=1) # ── Vendor-scoped filtering tests ───────────────────────────────────────── @@ -2012,7 +2011,7 @@ def test_single_vendor_filter(self, mock_post, slugs): mock_post.side_effect = _make_paged_responses(data, "device_type_list") client = self._make_client() - by_model, by_slug = client.get_device_types(manufacturer_slugs=slugs) + by_model, _by_slug = client.get_device_types(manufacturer_slugs=slugs) assert ("cisco", "Catalyst 3850") in by_model assert by_model[("cisco", "Catalyst 3850")].model == "Catalyst 3850" @@ -2068,7 +2067,7 @@ def test_multiple_vendor_filter(self, mock_post, slugs): mock_post.side_effect = _make_paged_responses(data, "device_type_list") client = self._make_client() - by_model, by_slug = client.get_device_types(manufacturer_slugs=slugs) + by_model, _by_slug = client.get_device_types(manufacturer_slugs=slugs) assert ("cisco", "Catalyst 3850") in by_model assert ("juniper", "EX4300") in by_model @@ -2106,7 +2105,7 @@ def test_none_manufacturer_slugs_unfiltered(self, mock_post): mock_post.side_effect = _make_paged_responses(data, "device_type_list") client = self._make_client() - by_model, by_slug = client.get_device_types(manufacturer_slugs=None) + _by_model, _by_slug = client.get_device_types(manufacturer_slugs=None) # Verify no filter in the query and no manufacturer variable sent call_payload = mock_post.call_args_list[0][1]["json"] @@ -2650,7 +2649,7 @@ def test_get_device_types_includes_last_updated(self, mocker): } ] client = self._mock_graphql(mocker, items) - by_model, by_slug = client.get_device_types() + by_model, _by_slug = client.get_device_types() record = by_model[("acme", "M")] assert record.last_updated == "2024-01-15T10:00:00Z" @@ -2922,7 +2921,8 @@ def test_schema_error_still_falls_back_to_the_older_field_tier(self): server.shutdown() assert result == [] - assert "T1" in attempts and "T2" in attempts + assert "T1" in attempts + assert "T2" in attempts @pytest.mark.real_http @@ -2992,7 +2992,8 @@ def test_schema_error_still_triggers_the_unfiltered_fallback(self, method): finally: server.shutdown() - assert "filtered" in attempts and "unfiltered" in attempts + assert "filtered" in attempts + assert "unfiltered" in attempts class TestErrorHandlingStandards: diff --git a/tests/test_import_run.py b/tests/test_import_run.py index 9c9d56947..def1d1a51 100644 --- a/tests/test_import_run.py +++ b/tests/test_import_run.py @@ -2,12 +2,12 @@ from collections import Counter from contextlib import contextmanager +from types import SimpleNamespace import pytest +from helpers import recording_handle as _recording_handle from core.errors import VendorSelectionError -from types import SimpleNamespace - from core.import_run import ( ImportRun, RunSummary, @@ -20,7 +20,6 @@ from core.log_handler import LogHandler from core.outcomes import EntityKind, Outcome from core.repo import DTLRepo -from helpers import recording_handle as _recording_handle class _ComponentCache: @@ -118,7 +117,7 @@ def discover_vendors(self, _devices_path, _modules_path, _racks_path): @staticmethod def resolve_slug_files(_slugs): """Force the pipeline to use its full file scan.""" - return None + return @staticmethod def get_devices(path, vendors): diff --git a/tests/test_log_handler.py b/tests/test_log_handler.py index 3eaf98a51..4e02de920 100644 --- a/tests/test_log_handler.py +++ b/tests/test_log_handler.py @@ -1,9 +1,9 @@ -from types import SimpleNamespace from inspect import signature +from types import SimpleNamespace from unittest.mock import MagicMock, patch -from core.log_handler import LogHandler from core.graphql_client import NetBoxGraphQLClient +from core.log_handler import LogHandler from core.netbox_api import DeviceTypes, NetBox from core.repo import DTLRepo diff --git a/tests/test_module_bay_type_sync.py b/tests/test_module_bay_type_sync.py index afe276260..b2084a1b3 100644 --- a/tests/test_module_bay_type_sync.py +++ b/tests/test_module_bay_type_sync.py @@ -12,6 +12,7 @@ import pynetbox import pytest +from helpers import FakeNetBox, recording_handle, write_module_bay_type from core.change_detector import ChangeType, ComponentChange, PropertyChange from core.component_registry import BY_YAML_KEY @@ -19,7 +20,6 @@ from core.module_bay_types import ModuleBayTypeCatalog from core.netbox_api import DeviceTypes, NetBox from core.outcomes import EntityKind, Outcome -from helpers import FakeNetBox, recording_handle, write_module_bay_type # The suite patches requests.Session by default, which would stop every request below. pytestmark = pytest.mark.real_http @@ -484,7 +484,8 @@ def test_an_older_server_creates_the_bay_without_the_field(self, make_device_typ ) posted = server.sent("POST", "module_bay_templates") - assert posted and "module_bay_types" not in posted[0] + assert posted + assert "module_bay_types" not in posted[0] def test_an_unresolvable_bay_is_never_posted(self, make_device_types, server): device_types, _ = make_device_types() diff --git a/tests/test_module_bay_types.py b/tests/test_module_bay_types.py index 82a2e4d14..e629b3937 100644 --- a/tests/test_module_bay_types.py +++ b/tests/test_module_bay_types.py @@ -7,9 +7,9 @@ """ import pytest +from helpers import FakeNetBox, write_module_bay_type from core.module_bay_types import ModuleBayCatalogError, ModuleBayTypeCatalog, ModuleBayTypeError -from helpers import FakeNetBox, write_module_bay_type class Handle: @@ -142,7 +142,8 @@ def test_same_name_different_slug_is_an_error_not_a_rename(self, catalog): ) with pytest.raises(ModuleBayTypeError) as exc: cat.ids_for("juniper", ["MX304-RE"]) - assert "mx304_re" in str(exc.value) and "mx304-re" in str(exc.value) + assert "mx304_re" in str(exc.value) + assert "mx304-re" in str(exc.value) assert not server.sent("POST", "module_bay_types") @@ -185,7 +186,8 @@ def test_an_entry_missing_a_required_field_is_refused_at_load(self, tmp_path, ca with pytest.raises(ModuleBayCatalogError) as exc: cat.identities_for("generic", ["SFP"]) - assert "slug" in str(exc.value) and "sfp.yaml" in str(exc.value) + assert "slug" in str(exc.value) + assert "sfp.yaml" in str(exc.value) def test_unparseable_yaml_is_refused_as_a_catalog_error(self, tmp_path, catalog): """A parser error escaping the boundary ends the run before it reports anything.""" @@ -222,7 +224,8 @@ def test_duplicate_scoped_entry_is_refused(self, tmp_path, catalog): cat, _ = catalog(root=tmp_path) with pytest.raises(ModuleBayCatalogError) as exc: cat.ids_for("generic", ["SFP"]) - assert "Duplicate" in str(exc.value) and "SFP" in str(exc.value) + assert "Duplicate" in str(exc.value) + assert "SFP" in str(exc.value) @pytest.mark.real_http @@ -235,7 +238,8 @@ def test_a_bare_key_is_refused_rather_than_clearing(self, catalog): with pytest.raises(ModuleBayTypeError) as exc: cat.ids_for("juniper", None) assert "list of names" in str(exc.value) - assert not server.sent("POST", "module_bay_types") and not server.sent("POST", "manufacturers") + assert not server.sent("POST", "module_bay_types") + assert not server.sent("POST", "manufacturers") def test_a_non_string_entry_is_refused(self, catalog): cat, _ = catalog() @@ -250,7 +254,8 @@ def test_an_empty_list_is_allowed_and_holds_no_classes(self, catalog): """`module_bay_types: []` is an explicit instruction, not a malformed value.""" cat, server = catalog() assert cat.ids_for("juniper", []) == [] - assert not server.sent("POST", "module_bay_types") and not server.sent("POST", "manufacturers") + assert not server.sent("POST", "module_bay_types") + assert not server.sent("POST", "manufacturers") def test_duplicate_names_collapse(self, catalog): """The relationship is a set, so a repeated name must not produce a repeated id.""" diff --git a/tests/test_nb_dt_import.py b/tests/test_nb_dt_import.py index fea59e41d..896b829e3 100644 --- a/tests/test_nb_dt_import.py +++ b/tests/test_nb_dt_import.py @@ -826,9 +826,9 @@ def test_missing_env_var_triggers_system_exit(self, nb_dt_import): patch("nb_dt_import.NetBox"), patch.dict(os.environ, {}, clear=True), patch("core.config.load_dotenv"), + pytest.raises(SystemExit), ): - with pytest.raises(SystemExit): - nb_dt_import.main() + nb_dt_import.main() def test_vendors_and_slugs_flags_log_lines(self, nb_dt_import, capsys): """--vendors and --slugs args cause their respective log lines to execute.""" @@ -1088,9 +1088,8 @@ def test_entry_point_connection_error_message_references_netbox(self, capsys): with patch("core.repo.DTLRepo") as MockDTLRepo, patch("core.netbox_api.NetBox"): MockDTLRepo.side_effect = _requests.exceptions.ConnectionError("Remote end closed") - with patch.object(sys, "argv", ["nb-dt-import.py", "--only-new"]): - with pytest.raises(SystemExit): - runpy.run_path(_NB_DT_IMPORT_PATH, run_name="__main__") + with patch.object(sys, "argv", ["nb-dt-import.py", "--only-new"]), pytest.raises(SystemExit): + runpy.run_path(_NB_DT_IMPORT_PATH, run_name="__main__") captured = capsys.readouterr() assert "connection" in captured.err.lower() or "netbox" in captured.err.lower() @@ -1481,9 +1480,9 @@ def test_main_turns_a_fatal_error_into_system_exit(self, nb_dt_import, make_conf with ( patch.object(nb_dt_import, "resolve_run_config", return_value=make_config(verbose=False)), patch.object(nb_dt_import, "_run", side_effect=error), + pytest.raises(SystemExit) as exc_info, ): - with pytest.raises(SystemExit) as exc_info: - nb_dt_import.main() + nb_dt_import.main() assert str(exc_info.value) == 'An unknown error occurred: "Git Repository Error"' assert capsys.readouterr().out == "" @@ -1498,9 +1497,9 @@ def test_main_prints_the_cause_traceback_only_in_verbose_mode(self, nb_dt_import with ( patch.object(nb_dt_import, "resolve_run_config", return_value=make_config(verbose=True)), patch.object(nb_dt_import, "_run", side_effect=error), + pytest.raises(SystemExit), ): - with pytest.raises(SystemExit): - nb_dt_import.main() + nb_dt_import.main() output = capsys.readouterr().out assert "Traceback (most recent call last)" in output @@ -1895,9 +1894,9 @@ def __exit__(self, exc_type, exc, tb): patch("nb_dt_import.NetBox", return_value=mock_nb), patch("core.import_run._process_device_types", side_effect=RuntimeError("boom")), patch("nb_dt_import.get_progress_panel", return_value=_Ctx()), + pytest.raises(RuntimeError, match="boom"), ): - with pytest.raises(RuntimeError, match="boom"): - nb_dt_import.main() + nb_dt_import.main() mock_nb.device_types.components.close.assert_called() @@ -2035,16 +2034,16 @@ def test_import_run_stops_preload_in_finally_on_error(self, make_config): with ( patch("core.import_run._process_device_types", side_effect=RuntimeError("boom")), patch("core.import_run._finalize_task_registry") as mock_finalize, + pytest.raises(RuntimeError, match="boom"), ): - with pytest.raises(RuntimeError, match="boom"): - run = import_run_module.ImportRun( - config, - dtl_repo, - netbox, - handle, - lambda _show_remaining_time: nullcontext(progress), - ) - run.execute() + run = import_run_module.ImportRun( + config, + dtl_repo, + netbox, + handle, + lambda _show_remaining_time: nullcontext(progress), + ) + run.execute() netbox.device_types.components.close.assert_called() mock_finalize.assert_called_once_with(progress, {}) @@ -2239,7 +2238,8 @@ def test_a_47_server_is_asked_for_the_module_bay_type_relation( queries = self._queries_for_version(nb_dt_import, monkeypatch, tmp_path, _real_library_root, "4.7.0") module_types = [q for q in queries if "module_type_list(" in q] - assert module_types and all("module_bay_types" in q for q in module_types) + assert module_types + assert all("module_bay_types" in q for q in module_types) def test_an_older_server_is_never_asked_for_it(self, nb_dt_import, monkeypatch, tmp_path, _real_library_root): """Selecting a field the schema lacks fails the whole query.""" diff --git a/tests/test_netbox_api.py b/tests/test_netbox_api.py index 2423e1064..d7236b78d 100644 --- a/tests/test_netbox_api.py +++ b/tests/test_netbox_api.py @@ -3,22 +3,22 @@ from types import SimpleNamespace from contextlib import contextmanager from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from unittest.mock import MagicMock, patch import pytest +from helpers import paginate_dispatch, recording_handle from requests.exceptions import ConnectionError as RequestsConnectionError -from unittest.mock import MagicMock, patch + from core.component_registry import BY_YAML_KEY, COMPONENT_TYPES -from core.outcomes import EntityKind, Outcome from core.netbox_api import ( + DeviceTypes, NetBox, NetBoxError, - DeviceTypes, SSLVerificationError, _delete_image_attachment, _FrontPortRecordWithMappings, ) -from helpers import paginate_dispatch, recording_handle - +from core.outcomes import EntityKind, Outcome # All component list keys used by the GraphQL client for empty-response fallback. _ALL_COMPONENT_KEYS = [component.list_key for component in COMPONENT_TYPES] @@ -806,8 +806,10 @@ def test_per_item_error_reports_every_failing_item( messages = [call.args[0] for call in mock_handle.log.call_args_list] assert len(messages) == 2 - assert "eth0" in messages[0] and "This field must be unique." in messages[0] - assert "eth2" in messages[1] and "Invalid choice." in messages[1] + assert "eth0" in messages[0] + assert "This field must be unique." in messages[0] + assert "eth2" in messages[1] + assert "Invalid choice." in messages[1] def test_non_list_error_logs_failed_items( self, mock_settings, mock_pynetbox, graphql_client, make_device_types, mock_handle @@ -1053,7 +1055,7 @@ def test_empty_src_returns_none(self): assert _image_dir_for_yaml("", "device-types", "elevation-images") is None def test_unknown_src_returns_none(self): - from core.netbox_api import _image_dir_for_yaml, _UNKNOWN_SRC + from core.netbox_api import _UNKNOWN_SRC, _image_dir_for_yaml assert _image_dir_for_yaml(_UNKNOWN_SRC, "device-types", "elevation-images") is None @@ -1921,6 +1923,7 @@ def test_update_property_change_request_error_logged( "Device Type Updated" log MUST NOT be emitted. """ import pynetbox as real_pynb2 + from core.change_detector import ChangeReport, DeviceTypeChange, PropertyChange mock_pynetbox.RequestError = real_pynb2.RequestError @@ -1991,6 +1994,7 @@ def _build_subdevice_role_flip_setup( report zero dependent devices and one blocking device-bay template. """ import pynetbox as real_pynb2 + from core.change_detector import ChangeReport, DeviceTypeChange, PropertyChange mock_pynetbox.RequestError = real_pynb2.RequestError @@ -2024,8 +2028,10 @@ def _build_subdevice_role_flip_setup( # Force the .error property to return the parsed dict (pynetbox usually does this). err.error = { "subdevice_role": [ - "Must delete all device bay templates associated with this device " - "before declassifying it as a parent device." + ( + "Must delete all device bay templates associated with this device " + "before declassifying it as a parent device." + ) ] } @@ -2049,7 +2055,7 @@ def test_constraint_failure_logs_hint_when_flag_off( self, mock_settings, mock_pynetbox, graphql_client, make_device_types, mock_handle ): """Without --force-resolve-conflicts, classifier hint is logged but no auto-resolve runs.""" - nb, dt, mock_nb_api, report, device_type, blocking_template, err = self._build_subdevice_role_flip_setup( + nb, _dt, mock_nb_api, report, device_type, blocking_template, err = self._build_subdevice_role_flip_setup( mock_settings, mock_handle, mock_pynetbox, make_device_types, force=False ) mock_nb_api.dcim.device_types.update.side_effect = err @@ -2076,13 +2082,14 @@ def test_constraint_failure_logs_hint_when_flag_off( assert failures[0].outcome == Outcome.FAILED assert "SuperServer" in failures[0].identity assert "module-bay-1" in failures[0].blocking_objects - assert failures[0].hint and "--force-resolve-conflicts" in failures[0].hint + assert failures[0].hint + assert "--force-resolve-conflicts" in failures[0].hint def test_constraint_failure_auto_resolves_when_flag_on_and_safe( self, mock_settings, mock_pynetbox, graphql_client, make_device_types, mock_handle ): """With flag on + zero dependents, blocking templates are deleted and PATCH retried.""" - nb, dt, mock_nb_api, report, device_type, blocking_template, err = self._build_subdevice_role_flip_setup( + nb, _dt, mock_nb_api, report, device_type, blocking_template, err = self._build_subdevice_role_flip_setup( mock_settings, mock_handle, mock_pynetbox, make_device_types, force=True ) # First update call fails; second (after auto-resolve) succeeds. @@ -2103,7 +2110,7 @@ def test_constraint_failure_auto_resolve_retry_still_fails( """If the retried PATCH after auto-resolve still fails, count it as a failure exactly once.""" import pynetbox as real_pynb2 - nb, dt, mock_nb_api, report, device_type, blocking_template, err = self._build_subdevice_role_flip_setup( + nb, _dt, mock_nb_api, report, device_type, blocking_template, err = self._build_subdevice_role_flip_setup( mock_settings, mock_handle, mock_pynetbox, make_device_types, force=True ) err2 = real_pynb2.RequestError(MagicMock(status_code=500, content=b'{"detail":"still bad"}')) @@ -2124,7 +2131,7 @@ def test_constraint_failure_blocked_when_devices_in_use( self, mock_settings, mock_pynetbox, graphql_client, make_device_types, mock_handle ): """With flag on but live devices reference the type, no remediation runs (safety gate).""" - nb, dt, mock_nb_api, report, device_type, blocking_template, err = self._build_subdevice_role_flip_setup( + nb, _dt, mock_nb_api, report, device_type, blocking_template, err = self._build_subdevice_role_flip_setup( mock_settings, mock_handle, mock_pynetbox, make_device_types, force=True ) live_device = MagicMock() @@ -2694,14 +2701,14 @@ def test_existing_module_with_new_image_is_actionable( } # existing images are empty → the image is "new" → actionable - with patch.object(nb, "_fetch_module_type_existing_images", return_value={42: set()}): - with patch( + with ( + patch.object(nb, "_fetch_module_type_existing_images", return_value={42: set()}), + patch( "core.netbox_api.NetBox._discover_module_image_files", return_value=[str(img)], - ): - result, existing_images_map, _ = nb.filter_actionable_module_types( - [module_type], all_mts, only_new=False - ) + ), + ): + result, existing_images_map, _ = nb.filter_actionable_module_types([module_type], all_mts, only_new=False) assert result == [module_type] # Verify the upload worklist is propagated so create_module_types can upload the image. assert existing_images_map == {42: set()} @@ -3534,6 +3541,7 @@ def test_property_update_success_counter_incremented( ): """Successful property update increments components_updated counter.""" from collections import Counter as _Counter + from core.change_detector import ChangeType, ComponentChange, PropertyChange mock_nb_api = mock_pynetbox.api.return_value @@ -3562,6 +3570,7 @@ def test_property_update_request_error_logged( ): """RequestError during property update is caught and logged.""" import pynetbox as real_pynb2 + from core.change_detector import ChangeType, ComponentChange, PropertyChange mock_pynetbox.RequestError = real_pynb2.RequestError @@ -3695,6 +3704,7 @@ def test_request_error_on_delete_is_logged( ): """RequestError during component deletion is caught and logged.""" import pynetbox as real_pynb2 + from core.change_detector import ChangeType, ComponentChange mock_pynetbox.RequestError = real_pynb2.RequestError @@ -4100,12 +4110,11 @@ def test_file_close_exception_swallowed( fake_fh = MagicMock() fake_fh.close.side_effect = OSError("cannot close") - with patch("builtins.open", return_value=fake_fh): - with patch("core.netbox_api.requests") as mock_req: - mock_req.RequestException = _req2.RequestException - mock_req.patch.side_effect = _req2.RequestException("server error") - # Should NOT raise despite close() raising - dt.upload_images("http://nb", "token", {"front_image": str(img)}, 1) + with patch("builtins.open", return_value=fake_fh), patch("core.netbox_api.requests") as mock_req: + mock_req.RequestException = _req2.RequestException + mock_req.patch.side_effect = _req2.RequestException("server error") + # Should NOT raise despite close() raising + dt.upload_images("http://nb", "token", {"front_image": str(img)}, 1) # The RequestException log should still have been called assert mock_handle.log.called @@ -4556,6 +4565,7 @@ def test_request_error_on_create_logged_no_crash(self, mock_settings, mock_pynet def test_request_error_on_update_logged_no_crash(self, mock_settings, mock_pynetbox, mock_handle): """RequestError during update is logged; processing continues.""" import pynetbox + from core.graphql_client import DotDict mock_pynetbox.api.return_value.version = "4.3" @@ -5023,6 +5033,7 @@ def test_load_for_vendor_populates_existing_device_types( ): """load_for_vendor populates existing_device_types and existing_device_types_by_slug.""" from unittest.mock import patch as _patch + from core.graphql_client import DotDict mock_nb_api = mock_pynetbox.api.return_value @@ -5055,6 +5066,7 @@ def test_load_for_vendor_replaces_prior_data( ): """A second call to load_for_vendor replaces data from the first call.""" from unittest.mock import patch as _patch + from core.graphql_client import DotDict mock_nb_api = mock_pynetbox.api.return_value @@ -5086,6 +5098,7 @@ def test_load_for_vendor_resets_state_before_fetch_on_failure( ): """State is reset before the fetch so a raised exception leaves a clean slate.""" from unittest.mock import patch as _patch + import pytest mock_nb_api = mock_pynetbox.api.return_value @@ -5307,6 +5320,7 @@ def test_returns_false_when_no_cache_entry(self, tmp_path): def test_returns_false_when_hash_matches(self, tmp_path): import hashlib + from core.netbox_api import _is_image_hash_changed data = b"unchanged_content" @@ -5317,6 +5331,7 @@ def test_returns_false_when_hash_matches(self, tmp_path): def test_returns_true_when_hash_differs(self, tmp_path): import hashlib + from core.netbox_api import _is_image_hash_changed img = tmp_path / "front.png" @@ -5326,6 +5341,7 @@ def test_returns_true_when_hash_differs(self, tmp_path): def test_returns_false_on_file_read_error(self, tmp_path): import hashlib + from core.netbox_api import _is_image_hash_changed path = str(tmp_path / "missing.png") @@ -5381,9 +5397,8 @@ def test_verify_images_reuploads_missing_image( mock_resp = MagicMock() mock_resp.ok = False # image missing on server → "missing" - with patch("glob.glob", return_value=[str(img)]): - with patch("requests.get", return_value=mock_resp): - nb.create_device_types([device_type]) + with patch("glob.glob", return_value=[str(img)]), patch("requests.get", return_value=mock_resp): + nb.create_device_types([device_type]) nb.device_types.upload_images.assert_called_once() @@ -5466,9 +5481,8 @@ def test_verify_images_skips_ok_image( mock_resp = MagicMock() mock_resp.ok = True # image accessible on server mock_resp.headers = {"Content-Type": "image/png"} - with patch("glob.glob", return_value=[str(img)]): - with patch("requests.get", return_value=mock_resp): - nb.create_device_types([device_type]) + with patch("glob.glob", return_value=[str(img)]), patch("requests.get", return_value=mock_resp): + nb.create_device_types([device_type]) nb.device_types.upload_images.assert_not_called() @@ -5503,9 +5517,8 @@ def test_default_mode_still_skips_without_http( "src": str(dev_types_dir / "ap.yaml"), } - with patch("glob.glob", return_value=[str(img)]): - with patch("requests.get") as mock_get: - nb.create_device_types([device_type]) + with patch("glob.glob", return_value=[str(img)]), patch("requests.get") as mock_get: + nb.create_device_types([device_type]) mock_get.assert_not_called() nb.device_types.upload_images.assert_not_called() @@ -5549,10 +5562,9 @@ def test_verify_ok_seeds_hash_cache( mock_resp = MagicMock() mock_resp.ok = True mock_resp.headers = {"Content-Type": "image/png"} - with patch("glob.glob", return_value=[str(img)]): - with patch("requests.get", return_value=mock_resp): - with patch("core.netbox_api._save_image_hash_cache") as mock_save: - nb.create_device_types([device_type]) + with patch("glob.glob", return_value=[str(img)]), patch("requests.get", return_value=mock_resp): + with patch("core.netbox_api._save_image_hash_cache") as mock_save: + nb.create_device_types([device_type]) # Hash cache must have been seeded with the local file's hash assert str(img) in nb._image_hash_cache @@ -5598,6 +5610,7 @@ def test_delete_image_attachment_success(self, mock_settings, mock_handle): def test_delete_image_attachment_logs_request_errors(self, mock_settings, mock_handle): import requests + from core.netbox_api import _delete_image_attachment with patch("core.netbox_api.requests.delete", side_effect=requests.RequestException("boom")): @@ -5616,6 +5629,7 @@ def test_load_module_type_properties_falls_back_without_a_schema(self, tmp_path) def test_fmt_connection_error_contains_url_and_hint(self): """_fmt_connection_error returns a message with the URL and a reachability hint.""" import requests as _requests + from core.netbox_api import _fmt_connection_error url = "http://netbox.example.com" @@ -5627,9 +5641,10 @@ def test_fmt_connection_error_contains_url_and_hint(self): def test_fmt_connection_error_verify_compatibility_uses_it(self, mock_settings, mock_pynetbox, mock_handle): """verify_compatibility uses _fmt_connection_error for ConnectionError.""" - import requests as _requests from unittest.mock import PropertyMock + import requests as _requests + type(mock_pynetbox.api.return_value).version = PropertyMock( side_effect=_requests.exceptions.ConnectionError("drop") ) @@ -5638,7 +5653,8 @@ def test_fmt_connection_error_verify_compatibility_uses_it(self, mock_settings, NetBox(mock_settings, mock_handle) exc_msg = str(exc_info.value.args[0]) if exc_info.value.args else "" - assert mock_settings.netbox_url in exc_msg and "Connection" in exc_msg + assert mock_settings.netbox_url in exc_msg + assert "Connection" in exc_msg def test_check_image_url_reports_the_transport_error_when_log_fn_provided(self): """The wire detail goes to log_fn; the verdict goes to the return value.""" @@ -5702,7 +5718,8 @@ def test_a_corrupt_cache_file_is_reported_by_path(self, mock_settings, mock_pyne assert nb._image_hash_cache == {} logged = " ".join(str(call) for call in mock_handle.log.call_args_list) - assert str(path) in logged and "unreadable image hash cache" in logged + assert str(path) in logged + assert "unreadable image hash cache" in logged def test_a_cache_file_holding_the_wrong_shape_is_reported( self, mock_settings, mock_pynetbox, mock_handle, tmp_path @@ -6050,6 +6067,7 @@ def test_try_resolve_update_logs_classifier_exception(self, mock_settings, mock_ def test_try_resolve_update_truncates_blocker_list(self, mock_settings, mock_pynetbox, mock_handle): from types import SimpleNamespace + from core.update_failure_resolver import FailureKind mock_pynetbox.api.return_value.version = "4.3" @@ -6073,6 +6091,7 @@ def test_try_resolve_update_truncates_blocker_list(self, mock_settings, mock_pyn def test_try_resolve_update_logs_auto_resolve_failure(self, mock_settings, mock_pynetbox, mock_handle): from types import SimpleNamespace + from core.update_failure_resolver import FailureKind mock_pynetbox.api.return_value.version = "4.3" @@ -6102,9 +6121,11 @@ def boom(): def test_try_resolve_update_logs_retryable_exception_after_auto_resolve( self, mock_settings, mock_pynetbox, mock_handle ): + from types import SimpleNamespace + import pynetbox as real_pynb import requests - from types import SimpleNamespace + from core.update_failure_resolver import FailureKind mock_pynetbox.RequestError = real_pynb.RequestError @@ -6194,6 +6215,7 @@ def test_handle_existing_device_type_logs_retryable_property_update_error( ): import pynetbox as real_pynb import requests + from core.change_detector import PropertyChange mock_pynetbox.RequestError = real_pynb.RequestError @@ -6434,6 +6456,7 @@ def test_apply_updates_for_type_logs_retryable_exception( ): import pynetbox as real_pynb import requests + from core.change_detector import ChangeType, ComponentChange, PropertyChange mock_pynetbox.RequestError = real_pynb.RequestError @@ -6460,6 +6483,7 @@ def test_remove_components_logs_retryable_exception( ): import pynetbox as real_pynb import requests + from core.change_detector import ChangeType, ComponentChange mock_pynetbox.RequestError = real_pynb.RequestError @@ -6510,6 +6534,7 @@ class TestRemainingCoverageBranches: def test_create_rack_types_logs_retryable_update_error(self, mock_settings, mock_pynetbox, mock_handle): import pynetbox as real_pynb import requests + from core.graphql_client import DotDict mock_pynetbox.RequestError = real_pynb.RequestError @@ -7249,14 +7274,14 @@ class TestSkippedComponentReasonReachesTheReport: def _dt(self, mock_pynetbox, make_device_types, parent_id=1): mock_nb_api = mock_pynetbox.api.return_value mock_nb_api.version = "4.3" - dt = make_device_types(nb_api=mock_nb_api) - return dt + return make_device_types(nb_api=mock_nb_api) def test_unresolvable_power_port_reaches_the_report( self, mock_settings, mock_pynetbox, graphql_client, make_device_types, mock_handle ): """The Powerman case: an outlet dropped for a bad power_port must name the reason.""" import pynetbox as real_pynb + from core.change_detector import ChangeReport, ChangeType, ComponentChange, DeviceTypeChange mock_pynetbox.RequestError = real_pynb.RequestError @@ -7342,6 +7367,7 @@ def test_component_update_transport_failure_reports_the_transport_error( """A retry-exhausted update must name the transport error, not the generic label.""" import pynetbox as real_pynb import requests + from core.change_detector import ( ChangeReport, ChangeType, @@ -7407,6 +7433,7 @@ def test_component_removal_transport_failure_reports_the_transport_error( """A retry-exhausted removal must name the transport error too.""" import pynetbox as real_pynb import requests + from core.change_detector import ChangeType, ComponentChange mock_pynetbox.RequestError = real_pynb.RequestError @@ -7436,6 +7463,7 @@ def test_failed_component_create_reports_the_netbox_message( ): """The report must name the constraint, not the generic failure label.""" import pynetbox as real_pynb + from core.change_detector import ChangeReport, ChangeType, ComponentChange, DeviceTypeChange mock_pynetbox.RequestError = real_pynb.RequestError diff --git a/tests/test_normalization.py b/tests/test_normalization.py index 6d759ea16..1aba6fe8c 100644 --- a/tests/test_normalization.py +++ b/tests/test_normalization.py @@ -8,7 +8,7 @@ class TestNormalizeValues: def test_netbox_choice_object_reads_value(self): choice = SimpleNamespace(value="1000base-t") - y, n = normalize_values("1000base-t", choice) + _y, n = normalize_values("1000base-t", choice) assert n == "1000base-t" def test_empty_string_normalized_to_none(self): @@ -32,26 +32,26 @@ def test_trailing_spaces_stripped(self): assert n == "def" def test_numeric_yaml_coerces_string_netbox(self): - y, n = normalize_values(1.0, "1.0") + _y, n = normalize_values(1.0, "1.0") assert n == 1.0 def test_numeric_netbox_coerces_string_yaml(self): - y, n = normalize_values("2.5", 2.5) + y, _n = normalize_values("2.5", 2.5) assert y == 2.5 def test_int_yaml_preserves_int_type(self): """YAML int 166 vs NetBox '166.00' should normalize nb to int 166, not float 166.0.""" - y, n = normalize_values(166, "166.00") + _y, n = normalize_values(166, "166.00") assert n == 166 assert isinstance(n, int) def test_int_netbox_preserves_int_type(self): - y, n = normalize_values("166", 166) + y, _n = normalize_values("166", 166) assert y == 166 assert isinstance(y, int) def test_float_yaml_stays_float(self): - y, n = normalize_values(26.1, "26.10") + _y, n = normalize_values(26.1, "26.10") assert n == 26.1 def test_bool_not_coerced(self): @@ -66,11 +66,11 @@ def test_bool_as_int_trap(self): assert n == "1.0" def test_non_numeric_string_netbox_stays_string(self): - y, n = normalize_values(1.0, "notanumber") + _y, n = normalize_values(1.0, "notanumber") assert n == "notanumber" def test_non_numeric_string_yaml_stays_string(self): - y, n = normalize_values("notanumber", 1.0) + y, _n = normalize_values("notanumber", 1.0) assert y == "notanumber" diff --git a/tests/test_repo.py b/tests/test_repo.py index 09714d486..95e9e9160 100644 --- a/tests/test_repo.py +++ b/tests/test_repo.py @@ -1,9 +1,14 @@ import os +from unittest.mock import MagicMock, call, mock_open, patch import pytest import yaml -from unittest.mock import MagicMock, call, mock_open, patch -from git import Actor, Repo as GitRepo, exc as git_exc +from git import Actor +from git import Repo as GitRepo +from git import exc as git_exc + +from core.errors import UnknownError +from core.log_handler import LogHandler from core.repo import ( DTLRepo, GitBranchNotFoundError, @@ -15,11 +20,9 @@ _safe_index_load, _safe_json_load, _safe_pickle_load, - validate_git_url, normalize_port_mappings, + validate_git_url, ) -from core.errors import UnknownError -from core.log_handler import LogHandler def _dtl_repo(config, repo_path, handle): @@ -46,31 +49,31 @@ def test_https_valid(self): assert err is None def test_https_no_hostname_invalid(self): - ok, err = validate_git_url("https://") + ok, _err = validate_git_url("https://") assert ok is False def test_git_at_scp_valid(self): - ok, err = validate_git_url("git@github.com:org/repo.git") + ok, _err = validate_git_url("git@github.com:org/repo.git") assert ok is True def test_git_at_no_colon_invalid(self): - ok, err = validate_git_url("git@github.com/org/repo.git") + ok, _err = validate_git_url("git@github.com/org/repo.git") assert ok is False def test_ssh_valid(self): - ok, err = validate_git_url("ssh://git@github.com/org/repo.git") + ok, _err = validate_git_url("ssh://git@github.com/org/repo.git") assert ok is True def test_ssh_no_hostname_invalid(self): - ok, err = validate_git_url("ssh://") + ok, _err = validate_git_url("ssh://") assert ok is False def test_file_valid(self): - ok, err = validate_git_url("file:///tmp/repo") + ok, _err = validate_git_url("file:///tmp/repo") assert ok is True def test_file_empty_path_invalid(self): - ok, err = validate_git_url("file://") + ok, _err = validate_git_url("file://") assert ok is False def test_empty_url_invalid(self): @@ -79,11 +82,11 @@ def test_empty_url_invalid(self): assert "Empty" in err def test_ftp_invalid(self): - ok, err = validate_git_url("ftp://example.com/repo.git") + ok, _err = validate_git_url("ftp://example.com/repo.git") assert ok is False def test_none_invalid(self): - ok, err = validate_git_url(None) + ok, _err = validate_git_url(None) assert ok is False @@ -358,8 +361,7 @@ def _make_repo(self): ref.name = "origin/master" mock_git_repo.remotes.origin.refs = [ref] MockRepo.return_value = mock_git_repo - repo = _dtl_repo(mock_args, "/tmp/repo", mock_handle) - return repo + return _dtl_repo(mock_args, "/tmp/repo", mock_handle) def test_get_relative_path(self): repo = self._make_repo() @@ -547,8 +549,7 @@ def _make_repo(self, tmp_path): ref.name = "origin/master" mock_git_repo.remotes.origin.refs = [ref] MockRepo.return_value = mock_git_repo - repo = _dtl_repo(mock_args, str(tmp_path / "repo"), mock_handle) - return repo + return _dtl_repo(mock_args, str(tmp_path / "repo"), mock_handle) def test_get_devices_all_vendors(self, tmp_path): repo = self._make_repo(tmp_path) @@ -556,7 +557,7 @@ def test_get_devices_all_vendors(self, tmp_path): devices.mkdir() (devices / "Cisco").mkdir() (devices / "Juniper").mkdir() - files, vendors = repo.get_devices(str(devices)) + _files, vendors = repo.get_devices(str(devices)) assert len(vendors) == 2 assert any(v["name"] == "Cisco" for v in vendors) @@ -566,7 +567,7 @@ def test_get_devices_filters_vendors(self, tmp_path): devices.mkdir() (devices / "Cisco").mkdir() (devices / "Juniper").mkdir() - files, vendors = repo.get_devices(str(devices), vendors=["cisco"]) + _files, vendors = repo.get_devices(str(devices), vendors=["cisco"]) assert len(vendors) == 1 assert vendors[0]["name"] == "Cisco" @@ -576,7 +577,7 @@ def test_get_devices_skips_testing_folder(self, tmp_path): devices.mkdir() (devices / "Cisco").mkdir() (devices / "testing").mkdir() - files, vendors = repo.get_devices(str(devices)) + _files, vendors = repo.get_devices(str(devices)) assert not any(v["name"] == "testing" for v in vendors) @@ -596,8 +597,7 @@ def _make_repo(self, tmp_path): ref.name = "origin/master" mock_git_repo.remotes.origin.refs = [ref] MockRepo.return_value = mock_git_repo - repo = _dtl_repo(mock_args, str(tmp_path / "repo"), mock_handle) - return repo + return _dtl_repo(mock_args, str(tmp_path / "repo"), mock_handle) def test_discovers_vendors_from_single_path(self, tmp_path): """Test discovery from a single existing path.""" @@ -677,7 +677,7 @@ def test_handles_os_errors_gracefully(self, tmp_path): def mock_listdir(path): if "devices" in path: raise OSError("Permission denied") - elif "modules" in path: + if "modules" in path: return ["Cisco"] return [] @@ -1253,7 +1253,7 @@ def test_valid_existing_dir_returns_true(self, tmp_path): """Existing writable directory returns True.""" from core.repo import validate_repo_path - ok, msg = validate_repo_path(str(tmp_path)) + ok, _msg = validate_repo_path(str(tmp_path)) assert ok is True @@ -1268,6 +1268,7 @@ def test_parse_device_type_returns_error_when_normalize_fails(tmp_path): Covers repo.py lines 225-226: 'if err: return err'. """ from unittest.mock import patch + from core.repo import parse_single_file yaml_file = tmp_path / "test.yaml" @@ -1345,8 +1346,7 @@ def _make_repo(self): ref.name = "origin/master" mock_git_repo.remotes.origin.refs = [ref] MockRepo.return_value = mock_git_repo - repo = _dtl_repo(mock_args, "/tmp/repo", mock_handle) - return repo + return _dtl_repo(mock_args, "/tmp/repo", mock_handle) def test_get_racks_path_ends_with_rack_types(self): repo = self._make_repo() @@ -1398,8 +1398,7 @@ def _make_repo(self): ref.name = "origin/master" mock_git_repo.remotes.origin.refs = [ref] MockRepo.return_value = mock_git_repo - repo = _dtl_repo(mock_args, "/tmp/repo", mock_handle) - return repo + return _dtl_repo(mock_args, "/tmp/repo", mock_handle) def test_keyboard_interrupt_is_reraised(self): import pytest @@ -1434,8 +1433,7 @@ def _make_repo(self): ref.name = "origin/master" mock_git_repo.remotes.origin.refs = [ref] MockRepo.return_value = mock_git_repo - repo = _dtl_repo(mock_args, "/tmp/repo", mock_handle) - return repo + return _dtl_repo(mock_args, "/tmp/repo", mock_handle) def test_item_without_manufacturer_is_included_without_dedup(self): """Item missing 'manufacturer' key skips dedup and is appended as-is.""" @@ -1536,8 +1534,7 @@ def _make_repo(self): ref.name = "origin/master" mock_git_repo.remotes.origin.refs = [ref] MockRepo.return_value = mock_git_repo - repo = _dtl_repo(mock_args, "/tmp/repo", mock_handle) - return repo + return _dtl_repo(mock_args, "/tmp/repo", mock_handle) def test_returns_none_when_pickle_missing(self, tmp_path): """Returns None gracefully when the device pickle doesn't exist.""" diff --git a/tests/test_update_failure_resolver.py b/tests/test_update_failure_resolver.py index 93a4bf2fd..1185c0998 100644 --- a/tests/test_update_failure_resolver.py +++ b/tests/test_update_failure_resolver.py @@ -9,11 +9,10 @@ from core.update_failure_resolver import ( FailureKind, - extract_error_payload, classify_device_type_update_failure, + extract_error_payload, ) - SUBDEVICE_ROLE_ERROR_DICT = { "subdevice_role": [ "Must delete all device bay templates associated with this device before declassifying it as a parent device." From d457df038e52858f9e99c1e92ec529f831cbb12a Mon Sep 17 00:00:00 2001 From: Marcin Zieba Date: Wed, 9 Sep 2026 15:06:01 +0200 Subject: [PATCH 2/4] chore(lint): widen the ruff rule set and clear what it found Adds the rule families that catch bug classes rather than style: DTZ (timezone-aware datetimes), BLE (blind except), B (bugbear), S (bandit), ERA (commented-out code), PT (pytest style), plus SIM/RET/PERF/C4/RUF/ISC and ten families that were already clean and now stay that way (PIE, FLY, RSE, PGH, TID, INP, A, LOG, G). The changes it required: - Timestamps are timezone-aware. Duration arithmetic uses UTC, which no DST change can skew. Displayed timestamps use .astimezone(), so the operator still reads local wall-clock time; a test pins that, because reaching for UTC here would silently shift every logged line. - SystemExit and the DotDict AttributeError carry their cause. - pytest.raises patterns that were meant literally are escaped. They were matching as regexes, so "bad.url" and a trailing "." accepted more than the test intended. - The _real_library_root fixture is renamed: it returns a value the tests use, so the underscore was telling readers and linters otherwise. - One dead commented import removed, and a commented-out allowlist in repo.py replaced with a line saying why any HTTPS host is accepted. Deliberately not adopted, with the reason recorded in the config: UP042 (StrEnum changes __str__ output), ISC001 and COM812 (the formatter owns that), SIM117 and A002 in tests (style only). The twelve blind excepts are deliberate defensive boundaries around futures, pynetbox calls and YAML parsing. Narrowing them changes runtime behaviour and needs its own analysis, so each is marked with the reason it exists. The rule still applies to new code. ANN, PTH, FBT, TRY003/EM and the PLR refactor rules stay off: ~1000 findings between them, all churn or wide-radius refactors, none of them a bug class this code has hit. --- core/change_detector.py | 4 +-- core/compat.py | 2 +- core/component_cache.py | 7 ++--- core/export.py | 2 +- core/graphql_client.py | 5 +-- core/import_run.py | 6 ++-- core/log_handler.py | 2 +- core/nb_serializer.py | 4 +-- core/netbox_api.py | 23 ++++++-------- core/repo.py | 10 +++--- core/update_failure_resolver.py | 8 ++--- nb-dt-import.py | 34 ++++++++++----------- pyproject.toml | 30 ++++++++++++++++-- tests/integration/test_import.py | 18 +++++------ tests/test_component_cache.py | 7 +++-- tests/test_config.py | 5 ++- tests/test_graphql_client.py | 6 ++-- tests/test_import_run.py | 30 +++++++++--------- tests/test_log_handler.py | 22 ++++++++++++++ tests/test_nb_dt_import.py | 52 +++++++++++++++++--------------- tests/test_netbox_api.py | 6 ++-- tests/test_repo.py | 11 ++++--- tests/test_suite_hygiene.py | 9 ++++-- 23 files changed, 176 insertions(+), 127 deletions(-) diff --git a/core/change_detector.py b/core/change_detector.py index 06700f12d..57ca4d3fc 100644 --- a/core/change_detector.py +++ b/core/change_detector.py @@ -418,9 +418,9 @@ def _compare_components( # same as an empty list so chassis YAMLs that omit (e.g.) interfaces can # still drive cleanup of stale templates in NetBox. if yaml_key in yaml_data or self.remove_unmanaged_types: - for existing_name in existing_components.keys(): + for existing_name in existing_components: if existing_name not in yaml_component_names: - changes.append( + changes.append( # noqa: PERF401 ComponentChange( component_type=yaml_key, component_name=existing_name, diff --git a/core/compat.py b/core/compat.py index 8ebcbf03a..2b35eb0a0 100644 --- a/core/compat.py +++ b/core/compat.py @@ -19,7 +19,7 @@ def parse_netbox_version(version) -> tuple[int, int]: suffixes NetBox ships ("4.7.0-beta2"). """ raw = [int(re.sub(r"\D.*", "", part.strip()) or "0") for part in str(version).split(".")] - return tuple((raw + [0, 0])[:2]) # type: ignore[return-value] + return tuple(([*raw, 0, 0])[:2]) # type: ignore[return-value] def supports_module_bay_types(version) -> bool: diff --git a/core/component_cache.py b/core/component_cache.py index f314ecf79..61ddbe834 100644 --- a/core/component_cache.py +++ b/core/component_cache.py @@ -304,10 +304,7 @@ def get(self, endpoint_name, parent_type, parent_id, endpoint): if key in cached: return cached[key] - if parent_type == "device": - filter_kwargs = {"device_type_id": parent_id} - else: - filter_kwargs = {"module_type_id": parent_id} + filter_kwargs = {"device_type_id": parent_id} if parent_type == "device" else {"module_type_id": parent_id} result = {item.name: item for item in endpoint.filter(**filter_kwargs)} self.record(endpoint_name, parent_type, parent_id, result) return result @@ -378,7 +375,7 @@ def _finish_endpoint(self, endpoint_name, future): """Mark *endpoint_name* complete on the display, sizing the bar to the result.""" try: total = max(len(future.result()), 1) - except Exception: + except Exception: # a worker error must not break the progress display # noqa: BLE001 total = 1 self._job["display"].finish(endpoint_name, total) self._job["done"].add(endpoint_name) diff --git a/core/export.py b/core/export.py index 9c3dd853c..d08b2d4cc 100644 --- a/core/export.py +++ b/core/export.py @@ -3,6 +3,7 @@ Entry point: ``Exporter(config, handle, export_dir, force_overwrite, vendor_slugs).run()`` """ +import contextlib import hashlib import os import re @@ -32,7 +33,6 @@ ) from core.netbox_api import IMAGE_EXTENSIONS, _build_auth_header from core.repo import LIBRARY_TYPE_DIRS, library_dirs_present -import contextlib class _SkipSentinel: diff --git a/core/graphql_client.py b/core/graphql_client.py index 98b71d3cd..d66c376af 100644 --- a/core/graphql_client.py +++ b/core/graphql_client.py @@ -56,7 +56,8 @@ def __getattr__(self, key): try: value = self[key] except KeyError: - raise AttributeError(f"'DotDict' has no attribute '{key}'") + # The dict miss is an implementation detail of attribute lookup, not the error. + raise AttributeError(f"'DotDict' has no attribute '{key}'") from None if isinstance(value, dict) and not isinstance(value, DotDict): value = DotDict(value) self[key] = value @@ -114,7 +115,7 @@ def _response_body_detail(response): return "" try: body = response.text.strip() - except Exception: # pragma: no cover - a body that cannot be decoded is not worth failing on + except Exception: # pragma: no cover - a body that cannot be decoded is not worth failing on # noqa: BLE001 return "" if not body: return "" diff --git a/core/import_run.py b/core/import_run.py index 6bc35f528..ab770c1b3 100644 --- a/core/import_run.py +++ b/core/import_run.py @@ -4,7 +4,7 @@ from collections import Counter from contextlib import contextmanager from dataclasses import dataclass -from datetime import datetime, timedelta +from datetime import UTC, datetime, timedelta from typing import Any from core.change_detector import IMAGE_PROPERTIES, ChangeDetector, ChangeType @@ -66,7 +66,7 @@ def capture(cls, netbox, repo, started_at): outcome_counts=netbox.outcomes.summary_by_kind(), failure_lines=tuple(netbox.outcomes.render_failure_report()), duplicate_definitions=tuple(repo.duplicate_definitions), - elapsed=datetime.now() - started_at, + elapsed=datetime.now(UTC) - started_at, ) def outcome_count(self, kind, outcome): @@ -588,7 +588,7 @@ def __init__(self, config, repo, netbox, reporter, progress_factory, *, started_ self.netbox = netbox self.reporter = reporter self.progress_factory = progress_factory - self.started_at = started_at or datetime.now() + self.started_at = started_at or datetime.now(UTC) self.progress: Any = None self.task_registry = None self.vendor_task_id = None diff --git a/core/log_handler.py b/core/log_handler.py index 6c543abac..3842cd3c3 100644 --- a/core/log_handler.py +++ b/core/log_handler.py @@ -19,7 +19,7 @@ def __init__(self, verbose: bool): def _timestamp(self): """Return the current time formatted as HH:MM:SS.""" - return datetime.now().strftime("%H:%M:%S") + return datetime.now().astimezone().strftime("%H:%M:%S") def set_console(self, console): """Set the Rich Console instance used for output, or None to fall back to print().""" diff --git a/core/nb_serializer.py b/core/nb_serializer.py index 0e10446b7..9ca1094a6 100644 --- a/core/nb_serializer.py +++ b/core/nb_serializer.py @@ -108,9 +108,7 @@ def _should_include(field: str, val: Any) -> bool: return False if isinstance(val, str) and val == "": return False - if field in _OMIT_IF_EQUAL and val == _OMIT_IF_EQUAL[field]: - return False - return True + return not (field in _OMIT_IF_EQUAL and val == _OMIT_IF_EQUAL[field]) def _serialize_component(record: Any, fields: Sequence[str]) -> dict: diff --git a/core/netbox_api.py b/core/netbox_api.py index 237359088..02f2566c8 100644 --- a/core/netbox_api.py +++ b/core/netbox_api.py @@ -432,9 +432,6 @@ def _image_dir_for_yaml(src_file: str, src_segment: str, dst_segment: str) -> "P return Path(*parts) -# from pynetbox import RequestError as APIRequestError - - def _is_mapping_removal(prop_change): """Return True when a ``_mappings`` change only takes mappings away.""" return ( @@ -800,7 +797,7 @@ def _try_resolve_and_retry_device_type_update(self, dt, device_type, updates, er device_type_id=dt.id, device_type_yaml=device_type, ) - except Exception as exc: # defensive: classifier must never break the run + except Exception as exc: # defensive: classifier must never break the run # noqa: BLE001 self.handle.verbose_log(f"Failure classifier raised {type(exc).__name__}: {exc}") return False, None @@ -833,7 +830,7 @@ def _try_resolve_and_retry_device_type_update(self, dt, device_type, updates, er try: for step in resolution.remediation_steps: step() - except Exception as exc: + except Exception as exc: # a failed remediation is reported, not raised # noqa: BLE001 self.handle.log(f"Auto-resolve failed for {dt.model}: {exc}") return False, resolution @@ -1476,11 +1473,11 @@ def filter_new_module_types(module_types, all_module_types): Returns: list[dict]: Module types not found in *all_module_types*. """ - new_module_types = [] - for module_type in module_types: - if NetBox._find_existing_module_type(module_type, all_module_types) is None: - new_module_types.append(module_type) - return new_module_types + return [ + module_type + for module_type in module_types + if NetBox._find_existing_module_type(module_type, all_module_types) is None + ] def _log_module_property_diffs(self, mfr_slug, model, fields_info, component_changes=None): """Emit diff-u style lines for changed module type properties and component changes. @@ -2511,7 +2508,7 @@ def _create_generic( payload = extract_error_payload(excep.error) per_item = payload if isinstance(payload, list) and len(payload) == len(to_create) else [] reported = 0 - for item, error in zip(to_create, per_item): + for item, error in zip(to_create, per_item, strict=False): if error: reported += 1 self._log_component_error( @@ -3154,7 +3151,7 @@ def upload_images(self, baseurl, token, images, device_type): file_handles = {} try: for field, path in images.items(): - file_handles[field] = (os.path.basename(path), open(path, "rb")) + file_handles[field] = (os.path.basename(path), open(path, "rb")) # noqa: SIM115 response = requests.patch( url, headers=headers, @@ -3172,7 +3169,7 @@ def upload_images(self, baseurl, token, images, device_type): except OSError as e: self.handle.log(f"Error reading image file for device type {device_type}: {e}") finally: - for _, (_, fh) in file_handles.items(): + for _, fh in file_handles.values(): with suppress(Exception): fh.close() diff --git a/core/repo.py b/core/repo.py index 11c47ceb8..4c368911a 100644 --- a/core/repo.py +++ b/core/repo.py @@ -110,7 +110,7 @@ def _vendor_slugs_from_index( return None try: entries = _safe_index_load(index_path) - except Exception: + except Exception: # an unreadable index or definition is skipped, not fatal # noqa: BLE001 return None result = set() for model_name, vendor_dir in entries: @@ -216,9 +216,7 @@ def validate_git_url(url): if url.startswith("https://"): parsed = urlparse(url) if parsed.scheme == "https" and parsed.hostname: - # Optional: enforce an allowlist if desired - # if parsed.hostname not in ("github.com", "gitlab.com"): - # return False, f"Host not allowed: {parsed.hostname}" + # Any HTTPS host is accepted: the library is forked and self-hosted widely. return True, None return False, "Invalid HTTPS URL" @@ -451,7 +449,7 @@ def parse_single_file(file): return data except yaml.YAMLError as excep: return f"Error: {excep}" - except Exception as e: + except Exception as e: # an unreadable index or definition is skipped, not fatal # noqa: BLE001 return f"Error: {e}" @@ -703,7 +701,7 @@ def resolve_slug_files(self, slugs): device_files: dict = {} # vendor_slug -> [abs_path] try: known_slugs = _safe_index_load(device_index) - except Exception: + except Exception: # an unreadable index or definition is skipped, not fatal # noqa: BLE001 return None for entry_slug, relpath in known_slugs: diff --git a/core/update_failure_resolver.py b/core/update_failure_resolver.py index 54bb6f104..205610e43 100644 --- a/core/update_failure_resolver.py +++ b/core/update_failure_resolver.py @@ -92,7 +92,7 @@ def extract_error_payload(error: Any) -> Any: if isinstance(error, (bytes, bytearray)): try: error = error.decode("utf-8", errors="replace") - except Exception: + except Exception: # pynetbox raises many types; a probe failure is not fatal # noqa: BLE001 return error if isinstance(error, str): try: @@ -136,7 +136,7 @@ def _count_dependent_devices(netbox: Any, device_type_id: int) -> tuple[int, lis filter_kwargs = {"device_type_id": device_type_id} try: devices = list(netbox.dcim.devices.filter(**filter_kwargs, limit=5)) - except Exception: + except Exception: # pynetbox raises many types; a probe failure is not fatal # noqa: BLE001 return -1, [] sample = [getattr(d, "name", None) or str(getattr(d, "id", "?")) for d in devices[:5]] if len(devices) < 5: @@ -144,7 +144,7 @@ def _count_dependent_devices(netbox: Any, device_type_id: int) -> tuple[int, lis # We capped at limit=5; query the real total separately. try: total = netbox.dcim.devices.count(**filter_kwargs) - except Exception: + except Exception: # pynetbox raises many types; a probe failure is not fatal # noqa: BLE001 total = len(devices) return total, sample @@ -161,7 +161,7 @@ def _list_device_bay_templates(netbox: Any, device_type_id: int) -> list[Any] | """ try: return list(netbox.dcim.device_bay_templates.filter(device_type_id=device_type_id)) - except Exception: + except Exception: # pynetbox raises many types; a probe failure is not fatal # noqa: BLE001 return None diff --git a/nb-dt-import.py b/nb-dt-import.py index 61946e167..cbd5cb68f 100644 --- a/nb-dt-import.py +++ b/nb-dt-import.py @@ -3,7 +3,7 @@ import sys from contextlib import contextmanager -from datetime import datetime +from datetime import UTC, datetime import requests from pynetbox.core.query import RequestError as NetBoxRequestError @@ -80,10 +80,7 @@ def _effective_speed(task, primary_attr): def render(self, task): """Render the current or finished speed.""" - if task.finished: - speed = self._effective_speed(task, "finished_speed") - else: - speed = self._effective_speed(task, "speed") + speed = self._effective_speed(task, "finished_speed") if task.finished else self._effective_speed(task, "speed") if speed is None: return Text("- it/s") return Text(f"{speed:.1f} it/s") @@ -135,7 +132,7 @@ def _run_export_diff(config: RunConfig, handle): def _run(config: RunConfig): """Build and execute the selected run pipeline.""" - started_at = datetime.now() + started_at = datetime.now(UTC) handle = LogHandler(config.verbose) for notice in config.notices: handle.log(notice) @@ -161,40 +158,41 @@ def main(): try: config = resolve_run_config() except ConfigError as exc: - raise SystemExit(str(exc)) + raise SystemExit(str(exc)) from exc try: return _run(config) except FatalError as exc: if config.verbose and exc.formatted_traceback: print(exc.formatted_traceback, end="") - raise SystemExit(str(exc)) + raise SystemExit(str(exc)) from exc except requests.exceptions.ConnectionError as exc: + detail = _fmt_connection_error(config.netbox_url, exc) print( - f"[{datetime.now().strftime('%H:%M:%S')}] Error: {_fmt_connection_error(config.netbox_url, exc)}", + f"[{datetime.now().astimezone().strftime('%H:%M:%S')}] Error: {detail}", file=sys.stderr, ) - raise SystemExit(1) + raise SystemExit(1) from exc except GraphQLError as exc: print( - f"[{datetime.now().strftime('%H:%M:%S')}] Error: NetBox GraphQL request failed — {exc}\n" - f"[{datetime.now().strftime('%H:%M:%S')}] This may be a temporary connectivity issue. " + f"[{datetime.now().astimezone().strftime('%H:%M:%S')}] Error: NetBox GraphQL request failed — {exc}\n" + f"[{datetime.now().astimezone().strftime('%H:%M:%S')}] This may be a temporary connectivity issue. " "Check that NetBox is reachable and try again.", file=sys.stderr, ) - raise SystemExit(1) + raise SystemExit(1) from exc except NetBoxRequestError as exc: print( - f"[{datetime.now().strftime('%H:%M:%S')}] Error: NetBox REST API request failed — {exc}\n" - f"[{datetime.now().strftime('%H:%M:%S')}] Check that NetBox is reachable and" + f"[{datetime.now().astimezone().strftime('%H:%M:%S')}] Error: NetBox REST API request failed — {exc}\n" + f"[{datetime.now().astimezone().strftime('%H:%M:%S')}] Check that NetBox is reachable and" " the API token has the required permissions.", file=sys.stderr, ) - raise SystemExit(1) + raise SystemExit(1) from exc if __name__ == "__main__": try: main() except KeyboardInterrupt: - print(f"[{datetime.now().strftime('%H:%M:%S')}] Interrupted by user (Ctrl-C). Exiting.") - raise SystemExit(130) + print(f"[{datetime.now().astimezone().strftime('%H:%M:%S')}] Interrupted by user (Ctrl-C). Exiting.") + raise SystemExit(130) from None diff --git a/pyproject.toml b/pyproject.toml index 99ff40817..c1cad3ccf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,8 +31,25 @@ dev = [ line-length = 120 [tool.ruff.lint] -select = ["E", "F", "W", "D", "C901"] +select = [ + "E", "F", "W", "D", "C901", + "I", # isort + "UP", # pyupgrade; safe at the 3.12 floor + "B", # bugbear + "DTZ", # datetimes must carry a timezone + "BLE", # no bare `except Exception` + "S", # bandit + "ERA", # no commented-out code + "SIM", "RET", "PERF", "C4", "RUF", "ISC", + "PIE", "FLY", "RSE", "PGH", "TID", "INP", "A", "LOG", "G", + "PT", # pytest style +] ignore = [ + # Conflicts with the formatter, which owns implicit string concatenation. + "ISC001", + # StrEnum.__str__ returns the bare value where (str, Enum) returns ClassName.MEMBER, + # so the rewrite would change logged and serialized output. + "UP042", # Docstring style choices (pick D211+D212 convention) "D203", # no-blank-line-before-class (incompatible with D211) "D213", # multi-line-summary-second-line (incompatible with D212) @@ -52,7 +69,16 @@ ignore = [ max-complexity = 15 [tool.ruff.lint.per-file-ignores] -"tests/**" = ["D102", "D103"] # test functions don't need docstrings +"tests/**" = [ + "D102", "D103", # test functions don't need docstrings + "S101", # asserting is what a test does + "S105", "S106", "S107", # fixture credentials are not secrets + "S108", "S603", "S607", # tmp paths and subprocess calls are the test's subject + "INP001", # tests/ is not an importable package + "ERA001", # commented sample payloads document the case under test + "SIM117", # nested `with` keeps one patch per line and reads better here + "A002", # `id` mirrors the NetBox field name the stub stands in for +] [tool.mypy] python_version = "3.12" diff --git a/tests/integration/test_import.py b/tests/integration/test_import.py index 429717d81..961697e53 100644 --- a/tests/integration/test_import.py +++ b/tests/integration/test_import.py @@ -19,29 +19,29 @@ Test scenarios -------------- A. All component types created with correct field values - – interfaces (including mgmt_only), power-ports (draw values), + - interfaces (including mgmt_only), power-ports (draw values), console-ports, console-server-ports, power-outlets (power_port link), rear-ports (positions), front-ports (M2M rear_ports), device-bays, module-bays (position). B. Device-type properties stored correctly - – u_height (decimal), is_full_depth (bool), weight, weight_unit, + - u_height (decimal), is_full_depth (bool), weight, weight_unit, airflow, part_number, comments. C. Image linkage - – front_image and rear_image URLs are set on the device type (not just + - front_image and rear_image URLs are set on the device type (not just "uploaded" as orphan files) and the URLs return HTTP 200. D. GraphQL schema consistency - – Query every device-type schema field and every + - Query every device-type schema field and every registry field directly through the GraphQL client so a removed/renamed schema field raises an explicit error rather than a silent false-positive. E. Front-port multi-position linkage - – FP1 → RP1 position 1; FP2 → RP1 position 2 (same rear port). + - FP1 → RP1 position 1; FP2 → RP1 position 2 (same rear port). F. Module-type component creation - – All module component types created; front-port rear_port mapping set. + - All module component types created; front-port rear_port mapping set. G. Idempotency - – Second run: 0 new, 0 modified device types and module types. + - Second run: 0 new, 0 modified device types and module types. H. Update mode - – Delete one interface via REST API; re-run with --update; verify it is + - Delete one interface via REST API; re-run with --update; verify it is recreated with the original type value. Usage:: @@ -426,7 +426,7 @@ def test_graphql_schema() -> None: expected_fields = component.graphql_fields try: records = client.get_component_templates(endpoint_name) - except Exception as exc: + except Exception as exc: # the test reports whatever the client raised # noqa: BLE001 fail( f"get_component_templates({endpoint_name!r}) raised {type(exc).__name__}: {exc} — " f"likely a GraphQL schema error (field removed or renamed)." diff --git a/tests/test_component_cache.py b/tests/test_component_cache.py index 181d77df1..49387cb34 100644 --- a/tests/test_component_cache.py +++ b/tests/test_component_cache.py @@ -7,6 +7,7 @@ import threading import time +from typing import ClassVar import pytest @@ -349,7 +350,7 @@ def test_each_prefetch_worker_owns_and_closes_its_graphql_client(self): """Concurrent endpoint requests must not share a requests session.""" class WorkerClient: - instances = [] + instances: ClassVar[list] = [] def __init__(self, *args, **kwargs): self.thread_id = None @@ -377,7 +378,7 @@ def test_close_waits_for_workers_and_closes_their_graphql_clients(self): """Cancelling a prefetch must not leak the sessions of workers already running.""" class WorkerClient: - instances = [] + instances: ClassVar[list] = [] started = threading.Event() def __init__(self, *args, **kwargs): @@ -417,7 +418,7 @@ def test_a_prefetch_cannot_be_collected_for_another_vendor(self): cache = make_cache() cache.begin_prefetch(manufacturer_slug="cisco") - with pytest.raises(ValueError, match="cisco.*juniper"): + with pytest.raises(ValueError, match=r"cisco.*juniper"): cache.ensure_ready(manufacturer_slug="juniper") cache.close() diff --git a/tests/test_config.py b/tests/test_config.py index e289960c8..0c77c2493 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -2,6 +2,7 @@ import os import pathlib +import re import subprocess import sys @@ -92,7 +93,9 @@ class TestRequiredVariables: """Required environment values fail with a typed configuration error.""" def test_one_missing_variable_uses_the_catalogue_error(self): - with pytest.raises(EnvironmentVariableError, match='Environment variable "NETBOX_TOKEN" is not set.'): + with pytest.raises( + EnvironmentVariableError, match=re.escape('Environment variable "NETBOX_TOKEN" is not set.') + ): resolve_run_config(argv=[], env={"NETBOX_URL": "http://netbox.local"}) def test_all_missing_variables_use_one_typed_error(self): diff --git a/tests/test_graphql_client.py b/tests/test_graphql_client.py index 9eab667eb..cd1fbf32f 100644 --- a/tests/test_graphql_client.py +++ b/tests/test_graphql_client.py @@ -2476,12 +2476,12 @@ def test_vendor_scoped_two_queries(self, mock_post): # Verify both filters were applied via GraphQL variables (not string interpolation) calls = mock_post.call_args_list - # calls[0]: device_type filter query – data page + # calls[0]: device_type filter query - data page device_payload = calls[0][1]["json"] device_query = device_payload["query"] device_vars = device_payload["variables"] - # calls[1]: device_type filter query – empty terminator (pagination ends) - # calls[2]: module_type filter query – data page + # calls[1]: device_type filter query - empty terminator (pagination ends) + # calls[2]: module_type filter query - data page module_payload = calls[2][1]["json"] module_query = module_payload["query"] module_vars = module_payload["variables"] diff --git a/tests/test_import_run.py b/tests/test_import_run.py index def1d1a51..2a242dce3 100644 --- a/tests/test_import_run.py +++ b/tests/test_import_run.py @@ -434,31 +434,31 @@ def __init__(self, *, counter=None): from core.outcomes import OutcomeRegistry self.outcomes = OutcomeRegistry() - base = dict( - added=0, - properties_updated=0, - components_updated=0, - components_added=0, - components_removed=0, - images=0, - manufacturer=0, - module_added=0, - module_updated=0, - rack_type_added=0, - rack_type_updated=0, - ) + base = { + "added": 0, + "properties_updated": 0, + "components_updated": 0, + "components_added": 0, + "components_removed": 0, + "images": 0, + "manufacturer": 0, + "module_added": 0, + "module_updated": 0, + "rack_type_added": 0, + "rack_type_updated": 0, + } base.update(counter or {}) self.counter = Counter(base) def _summary_lines(netbox): """Render a run summary through the real LogHandler and return its lines.""" - from datetime import datetime + from datetime import UTC, datetime handle, console = _recording_handle() handle.verbose = True repo = SimpleNamespace(duplicate_definitions=[]) - summary = RunSummary.capture(netbox, repo, datetime.now()) + summary = RunSummary.capture(netbox, repo, datetime.now(UTC)) _log_run_summary(handle, summary) return console.lines diff --git a/tests/test_log_handler.py b/tests/test_log_handler.py index 4e02de920..1d54245b8 100644 --- a/tests/test_log_handler.py +++ b/tests/test_log_handler.py @@ -166,3 +166,25 @@ def test_progress_group_supports_nested_blocks(): handle.end_progress_group() print_mock.assert_called_once_with("[12:00:00] Nested message") + + +class TestTimestampStaysOnLocalWallClock: + """The timestamp is timezone-aware now, which must not move what the operator reads to UTC.""" + + def test_the_timestamp_reads_local_time_not_utc(self, monkeypatch): + """A UTC-based fix for DTZ005 would silently shift every logged line by the local offset.""" + import time + from datetime import UTC, datetime + + monkeypatch.setenv("TZ", "Asia/Tokyo") + time.tzset() + try: + before = datetime.now().astimezone().strftime("%H:%M:%S") + stamp = LogHandler(False)._timestamp() + after = datetime.now().astimezone().strftime("%H:%M:%S") + + assert stamp in {before, after}, "the timestamp must follow the local clock" + assert stamp != datetime.now(UTC).strftime("%H:%M:%S"), "Asia/Tokyo is UTC+9, so UTC would differ" + finally: + monkeypatch.undo() + time.tzset() diff --git a/tests/test_nb_dt_import.py b/tests/test_nb_dt_import.py index 896b829e3..bb3af27d0 100644 --- a/tests/test_nb_dt_import.py +++ b/tests/test_nb_dt_import.py @@ -312,14 +312,14 @@ def test_items_per_second_column_uses_elapsed_fallback_when_finished_speed_missi @pytest.fixture(scope="session", autouse=True) -def _real_library_root(tmp_path_factory): +def real_library_root(tmp_path_factory): """Point the repo mocks at a real library tree: the pipeline stats these paths before reading them.""" global _LIBRARY_ROOT root = tmp_path_factory.mktemp("library") for name in ("device-types", "module-types", "rack-types"): (root / name).mkdir() _LIBRARY_ROOT = root - yield root + return root def _make_mock_repo(device_types=None): @@ -479,7 +479,7 @@ def test_explicit_total_skips_update_in_finally(self, nb_dt_import): # --------------------------------------------------------------------------- -# filter_device_types_by_change_keys – empty-keys branch +# filter_device_types_by_change_keys - empty-keys branch # --------------------------------------------------------------------------- @@ -494,7 +494,7 @@ def test_empty_change_keys_returns_empty_list(self, nb_dt_import): # --------------------------------------------------------------------------- -# select_device_types_* – None report branches +# select_device_types_* - None report branches # --------------------------------------------------------------------------- @@ -558,15 +558,17 @@ def test_cleanup_on_exception(self, nb_dt_import): mock_progress.add_task.return_value = 1 mock_dt = MagicMock() - with pytest.raises(ValueError): - with import_run_module._image_progress_scope(mock_progress, mock_dt, total=3): - raise ValueError("boom") + with ( + pytest.raises(ValueError, match="boom"), + import_run_module._image_progress_scope(mock_progress, mock_dt, total=3), + ): + raise ValueError("boom") assert mock_dt._image_progress is None # --------------------------------------------------------------------------- -# main() – comprehensive branch coverage +# main() - comprehensive branch coverage # --------------------------------------------------------------------------- @@ -1351,7 +1353,7 @@ class TestLogRunSummary: def test_rack_types_counters_are_logged(self, nb_dt_import): """When netbox.rack_types is True, rack_type_added/updated counters are logged.""" - from datetime import datetime + from datetime import UTC, datetime handle = MagicMock() mock_nb = MagicMock() @@ -1372,7 +1374,7 @@ def test_rack_types_counters_are_logged(self, nb_dt_import): ) mock_nb.outcomes.render_failure_report.return_value = [] mock_repo = SimpleNamespace(duplicate_definitions=[]) - summary = import_run_module.RunSummary.capture(mock_nb, mock_repo, datetime.now()) + summary = import_run_module.RunSummary.capture(mock_nb, mock_repo, datetime.now(UTC)) import_run_module._log_run_summary(handle, summary) @@ -1382,7 +1384,7 @@ def test_rack_types_counters_are_logged(self, nb_dt_import): def test_duplicate_definitions_are_logged(self, nb_dt_import): """When dtl_repo has duplicate_definitions, each entry is logged with kept/ignored.""" - from datetime import datetime + from datetime import UTC, datetime handle = MagicMock() mock_nb = MagicMock() @@ -1410,7 +1412,7 @@ def test_duplicate_definitions_are_logged(self, nb_dt_import): } ] mock_nb.outcomes.render_failure_report.return_value = [] - summary = import_run_module.RunSummary.capture(mock_nb, mock_repo, datetime.now()) + summary = import_run_module.RunSummary.capture(mock_nb, mock_repo, datetime.now(UTC)) import_run_module._log_run_summary(handle, summary) @@ -2031,7 +2033,7 @@ def test_import_run_stops_preload_in_finally_on_error(self, make_config): ) netbox = _make_mock_netbox() - with ( + with ( # noqa: PT012 patch("core.import_run._process_device_types", side_effect=RuntimeError("boom")), patch("core.import_run._finalize_task_registry") as mock_finalize, pytest.raises(RuntimeError, match="boom"), @@ -2124,9 +2126,9 @@ class TestExportDiffVendorFilterEndToEnd: """ def test_single_vendor_reaches_the_graphql_request( - self, nb_dt_import, monkeypatch, tmp_path, capsys, mock_post, _real_library_root + self, nb_dt_import, monkeypatch, tmp_path, capsys, mock_post, real_library_root ): - _run_export_diff_cli(nb_dt_import, monkeypatch, tmp_path, _real_library_root, "Cisco") + _run_export_diff_cli(nb_dt_import, monkeypatch, tmp_path, real_library_root, "Cisco") payloads = [call.kwargs["json"] for call in mock_post.call_args_list] filters = _filters_by_list_field(payloads) @@ -2137,10 +2139,10 @@ def test_single_vendor_reaches_the_graphql_request( assert "Nothing to export" in capsys.readouterr().out def test_multiple_vendors_are_sent_as_a_json_list( - self, nb_dt_import, monkeypatch, tmp_path, capsys, mock_post, _real_library_root + self, nb_dt_import, monkeypatch, tmp_path, capsys, mock_post, real_library_root ): """A tuple would not compare equal here, and NetBox would not accept it as a list variable.""" - _run_export_diff_cli(nb_dt_import, monkeypatch, tmp_path, _real_library_root, "Cisco,Juniper") + _run_export_diff_cli(nb_dt_import, monkeypatch, tmp_path, real_library_root, "Cisco,Juniper") payloads = [call.kwargs["json"] for call in mock_post.call_args_list] filters = _filters_by_list_field(payloads) @@ -2151,10 +2153,10 @@ def test_multiple_vendors_are_sent_as_a_json_list( assert "Nothing to export" in capsys.readouterr().out def test_no_vendors_queries_every_manufacturer( - self, nb_dt_import, monkeypatch, tmp_path, capsys, mock_post, _real_library_root + self, nb_dt_import, monkeypatch, tmp_path, capsys, mock_post, real_library_root ): """Without --vendors, config.vendors is (), which the GraphQL layer rejects if it reaches it.""" - _run_export_diff_cli(nb_dt_import, monkeypatch, tmp_path, _real_library_root) + _run_export_diff_cli(nb_dt_import, monkeypatch, tmp_path, real_library_root) payloads = [call.kwargs["json"] for call in mock_post.call_args_list] filters = _filters_by_list_field(payloads) @@ -2202,11 +2204,11 @@ def log_message(self, *args): threading.Thread(target=server.serve_forever, daemon=True).start() return f"http://127.0.0.1:{server.server_port}", server, bodies - def test_vendor_filter_is_serialized_as_a_json_list(self, nb_dt_import, monkeypatch, tmp_path, _real_library_root): + def test_vendor_filter_is_serialized_as_a_json_list(self, nb_dt_import, monkeypatch, tmp_path, real_library_root): url, server, bodies = self._serve() monkeypatch.setenv("NETBOX_URL", url) try: - _run_export_diff_cli(nb_dt_import, monkeypatch, tmp_path, _real_library_root, "Cisco,Juniper") + _run_export_diff_cli(nb_dt_import, monkeypatch, tmp_path, real_library_root, "Cisco,Juniper") finally: server.shutdown() server.server_close() @@ -2228,21 +2230,21 @@ def _queries_for_version(self, nb_dt_import, monkeypatch, tmp_path, library_root return [payload["query"] for payload in bodies] def test_a_47_server_is_asked_for_the_module_bay_type_relation( - self, nb_dt_import, monkeypatch, tmp_path, _real_library_root + self, nb_dt_import, monkeypatch, tmp_path, real_library_root ): """Not selecting it exports every bay without its restriction, silently. Asserted on the module-type query, which this run always issues; the component queries only run once there is something to export. """ - queries = self._queries_for_version(nb_dt_import, monkeypatch, tmp_path, _real_library_root, "4.7.0") + queries = self._queries_for_version(nb_dt_import, monkeypatch, tmp_path, real_library_root, "4.7.0") module_types = [q for q in queries if "module_type_list(" in q] assert module_types assert all("module_bay_types" in q for q in module_types) - def test_an_older_server_is_never_asked_for_it(self, nb_dt_import, monkeypatch, tmp_path, _real_library_root): + def test_an_older_server_is_never_asked_for_it(self, nb_dt_import, monkeypatch, tmp_path, real_library_root): """Selecting a field the schema lacks fails the whole query.""" - queries = self._queries_for_version(nb_dt_import, monkeypatch, tmp_path, _real_library_root, "4.6.9") + queries = self._queries_for_version(nb_dt_import, monkeypatch, tmp_path, real_library_root, "4.6.9") assert not [q for q in queries if "module_bay_types" in q] diff --git a/tests/test_netbox_api.py b/tests/test_netbox_api.py index d7236b78d..78e86d986 100644 --- a/tests/test_netbox_api.py +++ b/tests/test_netbox_api.py @@ -4628,7 +4628,7 @@ class TestVerifyCompatibility: """Tests for NetBox.verify_compatibility() version thresholds.""" @pytest.mark.parametrize( - "version_str, expected_m2m, expected_module_bay_types", + ("version_str", "expected_m2m", "expected_module_bay_types"), [ ("4.3", False, False), ("4.4", False, False), @@ -7199,13 +7199,13 @@ class TestSummaryWordingMatchesTheFailedOperation: def _summary_text(self, nb): """Render the real end-of-run summary for *nb* and return it as one string.""" - from datetime import datetime + from datetime import UTC, datetime from types import SimpleNamespace from core.import_run import RunSummary, _log_run_summary handle, console = recording_handle() - summary = RunSummary.capture(nb, SimpleNamespace(duplicate_definitions=[]), datetime.now()) + summary = RunSummary.capture(nb, SimpleNamespace(duplicate_definitions=[]), datetime.now(UTC)) _log_run_summary(handle, summary) return "\n".join(console.lines) diff --git a/tests/test_repo.py b/tests/test_repo.py index 95e9e9160..7d1dee359 100644 --- a/tests/test_repo.py +++ b/tests/test_repo.py @@ -1,4 +1,5 @@ import os +import re from unittest.mock import MagicMock, call, mock_open, patch import pytest @@ -146,7 +147,7 @@ def test_invalid_url_raises_before_the_clone_path(self): mock_args.repo_url = "ftp://bad.url" mock_args.repo_branch = "master" with _clone_present(False), patch("core.repo.Repo"): - with pytest.raises(InvalidGitURLError, match="Invalid Git URL: ftp://bad.url"): + with pytest.raises(InvalidGitURLError, match=re.escape("Invalid Git URL: ftp://bad.url")): _dtl_repo(mock_args, "/tmp/repo", LogHandler(False)) def test_invalid_path_raises_before_repository_access(self, tmp_path): @@ -258,7 +259,7 @@ class TestDTLRepoRealGit: @pytest.fixture(autouse=True) def mock_git_repo(self): """Override the global autouse git mock so these tests exercise real git.""" - yield None + return @pytest.fixture(autouse=True) def clear_ambient_git_env(self, monkeypatch): @@ -475,7 +476,7 @@ def test_repo_open_git_error_uses_the_configured_url_without_reading_self_repo(s failure = git_exc.GitCommandError("status", 1) with _clone_present(), patch("core.repo.Repo", side_effect=failure): - with pytest.raises(GitCommandError, match="https://example.invalid/repo.git") as exc_info: + with pytest.raises(GitCommandError, match=re.escape("https://example.invalid/repo.git")) as exc_info: _dtl_repo(mock_args, "/tmp/repo", LogHandler(False)) assert "cmdline: status" in exc_info.value.formatted_traceback @@ -501,7 +502,9 @@ def test_invalid_git_repository_raises_its_catalogue_error(self): invalid = git_exc.InvalidGitRepositoryError("/tmp/repo") with _clone_present(), patch("core.repo.Repo", side_effect=invalid): - with pytest.raises(GitInvalidRepositoryError, match='The repo "/tmp/repo" is not a valid git repo.'): + with pytest.raises( + GitInvalidRepositoryError, match=re.escape('The repo "/tmp/repo" is not a valid git repo.') + ): _dtl_repo(mock_args, "/tmp/repo", LogHandler(False)) diff --git a/tests/test_suite_hygiene.py b/tests/test_suite_hygiene.py index e85d411ec..c456b12be 100644 --- a/tests/test_suite_hygiene.py +++ b/tests/test_suite_hygiene.py @@ -39,9 +39,12 @@ def _orphaned_docstrings(tree): if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): continue for statement in node.body[1:]: - if isinstance(statement, ast.Expr) and isinstance(statement.value, ast.Constant): - if isinstance(statement.value.value, str): - yield node.name, statement.lineno + if ( + isinstance(statement, ast.Expr) + and isinstance(statement.value, ast.Constant) + and isinstance(statement.value.value, str) + ): + yield node.name, statement.lineno def test_the_scan_follows_pytest_discovery(tmp_path): From feecfe7b936765759392a415bdf44327afcb21c7 Mon Sep 17 00:00:00 2001 From: Marcin Zieba Date: Wed, 9 Sep 2026 16:30:54 +0200 Subject: [PATCH 3/4] fix(import): normalize a naive started_at at construction ImportRun stored whatever started_at it was given. RunSummary.capture then subtracts it from datetime.now(UTC), so a naive value raised TypeError at the end of the run rather than at the boundary that accepted it. Attach the local timezone when the value is naive, which is what Python means by a naive datetime, and leave an aware value untouched. The field only feeds elapsed-time reporting, so rejecting a plain datetime.now() would cost callers more than the conversion does. Also replaces the endpoint sampling in the log timestamp test with a frozen clock. It read the clock three times and asserted the middle read matched one of the outer two, which a scheduler delay between reads could break. The stub models a real clock (naive now() is local, now(tz) converts the same instant) rather than using replace(), which reinterprets the wall clock and made the test pass against a UTC implementation too. Review findings: https://github.com/marcinpsk/Device-Type-Library-Import/pull/137#discussion_r3969536115 https://github.com/marcinpsk/Device-Type-Library-Import/pull/137#discussion_r3969536124 --- core/import_run.py | 10 ++++++-- tests/test_import_run.py | 51 +++++++++++++++++++++++++++++++++++++++ tests/test_log_handler.py | 19 +++++++++++---- 3 files changed, 73 insertions(+), 7 deletions(-) diff --git a/core/import_run.py b/core/import_run.py index ab770c1b3..c99a6ef05 100644 --- a/core/import_run.py +++ b/core/import_run.py @@ -16,6 +16,11 @@ _PROGRESS_DESC_WIDTH = 28 +def _as_aware(moment): + """Return *moment* with a timezone attached; a naive value means local time.""" + return moment if moment.tzinfo is not None else moment.astimezone() + + @dataclass(frozen=True) class RunSelection: """Selected vendors and source paths for one import run.""" @@ -579,7 +584,7 @@ def __init__(self, config, repo, netbox, reporter, progress_factory, *, started_ netbox (NetBox): Connected NetBox interface. reporter (LogHandler): Run message sink. progress_factory: Context manager factory for the Rich progress display. - started_at (datetime | None): Start time used for elapsed-time reporting. + started_at (datetime | None): Start time for elapsed-time reporting; naive means local. """ if not isinstance(config, RunConfig): raise TypeError("config must be a RunConfig") @@ -588,7 +593,8 @@ def __init__(self, config, repo, netbox, reporter, progress_factory, *, started_ self.netbox = netbox self.reporter = reporter self.progress_factory = progress_factory - self.started_at = started_at or datetime.now(UTC) + # Normalize here: a naive value would otherwise survive the run and raise in capture(). + self.started_at = _as_aware(started_at) if started_at is not None else datetime.now(UTC) self.progress: Any = None self.task_registry = None self.vendor_task_id = None diff --git a/tests/test_import_run.py b/tests/test_import_run.py index 2a242dce3..26b093527 100644 --- a/tests/test_import_run.py +++ b/tests/test_import_run.py @@ -533,3 +533,54 @@ def test_headline_partials_match_the_itemised_rows(): assert "2 device types partially updated" in text assert "1 modules partially updated" in text assert "Partial updates: 3" in text + + +class TestNaiveStartedAtIsNormalized: + """A naive started_at used to survive construction and fail much later, in the summary.""" + + def test_a_naive_started_at_is_stored_timezone_aware(self, make_config, tmp_path): + from datetime import datetime + + run = ImportRun( + make_config(), + _RepositoryBoundary(tmp_path), + _NetBoxBoundary(), + LogHandler(False), + _ProgressFactory(), + started_at=datetime(2020, 1, 1, 12, 0, 0), # noqa: DTZ001 - a naive value is the subject of this test + ) + + assert run.started_at.tzinfo is not None, "a naive value cannot be subtracted from an aware now()" + + def test_a_run_started_naive_still_produces_a_summary(self, make_config, tmp_path): + """The real failure: RunSummary.capture subtracts started_at from an aware now().""" + from datetime import datetime + + run = ImportRun( + make_config(), + _RepositoryBoundary(tmp_path), + _NetBoxBoundary(), + LogHandler(False), + _ProgressFactory(), + started_at=datetime(2020, 1, 1, 12, 0, 0), # noqa: DTZ001 - a naive value is the subject of this test + ) + + summary = RunSummary.capture(_NetBoxBoundary(), SimpleNamespace(duplicate_definitions=[]), run.started_at) + + assert summary.elapsed.total_seconds() > 0 + + def test_an_aware_started_at_is_left_alone(self, make_config, tmp_path): + from datetime import UTC, datetime + + given = datetime(2020, 1, 1, 12, 0, 0, tzinfo=UTC) + + run = ImportRun( + make_config(), + _RepositoryBoundary(tmp_path), + _NetBoxBoundary(), + LogHandler(False), + _ProgressFactory(), + started_at=given, + ) + + assert run.started_at == given diff --git a/tests/test_log_handler.py b/tests/test_log_handler.py index 1d54245b8..5b5d9d7fa 100644 --- a/tests/test_log_handler.py +++ b/tests/test_log_handler.py @@ -179,12 +179,21 @@ def test_the_timestamp_reads_local_time_not_utc(self, monkeypatch): monkeypatch.setenv("TZ", "Asia/Tokyo") time.tzset() try: - before = datetime.now().astimezone().strftime("%H:%M:%S") - stamp = LogHandler(False)._timestamp() - after = datetime.now().astimezone().strftime("%H:%M:%S") + # One absolute instant: 18:04:05 UTC is 03:04:05 the next day in Tokyo. + instant = datetime(2026, 1, 1, 18, 4, 5, tzinfo=UTC) - assert stamp in {before, after}, "the timestamp must follow the local clock" - assert stamp != datetime.now(UTC).strftime("%H:%M:%S"), "Asia/Tokyo is UTC+9, so UTC would differ" + class _FrozenClock: + """Models a real clock: naive now() is local, now(tz) converts the same instant.""" + + @staticmethod + def now(tz=None): + if tz is not None: + return instant.astimezone(tz) + return instant.astimezone().replace(tzinfo=None) + + monkeypatch.setattr("core.log_handler.datetime", _FrozenClock) + + assert LogHandler(False)._timestamp() == "03:04:05", "local wall-clock, not the 18:04:05 UTC reading" finally: monkeypatch.undo() time.tzset() From 3f362443dd279a49130335018252881eb3ae67bf Mon Sep 17 00:00:00 2001 From: Marcin Zieba Date: Fri, 11 Sep 2026 12:52:24 +0200 Subject: [PATCH 4/4] chore(lint): satisfy the widened rules for code added on develop The wider rule set lands on this branch, so tests added on develop after it was cut are the first to break it. Import blocks inside three test functions are sorted, and the module-type query assertion binds its expected text to a name. SIM905 wanted that literal replaced with a list of tokens, which would leave the expected GraphQL unreadable. Binding it first keeps the query legible and the rule satisfied. --- tests/test_graphql_client.py | 13 +++++++------ tests/test_netbox_api.py | 4 ++-- tests/test_repo.py | 1 + 3 files changed, 10 insertions(+), 8 deletions(-) diff --git a/tests/test_graphql_client.py b/tests/test_graphql_client.py index cd1fbf32f..7f538d09a 100644 --- a/tests/test_graphql_client.py +++ b/tests/test_graphql_client.py @@ -1159,9 +1159,10 @@ def _make_client(self): @pytest.mark.real_http @pytest.mark.parametrize("caller", ["graphql_relation_fields", "get_module_types"]) def test_relation_selections_use_the_shared_helper(self, caller): - from core.component_registry import BY_ENDPOINT from helpers import FakeNetBox + from core.component_registry import BY_ENDPOINT + expected = "module_bay_types { id name slug manufacturer { slug } }" if caller == "graphql_relation_fields": assert BY_ENDPOINT["module_bay_templates"].graphql_relation_fields == [expected] @@ -1177,9 +1178,10 @@ def test_relation_selections_use_the_shared_helper(self, caller): @pytest.mark.real_http def test_module_type_query_contains_the_complete_relation_selection(self): - from core.component_registry import BY_ENDPOINT from helpers import FakeNetBox + from core.component_registry import BY_ENDPOINT + server = FakeNetBox() try: client = NetBoxGraphQLClient(server.url, "test-token", supports_module_bay_types=True) @@ -1190,8 +1192,7 @@ def test_module_type_query_contains_the_complete_relation_selection(self): selection = "module_bay_types { id name slug manufacturer { slug } }" assert BY_ENDPOINT["module_bay_templates"].graphql_relation_fields == [selection] - assert " ".join(query.split()) == " ".join( - """ + expected_query = """ query($pagination: OffsetPaginationInput) { module_type_list(pagination: $pagination) { id model part_number airflow description comments weight weight_unit last_updated @@ -1199,8 +1200,8 @@ def test_module_type_query_contains_the_complete_relation_selection(self): manufacturer { id name slug } } } -""".split() - ) +""" + assert " ".join(query.split()) == " ".join(expected_query.split()) def test_a_clone_still_selects_the_module_bay_type_relation(self, mock_post): """The prefetch runs on clones, not on the client it was cloned from. diff --git a/tests/test_netbox_api.py b/tests/test_netbox_api.py index 78e86d986..a2f61af4d 100644 --- a/tests/test_netbox_api.py +++ b/tests/test_netbox_api.py @@ -1,8 +1,8 @@ import os import threading -from types import SimpleNamespace from contextlib import contextmanager from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from types import SimpleNamespace from unittest.mock import MagicMock, patch import pytest @@ -7645,7 +7645,7 @@ def _mapping_clears_sent(self, nb): sent += [pc for pc in change.property_changes if pc.property_name == "_mappings"] return sent - @pytest.mark.parametrize("remove_components, expected", [(False, 0), (True, 1)]) + @pytest.mark.parametrize(("remove_components", "expected"), [(False, 0), (True, 1)]) def test_the_flag_decides_whether_a_clear_reaches_netbox( self, mock_settings, mock_pynetbox, mock_graphql_requests, mock_handle, remove_components, expected ): diff --git a/tests/test_repo.py b/tests/test_repo.py index 7d1dee359..c33af6b27 100644 --- a/tests/test_repo.py +++ b/tests/test_repo.py @@ -1884,6 +1884,7 @@ class TestAStanzaThatDoesNotListAFrontPort: def test_a_nonempty_stanza_clears_an_omitted_front_port_mapping(self): from types import SimpleNamespace + from core.change_detector import ChangeDetector data = yaml.safe_load("""