diff --git a/core/change_detector.py b/core/change_detector.py index 5e2ecf78..57ca4d3f 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: @@ -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, @@ -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/compat.py b/core/compat.py index 8ebcbf03..2b35eb0a 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 f314ecf7..61ddbe83 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/component_registry.py b/core/component_registry.py index 8fb4e474..10609fb6 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 9dd48507..d08b2d4c 100644 --- a/core/export.py +++ b/core/export.py @@ -3,13 +3,15 @@ Entry point: ``Exporter(config, handle, export_dir, force_overwrite, vendor_slugs).run()`` """ +import contextlib import hashlib 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 @@ -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 a188ea6e..d66c376a 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 "" @@ -299,6 +300,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 cdac353b..c99a6ef0 100644 --- a/core/import_run.py +++ b/core/import_run.py @@ -1,22 +1,26 @@ """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 datetime import UTC, datetime, timedelta 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 +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.""" @@ -67,7 +71,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): @@ -580,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") @@ -589,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() + # 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/core/log_handler.py b/core/log_handler.py index 2a53eab9..3842cd3c 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().""" @@ -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 4977553b..9ca1094a 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 @@ -107,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 3321d45d..02f2566c 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. @@ -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 ( @@ -542,7 +539,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 " @@ -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. @@ -1900,7 +1897,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 +1937,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 +2280,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 +2291,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 = [] @@ -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,11 +3169,9 @@ 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(): - try: + for _, fh in file_handles.values(): + 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 cf0b75e8..a9a9c798 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 62472f7a..4c368911 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 @@ -109,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: @@ -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 @@ -215,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" @@ -434,7 +433,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"] @@ -450,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}" @@ -625,7 +624,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: @@ -702,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: @@ -769,7 +768,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 82543a5d..205610e4 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 @@ -91,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: @@ -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 @@ -135,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: @@ -143,12 +144,12 @@ 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 -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.) @@ -160,7 +161,7 @@ def _list_device_bay_templates(netbox: Any, device_type_id: int) -> Optional[Lis """ 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 6d53f2da..cbd5cb68 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.""" -from contextlib import contextmanager -from datetime import datetime import sys +from contextlib import contextmanager +from datetime import UTC, datetime -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, @@ -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 99ff4081..c1cad3cc 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/conftest.py b/tests/conftest.py index a2d55c4e..32544ae5 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 acb110f7..961697e5 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:: @@ -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") @@ -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_change_detector.py b/tests/test_change_detector.py index f56450d3..bb79c675 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 a915936f..49387cb3 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 @@ -19,7 +20,6 @@ from core.component_registry import COMPONENT_TYPES from core.graphql_client import GraphQLCountMismatchError, GraphQLSchemaError - # ── Fakes ───────────────────────────────────────────────────────────────────── @@ -267,7 +267,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): @@ -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 e289960c..0c77c249 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_docker_workflow.py b/tests/test_docker_workflow.py index 8b8ce602..19fb5d4f 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 16a91d64..cbd7e0d6 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 c646a241..0cfddd01 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 8e063884..7bfa5bd1 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 378a83e1..7f538d09 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 = { @@ -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. @@ -1216,7 +1217,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 +1774,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 +1891,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 +1904,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 +2012,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 +2068,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 +2106,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"] @@ -2477,12 +2477,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"] @@ -2650,7 +2650,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 +2922,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 +2993,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 9c9d5694..26b09352 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): @@ -435,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 @@ -534,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 3eaf98a5..5b5d9d7f 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 @@ -166,3 +166,34 @@ 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: + # 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) + + 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() diff --git a/tests/test_module_bay_type_sync.py b/tests/test_module_bay_type_sync.py index afe27626..b2084a1b 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 82a2e4d1..e629b393 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 fea59e41..bb3af27d 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 # --------------------------------------------------------------------------- @@ -826,9 +828,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 +1090,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() @@ -1352,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() @@ -1373,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) @@ -1383,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() @@ -1411,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) @@ -1481,9 +1482,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 +1499,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 +1896,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() @@ -2032,19 +2033,19 @@ 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"), ): - 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, {}) @@ -2125,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) @@ -2138,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) @@ -2152,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) @@ -2203,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() @@ -2229,20 +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 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): + 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 2423e106..a2f61af4 100644 --- a/tests/test_netbox_api.py +++ b/tests/test_netbox_api.py @@ -1,24 +1,24 @@ 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 +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" @@ -4618,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), @@ -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 @@ -7174,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) @@ -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 @@ -7617,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_normalization.py b/tests/test_normalization.py index 6d759ea1..1aba6fe8 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 09714d48..c33af6b2 100644 --- a/tests/test_repo.py +++ b/tests/test_repo.py @@ -1,9 +1,15 @@ import os +import re +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 +21,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 +50,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 +83,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 @@ -143,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): @@ -255,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): @@ -358,8 +362,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() @@ -473,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 @@ -499,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)) @@ -547,8 +552,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 +560,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 +570,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 +580,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 +600,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 +680,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 +1256,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 +1271,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 +1349,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 +1401,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 +1436,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 +1537,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.""" @@ -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(""" diff --git a/tests/test_suite_hygiene.py b/tests/test_suite_hygiene.py index e85d411e..c456b12b 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): diff --git a/tests/test_update_failure_resolver.py b/tests/test_update_failure_resolver.py index 93a4bf2f..1185c099 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."