Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions .env.example
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
25 changes: 24 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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`) |
Expand Down Expand Up @@ -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
Expand Down
14 changes: 13 additions & 1 deletion core/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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."""

Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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"),
Expand Down
8 changes: 3 additions & 5 deletions core/export.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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."
)

Expand Down
25 changes: 15 additions & 10 deletions core/import_run.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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 = []

Expand All @@ -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"],
Expand Down
47 changes: 42 additions & 5 deletions core/repo.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down Expand Up @@ -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://).

Expand Down Expand Up @@ -419,11 +428,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 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).

Args:
config (RunConfig): Supplies `repo_url`, `repo_branch`, and `repo_path`.
Expand All @@ -441,6 +452,10 @@ def __init__(self, config, handle):
self.repo = None
self.cwd = os.getcwd()

if is_local_repo_url(self.url):
self._use_local_checkout()
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)
Expand All @@ -457,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.

Expand Down
19 changes: 19 additions & 0 deletions tests/test_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
30 changes: 30 additions & 0 deletions tests/test_exporter.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
31 changes: 31 additions & 0 deletions tests/test_import_run.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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
Loading