diff --git a/.github/workflows/test-netbox-main.yaml b/.github/workflows/test-netbox-main.yaml index 5ee5311a..a4b228e6 100644 --- a/.github/workflows/test-netbox-main.yaml +++ b/.github/workflows/test-netbox-main.yaml @@ -156,7 +156,7 @@ jobs: NETBOX_TOKEN: ${{ env.NETBOX_TOKEN }} REPO_URL: file:///tmp/test-fixtures REPO_BRANCH: main - run: uv run python tests/integration/test_import.py + run: uv run pytest tests/integration/ -m integration -x -v --timeout=600 - name: Print NetBox version on failure if: failure() diff --git a/pyproject.toml b/pyproject.toml index 3f9f8daf..fcad6733 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -70,6 +70,9 @@ testpaths = [ "tests" ] norecursedirs = ["tests/integration"] +markers = [ + "integration: end-to-end tests that require a live NetBox instance (set NETBOX_URL + NETBOX_TOKEN)", +] [tool.coverage.run] source = ["."] diff --git a/tests/conftest.py b/tests/conftest.py index 587e25cc..bd074819 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,8 +4,11 @@ @pytest.fixture(autouse=True) -def reset_graphql_clamping_warned(mock_env_vars): +def reset_graphql_clamping_warned(mock_env_vars, request): """Reset the module-level page-size clamping warning dedup set before each test.""" + if request.node.get_closest_marker("integration"): + yield + return import core.graphql_client as _gql _gql._CLAMPING_WARNED.clear() @@ -14,8 +17,11 @@ def reset_graphql_clamping_warned(mock_env_vars): @pytest.fixture(autouse=True) -def mock_env_vars(): +def mock_env_vars(request): """Set mandatory environment variables to prevent settings.py from exiting.""" + if request.node.get_closest_marker("integration"): + yield + return with patch.dict( os.environ, { @@ -31,8 +37,11 @@ def mock_env_vars(): @pytest.fixture(autouse=True) -def mock_git_repo(): +def mock_git_repo(request): """Mock git.Repo to prevent actual git operations during settings import.""" + if request.node.get_closest_marker("integration"): + yield None + return with patch("core.repo.Repo") as mock_repo: mock_remote = MagicMock() mock_remote.url = "https://example.com/repo.git" @@ -49,13 +58,16 @@ def mock_pynetbox(): @pytest.fixture(autouse=True) -def mock_graphql_requests(): +def mock_graphql_requests(request): """Mock the HTTP session used by NetBoxGraphQLClient to prevent real calls. Patches ``requests.Session`` in ``core.graphql_client`` so any client created during a test uses a mock session. Returns empty lists for all GraphQL list queries by default. """ + if request.node.get_closest_marker("integration"): + yield None + return with patch("core.graphql_client.requests.Session") as MockSession: mock_session = MockSession.return_value response = MagicMock() diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py new file mode 100644 index 00000000..f7dddb60 --- /dev/null +++ b/tests/integration/conftest.py @@ -0,0 +1,39 @@ +"""Conftest for integration tests. + +All tests in this package require a live NetBox instance reachable at +``NETBOX_URL`` with a valid ``NETBOX_TOKEN``. Normal ``pytest`` runs skip this +package automatically via ``norecursedirs`` in pyproject.toml; the CI workflow +invokes it explicitly with ``pytest tests/integration/ -m integration``. + +When ``NETBOX_URL`` or ``NETBOX_TOKEN`` are absent every test in the package is +marked as skipped during collection, so the suite shows "s" markers instead of +errors. + +Note: the env check reads ``os.environ``, which ``core.settings.load_dotenv()`` +(run when the test module is imported during collection) populates from any +local ``.env``. So a developer with a configured ``.env`` will have these +tests run even with the shell vars unset; the skip path fires only in a clean +checkout (e.g. CI before its real ``NETBOX_*`` vars are exported). +""" + +import os + +import pytest + + +def pytest_collection_modifyitems(items): + """Attach ``integration`` mark to every test in this package; skip all if env is missing.""" + url = os.environ.get("NETBOX_URL", "").strip() + token = os.environ.get("NETBOX_TOKEN", "").strip() + missing_env = not url or not token + skip_marker = pytest.mark.skip( + reason=( + "Integration tests require NETBOX_URL and NETBOX_TOKEN environment variables. " + "Run: export NETBOX_URL=http://localhost:8000 NETBOX_TOKEN=" + ) + ) + for item in items: + if "integration" in str(item.fspath): + item.add_marker(pytest.mark.integration) + if missing_env: + item.add_marker(skip_marker) diff --git a/tests/integration/test_import.py b/tests/integration/test_import.py index 525d0016..869aaa70 100644 --- a/tests/integration/test_import.py +++ b/tests/integration/test_import.py @@ -11,9 +11,10 @@ associated data in NetBox; every subsequent test (``test_front_port_multiposition``, ``test_module_types``, ``test_graphql_schema``, ``test_idempotency``, ``test_update_mode``) depends on that initial import having completed successfully. -If an early test fails, ``sys.exit(1)`` stops the suite immediately so later tests -do not run against incomplete state. Re-running individual tests requires a fully -provisioned NetBox environment (i.e. with the test fixtures already imported). +Run the suite with ``pytest -x`` so an early failure stops it immediately and +later tests do not run against incomplete state. Re-running individual tests +requires a fully provisioned NetBox environment (i.e. with the test fixtures +already imported). Test scenarios -------------- @@ -49,7 +50,7 @@ export NETBOX_TOKEN= export REPO_URL=file:///tmp/test-fixtures # local git repo built from tests/fixtures/ export REPO_BRANCH=main - uv run python tests/integration/test_import.py + uv run pytest tests/integration/ -m integration -x -v """ from __future__ import annotations @@ -61,22 +62,23 @@ from pathlib import Path from typing import Any, NoReturn +import pytest import requests import urllib3 -urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - -REPO_ROOT = Path(__file__).resolve().parents[2] -sys.path.insert(0, str(REPO_ROOT)) - -# Import after sys.path manipulation so local modules resolve correctly. -from core.change_detector import DEVICE_TYPE_PROPERTIES # noqa: E402 -from core.graphql_client import ( # noqa: E402 +from core.change_detector import DEVICE_TYPE_PROPERTIES +from core.graphql_client import ( COMPONENT_TEMPLATE_FIELDS, NetBoxGraphQLClient, _NO_MODULE_TYPE, ) +urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + +pytestmark = pytest.mark.integration + +REPO_ROOT = Path(__file__).resolve().parents[2] + NETBOX_URL = (os.environ.get("NETBOX_URL") or "").rstrip("/") or None NETBOX_TOKEN = os.environ.get("NETBOX_TOKEN") IGNORE_SSL = os.environ.get("IGNORE_SSL_ERRORS", "False").lower() == "true" @@ -95,8 +97,7 @@ def fail(msg: str) -> NoReturn: """Record a failure and exit immediately.""" - print(f"\n ✗ FAIL: {msg}", file=sys.stderr) - sys.exit(1) + pytest.fail(msg) def ok(msg: str) -> None: @@ -551,32 +552,3 @@ def test_update_mode() -> None: if not re.search(r"(? None: - if not NETBOX_URL or not NETBOX_TOKEN: - print( - "ERROR: NETBOX_URL and NETBOX_TOKEN environment variables are required.", - file=sys.stderr, - ) - sys.exit(1) - print(f"NetBox URL : {NETBOX_URL}") - print(f"Repo root : {REPO_ROOT}") - - test_first_import() - test_front_port_multiposition() - test_module_types() - test_graphql_schema() - test_idempotency() - test_update_mode() - - print("\n=== All integration tests passed ✓ ===\n") - - -if __name__ == "__main__": - main() diff --git a/tests/test_change_detector.py b/tests/test_change_detector.py index 3535119c..a525edb9 100644 --- a/tests/test_change_detector.py +++ b/tests/test_change_detector.py @@ -1,6 +1,7 @@ from types import SimpleNamespace from unittest.mock import MagicMock +from core.graphql_client import DotDict from core.change_detector import ( ChangeDetector, ChangeReport, @@ -68,15 +69,13 @@ def _make_detector(self): def test_no_changes_when_values_match(self): detector = self._make_detector() - netbox_dt = MagicMock() - netbox_dt.u_height = 2 + netbox_dt = DotDict({"u_height": 2}) changes = detector._compare_device_type_properties({"u_height": 2}, netbox_dt) assert changes == [] def test_change_detected(self): detector = self._make_detector() - netbox_dt = MagicMock() - netbox_dt.u_height = 1 + netbox_dt = DotDict({"u_height": 1}) changes = detector._compare_device_type_properties({"u_height": 2}, netbox_dt) assert len(changes) == 1 assert changes[0].property_name == "u_height" @@ -85,17 +84,14 @@ def test_change_detected(self): def test_omitted_property_not_compared(self): detector = self._make_detector() - netbox_dt = MagicMock() - netbox_dt.u_height = 99 + netbox_dt = DotDict({"u_height": 99}) # u_height not in yaml_data → should not be compared changes = detector._compare_device_type_properties({"model": "X"}, netbox_dt) assert changes == [] def test_multiple_changes(self): detector = self._make_detector() - netbox_dt = MagicMock() - netbox_dt.u_height = 1 - netbox_dt.is_full_depth = False + netbox_dt = DotDict({"u_height": 1, "is_full_depth": False}) changes = detector._compare_device_type_properties({"u_height": 2, "is_full_depth": True}, netbox_dt) assert len(changes) == 2 @@ -117,8 +113,7 @@ def test_component_added_when_missing_in_netbox(self): assert any(c.component_name == "eth0" and c.change_type == ChangeType.COMPONENT_ADDED for c in changes) def test_component_removed_when_missing_in_yaml(self): - existing_comp = MagicMock() - existing_comp.name = "eth99" + existing_comp = DotDict({"name": "eth99"}) detector = self._make_detector() detector.device_types.cached_components = {"interface_templates": {("device", 1): {"eth99": existing_comp}}} yaml_data = {"interfaces": []} # key present but empty → removal detected @@ -126,7 +121,7 @@ def test_component_removed_when_missing_in_yaml(self): assert any(c.component_name == "eth99" and c.change_type == ChangeType.COMPONENT_REMOVED for c in changes) def test_no_removal_when_key_absent(self): - existing_comp = MagicMock() + existing_comp = DotDict({}) detector = self._make_detector() detector.device_types.cached_components = {"interface_templates": {("device", 1): {"eth99": existing_comp}}} yaml_data = {} # 'interfaces' key absent → YAML doesn't manage this type @@ -135,8 +130,7 @@ def test_no_removal_when_key_absent(self): def test_removal_when_key_absent_with_remove_unmanaged_types(self): """remove_unmanaged_types=True flags removals even when the YAML omits the section entirely.""" - existing_comp = MagicMock() - existing_comp.name = "eth99" + existing_comp = DotDict({"name": "eth99"}) dt_instance = MagicMock() dt_instance.cached_components = {"interface_templates": {("device", 1): {"eth99": existing_comp}}} detector = ChangeDetector(dt_instance, MagicMock(), remove_unmanaged_types=True) @@ -145,8 +139,7 @@ def test_removal_when_key_absent_with_remove_unmanaged_types(self): assert any(c.component_name == "eth99" and c.change_type == ChangeType.COMPONENT_REMOVED for c in changes) def test_component_changed_when_property_differs(self): - existing_comp = MagicMock() - existing_comp.type = "virtual" + existing_comp = DotDict({"type": "virtual"}) detector = self._make_detector() detector.device_types.cached_components = {"interface_templates": {("device", 1): {"eth0": existing_comp}}} yaml_data = {"interfaces": [{"name": "eth0", "type": "1000base-t"}]} @@ -170,22 +163,20 @@ def _make_detector(self): def test_name_property_skipped(self): detector = self._make_detector() - netbox_comp = MagicMock() + netbox_comp = DotDict({}) changes = detector._compare_component_properties({"name": "eth0"}, netbox_comp, ["name"]) assert changes == [] def test_change_detected(self): detector = self._make_detector() - netbox_comp = MagicMock() - netbox_comp.type = "virtual" + netbox_comp = DotDict({"type": "virtual"}) changes = detector._compare_component_properties({"type": "1000base-t"}, netbox_comp, ["name", "type"]) assert len(changes) == 1 assert changes[0].property_name == "type" def test_omitted_prop_not_compared(self): detector = self._make_detector() - netbox_comp = MagicMock() - netbox_comp.type = "virtual" + netbox_comp = DotDict({"type": "virtual"}) changes = detector._compare_component_properties({}, netbox_comp, ["name", "type"]) assert changes == [] @@ -209,8 +200,7 @@ def test_new_device_type(self): assert report.new_device_types[0].model == "X" def test_existing_unchanged_increments_count(self): - existing = MagicMock() - existing.id = 1 + existing = DotDict({"id": 1}) detector = self._make_detector_with_cache( existing_by_model={("cisco", "X"): existing}, cached_components={}, @@ -221,9 +211,7 @@ def test_existing_unchanged_increments_count(self): assert len(report.modified_device_types) == 0 def test_existing_with_change_goes_to_modified(self): - existing = MagicMock() - existing.id = 1 - existing.u_height = 1 + existing = DotDict({"id": 1, "u_height": 1}) detector = self._make_detector_with_cache( existing_by_model={("cisco", "X"): existing}, cached_components={}, @@ -240,8 +228,7 @@ def test_existing_with_change_goes_to_modified(self): assert len(report.modified_device_types) == 1 def test_slug_fallback_lookup(self): - existing = MagicMock() - existing.id = 1 + existing = DotDict({"id": 1}) detector = self._make_detector_with_cache( existing_by_model={}, existing_by_slug={("cisco", "x"): existing}, @@ -329,8 +316,7 @@ class TestCompareImageProperties: def test_missing_front_image_detected(self): """YAML=true, NetBox=None → should report a missing image.""" yaml_data = {"front_image": True} - netbox_dt = MagicMock() - netbox_dt.front_image = None + netbox_dt = DotDict({"front_image": None}) changes = ChangeDetector._compare_image_properties(yaml_data, netbox_dt) @@ -342,8 +328,7 @@ def test_missing_front_image_detected(self): def test_missing_rear_image_detected(self): """YAML=true, NetBox=None → should report a missing image.""" yaml_data = {"rear_image": True} - netbox_dt = MagicMock() - netbox_dt.rear_image = None + netbox_dt = DotDict({"rear_image": None}) changes = ChangeDetector._compare_image_properties(yaml_data, netbox_dt) @@ -355,9 +340,7 @@ def test_missing_rear_image_detected(self): def test_both_images_missing(self): """Both images defined in YAML but missing in NetBox.""" yaml_data = {"front_image": True, "rear_image": True} - netbox_dt = MagicMock() - netbox_dt.front_image = None - netbox_dt.rear_image = None + netbox_dt = DotDict({"front_image": None, "rear_image": None}) changes = ChangeDetector._compare_image_properties(yaml_data, netbox_dt) @@ -368,8 +351,7 @@ def test_both_images_missing(self): def test_existing_image_not_flagged(self): """YAML=true, NetBox=URL → no change reported.""" yaml_data = {"front_image": True} - netbox_dt = MagicMock() - netbox_dt.front_image = "http://netbox/media/devicetypes/front.jpg" + netbox_dt = DotDict({"front_image": "http://netbox/media/devicetypes/front.jpg"}) changes = ChangeDetector._compare_image_properties(yaml_data, netbox_dt) @@ -378,8 +360,7 @@ def test_existing_image_not_flagged(self): def test_yaml_false_no_change(self): """YAML=false → no change reported regardless of NetBox state.""" yaml_data = {"front_image": False} - netbox_dt = MagicMock() - netbox_dt.front_image = None + netbox_dt = DotDict({"front_image": None}) changes = ChangeDetector._compare_image_properties(yaml_data, netbox_dt) @@ -388,9 +369,7 @@ def test_yaml_false_no_change(self): def test_yaml_omitted_no_change(self): """Image key omitted from YAML → no change reported.""" yaml_data = {"model": "Test"} - netbox_dt = MagicMock() - netbox_dt.front_image = None - netbox_dt.rear_image = None + netbox_dt = DotDict({"front_image": None, "rear_image": None}) changes = ChangeDetector._compare_image_properties(yaml_data, netbox_dt) @@ -399,8 +378,7 @@ def test_yaml_omitted_no_change(self): def test_empty_string_treated_as_missing(self): """NetBox returns empty string instead of None → still flagged as missing.""" yaml_data = {"front_image": True} - netbox_dt = MagicMock() - netbox_dt.front_image = "" + netbox_dt = DotDict({"front_image": ""}) changes = ChangeDetector._compare_image_properties(yaml_data, netbox_dt) diff --git a/tests/test_exporter.py b/tests/test_exporter.py index 4d92450c..978682fc 100644 --- a/tests/test_exporter.py +++ b/tests/test_exporter.py @@ -21,6 +21,18 @@ # ── Fixtures ────────────────────────────────────────────────────────────────── +class _Stub: + """Lightweight data stub: only has attributes for explicitly-set kwargs. + + Unlike MagicMock, accessing an absent attribute raises AttributeError, so + "field absent → don't emit it" branches in the exporter are exercised correctly. + """ + + def __init__(self, **kwargs): + for k, v in kwargs.items(): + setattr(self, k, v) + + def _make_settings(tmp_path): s = MagicMock() s.NETBOX_URL = "http://localhost:8000/" @@ -38,10 +50,7 @@ def _make_handle(): def _make_mfr(name="Nokia", slug="nokia"): - m = MagicMock() - m.name = name - m.slug = slug - return m + return _Stub(name=name, slug=slug) def _make_dt( @@ -61,24 +70,24 @@ def _make_dt( front_image=None, rear_image=None, ): - r = MagicMock() - r.id = id - r.model = model - r.slug = slug - r.last_updated = last_updated - r.u_height = u_height - r.is_full_depth = is_full_depth - r.part_number = part_number - r.airflow = airflow - r.weight = weight - r.weight_unit = weight_unit - r.description = description - r.comments = comments - r.subdevice_role = subdevice_role - r.front_image = front_image - r.rear_image = rear_image - r.manufacturer = _make_mfr() - return r + return _Stub( + id=id, + model=model, + slug=slug, + last_updated=last_updated, + u_height=u_height, + is_full_depth=is_full_depth, + part_number=part_number, + airflow=airflow, + weight=weight, + weight_unit=weight_unit, + description=description, + comments=comments, + subdevice_role=subdevice_role, + front_image=front_image, + rear_image=rear_image, + manufacturer=_make_mfr(), + ) def _make_mt( @@ -92,18 +101,18 @@ def _make_mt( description="", comments="", ): - r = MagicMock() - r.id = id - r.model = model - r.last_updated = last_updated - r.part_number = part_number - r.airflow = airflow - r.weight = weight - r.weight_unit = weight_unit - r.description = description - r.comments = comments - r.manufacturer = _make_mfr() - return r + return _Stub( + id=id, + model=model, + last_updated=last_updated, + part_number=part_number, + airflow=airflow, + weight=weight, + weight_unit=weight_unit, + description=description, + comments=comments, + manufacturer=_make_mfr(), + ) def _make_rt( @@ -127,28 +136,28 @@ def _make_rt( desc_units=False, comments="", ): - r = MagicMock() - r.id = id - r.model = model - r.slug = slug - r.last_updated = last_updated - r.form_factor = form_factor - r.description = description - r.width = width - r.u_height = u_height - r.starting_unit = starting_unit - r.outer_width = outer_width - r.outer_height = outer_height - r.outer_depth = outer_depth - r.outer_unit = outer_unit - r.mounting_depth = mounting_depth - r.weight = weight - r.max_weight = max_weight - r.weight_unit = weight_unit - r.desc_units = desc_units - r.comments = comments - r.manufacturer = _make_mfr() - return r + return _Stub( + id=id, + model=model, + slug=slug, + last_updated=last_updated, + form_factor=form_factor, + description=description, + width=width, + u_height=u_height, + starting_unit=starting_unit, + outer_width=outer_width, + outer_height=outer_height, + outer_depth=outer_depth, + outer_unit=outer_unit, + mounting_depth=mounting_depth, + weight=weight, + max_weight=max_weight, + weight_unit=weight_unit, + desc_units=desc_units, + comments=comments, + manufacturer=_make_mfr(), + ) # ── Tests ───────────────────────────────────────────────────────────────────── diff --git a/tests/test_log_handler.py b/tests/test_log_handler.py index 8a72527c..37170f07 100644 --- a/tests/test_log_handler.py +++ b/tests/test_log_handler.py @@ -137,10 +137,8 @@ def test_returns_count(self): handle = LogHandler(SimpleNamespace(verbose=False)) ports = [] for i in range(2): - p = MagicMock() - p.name = f"port{i}" - p.device_type = MagicMock(id=1) - p.id = i + # No `type` attr → exercises the `hasattr(port, 'type')` else-branch. + p = SimpleNamespace(name=f"port{i}", device_type=SimpleNamespace(id=1), id=i) ports.append(p) result = handle.log_device_ports_created(ports, "Interface") assert result == 2 @@ -151,11 +149,7 @@ def test_returns_zero_for_none(self): def test_verbose_logs_each_port(self): handle = LogHandler(SimpleNamespace(verbose=True)) - port = MagicMock() - port.name = "eth0" - port.type = "virtual" - port.device_type = MagicMock(id=5) - port.id = 10 + port = SimpleNamespace(name="eth0", type="virtual", device_type=SimpleNamespace(id=5), id=10) with patch.object(handle, "_emit") as mock_emit: handle.log_device_ports_created([port], "Interface") assert mock_emit.called @@ -166,10 +160,8 @@ class TestLogModulePortsCreated: def test_returns_count(self): handle = LogHandler(SimpleNamespace(verbose=False)) - port = MagicMock() - port.name = "port" - port.module_type = MagicMock(id=1) - port.id = 1 + # No `type` attr → exercises the `hasattr(port, 'type')` else-branch. + port = SimpleNamespace(name="port", module_type=SimpleNamespace(id=1), id=1) result = handle.log_module_ports_created([port], "Interface") assert result == 1 @@ -179,11 +171,7 @@ def test_returns_zero_for_none(self): def test_verbose_logs_each_port(self): handle = LogHandler(SimpleNamespace(verbose=True)) - port = MagicMock() - port.name = "xe-0/0/0" - port.type = "10gbase-x-sfpp" - port.module_type = MagicMock(id=3) - port.id = 7 + port = SimpleNamespace(name="xe-0/0/0", type="10gbase-x-sfpp", module_type=SimpleNamespace(id=3), id=7) with patch.object(handle, "_emit") as mock_emit: handle.log_module_ports_created([port], "Interface") assert mock_emit.called diff --git a/tests/test_nb_dt_import.py b/tests/test_nb_dt_import.py index ceea049d..3f5dda50 100644 --- a/tests/test_nb_dt_import.py +++ b/tests/test_nb_dt_import.py @@ -557,6 +557,9 @@ def test_only_new_creates_new_device_types(self, nb_dt_import): nb_dt_import.main() MockNetBox.return_value.create_device_types.assert_called_once() + call = MockNetBox.return_value.create_device_types.call_args + assert call.args[0] == dt + assert call.kwargs["only_new"] is True def test_default_mode_no_device_types(self, nb_dt_import): """Default mode with no discovered vendors completes without error.""" @@ -594,6 +597,9 @@ def test_default_mode_with_new_device_types(self, nb_dt_import): nb_dt_import.main() MockNetBox.return_value.create_device_types.assert_called_once() + call = MockNetBox.return_value.create_device_types.call_args + assert call.args[0] == dt + assert call.kwargs["only_new"] is True def test_update_mode_no_changes(self, nb_dt_import): """--update with no changes logs 'No device type changes to process'.""" @@ -832,6 +838,7 @@ def test_modules_with_types_to_process(self, nb_dt_import): nb_dt_import.main() mock_nb.create_module_types.assert_called_once() + assert mock_nb.create_module_types.call_args.args[0] == [module_type] def test_modules_update_mode_logs_change_detection_section(self, nb_dt_import): """--update with modules=True logs the MODULE TYPE CHANGE DETECTION header.""" @@ -950,6 +957,7 @@ def test_full_flow_calls_create_rack_types(self, nb_dt_import): nb_dt_import._process_rack_types(self._make_args(), netbox, handle, None, [rack_type]) netbox.create_rack_types.assert_called_once() + assert netbox.create_rack_types.call_args.args[0] == [rack_type] def test_existing_rack_type_shows_as_existing(self, nb_dt_import): """A rack type already in NetBox is counted as existing, not new.""" @@ -1819,8 +1827,15 @@ def test_run_vendor_loop_processes_slug_fast_path_and_skips_empty_vendor(self, n netbox.device_types.stop_component_preload.assert_called_once_with("job-1", progress=progress) netbox.create_manufacturers.assert_called_once_with([{"slug": "cisco", "name": "Cisco"}]) mock_process_device_types.assert_called_once() + assert mock_process_device_types.call_args.args[4] == [ + {"manufacturer": {"slug": "cisco"}, "model": "X", "slug": "x"} + ] mock_process_module_types.assert_called_once() + assert mock_process_module_types.call_args.args[4] == [ + {"manufacturer": {"slug": "cisco"}, "model": "M", "slug": "m"} + ] mock_process_rack_types.assert_called_once() + assert mock_process_rack_types.call_args.args[4] == [] # cisco not in rack_vendors → empty assert progress.advance.call_args_list == [((7,),), ((7,),)] mock_finalize.assert_called_once_with(progress, {}) assert any(call.args[0] == ["resolved.yaml"] for call in dtl_repo.parse_files.call_args_list) diff --git a/tests/test_normalization.py b/tests/test_normalization.py index f923b137..6d759ea1 100644 --- a/tests/test_normalization.py +++ b/tests/test_normalization.py @@ -1,4 +1,4 @@ -from unittest.mock import MagicMock +from types import SimpleNamespace from core.normalization import normalize_values, values_equal @@ -7,8 +7,7 @@ class TestNormalizeValues: """Tests for normalize_values.""" def test_netbox_choice_object_reads_value(self): - choice = MagicMock() - choice.value = "1000base-t" + choice = SimpleNamespace(value="1000base-t") y, n = normalize_values("1000base-t", choice) assert n == "1000base-t" @@ -134,8 +133,7 @@ def __float__(self): assert values_equal(1, "1.5") is False def test_netbox_choice_object(self): - choice = MagicMock() - choice.value = "kg" + choice = SimpleNamespace(value="kg") assert values_equal("kg", choice) choice.value = "lb" assert not values_equal("kg", choice) diff --git a/tests/test_repo.py b/tests/test_repo.py index 96821b44..2ab2b89f 100644 --- a/tests/test_repo.py +++ b/tests/test_repo.py @@ -308,114 +308,93 @@ def test_clone_repo_generic_error_calls_exception(self): class TestGetDevices: """Tests for DTLRepo.get_devices(): vendor filtering, YAML file discovery, and testing folder exclusion.""" - def _make_repo(self): + def _make_repo(self, tmp_path): + (tmp_path / "repo").mkdir() mock_args = MagicMock() mock_args.url = "https://github.com/org/repo.git" mock_args.branch = "master" mock_handle = MagicMock() - with ( - patch("os.path.isdir", return_value=True), - patch("core.repo.Repo") as MockRepo, - ): + with patch("core.repo.Repo") as MockRepo: mock_git_repo = MagicMock() mock_git_repo.remotes.origin.url = "https://github.com/org/repo.git" ref = MagicMock() ref.name = "origin/master" mock_git_repo.remotes.origin.refs = [ref] MockRepo.return_value = mock_git_repo - repo = DTLRepo(mock_args, "/tmp/repo", mock_handle) + repo = DTLRepo(mock_args, str(tmp_path / "repo"), mock_handle) return repo - def test_get_devices_all_vendors(self): - repo = self._make_repo() - with ( - patch("os.listdir", return_value=["Cisco", "Juniper"]), - patch("core.repo.glob", return_value=[]), - ): - files, vendors = repo.get_devices("/base/path") + def test_get_devices_all_vendors(self, tmp_path): + repo = self._make_repo(tmp_path) + devices = tmp_path / "devices" + devices.mkdir() + (devices / "Cisco").mkdir() + (devices / "Juniper").mkdir() + files, vendors = repo.get_devices(str(devices)) assert len(vendors) == 2 assert any(v["name"] == "Cisco" for v in vendors) - def test_get_devices_filters_vendors(self): - repo = self._make_repo() - with ( - patch("os.listdir", return_value=["Cisco", "Juniper"]), - patch("core.repo.glob", return_value=[]), - ): - files, vendors = repo.get_devices("/base/path", vendors=["cisco"]) + def test_get_devices_filters_vendors(self, tmp_path): + repo = self._make_repo(tmp_path) + devices = tmp_path / "devices" + devices.mkdir() + (devices / "Cisco").mkdir() + (devices / "Juniper").mkdir() + files, vendors = repo.get_devices(str(devices), vendors=["cisco"]) assert len(vendors) == 1 assert vendors[0]["name"] == "Cisco" - def test_get_devices_skips_testing_folder(self): - repo = self._make_repo() - with ( - patch("os.listdir", return_value=["Cisco", "testing"]), - patch("core.repo.glob", return_value=[]), - ): - files, vendors = repo.get_devices("/base/path") + def test_get_devices_skips_testing_folder(self, tmp_path): + repo = self._make_repo(tmp_path) + devices = tmp_path / "devices" + devices.mkdir() + (devices / "Cisco").mkdir() + (devices / "testing").mkdir() + files, vendors = repo.get_devices(str(devices)) assert not any(v["name"] == "testing" for v in vendors) class TestDiscoverVendors: """Tests for DTLRepo.discover_vendors(): vendor discovery across multiple paths with deduplication.""" - def _make_repo(self): + def _make_repo(self, tmp_path): + (tmp_path / "repo").mkdir() mock_args = MagicMock() mock_args.url = "https://github.com/org/repo.git" mock_args.branch = "master" mock_handle = MagicMock() - with ( - patch("os.path.isdir", return_value=True), - patch("core.repo.Repo") as MockRepo, - ): + with patch("core.repo.Repo") as MockRepo: mock_git_repo = MagicMock() mock_git_repo.remotes.origin.url = "https://github.com/org/repo.git" ref = MagicMock() ref.name = "origin/master" mock_git_repo.remotes.origin.refs = [ref] MockRepo.return_value = mock_git_repo - repo = DTLRepo(mock_args, "/tmp/repo", mock_handle) + repo = DTLRepo(mock_args, str(tmp_path / "repo"), mock_handle) return repo - def test_discovers_vendors_from_single_path(self): + def test_discovers_vendors_from_single_path(self, tmp_path): """Test discovery from a single existing path.""" - repo = self._make_repo() - - def mock_exists(path): - return "devices" in path - - with ( - patch("os.path.exists", side_effect=mock_exists), - patch("os.listdir", return_value=["Cisco", "Juniper"]), - patch("os.path.isdir", return_value=True), - ): - vendors = repo.discover_vendors("/devices", "/modules", "/racks") + repo = self._make_repo(tmp_path) + devices = tmp_path / "devices" + (devices / "Cisco").mkdir(parents=True) + (devices / "Juniper").mkdir() + vendors = repo.discover_vendors(str(devices), str(tmp_path / "modules"), str(tmp_path / "racks")) assert len(vendors) == 2 assert vendors[0]["name"] == "Cisco" assert vendors[0]["slug"] == "cisco" assert vendors[1]["name"] == "Juniper" assert vendors[1]["slug"] == "juniper" - def test_discovers_vendors_from_multiple_paths(self): + def test_discovers_vendors_from_multiple_paths(self, tmp_path): """Test discovery and merging from multiple paths.""" - repo = self._make_repo() - - def mock_listdir(path): - if "devices" in path: - return ["Cisco", "Juniper"] - elif "modules" in path: - return ["Arista", "Cisco"] - elif "racks" in path: - return ["Dell"] - return [] - - with ( - patch("os.path.exists", return_value=True), - patch("os.listdir", side_effect=mock_listdir), - patch("os.path.isdir", return_value=True), - ): - vendors = repo.discover_vendors("/devices", "/modules", "/racks") - + repo = self._make_repo(tmp_path) + (tmp_path / "devices" / "Cisco").mkdir(parents=True) + (tmp_path / "devices" / "Juniper").mkdir() + (tmp_path / "modules" / "Arista").mkdir(parents=True) + (tmp_path / "modules" / "Cisco").mkdir() + (tmp_path / "racks" / "Dell").mkdir(parents=True) + vendors = repo.discover_vendors(str(tmp_path / "devices"), str(tmp_path / "modules"), str(tmp_path / "racks")) assert len(vendors) == 4 vendor_names = [v["name"] for v in vendors] assert "Cisco" in vendor_names @@ -423,78 +402,51 @@ def mock_listdir(path): assert "Arista" in vendor_names assert "Dell" in vendor_names - def test_deduplicates_vendors_across_paths(self): + def test_deduplicates_vendors_across_paths(self, tmp_path): """Test that vendors appearing in multiple paths are deduplicated.""" - repo = self._make_repo() - - def mock_listdir(path): - # Cisco appears in all three paths - if "devices" in path: - return ["Cisco", "Juniper"] - elif "modules" in path: - return ["Cisco", "Arista"] - elif "racks" in path: - return ["Cisco"] - return [] - - with ( - patch("os.path.exists", return_value=True), - patch("os.listdir", side_effect=mock_listdir), - patch("os.path.isdir", return_value=True), - ): - vendors = repo.discover_vendors("/devices", "/modules", "/racks") - - # Cisco should appear only once despite being in all three paths + repo = self._make_repo(tmp_path) + (tmp_path / "devices" / "Cisco").mkdir(parents=True) + (tmp_path / "devices" / "Juniper").mkdir() + (tmp_path / "modules" / "Cisco").mkdir(parents=True) + (tmp_path / "modules" / "Arista").mkdir() + (tmp_path / "racks" / "Cisco").mkdir(parents=True) + vendors = repo.discover_vendors(str(tmp_path / "devices"), str(tmp_path / "modules"), str(tmp_path / "racks")) cisco_vendors = [v for v in vendors if v["slug"] == "cisco"] assert len(cisco_vendors) == 1 assert len(vendors) == 3 # Cisco, Juniper, Arista - def test_skips_testing_folder(self): + def test_skips_testing_folder(self, tmp_path): """Test that 'testing' folder (case-insensitive) is excluded.""" - repo = self._make_repo() - with ( - patch("os.path.exists", return_value=True), - patch("os.listdir", return_value=["Cisco", "testing", "Testing", "TESTING"]), - patch("os.path.isdir", return_value=True), - ): - vendors = repo.discover_vendors("/devices", "/modules", "/racks") - + repo = self._make_repo(tmp_path) + devices = tmp_path / "devices" + (devices / "Cisco").mkdir(parents=True) + (devices / "testing").mkdir() + (devices / "Testing").mkdir() + (devices / "TESTING").mkdir() + vendors = repo.discover_vendors(str(devices), str(tmp_path / "modules"), str(tmp_path / "racks")) assert len(vendors) == 1 assert vendors[0]["name"] == "Cisco" assert not any(v["name"].lower() == "testing" for v in vendors) - def test_handles_nonexistent_paths(self): + def test_handles_nonexistent_paths(self, tmp_path): """Test graceful handling of non-existent paths.""" - repo = self._make_repo() - - def mock_exists(path): - return "devices" in path # Only devices path exists - - def mock_listdir(path): - if "devices" in path: - return ["Cisco"] - return [] - - with ( - patch("os.path.exists", side_effect=mock_exists), - patch("os.listdir", side_effect=mock_listdir), - patch("os.path.isdir", return_value=True), - ): - vendors = repo.discover_vendors("/devices", "/modules", "/racks") - + repo = self._make_repo(tmp_path) + devices = tmp_path / "devices" + (devices / "Cisco").mkdir(parents=True) + # modules and racks paths are never created → os.path.exists returns False + vendors = repo.discover_vendors(str(devices), str(tmp_path / "modules"), str(tmp_path / "racks")) assert len(vendors) == 1 assert vendors[0]["name"] == "Cisco" - def test_handles_all_nonexistent_paths(self): + def test_handles_all_nonexistent_paths(self, tmp_path): """Test that all non-existent paths returns empty list.""" - repo = self._make_repo() - with patch("os.path.exists", return_value=False): - vendors = repo.discover_vendors("/devices", "/modules", "/racks") + repo = self._make_repo(tmp_path) + vendors = repo.discover_vendors(str(tmp_path / "devices"), str(tmp_path / "modules"), str(tmp_path / "racks")) assert vendors == [] - def test_handles_os_errors_gracefully(self): + def test_handles_os_errors_gracefully(self, tmp_path): """Test graceful handling of OS errors when listing directories.""" - repo = self._make_repo() + repo = self._make_repo(tmp_path) def mock_listdir(path): if "devices" in path: @@ -508,59 +460,46 @@ def mock_listdir(path): patch("os.listdir", side_effect=mock_listdir), patch("os.path.isdir", return_value=True), ): - vendors = repo.discover_vendors("/devices", "/modules", "/racks") + vendors = repo.discover_vendors( + str(tmp_path / "devices"), str(tmp_path / "modules"), str(tmp_path / "racks") + ) assert len(vendors) == 1 assert vendors[0]["name"] == "Cisco" - def test_skips_non_directory_entries(self): + def test_skips_non_directory_entries(self, tmp_path): """Test that files (non-directories) are skipped.""" - repo = self._make_repo() - - def mock_isdir(path): - # Only Cisco is a directory, README.md is a file - return "Cisco" in path - - with ( - patch("os.path.exists", return_value=True), - patch("os.listdir", return_value=["Cisco", "README.md"]), - patch("os.path.isdir", side_effect=mock_isdir), - ): - vendors = repo.discover_vendors("/devices", "/modules", "/racks") - + repo = self._make_repo(tmp_path) + devices = tmp_path / "devices" + (devices / "Cisco").mkdir(parents=True) + (devices / "README.md").write_text("readme") # file, not directory + vendors = repo.discover_vendors(str(devices), str(tmp_path / "modules"), str(tmp_path / "racks")) assert len(vendors) == 1 assert vendors[0]["name"] == "Cisco" - def test_returns_sorted_by_slug(self): + def test_returns_sorted_by_slug(self, tmp_path): """Test that vendors are returned sorted by slug.""" - repo = self._make_repo() - with ( - patch("os.path.exists", return_value=True), - patch("os.listdir", return_value=["Zebra", "Arista", "Cisco", "Dell"]), - patch("os.path.isdir", return_value=True), - ): - vendors = repo.discover_vendors("/devices", "/modules", "/racks") - + repo = self._make_repo(tmp_path) + devices = tmp_path / "devices" + for name in ["Zebra", "Arista", "Cisco", "Dell"]: + (devices / name).mkdir(parents=True) + vendors = repo.discover_vendors(str(devices), str(tmp_path / "modules"), str(tmp_path / "racks")) slugs = [v["slug"] for v in vendors] assert slugs == sorted(slugs) assert slugs == ["arista", "cisco", "dell", "zebra"] - def test_uses_slug_format_correctly(self): + def test_uses_slug_format_correctly(self, tmp_path): """Test that slug_format is applied correctly to vendor names.""" - repo = self._make_repo() - with ( - patch("os.path.exists", return_value=True), - patch("os.listdir", return_value=["Extreme Networks", "HPE-Aruba"]), - patch("os.path.isdir", return_value=True), - ): - vendors = repo.discover_vendors("/devices", "/modules", "/racks") - + repo = self._make_repo(tmp_path) + devices = tmp_path / "devices" + (devices / "Extreme Networks").mkdir(parents=True) + (devices / "HPE-Aruba").mkdir() + vendors = repo.discover_vendors(str(devices), str(tmp_path / "modules"), str(tmp_path / "racks")) assert len(vendors) == 2 - # slug_format lowercases and replaces non-word chars with hyphens assert any(v["slug"] == "extreme-networks" for v in vendors) assert any(v["slug"] == "hpe-aruba" for v in vendors) - def test_vendor_name_selection_is_deterministic_for_same_slug(self): + def test_vendor_name_selection_is_deterministic_for_same_slug(self, tmp_path): """Vendor name must be deterministic when multiple folders map to same slug. When two folder names produce the same slug, the alphabetically first @@ -569,19 +508,11 @@ def test_vendor_name_selection_is_deterministic_for_same_slug(self): Before the fix, os.listdir() was non-deterministic, so "Nokia" vs "nokia" folders could produce different vendor names across runs. """ - repo = self._make_repo() - - # Simulate os.listdir returning folders in reverse-alphabetical order - def mock_listdir_rev(_path): - return ["nokia", "Nokia"] # lowercase first → would win without sorting - - with ( - patch("os.path.exists", return_value=True), - patch("os.listdir", side_effect=mock_listdir_rev), - patch("os.path.isdir", return_value=True), - ): - vendors = repo.discover_vendors("/devices", "/modules", "/racks") - + repo = self._make_repo(tmp_path) + devices = tmp_path / "devices" + (devices / "Nokia").mkdir(parents=True) + (devices / "nokia").mkdir() + vendors = repo.discover_vendors(str(devices), str(tmp_path / "modules"), str(tmp_path / "racks")) # sorted("Nokia", "nokia") → "Nokia" < "nokia" (uppercase sorts first in ASCII) assert len(vendors) == 1 assert vendors[0]["name"] == "Nokia" # alphabetically first diff --git a/tests/test_update_failure_resolver.py b/tests/test_update_failure_resolver.py index b82088db..b5290f2a 100644 --- a/tests/test_update_failure_resolver.py +++ b/tests/test_update_failure_resolver.py @@ -2,6 +2,7 @@ from __future__ import annotations +from types import SimpleNamespace from unittest.mock import MagicMock import pytest @@ -58,10 +59,8 @@ def test_classifier_unhandled_for_unparseable_payload(): def test_classifier_recognises_subdevice_role_error_dict_safe_path(): """Dict payload + zero dependent devices + blocking templates -> SUBDEVICE_ROLE_FLIP, actionable.""" - t1 = MagicMock() - t1.name = "bay-1" - t2 = MagicMock() - t2.name = "bay-2" + t1 = SimpleNamespace(name="bay-1") + t2 = SimpleNamespace(name="bay-2") nb = _make_netbox(templates=[t1, t2], devices=[], device_count=0) res = classify_device_type_update_failure( @@ -81,8 +80,7 @@ def test_classifier_recognises_subdevice_role_error_dict_safe_path(): def test_classifier_accepts_json_string_payload(): """Pynetbox sometimes returns the body as a JSON string; classifier must handle it.""" - t = MagicMock() - t.name = "t" + t = SimpleNamespace(name="t") nb = _make_netbox(templates=[t], devices=[], device_count=0) res = classify_device_type_update_failure( SUBDEVICE_ROLE_ERROR_JSON, @@ -95,11 +93,9 @@ def test_classifier_accepts_json_string_payload(): def test_classifier_blocks_auto_resolve_when_dependent_devices_exist(): """Live devices reference the type -> MANUAL_REQUIRED, no remediation.""" - d1 = MagicMock() - d1.name = "router-1" - d2 = MagicMock() - d2.name = "router-2" - nb = _make_netbox(devices=[d1, d2], device_count=2, templates=[MagicMock()]) + d1 = SimpleNamespace(name="router-1") + d2 = SimpleNamespace(name="router-2") + nb = _make_netbox(devices=[d1, d2], device_count=2, templates=[SimpleNamespace(name="bay")]) res = classify_device_type_update_failure( SUBDEVICE_ROLE_ERROR_DICT, @@ -138,7 +134,7 @@ def test_classifier_returns_inspection_hint_when_no_blocking_templates(): def test_classifier_blocks_auto_resolve_when_yaml_still_lists_device_bays(): """YAML still declares device-bays -> auto-deleting them would loop; MANUAL_REQUIRED.""" - nb = _make_netbox(templates=[MagicMock()], devices=[], device_count=0) + nb = _make_netbox(templates=[SimpleNamespace(name="bay")], devices=[], device_count=0) res = classify_device_type_update_failure( SUBDEVICE_ROLE_ERROR_DICT, netbox=nb, @@ -152,7 +148,7 @@ def test_classifier_blocks_auto_resolve_when_yaml_still_lists_device_bays(): def test_classifier_dependent_count_unknown_blocks_resolve(): """If the count query raises, classifier must treat the type as unsafe (MANUAL_REQUIRED).""" nb = MagicMock() - nb.dcim.device_bay_templates.filter.return_value = [MagicMock()] + nb.dcim.device_bay_templates.filter.return_value = [SimpleNamespace(name="bay")] nb.dcim.devices.filter.side_effect = RuntimeError("API down") res = classify_device_type_update_failure( @@ -185,7 +181,7 @@ def test_classifier_remediation_step_calls_template_delete(): def test_classifier_handles_bytes_payload(): """RequestError content can be bytes; classifier must decode before parsing.""" - nb = _make_netbox(templates=[MagicMock()], devices=[], device_count=0) + nb = _make_netbox(templates=[SimpleNamespace(name="bay")], devices=[], device_count=0) res = classify_device_type_update_failure( SUBDEVICE_ROLE_ERROR_JSON.encode("utf-8"), netbox=nb, @@ -220,7 +216,7 @@ def test_extract_error_payload_returns_original_bytes_on_decode_failure(): ) def test_classifier_marker_matching_is_strict(payload, expected_kind): """Classifier matches only when both required markers are present in the message.""" - nb = _make_netbox(templates=[MagicMock()], devices=[], device_count=0) + nb = _make_netbox(templates=[SimpleNamespace(name="bay")], devices=[], device_count=0) res = classify_device_type_update_failure( payload, netbox=nb, @@ -232,7 +228,7 @@ def test_classifier_marker_matching_is_strict(payload, expected_kind): def test_classifier_recognises_subdevice_role_error_string_marker(): """Plain-string payload containing both markers must classify as SUBDEVICE_ROLE_FLIP.""" - nb = _make_netbox(templates=[MagicMock()], devices=[], device_count=0) + nb = _make_netbox(templates=[SimpleNamespace(name="bay")], devices=[], device_count=0) payload = ( "Must delete all device bay templates associated with this device before declassifying it as a parent device." ) @@ -252,7 +248,7 @@ class _BadObj: def __iter__(self): raise TypeError("not iterable as expected") - nb = _make_netbox(templates=[MagicMock()], devices=[], device_count=0) + nb = _make_netbox(templates=[SimpleNamespace(name="bay")], devices=[], device_count=0) res = classify_device_type_update_failure( {"subdevice_role": _BadObj()}, netbox=nb, @@ -288,10 +284,8 @@ def test_classifier_handles_template_query_failure(): def test_classifier_count_query_used_when_filter_returns_full_page(): """When filter returns 5 records (page cap), classifier must call .count() for the real total.""" - devs = [MagicMock() for i in range(5)] - for i, d in enumerate(devs): - d.name = f"router-{i}" - nb = _make_netbox(devices=devs, device_count=137, templates=[MagicMock()]) + devs = [SimpleNamespace(name=f"router-{i}") for i in range(5)] + nb = _make_netbox(devices=devs, device_count=137, templates=[SimpleNamespace(name="bay")]) res = classify_device_type_update_failure( SUBDEVICE_ROLE_ERROR_DICT, netbox=nb, @@ -305,11 +299,9 @@ def test_classifier_count_query_used_when_filter_returns_full_page(): def test_classifier_count_fallback_when_count_query_fails(): """If .count() raises, classifier falls back to len(filter_results).""" - devs = [MagicMock() for _ in range(5)] - for i, d in enumerate(devs): - d.name = f"router-{i}" + devs = [SimpleNamespace(name=f"router-{i}") for i in range(5)] nb = MagicMock() - nb.dcim.device_bay_templates.filter.return_value = [MagicMock()] + nb.dcim.device_bay_templates.filter.return_value = [SimpleNamespace(name="bay")] nb.dcim.devices.filter.return_value = devs nb.dcim.devices.count.side_effect = RuntimeError("boom") res = classify_device_type_update_failure(