From b32ec48dbace55537ffd30411e33dae87b83e23a Mon Sep 17 00:00:00 2001 From: Seth Grover <13872653+mmguero@users.noreply.github.com> Date: Tue, 25 Aug 2026 09:06:32 -0600 Subject: [PATCH 1/2] fix(repo): add REPO_URL=local sentinel to bypass git entirely (#124) DTLRepo always ran a clone or a fetch, even when REPO_PATH already held the library contents locally. validate_git_url() rejected the old 'local' value outright, and even a valid URL still triggered a real git fetch against the remote on every run. REPO_URL=local now skips Repo(), clone_from(), and fetch() entirely. REPO_PATH is used as-is and must already contain device-types/, module-types/, and rack-types/ (no .git required, and none is used). --- core/repo.py | 20 +++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/core/repo.py b/core/repo.py index 869bedd6..74688641 100644 --- a/core/repo.py +++ b/core/repo.py @@ -419,11 +419,13 @@ class DTLRepo: def __init__(self, config, handle): """Initialize repository management, updating an existing clone or creating a new one. - If the target path already holds a Git clone, the repository will be updated from - its configured remote; otherwise the provided URL is validated and a new clone is - created. The initializer sets instance attributes used by other methods (handler, - supported YAML extensions, URL, repo path, branch, repo reference, and current - working directory). + If REPO_URL is the sentinel "local", no git operation of any kind is performed — + REPO_PATH is used as-is and must already contain the device-type file tree. + Otherwise, if the target path already holds a Git clone, the repository will be + updated from its configured remote; if not, the provided URL is validated and a new + clone is created. The initializer sets instance attributes used by other methods + (handler, supported YAML extensions, URL, repo path, branch, repo reference, and + current working directory). Args: config (RunConfig): Supplies `repo_url`, `repo_branch`, and `repo_path`. @@ -441,6 +443,14 @@ def __init__(self, config, handle): self.repo = None self.cwd = os.getcwd() + if str(self.url).strip().casefold() == "local": + if not os.path.isdir(self.get_absolute_path()): + raise InvalidRepoPathError( + self.repo_path, reason="REPO_URL=local requires REPO_PATH to already contain the library files" + ) + self.handle.log(f"REPO_URL=local: using {self.get_absolute_path()} as-is, no git operations") + return + is_path_valid, path_error = validate_repo_path(self.repo_path) if not is_path_valid: raise InvalidRepoPathError(self.repo_path, reason=path_error) From 5694aaa8965afd3d88e5146649b2d78550b902b6 Mon Sep 17 00:00:00 2001 From: Marcin Zieba <49913098+marcinpsk@users.noreply.github.com> Date: Tue, 25 Aug 2026 18:35:07 +0200 Subject: [PATCH 2/2] fix(repo): validate the local library layout and share the check with export (#126) * fix(repo): validate the local library layout and share the check with export REPO_URL=local only checked that REPO_PATH was a directory. An unrelated or empty path passed that check, and discover_vendors skips type directories that are absent, so the run imported nothing and still exited 0. Export mode already had the check it needed, in Exporter._verify_repo_available. Both sides now read one definition of what makes a checkout a library: LIBRARY_TYPE_DIRS and library_dirs_present() in core/repo.py. The sentinel value moves to LOCAL_REPO_URL in core/config.py, beside the other REPO_* defaults, so config and repo cannot drift on it. Local mode still skips validate_repo_path on purpose: that check demands write access, which a read-only or air-gapped mount cannot give, and no import step writes to REPO_PATH. A test pins the read-only case. REPO_BRANCH is ignored under the sentinel, so a run that sets both now says so through the existing config notice mechanism instead of looking like it checked the branch out. Documents the mode in the README and .env.example, including the read-only Docker mount, which is the case that motivated the sentinel. * fix(import): treat an absent type root as empty instead of crashing The layout check accepts any one of device-types/, module-types/, and rack-types/, matching what export mode already accepted. plan_vendor then called get_devices() on all three regardless, and get_devices() lists the directory, so a local checkout holding only some of them raised FileNotFoundError before importing the types it did hold. A device-types-only library is the likeliest local layout, and it crashed on module-types. _parse_vendor_racks already had the guard this needed. It is now _parse_vendor_files and all three roots go through it, so the guard cannot apply to one root and not the others again. The repo mock in test_nb_dt_import pointed at /tmp/devices, /tmp/modules and /tmp/rack-types, paths that never existed. Nothing stat'd them, so the mock passed for a filesystem that was not there, which is why this went unseen. It now points at a real empty library tree, and the tests that matched on those literal paths match on the directory names instead. Found by CodeRabbit on #126. --- .env.example | 6 +++ README.md | 25 ++++++++++- core/config.py | 14 +++++- core/export.py | 8 ++-- core/import_run.py | 25 ++++++----- core/repo.py | 49 +++++++++++++++----- tests/test_config.py | 19 ++++++++ tests/test_exporter.py | 30 +++++++++++++ tests/test_import_run.py | 31 +++++++++++++ tests/test_nb_dt_import.py | 36 ++++++++++----- tests/test_repo.py | 91 ++++++++++++++++++++++++++++++++++++++ 11 files changed, 295 insertions(+), 39 deletions(-) diff --git a/.env.example b/.env.example index be9a7476..0c6b5dc4 100644 --- a/.env.example +++ b/.env.example @@ -4,6 +4,12 @@ NETBOX_URL=https://netbox.example.org REPO_BRANCH=master REPO_URL=https://github.com/netbox-community/devicetype-library.git +# Set REPO_URL=local to read REPO_PATH as it stands, with no git operation at all: +# no clone, no fetch, and no .git needed. REPO_PATH must already hold at least one of +# device-types/, module-types/, or rack-types/. REPO_BRANCH is ignored in this mode. +# Use it for air-gapped runs or a library you version yourself. +#REPO_URL=local + # Local path where the device-type library repository is cloned. # Defaults to a "repo" directory in the project root when not set. # Use an absolute path or a path relative to where you run the script. diff --git a/README.md b/README.md index 447a98e9..5b004a04 100644 --- a/README.md +++ b/README.md @@ -261,7 +261,7 @@ and device, creating anything that is missing from NetBox while skipping entries | --- | --- | --- | --- | | `NETBOX_URL` | ✅ | — | URL of your NetBox instance | | `NETBOX_TOKEN` | ✅ | — | API token with write access | -| `REPO_URL` | | community library | Git URL of the device-type library to clone | +| `REPO_URL` | | community library | Git URL of the device-type library to clone. Set to `local` to read `REPO_PATH` as it stands, with no git operation (see [Offline / local library](#offline--local-library)) | | `REPO_BRANCH` | | `master` | Branch to check out | | `REPO_PATH` | | `./repo` | Local path where the library is cloned. Accepts absolute or relative paths. | | `VENDORS` | | all | Comma-separated vendors to import (same effect as `--vendors`) | @@ -435,6 +435,29 @@ uv run nb-dt-import.py --vendors nokia --verify-images - After replacing a local image file with a higher-quality version and wanting NetBox to pick it up +#### Offline / local library + +Set `REPO_URL=local` to point the importer at a directory that already holds the library and +skip git completely: no clone, no fetch, and no `.git` needed. `REPO_PATH` must contain at +least one of `device-types/`, `module-types/`, or `rack-types/`, and an import stops with an +error if it does not, rather than reporting an empty library as nothing to do. + +Nothing in an import writes to `REPO_PATH`, so a read-only mount works. `REPO_BRANCH` is +ignored in this mode, and a run that sets both says so. + +```shell +REPO_URL=local REPO_PATH=/srv/devicetype-library uv run nb-dt-import.py +``` + +```shell +docker run --rm -e REPO_URL=local -e REPO_PATH=/library \ + -v /srv/devicetype-library:/library:ro \ + -e NETBOX_URL -e NETBOX_TOKEN ghcr.io/marcinpsk/device-type-library-import +``` + +**When to use**: air-gapped networks, a library you version yourself, or any run that must not +reach the network for the library. + #### Export Mode `--export-diff` runs in the opposite direction to every other mode: instead of importing the diff --git a/core/config.py b/core/config.py index f71d9074..7a67f694 100644 --- a/core/config.py +++ b/core/config.py @@ -11,6 +11,8 @@ DEFAULT_REPO_URL = "https://github.com/netbox-community/devicetype-library.git" DEFAULT_REPO_BRANCH = "master" +# REPO_URL value that reads REPO_PATH as it stands, with no git operation at all. +LOCAL_REPO_URL = "local" DEFAULT_GRAPHQL_PAGE_SIZE = 5000 DEFAULT_PRELOAD_THREADS = 8 @@ -20,6 +22,11 @@ _DEFAULT_REPO_PATH = f"{os.path.dirname(os.path.dirname(os.path.realpath(__file__)))}/repo" +def is_local_repo_url(url): + """Return True when *url* is the sentinel that turns off every git operation.""" + return str(url or "").strip().casefold() == LOCAL_REPO_URL + + class ConfigError(FatalError): """A configuration value the run cannot start with, phrased for the person who set it.""" @@ -102,7 +109,7 @@ def build_argument_parser(env): "--url", "--git", default=_text(env, "REPO_URL", DEFAULT_REPO_URL), - help="Git URL with valid Device Type YAML files", + help=f'Git URL with valid Device Type YAML files, or "{LOCAL_REPO_URL}" to read REPO_PATH with no git', ) parser.add_argument( "--slugs", @@ -264,6 +271,11 @@ def resolve_run_config(argv=None, env=None): # Only the environment can reach here: an explicit --slugs is rejected above. notices.append("Ignoring SLUGS from the environment: --export-diff does not filter by slug.") slugs = () + if is_local_repo_url(args.url) and args.branch != DEFAULT_REPO_BRANCH: + notices.append( + f"Ignoring REPO_BRANCH={args.branch}: REPO_URL={LOCAL_REPO_URL} reads REPO_PATH as it stands " + "and checks out no branch." + ) return RunConfig( netbox_url=_text(env, "NETBOX_URL"), diff --git a/core/export.py b/core/export.py index 02490761..c3e43aa1 100644 --- a/core/export.py +++ b/core/export.py @@ -28,12 +28,10 @@ serialize_rack_type, ) from core.netbox_api import IMAGE_EXTENSIONS, _build_auth_header +from core.repo import LIBRARY_TYPE_DIRS, library_dirs_present _SKIP = object() # sentinel: image already exists, no download needed -# Top-level directories that make a checkout a device-type library. -_LIBRARY_TYPE_DIRS = ("device-types", "module-types", "rack-types") - # Maps Content-Type to a canonical extension for extension-less attachments. _CONTENT_TYPE_EXT = { "image/png": ".png", @@ -474,11 +472,11 @@ def _write_export_items(self, items, manifest, manifest_path, progress) -> None: def _verify_repo_available(self) -> None: """Raise FileNotFoundError when the library is absent, which would otherwise read as an empty one.""" - if any((self.repo_path / name).is_dir() for name in _LIBRARY_TYPE_DIRS): + if library_dirs_present(self.repo_path): return raise FileNotFoundError( f"No device-type library found at {self.repo_path}: expected at least one of " - f"{', '.join(_LIBRARY_TYPE_DIRS)}. Export mode does not clone the library. " + f"{', '.join(LIBRARY_TYPE_DIRS)}. Export mode does not clone the library. " "Clone it to that path, or run an import first, which clones it for you." ) diff --git a/core/import_run.py b/core/import_run.py index 37a3caa9..00da6870 100644 --- a/core/import_run.py +++ b/core/import_run.py @@ -529,12 +529,15 @@ def _log_run_summary(handle, summary): handle.log("These duplicates would otherwise oscillate on every run. Please report/fix them upstream.") -def _parse_vendor_racks(repo, racks_path, vendor_name, slugs): - """Parse rack types for one vendor when the rack path exists.""" - if not os.path.isdir(racks_path): +def _parse_vendor_files(repo, base_path, vendor_name, slugs): + """Parse one vendor's types under *base_path*, treating an absent type root as empty. + + A local library may ship only some of the three type roots, which is a layout, not a fault. + """ + if not os.path.isdir(base_path): return [] - rack_files, _ = repo.get_devices(racks_path, [vendor_name.casefold()]) - return repo.parse_files(rack_files, slugs=slugs) + files, _ = repo.get_devices(base_path, [vendor_name.casefold()]) + return repo.parse_files(files, slugs=slugs) def _finalize_task_registry(progress, task_registry): @@ -641,16 +644,18 @@ def plan_vendor(self, selection, vendor): device_files = slug_resolved["device_files"].get(vendor["slug"], []) device_types = self.repo.parse_files(device_files) if device_files else [] else: - device_files, _ = self.repo.get_devices(selection.devices_path, [vendor["name"].casefold()]) - device_types = self.repo.parse_files(device_files, slugs=self.config.slugs or []) + device_types = _parse_vendor_files( + self.repo, selection.devices_path, vendor["name"], self.config.slugs or [] + ) if self.netbox.modules: module_hint = slug_resolved["module_vendors"] if slug_resolved is not None else None if module_hint is not None and vendor["slug"] not in module_hint: module_types = [] else: - module_files, _ = self.repo.get_devices(selection.modules_path, [vendor["name"].casefold()]) - module_types = self.repo.parse_files(module_files, slugs=self.config.slugs or []) + module_types = _parse_vendor_files( + self.repo, selection.modules_path, vendor["name"], self.config.slugs or [] + ) else: module_types = [] @@ -659,7 +664,7 @@ def plan_vendor(self, selection, vendor): if rack_hint is not None and vendor["slug"] not in rack_hint: rack_types = [] else: - rack_types = _parse_vendor_racks( + rack_types = _parse_vendor_files( self.repo, selection.racks_path, vendor["name"], diff --git a/core/repo.py b/core/repo.py index 74688641..57656278 100644 --- a/core/repo.py +++ b/core/repo.py @@ -11,8 +11,12 @@ from git import Repo, exc import yaml +from core.config import LOCAL_REPO_URL, is_local_repo_url from core.errors import FatalError, UnknownError +# Top-level directories that make a checkout a device-type library. +LIBRARY_TYPE_DIRS = ("device-types", "module-types", "rack-types") + class GitCommandError(FatalError): """A Git command that failed for a repository.""" @@ -184,6 +188,11 @@ def _safe_index_load(path: str): return _safe_pickle_load(path) +def library_dirs_present(path): + """Return True when *path* holds at least one device-type library directory.""" + return any(os.path.isdir(os.path.join(str(path), name)) for name in LIBRARY_TYPE_DIRS) + + def validate_git_url(url): """Determine whether a Git remote URL is allowed (HTTPS, SSH, or file://). @@ -419,11 +428,11 @@ class DTLRepo: def __init__(self, config, handle): """Initialize repository management, updating an existing clone or creating a new one. - If REPO_URL is the sentinel "local", no git operation of any kind is performed — - REPO_PATH is used as-is and must already contain the device-type file tree. - Otherwise, if the target path already holds a Git clone, the repository will be - updated from its configured remote; if not, the provided URL is validated and a new - clone is created. The initializer sets instance attributes used by other methods + If REPO_URL is the sentinel "local", no git operation runs at all: REPO_PATH is read + as it stands and must already hold the device-type file tree. Otherwise, if the target + path already holds a Git clone, the repository will be updated from its configured + remote; if not, the provided URL is validated and a new clone is created. The + initializer sets instance attributes used by other methods (handler, supported YAML extensions, URL, repo path, branch, repo reference, and current working directory). @@ -443,12 +452,8 @@ def __init__(self, config, handle): self.repo = None self.cwd = os.getcwd() - if str(self.url).strip().casefold() == "local": - if not os.path.isdir(self.get_absolute_path()): - raise InvalidRepoPathError( - self.repo_path, reason="REPO_URL=local requires REPO_PATH to already contain the library files" - ) - self.handle.log(f"REPO_URL=local: using {self.get_absolute_path()} as-is, no git operations") + if is_local_repo_url(self.url): + self._use_local_checkout() return is_path_valid, path_error = validate_repo_path(self.repo_path) @@ -467,6 +472,28 @@ def __init__(self, config, handle): raise InvalidGitURLError(self.url, reason=error_msg) self.clone_repo() + def _use_local_checkout(self): + """Accept REPO_PATH as it stands: no clone, no fetch, and no .git needed. + + Skips validate_repo_path on purpose: it demands write access, which a read-only + or air-gapped mount cannot give, and no import step writes to REPO_PATH. + """ + path = self.get_absolute_path() + if not os.path.isdir(path): + raise InvalidRepoPathError( + self.repo_path, + reason=f"REPO_URL={LOCAL_REPO_URL} needs REPO_PATH to be an existing directory", + ) + if not library_dirs_present(path): + raise InvalidRepoPathError( + self.repo_path, + reason=( + f"No device-type library found: expected at least one of {', '.join(LIBRARY_TYPE_DIRS)} " + f"inside it. REPO_URL={LOCAL_REPO_URL} does not clone the library." + ), + ) + self.handle.log(f"REPO_URL={LOCAL_REPO_URL}: reading {path} as it stands, with no git operation") + def get_relative_path(self): """Get the repository path configured for this instance relative to the current working directory. diff --git a/tests/test_config.py b/tests/test_config.py index 0bb65ee4..1aae1838 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -120,3 +120,22 @@ def test_an_explicit_repo_path_is_taken_as_given(self, tmp_path): config = _resolve(REPO_PATH=str(tmp_path)) assert config.repo_path == str(tmp_path) + + +class TestLocalRepoUrlIgnoresTheBranch: + """REPO_URL=local checks out nothing, so a REPO_BRANCH set beside it silently does nothing.""" + + def test_a_branch_set_beside_the_sentinel_is_reported_as_ignored(self): + config = _resolve(REPO_URL="local", REPO_BRANCH="feature") + + assert any("REPO_BRANCH" in notice for notice in config.notices), config.notices + + def test_the_sentinel_alone_needs_no_notice(self): + config = _resolve(REPO_URL="local") + + assert not any("REPO_BRANCH" in notice for notice in config.notices), config.notices + + def test_a_branch_set_beside_a_real_url_needs_no_notice(self): + config = _resolve(REPO_URL="https://example.com/repo.git", REPO_BRANCH="feature") + + assert not any("REPO_BRANCH" in notice for notice in config.notices), config.notices diff --git a/tests/test_exporter.py b/tests/test_exporter.py index ae5d1797..a356e528 100644 --- a/tests/test_exporter.py +++ b/tests/test_exporter.py @@ -1485,3 +1485,33 @@ def test_all_known_content_types_are_recognised(self): for ct, ext in _CONTENT_TYPE_EXT.items(): result = _sanitize_attachment_filename("img", "/media/img", ct) assert result.endswith(ext), f"Expected {ext} for {ct}, got {result}" + + +class TestRepoAvailability: + """Export mode and REPO_URL=local must agree on what makes a directory a library checkout.""" + + def _exporter(self, tmp_path, repo_path): + settings = replace(_make_settings(tmp_path), repo_path=str(repo_path)) + return Exporter(settings, _make_handle(), str(tmp_path / "extra"), False, None) + + @pytest.mark.parametrize("present", ["device-types", "module-types", "rack-types"]) + def test_one_library_directory_is_enough(self, present, tmp_path): + repo = tmp_path / "repo" + (repo / present).mkdir(parents=True) + + self._exporter(tmp_path, repo)._verify_repo_available() + + def test_a_directory_without_library_directories_stops_the_export(self, tmp_path): + repo = tmp_path / "repo" + repo.mkdir() + + with pytest.raises(FileNotFoundError, match="No device-type library found"): + self._exporter(tmp_path, repo)._verify_repo_available() + + def test_a_stray_file_named_like_a_library_directory_stops_the_export(self, tmp_path): + repo = tmp_path / "repo" + repo.mkdir() + (repo / "device-types").write_text("not a directory") + + with pytest.raises(FileNotFoundError, match="No device-type library found"): + self._exporter(tmp_path, repo)._verify_repo_available() diff --git a/tests/test_import_run.py b/tests/test_import_run.py index caeb96fa..59f14405 100644 --- a/tests/test_import_run.py +++ b/tests/test_import_run.py @@ -8,6 +8,7 @@ from core.errors import VendorSelectionError from core.import_run import ImportRun, RunSummary, VendorPlan, _process_device_types from core.log_handler import LogHandler +from core.repo import DTLRepo class _ComponentCache: @@ -277,3 +278,33 @@ def test_execute_releases_console_when_progress_setup_fails(make_config, tmp_pat assert progress_factory.exited is True assert handle.console is None assert netbox.device_types.components.close_count == 1 + + +class TestPartialLibraryLayouts: + """A local checkout may hold one type root only, so planning must not read the absent ones.""" + + TYPE_DIRS = ("device-types", "module-types", "rack-types") + + def _repo(self, tmp_path, present, make_config): + """Build a real DTLRepo over a checkout holding *present* alone.""" + vendor = tmp_path / present / "TestVendor" + vendor.mkdir(parents=True) + (vendor / "thing.yaml").write_text("manufacturer: TestVendor\nmodel: Test\nslug: test\n") + config = make_config(repo_url="local", repo_path=str(tmp_path)) + return DTLRepo(config, LogHandler(False)), config + + @pytest.mark.parametrize("present", TYPE_DIRS) + def test_one_type_root_plans_without_reading_the_absent_ones(self, present, make_config, tmp_path): + repo, config = self._repo(tmp_path, present, make_config) + run = ImportRun(config, repo, _NetBoxBoundary(), LogHandler(False), _ProgressFactory()) + + selection = run.discover() + plan = run.plan_vendor(selection, {"name": "TestVendor", "slug": "testvendor"}) + + parsed = { + "device-types": len(plan.device_types), + "module-types": len(plan.module_types), + "rack-types": len(plan.rack_types), + } + assert parsed[present] == 1, parsed + assert sum(parsed.values()) == 1, parsed diff --git a/tests/test_nb_dt_import.py b/tests/test_nb_dt_import.py index 71da96d2..929231ca 100644 --- a/tests/test_nb_dt_import.py +++ b/tests/test_nb_dt_import.py @@ -308,13 +308,27 @@ def test_items_per_second_column_uses_elapsed_fallback_when_finished_speed_missi _PROJECT_ROOT = Path(__file__).resolve().parents[1] +_LIBRARY_ROOT = None + + +@pytest.fixture(scope="session", autouse=True) +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 + + def _make_mock_repo(device_types=None): """Return a pre-configured DTLRepo mock with no files by default.""" mock_repo = MagicMock() mock_repo.get_devices.return_value = ([], []) - mock_repo.get_devices_path.return_value = "/tmp/devices" - mock_repo.get_modules_path.return_value = "/tmp/modules" - mock_repo.get_racks_path.return_value = "/tmp/rack-types" + mock_repo.get_devices_path.return_value = str(_LIBRARY_ROOT / "device-types") + mock_repo.get_modules_path.return_value = str(_LIBRARY_ROOT / "module-types") + mock_repo.get_racks_path.return_value = str(_LIBRARY_ROOT / "rack-types") mock_repo.discover_vendors.return_value = [] mock_repo.parse_files.return_value = device_types if device_types is not None else [] mock_repo.resolve_slug_files.return_value = None # no pickle available by default @@ -905,7 +919,7 @@ def test_modules_update_mode_logs_change_detection_section(self, nb_dt_import): repo = _make_mock_repo() repo.discover_vendors.return_value = [{"name": "Vendor One", "slug": "vendor-one"}] repo.get_devices.side_effect = lambda path, vendors=None: ( - (["module.yaml"], []) if "modules" in path else ([], []) + (["module.yaml"], []) if "module-types" in path else ([], []) ) repo.parse_files.side_effect = lambda files, slugs=None: [module_type] if files == ["module.yaml"] else [] MockRepo.return_value = repo @@ -1600,13 +1614,13 @@ def test_no_pulse_bar_column_uses_static_empty_bar_for_unknown_total(self, nb_dt assert bar.total == 1.0 assert bar.completed == 0.0 - def test_parse_vendor_racks_calls_repo_when_directory_exists(self, nb_dt_import): + def test_parse_vendor_files_calls_repo_when_directory_exists(self, nb_dt_import): repo = MagicMock() repo.get_devices.return_value = (["rack.yaml"], []) repo.parse_files.return_value = [{"model": "Rack"}] with patch("core.import_run.os.path.isdir", return_value=True): - result = import_run_module._parse_vendor_racks(repo, "/racks", "nokia", ["rack"]) + result = import_run_module._parse_vendor_files(repo, "/racks", "nokia", ["rack"]) assert result == [{"model": "Rack"}] repo.get_devices.assert_called_once_with("/racks", ["nokia"]) @@ -1844,7 +1858,7 @@ def _parse_files(files, slugs=None, progress=None): return [] mock_repo.get_devices.side_effect = lambda path, vendors=None: ( - (["device.yaml"], []) if path == "/tmp/devices" else ([], []) + (["device.yaml"], []) if path.endswith("device-types") else ([], []) ) mock_repo.parse_files.side_effect = _parse_files mock_nb = _make_mock_netbox() @@ -1878,7 +1892,7 @@ def test_main_stops_preload_job_in_finally_on_error(self, nb_dt_import): mock_repo = _make_mock_repo() mock_repo.discover_vendors.return_value = [{"name": "Cisco", "slug": "cisco"}] mock_repo.get_devices.side_effect = lambda path, vendors=None: ( - (["device.yaml"], []) if path == "/tmp/devices" else ([], []) + (["device.yaml"], []) if path.endswith("device-types") else ([], []) ) mock_repo.parse_files.side_effect = lambda files, slugs=None, progress=None: ( [{"manufacturer": {"slug": "cisco"}, "model": "X", "slug": "x"}] if files == ["device.yaml"] else [] @@ -1964,14 +1978,14 @@ def test_import_run_processes_slug_fast_path_and_skips_empty_vendor(self, make_c dtl_repo = _make_mock_repo() dtl_repo.get_devices.side_effect = lambda path, vendors=None: ( ([f"{vendors[0]}-{path.split('/')[-1]}.yaml"], []) - if vendors and vendors[0] == "cisco" and path in {"/tmp/modules", "/tmp/rack-types"} + if vendors and vendors[0] == "cisco" and path.endswith(("module-types", "rack-types")) else ([], []) ) dtl_repo.parse_files.side_effect = lambda files, slugs=None, progress=None: ( [{"manufacturer": {"slug": "cisco"}, "model": "X", "slug": "x"}] if files == ["resolved.yaml"] else [{"manufacturer": {"slug": "cisco"}, "model": "M", "slug": "m"}] - if files == ["cisco-modules.yaml"] + if files == ["cisco-module-types.yaml"] else [] ) netbox = _make_mock_netbox(modules=True) @@ -2030,7 +2044,7 @@ def test_import_run_stops_preload_in_finally_on_error(self, make_config): dtl_repo = _make_mock_repo() dtl_repo.discover_vendors.return_value = [{"slug": "cisco", "name": "Cisco"}] dtl_repo.get_devices.side_effect = lambda path, vendors=None: ( - (["device.yaml"], []) if path == "/tmp/devices" else ([], []) + (["device.yaml"], []) if path.endswith("device-types") else ([], []) ) dtl_repo.parse_files.side_effect = lambda files, slugs=None, progress=None: ( [{"manufacturer": {"slug": "cisco"}, "model": "X", "slug": "x"}] if files == ["device.yaml"] else [] diff --git a/tests/test_repo.py b/tests/test_repo.py index a260c8bb..ab4e695a 100644 --- a/tests/test_repo.py +++ b/tests/test_repo.py @@ -154,6 +154,97 @@ def test_invalid_path_raises_before_repository_access(self, tmp_path): _dtl_repo(config, str(invalid_path), LogHandler(False)) +class TestDTLRepoLocalMode: + """REPO_URL=local reads REPO_PATH as it stands: no clone, no fetch, no .git.""" + + @staticmethod + def _make_library(path, *dir_names): + """Create a library checkout at *path* holding one YAML file under each named directory.""" + path.mkdir(parents=True, exist_ok=True) + for name in dir_names: + vendor = path / name / "TestVendor" + vendor.mkdir(parents=True) + (vendor / "device.yaml").write_text("manufacturer: TestVendor\nmodel: Test\n") + return path + + def _init_local(self, repo_path, url="local", handle=None): + config = MagicMock(repo_url=url, repo_branch="master") + return _dtl_repo(config, str(repo_path), handle or LogHandler(False)) + + def test_reads_the_checkout_without_running_git(self, tmp_path, mock_git_repo): + """The whole point of the sentinel: the files are read and git is never reached.""" + library = self._make_library(tmp_path / "library", "device-types") + + repo = self._init_local(library) + files, vendors = repo.get_devices(repo.get_devices_path()) + + assert [os.path.basename(path) for path in files] == ["device.yaml"] + assert vendors == [{"name": "TestVendor", "slug": "testvendor"}] + assert [parsed["model"] for parsed in repo.parse_files(files)] == ["Test"] + assert not (library / ".git").exists() + mock_git_repo.assert_not_called() + mock_git_repo.clone_from.assert_not_called() + + @pytest.mark.parametrize("value", ["local", "LOCAL", "Local", " local "]) + def test_the_sentinel_ignores_case_and_padding(self, value, tmp_path, mock_git_repo): + library = self._make_library(tmp_path / "library", "device-types") + + self._init_local(library, url=value) + + mock_git_repo.assert_not_called() + mock_git_repo.clone_from.assert_not_called() + + @pytest.mark.parametrize("present", ["device-types", "module-types", "rack-types"]) + def test_one_library_directory_is_enough(self, present, tmp_path, mock_git_repo): + """Export mode accepts any one of the three, so local import mode must agree.""" + library = self._make_library(tmp_path / "library", present) + + self._init_local(library) + + mock_git_repo.assert_not_called() + + def test_a_missing_directory_is_reported_as_a_path_error(self, tmp_path): + with pytest.raises(InvalidRepoPathError, match="existing directory"): + self._init_local(tmp_path / "absent") + + def test_a_directory_without_library_directories_is_rejected(self, tmp_path): + """An unrelated path would otherwise import nothing and still exit successfully.""" + empty = tmp_path / "library" + empty.mkdir() + + with pytest.raises(InvalidRepoPathError, match="No device-type library found"): + self._init_local(empty) + + def test_a_stray_file_named_like_a_library_directory_is_rejected(self, tmp_path): + library = tmp_path / "library" + library.mkdir() + (library / "device-types").write_text("not a directory") + + with pytest.raises(InvalidRepoPathError, match="No device-type library found"): + self._init_local(library) + + @pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="root ignores the write bit") + def test_a_read_only_checkout_is_accepted(self, tmp_path): + """The air-gapped case: nothing writes to REPO_PATH, so a read-only mount must work.""" + library = self._make_library(tmp_path / "library", "device-types") + library.chmod(0o555) + + try: + repo = self._init_local(library) + assert repo.get_devices(repo.get_devices_path())[0] + finally: + library.chmod(0o755) + + def test_the_branch_is_left_unused(self, tmp_path, mock_git_repo): + """No checkout happens, so REPO_BRANCH must not reach git (config warns about it instead).""" + library = self._make_library(tmp_path / "library", "device-types") + config = MagicMock(repo_url="local", repo_branch="a-branch-that-does-not-exist") + + _dtl_repo(config, str(library), LogHandler(False)) + + mock_git_repo.assert_not_called() + + class TestDTLRepoRealGit: """Clone/pull branching driven against a real local Git repository (no git mocks)."""