diff --git a/.devcontainer/config/isolated_test_settings.py b/.devcontainer/config/isolated_test_settings.py index 5270c9bb..080f05de 100644 --- a/.devcontainer/config/isolated_test_settings.py +++ b/.devcontainer/config/isolated_test_settings.py @@ -1,7 +1,12 @@ # SPDX-License-Identifier: Apache-2.0 # Copyright (C) 2025 Marcin Zieba # -# Isolated test-database settings shim. +# Isolated test settings for the performance runner. +# +# The plugin's pytest suite uses ``netbox_interface_name_rules.tests.isolated_settings``, which +# gives every worker private databases. This shim serves ``manage.py test`` runs only: the +# performance runner measures the plugin under the devcontainer's full plugin list, so narrowing +# it here would change the environment the committed baselines were taken in. # # Django names the test database ``test_`` (here: ``test_netbox``), so two # ``manage.py test`` runs in the same devcontainer collide on a single test DB and @@ -22,3 +27,10 @@ _name = _os.environ.get("TEST_DB_NAME") if _name: DATABASES["default"].setdefault("TEST", {})["NAME"] = _name # noqa: F405 + +# NetBox writes the search cache inline only when no RQ worker serves the queue, so +# ``TEST_REDIS_DB`` moves the queues off the database the devcontainer's worker holds. +_redis_db = _os.environ.get("TEST_REDIS_DB") +if _redis_db: + for _queue in RQ_QUEUES.values(): # noqa: F405 + _queue["DB"] = int(_redis_db) diff --git a/.devcontainer/docker-compose.yml b/.devcontainer/docker-compose.yml index dabb9faf..7e1d6ed6 100644 --- a/.devcontainer/docker-compose.yml +++ b/.devcontainer/docker-compose.yml @@ -24,6 +24,9 @@ services: SUPERUSER_EMAIL: ${SUPERUSER_EMAIL:-admin@example.com} SUPERUSER_PASSWORD: ${SUPERUSER_PASSWORD:-admin} SKIP_SUPERUSER: ${SKIP_SUPERUSER:-false} + # The container is root and mounts the host checkout, so cached bytecode lands there owned + # by root. `-p no:cacheprovider` does not prevent it. + PYTHONDONTWRITEBYTECODE: "1" # Proxy settings (optional) HTTP_PROXY: ${HTTP_PROXY:-} HTTPS_PROXY: ${HTTPS_PROXY:-} @@ -31,9 +34,11 @@ services: https_proxy: ${HTTPS_PROXY:-} NO_PROXY: ${NO_PROXY:-} no_proxy: ${NO_PROXY:-} - REQUESTS_CA_BUNDLE: ${REQUESTS_CA_BUNDLE:-} - SSL_CERT_FILE: ${SSL_CERT_FILE:-} - CURL_CA_BUNDLE: ${CURL_CA_BUNDLE:-} + # setup.sh installs any ca-bundle.crt into this store, so a host path here + # would only name a file the container cannot open. + REQUESTS_CA_BUNDLE: /etc/ssl/certs/ca-certificates.crt + SSL_CERT_FILE: /etc/ssl/certs/ca-certificates.crt + CURL_CA_BUNDLE: /etc/ssl/certs/ca-certificates.crt depends_on: postgres: condition: service_healthy @@ -45,7 +50,12 @@ services: postgres: image: postgres:18 - command: postgres -c max_connections=200 + # PostgreSQL 18 allocates a dynamic shared memory segment at startup and fails + # to start on Docker's 64MB default /dev/shm. + shm_size: 1gb + # Parallel test runs open a connection per worker per test database, and this + # postgres serves several test databases at once. + command: postgres -c max_connections=800 environment: POSTGRES_DB: ${DB_NAME:-netbox} POSTGRES_USER: ${DB_USER:-netbox} @@ -89,3 +99,6 @@ networks: config: # Pinned to avoid Docker auto-assigning 172.30.x.x - subnet: "172.28.200.0/24" + # Dynamic assignment stays above .128 so no unpinned service (e.g. a local + # override) can take the static postgres/redis addresses setup.sh uses. + ip_range: "172.28.200.128/25" diff --git a/.devcontainer/scripts/load-aliases.sh b/.devcontainer/scripts/load-aliases.sh index 44f24b28..8390d8ec 100755 --- a/.devcontainer/scripts/load-aliases.sh +++ b/.devcontainer/scripts/load-aliases.sh @@ -119,14 +119,23 @@ netbox-shell() { cd /opt/netbox/netbox && source /opt/netbox/venv/bin/activate && python manage.py shell } +# Run the plugin suite with pytest, the runner CI uses. The settings module gives each xdist +# worker private PostgreSQL and Redis databases, so concurrent suites in the shared devcontainer +# do not collide. Override the targets with TEST_DB_NAME=... / TEST_REDIS_HOST=... . netbox-test() { - cd /opt/netbox/netbox && source /opt/netbox/venv/bin/activate && python manage.py test netbox_interface_name_rules "$@" + if [ "$#" -eq 0 ]; then + set -- netbox_interface_name_rules + fi + cd "$PLUGIN_DIR" && source /opt/netbox/venv/bin/activate && \ + TEST_DB_NAME="${TEST_DB_NAME:-test_netbox_interface_name_rules}" \ + TEST_REDIS_HOST="${TEST_REDIS_HOST:-redis}" \ + pytest "$@" } -# Run tests on a per-session ISOLATED test database, so concurrent suites in the -# shared devcontainer don't collide on test_netbox (which corrupts migrations and -# can leave lock-holding zombie connections). Pass the app(s) + any test flags, -# e.g. netbox-test-isolated netbox_nso_plugin --keepdb +# Run a Django-runner suite on an isolated test database. The plugin's own suite runs under +# pytest via netbox-test; this stays for the performance runner and for other apps. +# Sharing test_netbox corrupts migrations and can leave lock-holding zombie connections. +# Pass the app(s) + any test flags, e.g. netbox-test-isolated netbox_nso_plugin --keepdb # Override the DB name with TEST_DB_NAME=...; otherwise it's derived from the first # app argument (a stable name so --keepdb can reuse it). netbox-test-isolated() { diff --git a/.devcontainer/scripts/setup.sh b/.devcontainer/scripts/setup.sh index 8a12c09a..e45edbcb 100755 --- a/.devcontainer/scripts/setup.sh +++ b/.devcontainer/scripts/setup.sh @@ -137,7 +137,7 @@ fi echo "🔧 Installing development dependencies..." apt-get update -qq -apt-get install -y -qq net-tools git +apt-get install -y -qq net-tools git ripgrep # Dev tools used by the agent loop and pre-commit hooks. Keep in sync with # the `dev` extras in pyproject.toml — at minimum, anything invoked by: # - test + coverage runs: pytest, pytest-django, pytest-cov, pytest-xdist @@ -172,6 +172,7 @@ if [ -z "$PLUGIN_WS_DIR" ]; then fi echo "📂 Plugin workspace: $PLUGIN_WS_DIR" cd "$PLUGIN_WS_DIR" +$PIP_CMD install --group workflow-tests $PIP_CMD install -e . echo "✅ Installed $PLUGIN_NAME in editable mode" diff --git a/.devcontainer/scripts/tests/lib.sh b/.devcontainer/scripts/tests/lib.sh new file mode 100644 index 00000000..1f972e9b --- /dev/null +++ b/.devcontainer/scripts/tests/lib.sh @@ -0,0 +1,18 @@ +#!/bin/bash +# SPDX-License-Identifier: Apache-2.0 +# Copyright (C) 2025 Marcin Zieba + +# Populate COMPOSE_FILES with the `-f` arguments a real recreate would use. +# +# The override is untracked and per-developer, but Docker applies it to every recreate. A check that +# renders only the base file cannot see a value the override replaces, so it would pass while the +# container it protects comes up wrong. +compose_file_args() { + local root="$1" + local override="$root/.devcontainer/docker-compose.override.yml" + + COMPOSE_FILES=(-f "$root/.devcontainer/docker-compose.yml") + if [ -f "$override" ]; then + COMPOSE_FILES+=(-f "$override") + fi +} diff --git a/.devcontainer/scripts/tests/test-bytecode-writes.sh b/.devcontainer/scripts/tests/test-bytecode-writes.sh new file mode 100755 index 00000000..827a847b --- /dev/null +++ b/.devcontainer/scripts/tests/test-bytecode-writes.sh @@ -0,0 +1,46 @@ +#!/bin/bash +# SPDX-License-Identifier: Apache-2.0 +# Copyright (C) 2025 Marcin Zieba + +set -euo pipefail + +REPO_ROOT="$(cd "$(dirname "$0")/../../.." && pwd)" +# shellcheck source=.devcontainer/scripts/tests/lib.sh +source "$(dirname "$0")/lib.sh" +compose_file_args "$REPO_ROOT" + +# The container runs as root and mounts the host checkout, so any bytecode Python caches lands in +# the developer's tree owned by root. Those files can block `git worktree remove` and host tooling. +docker compose "${COMPOSE_FILES[@]}" config --format json | python3 -c " +import json +import sys + +configuration = json.load(sys.stdin) +services = configuration.get('services', {}) + +failures = [] +for name, service in services.items(): + # Only services that mount the host checkout can write into it. + mounts = service.get('volumes') or [] + if not any(str(mount.get('target', '')).startswith('/workspaces') for mount in mounts): + continue + value = (service.get('environment') or {}).get('PYTHONDONTWRITEBYTECODE') + if str(value) != '1': + failures.append(f'{name} mounts the checkout but sets PYTHONDONTWRITEBYTECODE={value!r}') + +if failures: + for failure in failures: + print(f'FAIL: {failure}', file=sys.stderr) + print( + 'Root-owned .pyc files would accumulate in the host tree. ' + 'Set PYTHONDONTWRITEBYTECODE: \"1\" on that service.', + file=sys.stderr, + ) + raise SystemExit(1) + +if not services: + print('FAIL: the compose file declares no services', file=sys.stderr) + raise SystemExit(1) + +print('Bytecode-write check passed (no checkout-mounting service caches bytecode)') +" diff --git a/.devcontainer/scripts/tests/test-ca-environment.sh b/.devcontainer/scripts/tests/test-ca-environment.sh new file mode 100755 index 00000000..a74c7791 --- /dev/null +++ b/.devcontainer/scripts/tests/test-ca-environment.sh @@ -0,0 +1,38 @@ +#!/bin/bash +# SPDX-License-Identifier: Apache-2.0 +# Copyright (C) 2025 Marcin Zieba + +set -euo pipefail + +REPO_ROOT="$(cd "$(dirname "$0")/../../.." && pwd)" +# shellcheck source=.devcontainer/scripts/tests/lib.sh +source "$(dirname "$0")/lib.sh" +compose_file_args "$REPO_ROOT" +CONTAINER_CA_BUNDLE="/etc/ssl/certs/ca-certificates.crt" +HOST_CA_BUNDLE="/host-only/ca.pem" + +config="$({ + REQUESTS_CA_BUNDLE="$HOST_CA_BUNDLE" \ + SSL_CERT_FILE="$HOST_CA_BUNDLE" \ + CURL_CA_BUNDLE="$HOST_CA_BUNDLE" \ + docker compose "${COMPOSE_FILES[@]}" config +})" + +# Compare the rendered value as a string. A grep pattern would read every "." in a certificate +# path as a wildcard, so a wrong path such as ca-certificatesXcrt would satisfy the check. +for variable in REQUESTS_CA_BUNDLE SSL_CERT_FILE CURL_CA_BUNDLE; do + value="$(awk -v key="$variable:" '$1 == key { print $2; exit }' <<< "$config")" + value="${value%\"}" + value="${value#\"}" + if [ "$value" != "$CONTAINER_CA_BUNDLE" ]; then + echo "FAIL: $variable is '$value', not the container trust store $CONTAINER_CA_BUNDLE" >&2 + exit 1 + fi +done + +if grep -qF "$HOST_CA_BUNDLE" <<< "$config"; then + echo "FAIL: a host CA path leaked into the container configuration" >&2 + exit 1 +fi + +echo "Container CA environment check passed" diff --git a/.devcontainer/scripts/tests/test-netbox-test-targets.sh b/.devcontainer/scripts/tests/test-netbox-test-targets.sh new file mode 100644 index 00000000..b6e5a4fc --- /dev/null +++ b/.devcontainer/scripts/tests/test-netbox-test-targets.sh @@ -0,0 +1,33 @@ +#!/bin/bash +# SPDX-License-Identifier: Apache-2.0 +# Copyright (C) 2025 Marcin Zieba + +set -eo pipefail + +REPO_ROOT="$(cd "$(dirname "$0")/../../.." && pwd)" +TEST_DIR="$(mktemp -d)" +trap 'rm -rf "$TEST_DIR"' EXIT +export PYTEST_ARGV="$TEST_DIR/argv" + +source() { + if [ "$1" = /opt/netbox/venv/bin/activate ]; then + return 0 + fi + builtin source "$@" +} +source "$REPO_ROOT/.devcontainer/scripts/load-aliases.sh" +cat > "$TEST_DIR/pytest" <<'STUB' +#!/bin/bash +printf '%s\n' "$@" > "$PYTEST_ARGV" +STUB +chmod +x "$TEST_DIR/pytest" +export PATH="$TEST_DIR:$PATH" + +netbox-test +printf '%s\n' netbox_interface_name_rules > "$TEST_DIR/expected" +diff -u "$TEST_DIR/expected" "$PYTEST_ARGV" + +netbox-test netbox_interface_name_rules/tests/test_rules.py -k 'a test name' +printf '%s\n' netbox_interface_name_rules/tests/test_rules.py -k 'a test name' > "$TEST_DIR/expected" +diff -u "$TEST_DIR/expected" "$PYTEST_ARGV" +printf '%s\n' 'NetBox test target check passed' diff --git a/.devcontainer/scripts/tests/test-network-pins.sh b/.devcontainer/scripts/tests/test-network-pins.sh new file mode 100755 index 00000000..d1567644 --- /dev/null +++ b/.devcontainer/scripts/tests/test-network-pins.sh @@ -0,0 +1,48 @@ +#!/bin/bash +# SPDX-License-Identifier: Apache-2.0 +# Copyright (C) 2025 Marcin Zieba + +set -euo pipefail + +REPO_ROOT="$(cd "$(dirname "$0")/../../.." && pwd)" +# shellcheck source=.devcontainer/scripts/tests/lib.sh +source "$(dirname "$0")/lib.sh" +compose_file_args "$REPO_ROOT" + +if [ -n "${NETWORK_PINS_CONFIG:-}" ]; then + config="$(cat -- "$NETWORK_PINS_CONFIG")" +else + config="$(docker compose "${COMPOSE_FILES[@]}" config)" +fi + +mapfile -t ranges < <(sed -n 's/^[[:space:]]*ip_range:[[:space:]]*"\{0,1\}\([^"]*\)"\{0,1\}$/\1/p' <<< "$config") +if [ "${#ranges[@]}" -eq 0 ]; then + echo "FAIL: the default network declares no ip_range, so Docker may hand a pinned address to another service" >&2 + exit 1 +fi + +if [ "${#ranges[@]}" -gt 1 ]; then + echo "FAIL: expected one ip_range, found ${#ranges[@]}" >&2 + exit 1 +fi +ip_range="${ranges[0]}" + +mapfile -t pinned < <(sed -n 's/^[[:space:]]*ipv4_address:[[:space:]]*"\{0,1\}\([^"]*\)"\{0,1\}$/\1/p' <<< "$config") +if [ "${#pinned[@]}" -eq 0 ]; then + echo "FAIL: no service pins an ipv4_address, so this check has nothing to protect" >&2 + exit 1 +fi + +# Every pinned address must sit outside the pool Docker assigns from. +for address in "${pinned[@]}"; do + verdict="$(python3 -c ' +import ipaddress, sys +print("inside" if ipaddress.ip_address(sys.argv[1]) in ipaddress.ip_network(sys.argv[2]) else "outside") +' "$address" "$ip_range")" + if [ "$verdict" = "inside" ]; then + echo "FAIL: pinned address $address lies inside the dynamic range $ip_range" >&2 + exit 1 + fi +done + +echo "Network pin check passed (${#pinned[@]} pinned addresses outside $ip_range)" diff --git a/.github/workflows/lint-format.yaml b/.github/workflows/lint-format.yaml index 79d2d3cb..1daac334 100644 --- a/.github/workflows/lint-format.yaml +++ b/.github/workflows/lint-format.yaml @@ -84,3 +84,6 @@ jobs: - name: Run devcontainer script tests run: | bash .devcontainer/scripts/tests/test-debug-toolbar-patches.sh + bash .devcontainer/scripts/tests/test-ca-environment.sh + bash .devcontainer/scripts/tests/test-network-pins.sh + bash .devcontainer/scripts/tests/test-bytecode-writes.sh diff --git a/.github/workflows/test-netbox-main.yaml b/.github/workflows/test-netbox-main.yaml index a7a0fb21..748446e6 100644 --- a/.github/workflows/test-netbox-main.yaml +++ b/.github/workflows/test-netbox-main.yaml @@ -73,7 +73,8 @@ jobs: working-directory: netbox-InterfaceNameRules-plugin run: | uv pip install --system -r ../netbox/requirements.txt - uv pip install --system --only-binary=:all: pytest==9.0.2 pytest-django==4.12.0 pytest-xdist==3.8.0 tblib==3.2.2 + uv pip install --system --only-binary=:all: pytest==9.0.2 pytest-cov==7.0.0 pytest-django==4.12.0 pytest-xdist==3.8.0 tblib==3.2.2 + uv pip install --system --only-binary=:all: --group workflow-tests uv pip install --system -e . - name: Set up NetBox configuration @@ -104,5 +105,7 @@ jobs: env: NETBOX_CONFIGURATION: netbox.configuration PYTHONPATH: ${{ github.workspace }}/netbox/netbox + TEST_DB_NAME: test_netbox_interface_name_rules_main_ci + TEST_REDIS_HOST: localhost run: | - pytest -n auto netbox_interface_name_rules -o pythonpath=../netbox/netbox + pytest -n auto netbox_interface_name_rules --no-cov -o pythonpath=../netbox/netbox diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index 5a51055f..0f0e451c 100644 --- a/.github/workflows/test.yaml +++ b/.github/workflows/test.yaml @@ -95,6 +95,7 @@ jobs: run: | uv pip install --system -r ../netbox/requirements.txt uv pip install --system --only-binary=:all: pytest==9.0.2 pytest-cov==7.0.0 pytest-django==4.12.0 pytest-xdist==3.8.0 tblib==3.2.2 + uv pip install --system --only-binary=:all: --group workflow-tests uv pip install --system -e . - name: Set up NetBox configuration @@ -126,6 +127,8 @@ jobs: NETBOX_CONFIGURATION: netbox.configuration PYTHONPATH: ${{ github.workspace }}/netbox/netbox COVERAGE_RCFILE: pyproject.toml + TEST_DB_NAME: test_netbox_interface_name_rules_ci + TEST_REDIS_HOST: localhost # Turns the channelization tests' skipUnless guard into an assertion on the leg that must # have the feature, so a broken probe cannot silently skip the whole file. EXPECT_NETBOX_CHANNELIZATION: ${{ (matrix.netbox-version == 'feature' || matrix.netbox-version == 'v4.7.0') && '1' || '' }} @@ -136,8 +139,7 @@ jobs: # condition, and tests/query_counts.json together. UPDATE_QUERY_COUNTS: ${{ matrix.netbox-version != 'v4.7.0' && '1' || '' }} run: | - pytest -n auto netbox_interface_name_rules --cov=netbox_interface_name_rules --cov-report=term-missing \ - -o pythonpath=../netbox/netbox + pytest -n auto netbox_interface_name_rules -o pythonpath=../netbox/netbox - name: Generate coverage report if: matrix.netbox-version == 'v4.5.3' diff --git a/REUSE.toml b/REUSE.toml index f48bb262..7c4031de 100644 --- a/REUSE.toml +++ b/REUSE.toml @@ -35,6 +35,12 @@ path = "netbox_interface_name_rules/tests/query_counts.json" SPDX-FileCopyrightText = "2025 Marcin Zieba " SPDX-License-Identifier = "Apache-2.0" +# The comment-block permit list is generated from the tree, and JSON takes no header comment. +[[annotations]] +path = "netbox_interface_name_rules/tests/comment_blocks.json" +SPDX-FileCopyrightText = "2025 Marcin Zieba " +SPDX-License-Identifier = "Apache-2.0" + # CODEOWNERS (auto-request review); added via API so it had no license info — annotate it here. [[annotations]] path = ".github/CODEOWNERS" diff --git a/conftest.py b/conftest.py index e12d9215..0ac11a9c 100644 --- a/conftest.py +++ b/conftest.py @@ -1,8 +1,15 @@ # SPDX-License-Identifier: Apache-2.0 # Copyright (C) 2025 Marcin Zieba -"""Root pytest configuration for shared-host parallelism.""" +"""Root pytest configuration for shared-host parallelism. -MAX_PARALLEL_WORKERS = 8 +pytest loads this file before pytest-xdist resolves the `-n auto` in the addopts, and it loads +`netbox_interface_name_rules/tests/conftest.py` only during collection, after the workers start. The +worker cap therefore has to live here to reach an invocation that names no test path. +""" + +import pytest + +from netbox_interface_name_rules.tests.parallel import MAX_PARALLEL_WORKERS def pytest_xdist_auto_num_workers(config): @@ -10,3 +17,25 @@ def pytest_xdist_auto_num_workers(config): from xdist.plugin import pytest_xdist_auto_num_workers as detected_num_workers return min(detected_num_workers(config), MAX_PARALLEL_WORKERS) + + +def _requested_worker_count(config): + """Use xdist's resolved specs so the refusal counts workers the way xdist does.""" + from xdist.workermanage import parse_tx_spec_config + + return len(parse_tx_spec_config(config)) + + +def pytest_configure(config): + """Refuse a worker count that cannot get private databases, before any database is created.""" + # The two conditions xdist itself starts workers on, and the list it reads them from. + if config.getoption("dist", "no") == "no" or config.getoption("collectonly", False): + return + if not config.getoption("tx", []): + return + requested = _requested_worker_count(config) + if requested > MAX_PARALLEL_WORKERS: + raise pytest.UsageError( + f"{requested} workers exceed the isolation limit: at most {MAX_PARALLEL_WORKERS} " + "pytest workers get private PostgreSQL and Redis databases." + ) diff --git a/docs/design/workflow-command-detection.md b/docs/design/workflow-command-detection.md new file mode 100644 index 00000000..83dd7a49 --- /dev/null +++ b/docs/design/workflow-command-detection.md @@ -0,0 +1,25 @@ +# Detect pytest commands in workflows + +The workflow plugin check reads YAML step `run` values. It uses Tree-sitter's +Bash grammar to distinguish executable commands from argument text, comments, +quoted operators, and heredoc bodies. It does not execute workflow code. + +Text matching cannot distinguish `echo pytest` from a pytest invocation. +Shell tokenization alone loses whether an operator token was quoted. +The Bash syntax tree supplies that distinction. `shlex` decodes static words +only after the parser establishes their boundaries. + +The detector recognizes literal `pytest` commands, Python module invocations, +and the `poetry run`, `uv run`, and `xvfb-run` prefixes. It inspects command +substitutions separately. GitHub expressions become opaque shell expansions +before parsing. Variable values, ANSI-C quoting, arbitrary wrappers, aliases, `eval`, and +programs embedded in `bash -c` are outside this static check. + +Syntax errors raise an error that names the workflow. The check does not fall +back to text matching. Real YAML fixtures cover command lists, quotations, +heredocs, metadata, installations, and invalid shell input. The checked-in +workflow test also verifies that only the two test workflows invoke pytest. + +The `workflow-tests` dependency group owns the parser versions. The development +group includes it. Both test workflows and devcontainer setup install it. +These are test dependencies. diff --git a/netbox_interface_name_rules/engine.py b/netbox_interface_name_rules/engine.py index 9d924d20..5fb43abc 100644 --- a/netbox_interface_name_rules/engine.py +++ b/netbox_interface_name_rules/engine.py @@ -13,7 +13,6 @@ from . import family as family_ops from . import naming, rule_selection -from .family import targets as family_targets from .family import template_names as family_template_names from .regex_safety import compile_module_type_pattern @@ -630,7 +629,7 @@ def _preview_plans(rule, plan_set) -> list: if installed: # pragma: no cover - requires a NetBox that models channelization return installed creations = [plan for plan in plan_set.plans if plan.base_name is not None] - kept = family_targets.one_family_per_name_set([(plan.base_name, plan.target_names) for plan in creations]) + kept = family_ops.one_family_per_name_set([(plan.base_name, plan.target_names) for plan in creations]) return [creations[index] for index in kept] diff --git a/netbox_interface_name_rules/migrations/0005_platform.py b/netbox_interface_name_rules/migrations/0005_platform.py index 8f8264ef..ffa11608 100644 --- a/netbox_interface_name_rules/migrations/0005_platform.py +++ b/netbox_interface_name_rules/migrations/0005_platform.py @@ -16,7 +16,6 @@ class Migration(migrations.Migration): dependencies = [ ("netbox_interface_name_rules", "0004_nulls_distinct"), - ("dcim", "0001_initial"), ] operations = [ diff --git a/netbox_interface_name_rules/migrations/0007_alter_optional_fks_set_null.py b/netbox_interface_name_rules/migrations/0007_alter_optional_fks_set_null.py index fb6f6bde..a3d22a0e 100644 --- a/netbox_interface_name_rules/migrations/0007_alter_optional_fks_set_null.py +++ b/netbox_interface_name_rules/migrations/0007_alter_optional_fks_set_null.py @@ -14,7 +14,6 @@ class Migration(migrations.Migration): dependencies = [ ("netbox_interface_name_rules", "0006_alter_interfacenamerule_options"), - ("dcim", "0001_initial"), ] operations = [ diff --git a/netbox_interface_name_rules/tests/comment_blocks.json b/netbox_interface_name_rules/tests/comment_blocks.json new file mode 100644 index 00000000..2a38b501 --- /dev/null +++ b/netbox_interface_name_rules/tests/comment_blocks.json @@ -0,0 +1,312 @@ +{ + "engine.py": [ + [ + "# A channelized parent is its own base: its channels are separate rows, so the name needs no\n# \":\"-splitting to find them.", + 1 + ], + [ + "# All interfaces already have the names the rule would produce — flag as\n# potentially obsolete (e.g., newer NetBox generates correct names natively).\n# Skipped when the 0-count was caused by name collisions (a different reason\n# than a no-op rule), so a collision never mislabels the rule as deprecated.\n# Skipped for families too: a structural skip, or a family whose parent deliberately\n# keeps its raw name, says nothing about the rule being obsolete.", + 1 + ], + [ + "# An earlier flat apply leaves N sibling interfaces where NetBox 4.7+ models a channelized parent\n# with N channel subinterfaces. Converting one rewrites rows an operator owns — cables, addresses,\n# tags — so it is never a side effect of applying a rule: the operator confirms it per family.", + 1 + ], + [ + "# Only the cheap half of the guard, so a rule that offers no conversion never reads its modules;\n# whether this release can hold a family is the family package's call.", + 1 + ], + [ + "# Sort Python-side: specificity_score descending, then module_type_pattern length\n# descending (for device-interface rules with ties), then pk ascending for stability.\n# (InterfaceNameRule has no DB 'priority' field; specificity_score is a property.)", + 1 + ], + [ + "# The guard runs while it can still see every claimed row, before two of them that intend\n# one family are collapsed into it.", + 1 + ] + ], + "family/batch.py": [ + [ + "# A member left with the name it had for a reason the operator can act on. An unsupported topology\n# is not one of them: the release cannot hold the family, so nothing was dropped by this batch.", + 1 + ] + ], + "family/installed.py": [ + [ + "# A device rule never builds a family, so its channel count says nothing about this one: the\n# members keep the suffixes they carry under whatever name the parent takes. A device-level\n# interface has no module template family, so there is no suffix to recover from one either.", + 1 + ] + ], + "family/structural.py": [ + [ + "# A flat family is N sibling interfaces on one module: the base takes the first name and the rest\n# are new rows. Unlike a channelized family there is no parent to cascade from, so a sibling whose\n# name is taken is skipped on its own while the family keeps the names it could take.", + 1 + ] + ], + "forms.py": [ + [ + "# Blank first choice: a bulk edit posts every rendered field, so without a \"no change\" option a\n# select rewrites the column on every selected rule.", + 1 + ], + [ + "# FK fields must declare to_field_name explicitly so YAML/CSV can reference\n# objects by their natural key instead of numeric PK.", + 1 + ] + ], + "graphql/filters.py": [ + [ + "# Degrade only when the module is genuinely absent (or one of its parents). Any other\n# ImportError is a real breakage and must not masquerade as \"old NetBox\".", + 1 + ], + [ + "# strawberry-graphql-django renamed FilterLookup to StrFilterLookup in 0.86; NetBox 4.5\n# pins 0.75. getattr falls back only when the symbol is absent, so an ImportError raised\n# while resolving it propagates instead of quietly selecting the legacy alias.", + 1 + ] + ], + "models.py": [ + [ + "# Django accepts any iterable. Reading a generator here would leave Django an empty\n# one, and it skips the write when update_fields is empty.", + 1 + ], + [ + "# Override inherited tags to avoid reverse accessor clash when co-installed\n# with another plugin that has a model of the same name.", + 1 + ], + [ + "# The implications _validate_breakout_topology() enforces over enum and integer\n# columns, written as ~P | Q. Its parent-template grammar rules stay in save().", + 1 + ] + ], + "rule_selection.py": [ + [ + "# A single dictionary read cannot race with another thread's memo clear between\n# a membership check and a later subscript.", + 1 + ], + [ + "# Device-interface rows reuse module_type_pattern as an interface-name filter, so they are\n# never module rules and never enter module selection.", + 1 + ], + [ + "# Publish each loaded rule set as one new dictionary. Concurrent readers then see\n# one complete version rather than a mixture of cache entries from two versions.", + 1 + ], + [ + "# These fields can change either matching or the selected rule's output. The row\n# identity prevents compensating edits across two rules from preserving the hash.", + 1 + ] + ], + "signals.py": [ + [ + "# Module.save()\n# → bulk_create() all interfaces (pre_save never fires)\n# → NetBox fires post_save(Module)\n# → on_module_saved → transaction.on_commit\n# → _apply_rules_deferred → apply_interface_name_rules()", + 1 + ], + [ + "# Adding a pre_save on Interface would be a no-op for the normal install\n# path and would create a false sense of security.", + 1 + ], + [ + "# NetBox's Module.save() creates interfaces via bulk_create() which bypasses\n# pre_save signals entirely. NetBox then manually dispatches post_save for\n# the Module object. The actual renaming path is therefore:", + 1 + ], + [ + "# When a Device's vc_position or virtual_chassis changes, any module-attached\n# interfaces with rules using {vc_position} need to be renamed. We capture\n# the old values in pre_save (stored on the instance) and compare in post_save.", + 1 + ], + [ + "# When netbox-librenms-plugin is installed it exposes a Signal that asks\n# subscribers to rewrite the list of interface names it predicts a module's\n# templates will produce. Without this rewrite, its module-adoption lookup\n# would only see NetBox's raw template output and miss interfaces we renamed\n# after install. The import is guarded so INR works fine when librenms-plugin\n# is absent.", + 1 + ] + ], + "tests/parallel.py": [ + [ + "# A stock Redis server serves databases 0 to 15. The devcontainer's own rqworker holds the first\n# two, so the rest divides into slots of one task and one cache database.", + 1 + ] + ], + "tests/test_channelization.py": [ + [ + "# A second family whose template does not feed the current name back in, so \"already correctly\n# named\" is a state the rule can actually reach (an \"et-{base}\" rule renames on every pass).", + 1 + ], + [ + "# A standalone interface alongside the family, so \"families counted once\" stays distinguishable\n# from \"interfaces not counted at all\".", + 1 + ] + ], + "tests/test_conversion.py": [ + [ + "# The scan orders modules by bay, so the cap has to land on exactly the families blocked above.\n# Asserting it here keeps a change of scan order from quietly turning this into a different test.", + 1 + ] + ], + "tests/test_device_rules.py": [ + [ + "# Derive the over-length name from the live field limit so the test still exercises the\n# validation branch if Interface.name's max_length ever changes. vc_position=1 adds one char,\n# so (max_length + 1) literal chars guarantees the rendered name exceeds the limit.", + 1 + ], + [ + "# assertLogs asserts the warning branch actually fired — without it, \"no rule matched\" would\n# also yield result==0 and a preserved name, so the test would pass without covering the path.", + 1 + ] + ], + "tests/test_documentation.py": [ + [ + "# Device-interface rules match any device interface, including a standalone one such as\n# mgmt0; saying \"family parent\" would send an operator to the wrong scope.", + 1 + ] + ], + "tests/test_e2e.py": [ + [ + "# Bay with numeric position — interface template creates \"5\",\n# rule renames to \"swp5\"", + 1 + ], + [ + "# The interface was created with the bay position as name\n# Apply the rename rules", + 1 + ] + ], + "tests/test_engine_advanced.py": [ + [ + "# Create with the final correct name (so the rule renames nothing for it)\n# But the name \"et-0/0/0\" is NOT in raw_names (raw = \"0\"), so this won't trigger deprecated.\n# Instead, set force_reapply so it's in unrenamed but produces 0 renames.", + 1 + ], + [ + "# The name the rule would produce for \"Gi0/2\" (vc_position=1, port=2) is already taken\n# on the device, so the device-scope pre-check skips the rename with a clean WARNING\n# (no full_clean ValidationError / ERROR traceback).", + 1 + ], + [ + "# force_reapply=True: base \"xe-0/0/0\" is matched (via last segment \"0\" in raw_names),\n# but template evaluates to the same names, so 0 renames occur.", + 1 + ], + [ + "# force_reapply=True: base \"xe-2/0/0\" is matched (last segment \"0\" in raw_names),\n# and vc_device.vc_position=1 → new name \"xe-1/0/0:0\" differs → rename occurs.", + 1 + ] + ], + "tests/test_installed_families.py": [ + [ + "# Plan the complete family first. Without it the module holds only the raw base interface,\n# which yields no plans on its own, so the assertion would pass at any channel_count.", + 1 + ] + ], + "tests/test_isolated_test_settings.py": [ + [ + "# Hand the child this interpreter's own import path. NetBox lives in a different place in\n# the devcontainer than in CI, and neither location may be assumed here.", + 1 + ], + [ + "# Read the settings module named on the command line, so the shim can be compared with the\n# NetBox settings it wraps.", + 1 + ], + [ + "# The suite's own settings module reaches the child through these, and Django would load it\n# lazily instead of the module under test.", + 1 + ] + ], + "tests/test_module_boundaries.py": [ + [ + "# The family package does not export the template-name helpers the engine needs, so the engine\n# reaches past the seam for them. Widening the package API is a change to its public surface and\n# belongs in its own commit; until then this is the one import allowed through.", + 1 + ] + ], + "tests/test_rule_validation_agreement.py": [ + [ + "# These rows predate the constraints, so they must reach the table the way they did then:\n# bulk_create() skips both clean() and save().", + 1 + ] + ], + "tests/test_rules.py": [ + [ + "# A genuine second thread. _pin is thread-local, so this worker must see depth 0 — it never\n# inherits the main thread's active pin. It then publishes a different rule-set version the\n# way _get_enabled_rules() does: one atomic rebind of the module global. (No DB access — a\n# separate thread has its own connection and cannot see this TestCase's uncommitted rows.)", + 1 + ], + [ + "# An unpinned thread clearing the shared memo at the cap must not disturb the pinned\n# batch. _pin is a threading.local, so the worker holds no pin of its own. The target is\n# a plain dict.clear(), so the worker touches no ORM and needs no connection.", + 1 + ], + [ + "# And the eviction actually fired: the final size is below the number of distinct contexts,\n# so the test isn't vacuously green on a memo that simply never reached the cap.", + 1 + ], + [ + "# Bulk .update() writes the column directly — no .save(), so auto_now/last_updated is NOT\n# bumped and the enabled-rule count and column sums are unchanged. Only the text differs.", + 1 + ], + [ + "# Check the bound after EVERY insert, not just once at the end. The memo size must never\n# exceed the cap at any point. A regression that let it leak toward 2*cap before clearing\n# would slip past an end-of-loop `<=` snapshot (which only sees the post-clear size) but\n# trips here on the very insert that crosses the cap.", + 1 + ], + [ + "# Forge a one-rule set whose name_template embeds r1's trailing columns, a row separator, and\n# r2's columns up to its name_template. Column order: id, module_type_id, is_regex, pattern,\n# parent_id, device_id, platform_id, name_template, channel_count, channel_start, adi — so r2's\n# rendered cells through its name are [pk, '', 'false', 'p2', '', '', '', 'b'].", + 1 + ], + [ + "# Inside the pin, the first lookup primes the set (one query) and the rest reuse it: a loop of\n# many lookups costs a single fingerprint query instead of one per call.", + 1 + ], + [ + "# Null the scope FK the way a SET_NULL cascade / bulk .update() does: a straight UPDATE that\n# does NOT bump last_updated and leaves the enabled-rule count unchanged. The fingerprint's\n# device_type-id sum still changes, so the next call reloads instead of serving the stale copy.", + 1 + ], + [ + "# Poison the shared memo for the same signature. A primed lookup that read the shared\n# memo instead of its private copy would return the decoy rather than recomputing, so\n# this is what separates \"the copy was used\" from \"the answer was recomputed\".", + 1 + ], + [ + "# Retention lives in the private copy. Asserting the shared memo is empty would prove\n# nothing: pinned writes never reach it, so it is already {} before the clear.", + 1 + ], + [ + "# Swap device_type between the two rules via bulk .update(): SUM(device_type) is unchanged\n# (x+y == y+x), count unchanged, last_updated unbumped. Only the per-rule pairing differs.", + 1 + ], + [ + "# These tests mutate rule_selection's module-level _RULE_CACHE / _pin, which TestCase does\n# NOT roll back (only the DB is). Reset them before each method so the class is order-independent\n# and a stale snapshot from a prior method can't be reused for a same-content rule set.", + 1 + ], + [ + "# This test deliberately rebinds the module global from another thread; restore a clean\n# sentinel so the simulated reload can't leak into sibling tests even if setUp() is weakened.", + 1 + ], + [ + "# Two device-level rules (module_type=None → every FK column renders ''). Distinct patterns keep\n# them unique (the device-rule constraint is on pattern/device/platform); created() bypasses model\n# validation, so the control chars in the forged template below are stored as-is.", + 1 + ], + [ + "# the fingerprint's own filter predicate (filter(enabled=True)) — toggling it adds/removes\n# the row from the aggregate, so the hash already changes; it must not also be a column", + 1 + ] + ], + "tests/test_standard_views.py": [ + [ + "# breakout_mode stays flat here: bulk edit runs full_clean(), and the channelized topology\n# is only valid for a rule that also defines channels.", + 1 + ] + ], + "tests/test_vc_drift.py": [ + [ + "# Every fixture spelling a name NetBox resolved from the token needs the release that resolves it:\n# on 4.5 and older the token stays literal in the interface name and the drift cannot even occur.", + 1 + ], + [ + "# One token template plus a plain one whose interface an earlier rename moved onto the\n# token template's fallback variant.", + 1 + ] + ], + "tests/test_workflows.py": [ + [ + "# The option each pytest plugin registers. `--no-cov` needs pytest-cov for the same reason `--cov`\n# does: an unregistered option is a parse error, whether it turns coverage on or off.", + 1 + ] + ], + "views.py": [ + [ + "# Skip duplicate detection when the user lacks view permission — we cannot\n# query existing rules without it, so add-only users always land on the\n# create form (potentially allowing duplicates).", + 1 + ], + [ + "# instance is intentionally omitted: InterfaceNameRule does not\n# inherit JobsMixin, so passing instance= would fail full_clean().\n# The job is still named and findable in Core → Jobs.", + 1 + ] + ] +} diff --git a/netbox_interface_name_rules/tests/conftest.py b/netbox_interface_name_rules/tests/conftest.py new file mode 100644 index 00000000..c05ae52e --- /dev/null +++ b/netbox_interface_name_rules/tests/conftest.py @@ -0,0 +1,22 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (C) 2025 Marcin Zieba +"""Pytest fixtures for isolated parallel test workers.""" + +import os + +import pytest + +from netbox_interface_name_rules.tests.parallel import isolated_test_database_name + + +@pytest.fixture(scope="session") +def django_db_modify_db_settings(django_db_modify_db_settings): + """Give each pytest worker a private PostgreSQL database.""" + from django.conf import settings + + test_settings = dict(settings.DATABASES["default"].get("TEST") or {}) + test_settings["NAME"] = isolated_test_database_name( + os.environ["TEST_DB_NAME"], + os.environ.get("PYTEST_XDIST_WORKER"), + ) + settings.DATABASES["default"]["TEST"] = test_settings diff --git a/netbox_interface_name_rules/tests/helpers.py b/netbox_interface_name_rules/tests/helpers.py new file mode 100644 index 00000000..cc5d7a5d --- /dev/null +++ b/netbox_interface_name_rules/tests/helpers.py @@ -0,0 +1,93 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (C) 2025 Marcin Zieba +"""Builders for the DCIM objects the tests need. + +Every builder takes a *prefix* and derives names and slugs from it. Test classes share one database +per worker, so a class that names its objects after itself cannot collide with another class, and a +failure still names the class it came from. +""" + +from dataclasses import dataclass + +from dcim.models import ( + Device, + DeviceRole, + DeviceType, + Manufacturer, + ModuleBayTemplate, + ModuleType, + Site, +) + + +def slug_for(prefix: str, suffix: str = "") -> str: + """Return a slug built from *prefix*, safe to use as a NetBox slug.""" + cleaned = "".join(character if character.isalnum() else "-" for character in prefix).strip("-").lower() + return f"{cleaned}-{suffix}" if suffix else cleaned + + +def make_manufacturer(prefix: str) -> Manufacturer: + """Return one manufacturer named after *prefix*.""" + return Manufacturer.objects.create(name=f"{prefix} Manufacturer", slug=slug_for(prefix, "mfg")) + + +def make_device_type(manufacturer: Manufacturer, prefix: str, model: str | None = None) -> DeviceType: + """Return one device type named after *prefix*.""" + model = model or f"{prefix} Device Type" + return DeviceType.objects.create(manufacturer=manufacturer, model=model, slug=slug_for(prefix, "type")) + + +def make_module_type( + manufacturer: Manufacturer, + prefix: str, + model: str | None = None, + part_number: str | None = None, +) -> ModuleType: + """Return one module type named after *prefix*, or after an explicit *model*.""" + model = model or f"{prefix} Module Type" + return ModuleType.objects.create(manufacturer=manufacturer, model=model, part_number=part_number or model) + + +def make_module_bay_templates(device_type: DeviceType, names: tuple[str, ...]) -> list[ModuleBayTemplate]: + """Return module bay templates on *device_type*, positioned in the order given. + + NetBox instantiates the bays when a device is created, so these must exist before the device. + """ + return [ + ModuleBayTemplate.objects.create(device_type=device_type, name=name, position=str(position)) + for position, name in enumerate(names) + ] + + +@dataclass(frozen=True) +class DevicePlacement: + """The role and site a device needs, kept together so a test can reuse them.""" + + role: DeviceRole + site: Site + + +def make_placement(prefix: str) -> DevicePlacement: + """Return the role and site for devices named after *prefix*.""" + return DevicePlacement( + role=DeviceRole.objects.create(name=f"{prefix} Role", slug=slug_for(prefix, "role")), + site=Site.objects.create(name=f"{prefix} Site", slug=slug_for(prefix, "site")), + ) + + +def make_device( + prefix: str, + device_type: DeviceType, + placement: DevicePlacement | None = None, + name: str | None = None, + **fields, +) -> Device: + """Return one device on *device_type*, creating a role and site when none is given.""" + placement = placement or make_placement(prefix) + return Device.objects.create( + name=name or slug_for(prefix, "01"), + device_type=device_type, + role=placement.role, + site=placement.site, + **fields, + ) diff --git a/netbox_interface_name_rules/tests/isolated_settings.py b/netbox_interface_name_rules/tests/isolated_settings.py new file mode 100644 index 00000000..54fcc042 --- /dev/null +++ b/netbox_interface_name_rules/tests/isolated_settings.py @@ -0,0 +1,45 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (C) 2025 Marcin Zieba +"""NetBox test settings that require caller-selected database and Redis targets. + +The devcontainer runs a live rqworker and a shared cache on the first two Redis databases, and other +sessions run their own suites against the same server. Every worker therefore takes a private pair +of Redis databases and a private PostgreSQL database instead of the defaults. +""" + +import os + +from netbox_interface_name_rules.tests.parallel import isolated_cache_location, isolated_redis_databases + +_worker_id = os.environ.get("PYTEST_XDIST_WORKER") +_tasks_database, _cache_database = isolated_redis_databases(_worker_id) + +_redis_host = os.environ.get("TEST_REDIS_HOST", "").strip() +if not _redis_host: + raise ValueError("TEST_REDIS_HOST must name the Redis server the tests may use.") + +os.environ["REDIS_HOST"] = _redis_host +os.environ["REDIS_CACHE_HOST"] = _redis_host +os.environ["REDIS_DATABASE"] = str(_tasks_database) +os.environ["REDIS_CACHE_DATABASE"] = str(_cache_database) +os.environ.setdefault("NETBOX_CONFIGURATION", "netbox_interface_name_rules.tests.netbox_configuration") + +from netbox.settings import * # noqa: E402, F403 + +_database_name = os.environ.get("TEST_DB_NAME", "") +if not _database_name.startswith("test_"): + raise ValueError("TEST_DB_NAME must be set and must start with 'test_'.") + +# The worker suffix is applied in the conftest fixture, after xdist resolves the worker identity. +DATABASES["default"].setdefault("TEST", {})["NAME"] = _database_name # noqa: F405 + +# Set here, not only through the environment: a configuration module that ignores it must not win. +for _queue in RQ_QUEUES.values(): # noqa: F405 + _queue["HOST"] = _redis_host + _queue["DB"] = _tasks_database + +CACHES["default"]["LOCATION"] = isolated_cache_location( # noqa: F405 + CACHES["default"]["LOCATION"], # noqa: F405 + _redis_host, + _cache_database, +) diff --git a/netbox_interface_name_rules/tests/netbox_configuration.py b/netbox_interface_name_rules/tests/netbox_configuration.py new file mode 100644 index 00000000..ebe83b91 --- /dev/null +++ b/netbox_interface_name_rules/tests/netbox_configuration.py @@ -0,0 +1,18 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (C) 2025 Marcin Zieba +"""NetBox configuration that loads this plugin alone. + +A development container installs several plugins into one virtualenv. Loading them all makes a run +depend on trees this repository does not control, so the tests pin the plugin list to this package. +""" + +from netbox import configuration as _configuration + +for _name in dir(_configuration): + if _name.isupper(): + globals()[_name] = getattr(_configuration, _name) + +PLUGINS = ["netbox_interface_name_rules"] +PLUGINS_CONFIG = { + "netbox_interface_name_rules": getattr(_configuration, "PLUGINS_CONFIG", {}).get("netbox_interface_name_rules", {}), +} diff --git a/netbox_interface_name_rules/tests/parallel.py b/netbox_interface_name_rules/tests/parallel.py new file mode 100644 index 00000000..76191b56 --- /dev/null +++ b/netbox_interface_name_rules/tests/parallel.py @@ -0,0 +1,58 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (C) 2025 Marcin Zieba +"""Isolation helpers for parallel pytest workers.""" + +import re +from urllib.parse import urlsplit, urlunsplit + +# A stock Redis server serves databases 0 to 15. The devcontainer's own rqworker holds the first +# two, so the rest divides into slots of one task and one cache database. +REDIS_DATABASE_COUNT = 16 +RESERVED_REDIS_DATABASES = 2 + +_REDIS_SLOT_COUNT = (REDIS_DATABASE_COUNT - RESERVED_REDIS_DATABASES) // 2 +# Slot 0 belongs to a serial run, so the worker ceiling is one below the slot count. +MAX_PARALLEL_WORKERS = _REDIS_SLOT_COUNT - 1 + +_POSTGRES_NAME_LIMIT = 63 +_WORKER_ID_PATTERN = re.compile(r"gw(?P\d+)") + + +def _worker_number(worker_id: str) -> int: + """Return the ordinal of *worker_id*.""" + match = _WORKER_ID_PATTERN.fullmatch(worker_id) + if match is None: + raise ValueError(f"Unsupported pytest worker ID: {worker_id!r}.") + number = int(match.group("number")) + if number >= MAX_PARALLEL_WORKERS: + raise ValueError(f"At most {MAX_PARALLEL_WORKERS} pytest workers are supported.") + return number + + +def isolated_test_database_name(base_name: str, worker_id: str | None) -> str: + """Return a PostgreSQL-safe test database name for one pytest worker.""" + suffix = f"_{worker_id}" if worker_id else "" + return f"{base_name[: _POSTGRES_NAME_LIMIT - len(suffix)]}{suffix}" + + +def isolated_cache_location(location: str, host: str, database: int) -> str: + """Return *location* pointed at *host* and *database*, keeping scheme, credentials and port.""" + parsed = urlsplit(location) + netloc = parsed.username or "" + if parsed.password is not None: + netloc += f":{parsed.password}" + if parsed.username is not None: + netloc += "@" + netloc += host if parsed.port is None else f"{host}:{parsed.port}" + return urlunsplit((parsed.scheme, netloc, f"/{database}", parsed.query, parsed.fragment)) + + +def isolated_redis_databases(worker_id: str | None) -> tuple[int, int]: + """Return the private task and cache Redis databases for one pytest worker. + + A serial run takes slot 0, so ``pytest -n 0`` never shares queues or cache entries with an + xdist worker running against the same Redis host. + """ + slot = 0 if worker_id is None else _worker_number(worker_id) + 1 + tasks = RESERVED_REDIS_DATABASES + slot + return tasks, tasks + _REDIS_SLOT_COUNT diff --git a/netbox_interface_name_rules/tests/signal_performance.py b/netbox_interface_name_rules/tests/signal_performance.py index b0e58472..70527ca3 100644 --- a/netbox_interface_name_rules/tests/signal_performance.py +++ b/netbox_interface_name_rules/tests/signal_performance.py @@ -521,7 +521,7 @@ def _cpu_model() -> str: """Return a stable processor description without recording the host identity.""" cpuinfo = Path("/proc/cpuinfo") if cpuinfo.exists(): - for line in cpuinfo.read_text().splitlines(): + for line in cpuinfo.read_text(encoding="utf-8").splitlines(): if line.lower().startswith("model name"): return line.partition(":")[2].strip() return platform.processor() or "unknown" @@ -1192,5 +1192,5 @@ def test_record_existing_signal_path_performance(self): "scenarios": scenario_results, } validate_artifact(artifact, "generated performance artifact") - output.write_text(json.dumps(artifact, indent=2, sort_keys=True) + "\n") - output.with_suffix(".md").write_text(_markdown_summary(artifact)) + output.write_text(json.dumps(artifact, indent=2, sort_keys=True) + "\n", encoding="utf-8") + output.with_suffix(".md").write_text(_markdown_summary(artifact), encoding="utf-8") diff --git a/netbox_interface_name_rules/tests/test_comment_style.py b/netbox_interface_name_rules/tests/test_comment_style.py new file mode 100644 index 00000000..bb90b355 --- /dev/null +++ b/netbox_interface_name_rules/tests/test_comment_style.py @@ -0,0 +1,106 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (C) 2025 Marcin Zieba +"""Reject a multi-line comment block that `comment_blocks.json` does not already record. + +A run of two or more whole-line `#` comments is a block. Prose belongs in the commit message or the +pull request, and the code keeps a single line pointing at the non-obvious part. The record is a +permit list of the blocks that already existed, so adding one fails until it is deliberately recorded. + +Removing a block needs no edit here: an entry with nothing left to permit grants nothing. Requiring +its removal would make every comment fix a two-file change and collide between branches. + +Banners, blank `#` lines and pragmas separate rather than explain, so they neither count as a block +nor join two blocks. Migrations are excluded, matching the ruff `per-file-ignores` carve-out. +""" + +import json +import pathlib +import tempfile +import tokenize +from collections import Counter + +from django.test import SimpleTestCase + +PACKAGE = pathlib.Path(__file__).resolve().parents[1] +BASELINE = pathlib.Path(__file__).resolve().parent / "comment_blocks.json" + +_PRAGMAS = ("# noqa", "# type:", "# ruff:", "# fmt:", "# pragma:", "# SPDX", "# Copyright") +_RULE_CHARACTERS = set("-=*_") + + +def _is_banner(text): + """Return True for a `# ---` rule or a bare `#`, which separate rather than explain.""" + return not set(text.lstrip("#").strip()) - _RULE_CHARACTERS + + +def _own_line_comments(path): + """Map each line number carrying a whole-line explanatory comment to its text.""" + lines = path.read_text(encoding="utf-8").splitlines() + found = {} + with path.open("rb") as handle: + for token in tokenize.tokenize(handle.readline): + if token.type != tokenize.COMMENT: + continue + row = token.start[0] + # A trailing comment explains one statement, so it never joins the block above it. + if not lines[row - 1].lstrip().startswith("#"): + continue + text = token.string.strip() + if not _is_banner(text) and not text.startswith(_PRAGMAS): + found[row] = text + return found + + +def _blocks(path): + """Yield every run of two or more consecutive whole-line comments, joined line by line.""" + comments = _own_line_comments(path) + for row in sorted(comments): + if row - 1 in comments: + continue + length = 0 + while row + length in comments: + length += 1 + if length > 1: + yield "\n".join(comments[row + offset] for offset in range(length)) + + +def blocks_in_package(): + """Return the multi-line comment blocks the package holds, keyed by repository path.""" + found = {} + for path in sorted(PACKAGE.rglob("*.py")): + if "migrations" in path.parts: + continue + blocks = list(_blocks(path)) + if blocks: + found[str(path.relative_to(PACKAGE))] = sorted(Counter(blocks).items()) + return found + + +class CommentStyleTest(SimpleTestCase): + """A new multi-line comment block has to be recorded before it is allowed.""" + + def test_no_unrecorded_multi_line_comment_block(self): + recorded = { + name: Counter({text: count for text, count in entries}) + for name, entries in json.loads(BASELINE.read_text(encoding="utf-8")).items() + } + unrecorded = [] + for name, entries in blocks_in_package().items(): + permitted = recorded.get(name, Counter()) + for text, count in entries: + if count > permitted[text]: + unrecorded.append(f"{name}: {text}") + + self.assertEqual( + unrecorded, + [], + "Move the explanation to the commit message and keep one line, or record the block.", + ) + + def test_a_block_key_covers_every_line_of_the_run(self): + """The key holds the whole run, so a later line cannot change without a baseline update.""" + with tempfile.TemporaryDirectory() as directory: + path = pathlib.Path(directory) / "sample.py" + path.write_text("# first line\n# second line\nvalue = 1\n", encoding="utf-8") + + self.assertEqual(list(_blocks(path)), ["# first line\n# second line"]) diff --git a/netbox_interface_name_rules/tests/test_device_rules.py b/netbox_interface_name_rules/tests/test_device_rules.py index 733e0fc9..ce46ca69 100644 --- a/netbox_interface_name_rules/tests/test_device_rules.py +++ b/netbox_interface_name_rules/tests/test_device_rules.py @@ -4,20 +4,19 @@ from dcim.models import ( Device, - DeviceRole, DeviceType, Interface, Manufacturer, ModuleBayTemplate, ModuleType, Platform, - Site, VirtualChassis, ) from django.test import TestCase from netbox_interface_name_rules.engine import apply_device_interface_rules from netbox_interface_name_rules.models import InterfaceNameRule +from netbox_interface_name_rules.tests.helpers import make_placement class ApplyDeviceInterfaceRulesTest(TestCase): @@ -31,15 +30,14 @@ def setUpTestData(cls): manufacturer=manufacturer, model="DevRule-Switch", slug="devrule-switch" ) cls.platform = Platform.objects.create(name="DevRule-IOS", slug="devrule-ios") - role = DeviceRole.objects.create(name="DevRuleRole", slug="devrulerole") - site = Site.objects.create(name="DevRuleSite", slug="devrulesit") + placement = make_placement("DevRule") cls.vc = VirtualChassis.objects.create(name="devrule-vc") cls.device1 = Device.objects.create( name="devrule-sw1", device_type=cls.device_type, - role=role, - site=site, + role=placement.role, + site=placement.site, virtual_chassis=cls.vc, vc_position=1, platform=cls.platform, @@ -47,16 +45,16 @@ def setUpTestData(cls): cls.device_no_vc = Device.objects.create( name="devrule-standalone", device_type=cls.device_type, - role=role, - site=site, + role=placement.role, + site=placement.site, ) # Another device with vc_position=None (edge case: VC set but position unset) cls.vc2 = VirtualChassis.objects.create(name="devrule-vc2") cls.device_no_pos = Device.objects.create( name="devrule-nopos", device_type=cls.device_type, - role=role, - site=site, + role=placement.role, + site=placement.site, virtual_chassis=cls.vc2, vc_position=None, ) @@ -304,14 +302,13 @@ def setUpTestData(cls): device_type = DeviceType.objects.create(manufacturer=manufacturer, model="DR2-Switch", slug="dr2-switch") ModuleBayTemplate.objects.create(device_type=device_type, name="Slot 0", position="0") cls.module_type = ModuleType.objects.create(manufacturer=manufacturer, model="DR2-LC", part_number="DR2-LC") - role = DeviceRole.objects.create(name="DR2Role", slug="dr2role") - site = Site.objects.create(name="DR2Site", slug="dr2site") + placement = make_placement("DR2") vc = VirtualChassis.objects.create(name="dr2-vc") cls.device = Device.objects.create( name="dr2-sw1", device_type=device_type, - role=role, - site=site, + role=placement.role, + site=placement.site, virtual_chassis=vc, vc_position=2, ) @@ -336,14 +333,13 @@ class DeviceInterfaceEdgeCaseNamesTest(TestCase): def setUpTestData(cls): mfg = Manufacturer.objects.create(name="EdgeMfg", slug="edgemfg") cls.device_type = DeviceType.objects.create(manufacturer=mfg, model="Edge-Dev", slug="edge-dev") - role = DeviceRole.objects.create(name="EdgeRole", slug="edgerole") - site = Site.objects.create(name="EdgeSite", slug="edgesite") + placement = make_placement("Edge") vc = VirtualChassis.objects.create(name="edge-vc") cls.device = Device.objects.create( name="edge-sw1", device_type=cls.device_type, - role=role, - site=site, + role=placement.role, + site=placement.site, virtual_chassis=vc, vc_position=1, ) diff --git a/netbox_interface_name_rules/tests/test_documentation.py b/netbox_interface_name_rules/tests/test_documentation.py index 48774d15..10aa13af 100644 --- a/netbox_interface_name_rules/tests/test_documentation.py +++ b/netbox_interface_name_rules/tests/test_documentation.py @@ -27,14 +27,14 @@ def _conversion_sentences(): """Return the conversion section as lowercased, whitespace-collapsed sentences.""" - guide = (_PROJECT_ROOT / "docs" / "template-variables.md").read_text() + guide = (_PROJECT_ROOT / "docs" / "template-variables.md").read_text(encoding="utf-8") section = guide.split("### Converting an installed flat family", 1)[1].split("### Converter Offset", 1)[0] return [sentence for sentence in " ".join(section.lower().split()).split(". ") if sentence] def _example_conversion_sentences(): """Return the conversion example as lowercased, whitespace-collapsed sentences.""" - guide = (_PROJECT_ROOT / "docs" / "examples.md").read_text() + guide = (_PROJECT_ROOT / "docs" / "examples.md").read_text(encoding="utf-8") section = guide.split("### Converting a flat family (NetBox 4.7+)", 1)[1].split("What the conversion does", 1)[0] return [sentence for sentence in " ".join(section.lower().split()).split(". ") if sentence] @@ -83,7 +83,9 @@ class PerformanceDocumentationTest(unittest.TestCase): """Keep the performance narrative consistent with the committed comparison.""" def test_statement_attribution_matches_the_comparison(self): - comparison = (_PROJECT_ROOT / "performance" / "comparisons" / "family-package-vs-existing.md").read_text() + comparison = (_PROJECT_ROOT / "performance" / "comparisons" / "family-package-vs-existing.md").read_text( + encoding="utf-8" + ) attribution = comparison.split("### Where those statements come from", 1)[1] changes_by_scenario = {} for line in attribution.splitlines(): @@ -94,7 +96,7 @@ def test_statement_attribution_matches_the_comparison(self): if change: changes_by_scenario.setdefault(cells[0], {})[cells[1]] = change - readme = (_PROJECT_ROOT / "performance" / "README.md").read_text() + readme = (_PROJECT_ROOT / "performance" / "README.md").read_text(encoding="utf-8") result = readme.split("## Result of the interface-family comparison", 1)[1] readme_changes = {} for line in result.split("Count the statements", 1)[0].splitlines(): @@ -118,7 +120,7 @@ class ReviewedDocumentationContractTest(unittest.TestCase): """Keep reviewed compatibility and transaction statements complete.""" def test_rule_priority_lists_every_specificity_score(self): - guide = (_PROJECT_ROOT / "docs" / "configuration.md").read_text() + guide = (_PROJECT_ROOT / "docs" / "configuration.md").read_text(encoding="utf-8") priority = guide.split("### Rule Priority", 1)[1].split("### RE2 Pattern Syntax", 1)[0] scopes = { 7: ( @@ -145,7 +147,9 @@ def test_rule_priority_lists_every_specificity_score(self): ) def test_transaction_adr_states_unrelated_failure_behavior(self): - adr = (_PROJECT_ROOT / "docs" / "adr" / "0005-execute-each-family-in-its-own-transaction.md").read_text() + adr = (_PROJECT_ROOT / "docs" / "adr" / "0005-execute-each-family-in-its-own-transaction.md").read_text( + encoding="utf-8" + ) self.assertIn( "An unrelated integrity or infrastructure failure rolls back its own family and propagates to the operation boundary.", @@ -153,7 +157,7 @@ def test_transaction_adr_states_unrelated_failure_behavior(self): ) def test_re2_upgrade_guide_separates_errors_from_warnings(self): - guide = (_PROJECT_ROOT / "docs" / "installation.md").read_text() + guide = (_PROJECT_ROOT / "docs" / "installation.md").read_text(encoding="utf-8") section = guide.split("## Run Database Migrations", 1)[1].split("## Restart NetBox", 1)[0] migration = " ".join(section.split()) @@ -167,7 +171,7 @@ def test_re2_upgrade_guide_separates_errors_from_warnings(self): self.assertIn("run the migration again", migration) def test_configuration_names_both_pattern_matching_contexts(self): - guide = (_PROJECT_ROOT / "docs" / "configuration.md").read_text() + guide = (_PROJECT_ROOT / "docs" / "configuration.md").read_text(encoding="utf-8") pattern_guidance = guide.split("### Rule Fields", 1)[1].split("### RE2 Pattern Syntax", 1)[0] self.assertIn("module type model name", pattern_guidance) @@ -187,7 +191,7 @@ def test_pattern_help_text_names_both_matching_contexts(self): self.assertIn("Applies to Device Interfaces", help_text) def test_readme_badge_matches_the_supported_netbox_floor(self): - readme = (_PROJECT_ROOT / "README.md").read_text() + readme = (_PROJECT_ROOT / "README.md").read_text(encoding="utf-8") self.assertIn("NetBox-%E2%89%A54.3.0-blue", readme) self.assertNotIn("NetBox-%E2%89%A54.2.0-blue", readme) @@ -209,9 +213,9 @@ def _shipped_patterns(): """Return every module-type pattern the plugin ships or documents, by source.""" found = [] for path in sorted((_PROJECT_ROOT / "contrib").glob("*.yaml")): - found.extend((path.name, pattern) for pattern in _patterns_in(yaml.safe_load(path.read_text()))) + found.extend((path.name, pattern) for pattern in _patterns_in(yaml.safe_load(path.read_text(encoding="utf-8")))) for path in sorted((_PROJECT_ROOT / "docs").glob("*.md")): - for raw in re.findall(r"^\s*-?\s*module_type_pattern:\s*(.+)$", path.read_text(), re.MULTILINE): + for raw in re.findall(r"^\s*-?\s*module_type_pattern:\s*(.+)$", path.read_text(encoding="utf-8"), re.MULTILINE): value = yaml.safe_load(raw) if isinstance(value, str): found.append((path.name, value)) diff --git a/netbox_interface_name_rules/tests/test_installed_families.py b/netbox_interface_name_rules/tests/test_installed_families.py index 46728312..8068f875 100644 --- a/netbox_interface_name_rules/tests/test_installed_families.py +++ b/netbox_interface_name_rules/tests/test_installed_families.py @@ -9,7 +9,6 @@ from dcim.choices import InterfaceTypeChoices from dcim.models import ( Device, - DeviceRole, DeviceType, Interface, InterfaceTemplate, @@ -18,7 +17,6 @@ ModuleBay, ModuleBayTemplate, ModuleType, - Site, VirtualChassis, ) from django.db import IntegrityError, connection @@ -41,6 +39,7 @@ from netbox_interface_name_rules.family.execution import _lock_family from netbox_interface_name_rules.models import InterfaceNameRule from netbox_interface_name_rules.naming import evaluate_name_template +from netbox_interface_name_rules.tests.helpers import make_placement from netbox_interface_name_rules.tests.out_of_band import rename_out_of_band CHANNEL_TYPE = getattr(InterfaceTypeChoices, "TYPE_CHANNEL", "channel") @@ -87,13 +86,12 @@ def setUpTestData(cls): name="{module}", type="100gbase-x-qsfp28", ) - role = DeviceRole.objects.create(name="FamilyPlanRole", slug="family-plan-role") - site = Site.objects.create(name="FamilyPlanSite", slug="family-plan-site") + placement = make_placement("FamilyPlan") cls.device = Device.objects.create( name="family-plan-device-01", device_type=device_type, - role=role, - site=site, + role=placement.role, + site=placement.site, ) cls.bay = ModuleBay.objects.get(device=cls.device, name="Bay 7") cls.rule = InterfaceNameRule.objects.create( @@ -640,13 +638,12 @@ def setUpTestData(cls): parent=parent, channel_id=channel_id, ) - role = DeviceRole.objects.create(name="InstalledChanRole", slug="installed-chan-role") - site = Site.objects.create(name="InstalledChanSite", slug="installed-chan-site") + placement = make_placement("InstalledChan") cls.device = Device.objects.create( name="installed-chan-device-01", device_type=device_type, - role=role, - site=site, + role=placement.role, + site=placement.site, ) cls.rule = InterfaceNameRule.objects.create( module_type=cls.module_type, diff --git a/netbox_interface_name_rules/tests/test_isolated_test_settings.py b/netbox_interface_name_rules/tests/test_isolated_test_settings.py new file mode 100644 index 00000000..9e6638ea --- /dev/null +++ b/netbox_interface_name_rules/tests/test_isolated_test_settings.py @@ -0,0 +1,80 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (C) 2025 Marcin Zieba +"""Checks for the isolated test-settings shim the devcontainer runs tests with.""" + +import json +import os +import subprocess +import sys +from pathlib import Path +from unittest import TestCase + +ROOT = Path(__file__).resolve().parents[2] +CONFIG_DIR = ROOT / ".devcontainer/config" + +# Read the settings module named on the command line, so the shim can be compared with the +# NetBox settings it wraps. +_PROBE = """ +import importlib +import json +import sys + +module = importlib.import_module(sys.argv[1]) +print(json.dumps({ + "test_db_name": module.DATABASES["default"].get("TEST", {}).get("NAME"), + "queue_databases": sorted({params["DB"] for params in module.RQ_QUEUES.values()}), + "default_queue_database": module.RQ_QUEUES["default"]["DB"], +})) +""" + + +class IsolatedTestSettingsTest(TestCase): + """Assemble the shim the way Django does and read back what it isolated.""" + + # The suite's own settings module reaches the child through these, and Django would load it + # lazily instead of the module under test. + _SUITE_VARIABLES = ( + "DJANGO_SETTINGS_MODULE", + "PYTEST_XDIST_WORKER", + "REDIS_HOST", + "REDIS_CACHE_HOST", + "REDIS_DATABASE", + "REDIS_CACHE_DATABASE", + ) + + def _load(self, module, environment): + """Import *module* in a clean interpreter and return the settings it produced.""" + env = {**os.environ, **environment} + for key in self._SUITE_VARIABLES: + env.pop(key, None) + for key, value in environment.items(): + if value is None: + env.pop(key, None) + # Hand the child this interpreter's own import path. NetBox lives in a different place in + # the devcontainer than in CI, and neither location may be assumed here. + env["PYTHONPATH"] = os.pathsep.join([str(CONFIG_DIR), *(entry for entry in sys.path if entry)]) + completed = subprocess.run( + [sys.executable, "-c", _PROBE, module], + env=env, + capture_output=True, + text=True, + check=False, + ) + # Report the child's own error: a CalledProcessError would hide why the import failed. + assert completed.returncode == 0, f"{module} failed to import:\n{completed.stderr[-2000:]}" + return json.loads(completed.stdout.strip().splitlines()[-1]) + + def test_it_isolates_the_task_queue_redis_database(self): + settings = self._load("isolated_test_settings", {"TEST_DB_NAME": "inr_probe", "TEST_REDIS_DB": "9"}) + + self.assertEqual(settings["test_db_name"], "inr_probe") + self.assertEqual(settings["queue_databases"], [9]) + + def test_it_leaves_the_task_queue_alone_without_the_variable(self): + environment = {"TEST_DB_NAME": "inr_probe", "TEST_REDIS_DB": None} + baseline = self._load("netbox.settings", environment) + + settings = self._load("isolated_test_settings", environment) + + self.assertEqual(settings["default_queue_database"], baseline["default_queue_database"]) + self.assertEqual(settings["queue_databases"], baseline["queue_databases"]) diff --git a/netbox_interface_name_rules/tests/test_migrations.py b/netbox_interface_name_rules/tests/test_migrations.py new file mode 100644 index 00000000..7db2b125 --- /dev/null +++ b/netbox_interface_name_rules/tests/test_migrations.py @@ -0,0 +1,40 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (C) 2025 Marcin Zieba +"""The migration graph resolves against the migrations NetBox actually ships.""" + +from django.db.migrations.loader import MigrationLoader +from django.test import SimpleTestCase + +APP_LABEL = "netbox_interface_name_rules" + + +class MigrationGraphTest(SimpleTestCase): + """A dependency on a migration a NetBox squash removed must not reach the graph.""" + + def test_the_graph_builds_with_replacements_disabled(self): + """`migrate` remaps a replaced node through the squash that lists it in `replaces`, so it + hides this. `sqlmigrate` disables replacements and fails outright. + """ + loader = MigrationLoader(None, replace_migrations=False) + + self.assertIn((APP_LABEL, "0001_initial"), loader.graph.nodes) + + def test_every_cross_app_dependency_exists_on_disk(self): + """Report every offending dependency, not just the one the graph happens to reach first. + + Building the graph raises on the first dangling node, so a tree with two of these shows the + second only after the first is fixed. `load=False` reads the migrations off disk without it. + """ + loader = MigrationLoader(None, load=False) + loader.load_disk() + available = set(loader.disk_migrations) + + missing = [ + f"{name} depends on {dependency}" + for (app, name), migration in loader.disk_migrations.items() + if app == APP_LABEL + for dependency in migration.dependencies + if dependency[0] != APP_LABEL and not dependency[1].startswith("__") and dependency not in available + ] + + self.assertEqual(missing, [], "NetBox squashes remove migrations; depend on one it still ships.") diff --git a/netbox_interface_name_rules/tests/test_module_boundaries.py b/netbox_interface_name_rules/tests/test_module_boundaries.py new file mode 100644 index 00000000..3b95f6fc --- /dev/null +++ b/netbox_interface_name_rules/tests/test_module_boundaries.py @@ -0,0 +1,92 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (C) 2025 Marcin Zieba +"""The family package is used through its own public seam. + +`family/__init__.py` re-exports what the rest of the plugin may use. A module outside the package +that imports a submodule instead binds itself to an internal layout the package is free to change. +""" + +import ast +import pathlib +import tempfile + +from django.test import SimpleTestCase + +PACKAGE = pathlib.Path(__file__).resolve().parents[1] +FAMILY_PACKAGE = "family" + +# The family package does not export the template-name helpers the engine needs, so the engine +# reaches past the seam for them. Widening the package API is a change to its public surface and +# belongs in its own commit; until then this is the one import allowed through. +PERMITTED_SUBMODULE_IMPORTS = {("engine.py", "template_names")} + + +def _family_submodules() -> set[str]: + """Return the module names the family package is made of.""" + package = PACKAGE / FAMILY_PACKAGE + return {path.stem for path in package.glob("*.py") if path.stem != "__init__"} + + +def _family_submodule_imports(path: pathlib.Path) -> set[str]: + """Return the family submodules *path* imports directly, by either spelling.""" + submodules = _family_submodules() + tree = ast.parse(path.read_text(encoding="utf-8")) + found = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + # `import netbox_interface_name_rules.family.batch` binds the same internal layout. + for alias in node.names: + imported = alias.name.removeprefix("netbox_interface_name_rules.") + if imported.startswith(f"{FAMILY_PACKAGE}."): + found.add(imported.split(".", 1)[1]) + continue + if not isinstance(node, ast.ImportFrom) or node.module is None: + continue + module = node.module.removeprefix("netbox_interface_name_rules.") + if module == FAMILY_PACKAGE: + # A submodule imported as a name reaches past the seam just as a dotted path does. + found.update(alias.name for alias in node.names if alias.name in submodules) + elif module.startswith(f"{FAMILY_PACKAGE}."): + found.add(module.split(".", 1)[1]) + return found + + +class FamilySeamTest(SimpleTestCase): + """Modules outside the family package import the package, not its parts.""" + + def test_no_module_reaches_past_the_family_seam(self): + violations = set() + for path in sorted(PACKAGE.rglob("*.py")): + relative = path.relative_to(PACKAGE) + if relative.parts[0] in {FAMILY_PACKAGE, "tests", "migrations"}: + continue + for submodule in _family_submodule_imports(path): + if (str(relative), submodule) not in PERMITTED_SUBMODULE_IMPORTS: + violations.add(f"{relative} imports family.{submodule}") + + self.assertEqual( + violations, + set(), + "Import these through `netbox_interface_name_rules.family`, or export them from it.", + ) + + def test_a_direct_import_statement_reaches_past_the_seam_too(self): + """`import netbox_interface_name_rules.family.batch` binds the layout as a `from` import does.""" + with tempfile.TemporaryDirectory() as directory: + path = pathlib.Path(directory) / "sample.py" + path.write_text( + "import netbox_interface_name_rules.family.batch\n" + "import netbox_interface_name_rules.family.conversion as conversion\n", + encoding="utf-8", + ) + + self.assertEqual(_family_submodule_imports(path), {"batch", "conversion"}) + + def test_every_permitted_import_still_exists(self): + """A permit that nothing uses any more must be removed, not left to grant something later.""" + for name, submodule in PERMITTED_SUBMODULE_IMPORTS: + self.assertIn( + submodule, + _family_submodule_imports(PACKAGE / name), + f"{name} no longer imports family.{submodule}: drop it from the permit list.", + ) diff --git a/netbox_interface_name_rules/tests/test_naming.py b/netbox_interface_name_rules/tests/test_naming.py index dbb24be0..5c3fceb1 100644 --- a/netbox_interface_name_rules/tests/test_naming.py +++ b/netbox_interface_name_rules/tests/test_naming.py @@ -4,20 +4,19 @@ from dcim.models import ( Device, - DeviceRole, DeviceType, Manufacturer, Module, ModuleBay, ModuleBayTemplate, ModuleType, - Site, VirtualChassis, ) from django.test import TestCase from netbox_interface_name_rules.models import InterfaceNameRule from netbox_interface_name_rules.naming import build_variables, evaluate_name_template +from netbox_interface_name_rules.tests.helpers import make_placement class NamingTest(TestCase): @@ -37,14 +36,13 @@ def setUpTestData(cls): model="NAMING-SFP", part_number="NAMING-SFP", ) - role = DeviceRole.objects.create(name="NamingRole", slug="naming-role") - site = Site.objects.create(name="NamingSite", slug="naming-site") + placement = make_placement("Naming") virtual_chassis = VirtualChassis.objects.create(name="naming-vc") cls.device = Device.objects.create( name="naming-device-01", device_type=device_type, - role=role, - site=site, + role=placement.role, + site=placement.site, virtual_chassis=virtual_chassis, vc_position=3, ) diff --git a/netbox_interface_name_rules/tests/test_network_pin_guard.py b/netbox_interface_name_rules/tests/test_network_pin_guard.py new file mode 100644 index 00000000..1f91da7d --- /dev/null +++ b/netbox_interface_name_rules/tests/test_network_pin_guard.py @@ -0,0 +1,60 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (C) 2025 Marcin Zieba +"""Exercise the network pin guard without Docker.""" + +import os +import shutil +import subprocess +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest import skipUnless + +from django.test import SimpleTestCase + +_SCRIPT = Path(__file__).resolve().parents[2] / ".devcontainer/scripts/tests/test-network-pins.sh" + + +@skipUnless(shutil.which("bash"), "bash is required") +class NetworkPinGuardTest(SimpleTestCase): + def _run(self, config): + with TemporaryDirectory() as directory: + fixture = Path(directory) / "config.yaml" + fixture.write_text(config, encoding="utf-8") + return subprocess.run( + ["bash", str(_SCRIPT)], + env={**os.environ, "NETWORK_PINS_CONFIG": str(fixture)}, + capture_output=True, + text=True, + check=False, + ) + + def test_pins_outside_one_range_are_accepted(self): + result = self._run('ip_range: "198.18.0.128/25"\nipv4_address: "198.18.0.2"\n') + self.assertEqual(result.returncode, 0, result.stderr) + + def test_a_pin_inside_the_range_is_rejected(self): + result = self._run("ip_range: 198.18.0.128/25\nipv4_address: 198.18.0.129\n") + self.assertNotEqual(result.returncode, 0) + self.assertIn("lies inside the dynamic range", result.stderr) + + def test_multiple_ranges_are_rejected(self): + result = self._run("ip_range: 198.18.0.128/25\nip_range: 198.18.1.128/25\nipv4_address: 198.18.0.2\n") + self.assertNotEqual(result.returncode, 0) + + def test_a_range_with_host_bits_is_rejected(self): + result = self._run("ip_range: 198.18.0.129/25\nipv4_address: 198.18.0.2\n") + self.assertNotEqual(result.returncode, 0) + + def test_a_missing_range_is_rejected(self): + result = self._run("ipv4_address: 198.18.0.2\n") + self.assertNotEqual(result.returncode, 0) + self.assertIn("no ip_range", result.stderr) + + def test_missing_pins_are_rejected(self): + result = self._run("ip_range: 198.18.0.128/25\n") + self.assertNotEqual(result.returncode, 0) + self.assertIn("no service pins", result.stderr) + + def test_a_quote_in_an_address_is_rejected(self): + result = self._run("ip_range: 198.18.0.128/25\nipv4_address: 198.18.0.2'\n") + self.assertNotEqual(result.returncode, 0) diff --git a/netbox_interface_name_rules/tests/test_parallel_isolation.py b/netbox_interface_name_rules/tests/test_parallel_isolation.py new file mode 100644 index 00000000..a60abf20 --- /dev/null +++ b/netbox_interface_name_rules/tests/test_parallel_isolation.py @@ -0,0 +1,180 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (C) 2025 Marcin Zieba +"""Tests for parallel test worker isolation.""" + +import os +import re +import subprocess +import sys +from importlib.util import module_from_spec, spec_from_file_location +from pathlib import Path + +import pytest + +from netbox_interface_name_rules.tests.parallel import ( + _REDIS_SLOT_COUNT, + MAX_PARALLEL_WORKERS, + RESERVED_REDIS_DATABASES, + isolated_cache_location, + isolated_redis_databases, + isolated_test_database_name, +) + +_PROJECT_ROOT = Path(__file__).resolve().parents[2] + + +def _root_conftest(): + """Load the root conftest by path: it is not importable as a package module.""" + spec = spec_from_file_location("interface_name_rules_root_conftest", _PROJECT_ROOT / "conftest.py") + module = module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _run_empty_pytest(*arguments, timeout=180): + """Run pytest without collecting this plugin's suite, so only the worker rules apply.""" + environment = {key: value for key, value in os.environ.items() if not key.startswith(("PYTEST_", "COV_"))} + environment["TEST_DB_NAME"] = "test_worker_pool_contract" + environment["TEST_REDIS_HOST"] = os.environ.get("TEST_REDIS_HOST", "redis") + return subprocess.run( + [ + sys.executable, + "-m", + "pytest", + *arguments, + "--no-cov", + "-p", + "no:cacheprovider", + "--ignore=netbox_interface_name_rules", + ], + capture_output=True, + text=True, + env=environment, + cwd=_PROJECT_ROOT, + check=False, + timeout=timeout, + ) + + +def test_a_worker_gets_one_database_and_two_redis_databases(): + assert isolated_test_database_name("test_inr", "gw3") == "test_inr_gw3" + assert isolated_redis_databases("gw3") == ( + RESERVED_REDIS_DATABASES + 4, + RESERVED_REDIS_DATABASES + 4 + _REDIS_SLOT_COUNT, + ) + + +def test_the_live_rqworker_databases_are_never_handed_out(): + """The devcontainer's own rqworker and cache hold the first two databases.""" + handed_out = set() + for worker in [None, *(f"gw{number}" for number in range(MAX_PARALLEL_WORKERS))]: + handed_out.update(isolated_redis_databases(worker)) + + assert handed_out.isdisjoint(range(RESERVED_REDIS_DATABASES)) + assert max(handed_out) < 16 + + +def test_every_worker_pair_is_distinct(): + """Two workers must never share a task or a cache database.""" + pairs = [isolated_redis_databases(f"gw{number}") for number in range(MAX_PARALLEL_WORKERS)] + assigned = [database for pair in pairs for database in pair] + + assert len(set(assigned)) == len(assigned) + + +def test_a_serial_run_never_shares_a_redis_pair_with_a_worker(): + """A serial session and an xdist worker on one Redis host must not share queues or cache.""" + pairs = [ + isolated_redis_databases(None), + *(isolated_redis_databases(f"gw{number}") for number in range(MAX_PARALLEL_WORKERS)), + ] + assigned = [database for pair in pairs for database in pair] + + assert len(set(assigned)) == len(assigned) + + +def test_a_serial_run_also_avoids_the_live_databases(): + assert isolated_test_database_name("test_inr", None) == "test_inr" + assert isolated_redis_databases(None) == (RESERVED_REDIS_DATABASES, RESERVED_REDIS_DATABASES + _REDIS_SLOT_COUNT) + + +def test_the_cache_location_keeps_everything_but_the_host_and_database(): + """Rewriting the URL must not drop the port, the scheme or the credentials a deployment sets.""" + assert isolated_cache_location("redis://localhost:6379/1", "redis", 11) == "redis://redis:6379/11" + assert isolated_cache_location("rediss://user:pw@old:6380/1", "new", 4) == "rediss://user:pw@new:6380/4" + assert isolated_cache_location("redis://user@old:6380/1", "new", 4) == "redis://user@new:6380/4" + assert isolated_cache_location("redis://:pw@old:6380/1", "new", 4) == "redis://:pw@new:6380/4" + assert isolated_cache_location("redis://us%40er@old:6380/1", "new", 4) == "redis://us%40er@new:6380/4" + assert isolated_cache_location("rediss://user:p%3Aw@old:6380/1", "new", 4) == "rediss://user:p%3Aw@new:6380/4" + assert isolated_cache_location("redis://localhost/1", "redis", 9) == "redis://redis/9" + + +def test_database_name_stays_within_the_postgresql_limit(): + database_name = isolated_test_database_name(f"test_{'x' * 70}", "gw5") + + assert len(database_name) == 63 + assert database_name.endswith("_gw5") + + +def test_a_worker_above_the_ceiling_is_rejected(): + with pytest.raises(ValueError, match=f"At most {MAX_PARALLEL_WORKERS} pytest workers"): + isolated_redis_databases(f"gw{MAX_PARALLEL_WORKERS}") + + +def test_an_unrecognised_worker_id_is_rejected(): + with pytest.raises(ValueError, match="Unsupported pytest worker ID"): + isolated_redis_databases("worker-3") + + +@pytest.mark.parametrize(("detected", "expected"), [("2", 2), ("32", MAX_PARALLEL_WORKERS)]) +def test_auto_worker_count_never_exceeds_the_ceiling(monkeypatch, detected, expected): + monkeypatch.setenv("PYTEST_XDIST_AUTO_NUM_WORKERS", detected) + + assert _root_conftest().pytest_xdist_auto_num_workers(None) == expected + + +def test_a_bare_run_caps_the_auto_worker_pool(): + """The cap must reach an invocation that names no test path, which the addopts `-n auto` targets.""" + result = _run_empty_pytest("-n", "auto", "-v") + + # `--ignore` leaves nothing to collect, so pytest exits 5. + assert result.returncode in (0, 5), f"exit {result.returncode}\n{result.stdout[-3000:]}" + created = re.search(r"created: (\d+)/\d+ workers", result.stdout) + assert created is not None, result.stdout[-3000:] + assert 0 < int(created.group(1)) <= MAX_PARALLEL_WORKERS + + +def test_an_explicit_worker_count_above_the_ceiling_is_refused(): + """`-n 16` never reaches the auto hook, so without the refusal it would start unisolated workers.""" + result = _run_empty_pytest("-n", str(MAX_PARALLEL_WORKERS + 1)) + + # 4 is pytest's usage-error status: it must refuse rather than run and collide later. + assert result.returncode == 4, f"exit {result.returncode}\n{(result.stdout + result.stderr)[-3000:]}" + assert f"at most {MAX_PARALLEL_WORKERS} pytest workers" in result.stdout + result.stderr + + +def test_collecting_without_running_is_left_alone(): + """xdist starts no worker for `--collect-only`, so the ceiling has nothing to refuse.""" + result = _run_empty_pytest( + "-o", "addopts=", "--collect-only", "--tx", f"{MAX_PARALLEL_WORKERS + 1}*popen", "--dist", "load" + ) + + assert result.returncode == 5, f"exit {result.returncode}\n{(result.stdout + result.stderr)[-3000:]}" + + +@pytest.mark.django_db +def test_the_running_worker_uses_its_private_targets(settings): + """Apply the worker identity to the real Django settings, not just to the helpers.""" + worker_id = os.environ.get("PYTEST_XDIST_WORKER") + tasks_database, cache_database = isolated_redis_databases(worker_id) + + assert settings.DATABASES["default"]["TEST"]["NAME"] == isolated_test_database_name( + os.environ["TEST_DB_NAME"], worker_id + ) + assert settings.RQ_QUEUES["default"]["DB"] == tasks_database + assert settings.CACHES["default"]["LOCATION"].endswith(f"/{cache_database}") + + +def test_only_this_plugin_loads_under_the_test_settings(settings): + """A co-installed plugin must not reach a run of this suite.""" + assert settings.PLUGINS == ["netbox_interface_name_rules"] diff --git a/netbox_interface_name_rules/tests/test_performance_compare.py b/netbox_interface_name_rules/tests/test_performance_compare.py index 9d9cdc79..10b3ac01 100644 --- a/netbox_interface_name_rules/tests/test_performance_compare.py +++ b/netbox_interface_name_rules/tests/test_performance_compare.py @@ -10,7 +10,6 @@ import unittest from pathlib import Path from tempfile import TemporaryDirectory -from unittest.mock import patch _PROJECT_ROOT = Path(__file__).resolve().parents[2] _COMPARE_PATH = _PROJECT_ROOT / "performance" / "compare.py" @@ -67,18 +66,20 @@ class PerformancePackageTest(unittest.TestCase): """The repository-only performance harness does not ship as a generic package.""" def test_package_discovery_excludes_the_performance_tools(self): - configuration = tomllib.loads((_PROJECT_ROOT / "pyproject.toml").read_text()) + configuration = tomllib.loads((_PROJECT_ROOT / "pyproject.toml").read_text(encoding="utf-8")) patterns = configuration["tool"]["setuptools"]["packages"]["find"]["include"] self.assertFalse(any(fnmatch.fnmatchcase("performance", pattern) for pattern in patterns)) def test_pytest_adds_the_repository_checkout_to_pythonpath(self): - configuration = tomllib.loads((_PROJECT_ROOT / "pyproject.toml").read_text()) + configuration = tomllib.loads((_PROJECT_ROOT / "pyproject.toml").read_text(encoding="utf-8")) self.assertIn(".", configuration["tool"]["pytest"]["ini_options"]["pythonpath"]) def test_recorded_direct_callbacks_reach_zero_shared_reads(self): - comparison = (_PROJECT_ROOT / "performance" / "comparisons" / "family-package-vs-existing.md").read_text() + comparison = (_PROJECT_ROOT / "performance" / "comparisons" / "family-package-vs-existing.md").read_text( + encoding="utf-8" + ) expected_scenarios = { "module.direct_callback.no_matching_rule", "module.direct_callback.plain_rename", @@ -100,12 +101,12 @@ def test_recorded_direct_callbacks_reach_zero_shared_reads(self): self.assertEqual({row[3] for row in direct_reads}, {"0"}) def test_readme_does_not_equate_shared_reads_with_all_disk_io(self): - readme = (_PROJECT_ROOT / "performance" / "README.md").read_text() + readme = (_PROJECT_ROOT / "performance" / "README.md").read_text(encoding="utf-8") self.assertNotIn("never goes to disk", readme) def test_readme_does_not_infer_planner_cost_from_statement_counts(self): - readme = (_PROJECT_ROOT / "performance" / "README.md").read_text() + readme = (_PROJECT_ROOT / "performance" / "README.md").read_text(encoding="utf-8") self.assertNotIn("less planner work", readme) self.assertIn( @@ -115,14 +116,16 @@ def test_readme_does_not_infer_planner_cost_from_statement_counts(self): def test_readme_shared_read_claim_matches_the_comparison(self): """Read the recorded shared reads rather than pinning what one pair of runs happened to show.""" - comparison = (_PROJECT_ROOT / "performance" / "comparisons" / "family-package-vs-existing.md").read_text() + comparison = (_PROJECT_ROOT / "performance" / "comparisons" / "family-package-vs-existing.md").read_text( + encoding="utf-8" + ) before_reads = after_reads = 0 for line in comparison.splitlines(): cells = [cell.strip().strip("`") for cell in line.strip().strip("|").split("|")] if len(cells) == 6 and cells[1] == "Shared reads" and ".direct_callback." in cells[0]: before_reads += int(cells[2]) after_reads += int(cells[3]) - readme = _unwrapped((_PROJECT_ROOT / "performance" / "README.md").read_text()) + readme = _unwrapped((_PROJECT_ROOT / "performance" / "README.md").read_text(encoding="utf-8")) self.assertNotIn("No shared-buffer reads were observed in any direct-callback scenario.", readme) scoped_claim = "No shared-buffer reads were observed in any direct-callback scenario after the refactor" @@ -137,7 +140,9 @@ def test_readme_shared_read_claim_matches_the_comparison(self): self.assertIn(before_claim, readme) def test_comparison_separates_deterministic_counts_from_cache_metrics(self): - comparison = (_PROJECT_ROOT / "performance" / "comparisons" / "family-package-vs-existing.md").read_text() + comparison = (_PROJECT_ROOT / "performance" / "comparisons" / "family-package-vs-existing.md").read_text( + encoding="utf-8" + ) introduction = comparison.split("## Environment", 1)[0] self.assertIn(compare._COMPARISON_INTRO, introduction) @@ -149,7 +154,9 @@ def test_comparison_separates_deterministic_counts_from_cache_metrics(self): self.assertNotIn("Database work is deterministic", introduction) def test_comparison_machine_time_note_matches_the_load_it_reports(self): - comparison = (_PROJECT_ROOT / "performance" / "comparisons" / "family-package-vs-existing.md").read_text() + comparison = (_PROJECT_ROOT / "performance" / "comparisons" / "family-package-vs-existing.md").read_text( + encoding="utf-8" + ) environment = comparison.split("## Environment", 1)[1].split("## Database work", 1)[0] machine_time = comparison.split("## Machine time", 1)[1].split("## Statement-count regressions", 1)[0] load_row = next(row for row in environment.splitlines() if row.startswith("| host load")) @@ -162,19 +169,6 @@ def test_comparison_machine_time_note_matches_the_load_it_reports(self): self.assertIn(expected, machine_time) -class XdistWorkerCapTest(unittest.TestCase): - """The shared host gets a bounded number of automatic pytest workers.""" - - def test_auto_worker_count_is_capped_at_eight(self): - configuration_path = _PROJECT_ROOT / "conftest.py" - spec = importlib.util.spec_from_file_location("project_conftest", configuration_path) - configuration = importlib.util.module_from_spec(spec) - spec.loader.exec_module(configuration) - - with patch("xdist.plugin.pytest_xdist_auto_num_workers", return_value=32): - self.assertEqual(configuration.pytest_xdist_auto_num_workers(object()), 8) - - class PlanIdentityTest(unittest.TestCase): """Equivalent plans group together, so runtime statistics must stay out of the identity.""" @@ -213,8 +207,8 @@ class ComparisonDestinationTest(unittest.TestCase): def test_main_validates_inputs_before_the_destination(self): with TemporaryDirectory(dir=_PROJECT_ROOT) as directory: before, after = Path(directory) / "before.json", Path(directory) / "after.json" - before.write_text(json.dumps(_timed_artifact(_MACHINE_TIME))) - after.write_text(json.dumps(_artifact({}))) + before.write_text(json.dumps(_timed_artifact(_MACHINE_TIME)), encoding="utf-8") + after.write_text(json.dumps(_artifact({})), encoding="utf-8") with self.assertRaisesRegex(ValueError, "scenarios"): compare.main(["compare.py", str(before), str(after), str(before)]) @@ -223,12 +217,12 @@ def test_main_writes_a_separate_report(self): root = Path(directory) before, after, destination = root / "before.json", root / "after.json", root / "report.md" contents = json.dumps(_timed_artifact(_MACHINE_TIME)) - before.write_text(contents) - after.write_text(contents) + before.write_text(contents, encoding="utf-8") + after.write_text(contents, encoding="utf-8") compare.main(["compare.py", str(before), str(after), str(destination)]) - self.assertIn("# Automatic naming performance comparison", destination.read_text()) - self.assertEqual(before.read_text(), contents) - self.assertEqual(after.read_text(), contents) + self.assertIn("# Automatic naming performance comparison", destination.read_text(encoding="utf-8")) + self.assertEqual(before.read_text(encoding="utf-8"), contents) + self.assertEqual(after.read_text(encoding="utf-8"), contents) def test_main_refuses_to_overwrite_either_input(self): for target in ("before.json", "after.json"): @@ -237,16 +231,16 @@ def test_main_refuses_to_overwrite_either_input(self): root = Path(directory) before, after = root / "before.json", root / "after.json" contents = json.dumps(_timed_artifact(_MACHINE_TIME)) - before.write_text(contents) - after.write_text(contents) + before.write_text(contents, encoding="utf-8") + after.write_text(contents, encoding="utf-8") destination = root / target if symlink: destination = root / "report.md" destination.symlink_to(root / target) with self.assertRaisesRegex(SystemExit, "destination.*input"): compare.main(["compare.py", str(before), str(after), str(destination)]) - self.assertEqual(before.read_text(), contents) - self.assertEqual(after.read_text(), contents) + self.assertEqual(before.read_text(encoding="utf-8"), contents) + self.assertEqual(after.read_text(encoding="utf-8"), contents) def test_main_refuses_a_hard_link_to_either_input(self): for target in ("before.json", "after.json"): diff --git a/netbox_interface_name_rules/tests/test_regex.py b/netbox_interface_name_rules/tests/test_regex.py index 208639b6..79a2b3ac 100644 --- a/netbox_interface_name_rules/tests/test_regex.py +++ b/netbox_interface_name_rules/tests/test_regex.py @@ -7,8 +7,6 @@ import re2 from dcim.models import ( - Device, - DeviceRole, DeviceType, Interface, Manufacturer, @@ -16,7 +14,6 @@ ModuleBay, ModuleBayTemplate, ModuleType, - Site, ) from django.core.exceptions import ValidationError from django.db import connection @@ -25,6 +22,7 @@ from netbox_interface_name_rules.engine import apply_interface_name_rules, find_matching_rule, has_applicable_interfaces from netbox_interface_name_rules.models import InterfaceNameRule +from netbox_interface_name_rules.tests.helpers import make_device _RE2_AUDIT = importlib.import_module("netbox_interface_name_rules.migrations.0014_validate_re2_patterns") @@ -390,9 +388,7 @@ def setUpTestData(cls): cls.mt_zr = ModuleType.objects.create(manufacturer=manufacturer, model="QSFP-DD-400G-ZR", part_number="APZR") ModuleBayTemplate.objects.create(device_type=cls.device_type, name="Transceiver 0", position="0") ModuleBayTemplate.objects.create(device_type=cls.device_type, name="Transceiver 1", position="1") - role = DeviceRole.objects.create(name="RxApplyRole", slug="rxapplyrole") - site = Site.objects.create(name="RxApplySite", slug="rxapplysite") - cls.device = Device.objects.create(name="rx-apply-01", device_type=cls.device_type, role=role, site=site) + cls.device = make_device("RxApply", cls.device_type, name="rx-apply-01") def test_regex_rule_renames_interface(self): """Regex rule matches and renames interface for LR4.""" diff --git a/netbox_interface_name_rules/tests/test_rule_validation_agreement.py b/netbox_interface_name_rules/tests/test_rule_validation_agreement.py index f2629d07..c1e0b5c0 100644 --- a/netbox_interface_name_rules/tests/test_rule_validation_agreement.py +++ b/netbox_interface_name_rules/tests/test_rule_validation_agreement.py @@ -245,7 +245,7 @@ def test_migrations_have_no_live_application_imports(self): migrations = Path(__file__).resolve().parents[1] / "migrations" for migration in sorted(migrations.rglob("*.py")): with self.subTest(migration=migration.name): - tree = ast.parse(migration.read_text()) + tree = ast.parse(migration.read_text(encoding="utf-8")) imports = [] for node in ast.walk(tree): if isinstance(node, ast.ImportFrom): diff --git a/netbox_interface_name_rules/tests/test_rules.py b/netbox_interface_name_rules/tests/test_rules.py index b5a7284a..736daaf3 100644 --- a/netbox_interface_name_rules/tests/test_rules.py +++ b/netbox_interface_name_rules/tests/test_rules.py @@ -8,8 +8,6 @@ import threading from dcim.models import ( - Device, - DeviceRole, DeviceType, Interface, Manufacturer, @@ -18,7 +16,6 @@ ModuleBayTemplate, ModuleType, Platform, - Site, ) from django.test import TestCase @@ -28,6 +25,7 @@ find_matching_rule, ) from netbox_interface_name_rules.models import InterfaceNameRule +from netbox_interface_name_rules.tests.helpers import make_device class FindMatchingRuleTest(TestCase): @@ -158,9 +156,7 @@ def setUpTestData(cls): cls.device_type = DeviceType.objects.create(manufacturer=manufacturer, model="VAR-DEV", slug="var-dev") # Templates must be created BEFORE devices (instantiated on device creation) ModuleBayTemplate.objects.create(device_type=cls.device_type, name="Transceiver 5", position="5") - role = DeviceRole.objects.create(name="VarRole", slug="varrole") - site = Site.objects.create(name="VarSite", slug="varsite") - cls.device = Device.objects.create(name="var-test-01", device_type=cls.device_type, role=role, site=site) + cls.device = make_device("Var", cls.device_type, name="var-test-01") def test_simple_bay_variables(self): bay = ModuleBay.objects.get(device=self.device, name="Transceiver 5") @@ -174,9 +170,7 @@ def test_non_numeric_position(self): manufacturer = Manufacturer.objects.create(name="NNMfg", slug="nnmfg") dt = DeviceType.objects.create(manufacturer=manufacturer, model="NN-DEV", slug="nn-dev") ModuleBayTemplate.objects.create(device_type=dt, name="Transceiver swp3", position="swp3") - role = DeviceRole.objects.create(name="NNRole", slug="nnrole") - site = Site.objects.create(name="NNSite", slug="nnsite") - device = Device.objects.create(name="nn-test-01", device_type=dt, role=role, site=site) + device = make_device("NN", dt, name="nn-test-01") bay = ModuleBay.objects.get(device=device, name="Transceiver swp3") variables = build_variables(bay) self.assertEqual(variables["bay_position"], "swp3") @@ -196,9 +190,7 @@ def setUpTestData(cls): # Templates before device ModuleBayTemplate.objects.create(device_type=cls.device_type, name="Transceiver 0", position="0") ModuleBayTemplate.objects.create(device_type=cls.device_type, name="Transceiver 1", position="1") - role = DeviceRole.objects.create(name="ApplyRole", slug="applyrole") - site = Site.objects.create(name="ApplySite", slug="applysite") - cls.device = Device.objects.create(name="apply-test-01", device_type=cls.device_type, role=role, site=site) + cls.device = make_device("Apply", cls.device_type, name="apply-test-01") def test_simple_rename(self): """Module install with matching rule renames the interface.""" diff --git a/netbox_interface_name_rules/tests/test_signals.py b/netbox_interface_name_rules/tests/test_signals.py index eb5501f8..8dd1f5cb 100644 --- a/netbox_interface_name_rules/tests/test_signals.py +++ b/netbox_interface_name_rules/tests/test_signals.py @@ -8,7 +8,6 @@ from dcim.models import ( Device, - DeviceRole, DeviceType, Interface, Manufacturer, @@ -16,7 +15,6 @@ ModuleBay, ModuleBayTemplate, ModuleType, - Site, VirtualChassis, ) from django.test import TestCase @@ -30,6 +28,7 @@ on_module_pre_save, on_module_saved, ) +from netbox_interface_name_rules.tests.helpers import make_device, make_placement _librenms_available = importlib.util.find_spec("netbox_librenms_plugin") is not None @@ -48,9 +47,7 @@ def setUpTestData(cls): ) ModuleBayTemplate.objects.create(device_type=cls.device_type, name="SigBay 0", position="0") ModuleBayTemplate.objects.create(device_type=cls.device_type, name="SigBay 1", position="1") - role = DeviceRole.objects.create(name="SigRole", slug="sigrole") - site = Site.objects.create(name="SigSite", slug="sigsite") - cls.device = Device.objects.create(name="sig-test-01", device_type=cls.device_type, role=role, site=site) + cls.device = make_device("Sig", cls.device_type, name="sig-test-01") cls.bay0 = ModuleBay.objects.get(device=cls.device, name="SigBay 0") cls.bay1 = ModuleBay.objects.get(device=cls.device, name="SigBay 1") @@ -159,15 +156,14 @@ def setUpTestData(cls): cls.module_type = ModuleType.objects.create( manufacturer=manufacturer, model="SigDev-SFP", part_number="SigDev-SFP" ) - role = DeviceRole.objects.create(name="SigDevRole", slug="sigdevrole") - site = Site.objects.create(name="SigDevSite", slug="sigdevsite") + placement = make_placement("SigDev") cls.vc = VirtualChassis.objects.create(name="sigdev-vc") cls.device = Device.objects.create( name="sigdev-sw1", device_type=device_type, - role=role, - site=site, + role=placement.role, + site=placement.site, virtual_chassis=cls.vc, vc_position=1, ) @@ -287,14 +283,13 @@ def test_deferred_device_no_modules_no_error(self): """_apply_rules_for_device_deferred with device having no modules runs without error.""" manufacturer = Manufacturer.objects.create(name="SigDevMfg2", slug="sigdevmfg2") dt = DeviceType.objects.create(manufacturer=manufacturer, model="SD2-Switch", slug="sd2-switch") - role = DeviceRole.objects.create(name="SD2Role", slug="sd2role") - site = Site.objects.create(name="SD2Site", slug="sd2site") + placement = make_placement("SD2") vc = VirtualChassis.objects.create(name="sd2-vc") device_no_modules = Device.objects.create( name="sd2-sw-nomod", device_type=dt, - role=role, - site=site, + role=placement.role, + site=placement.site, virtual_chassis=vc, vc_position=2, ) @@ -315,14 +310,13 @@ def setUpTestData(cls): cls.device_type = DeviceType.objects.create(manufacturer=manufacturer, model="SigX-Dev", slug="sigx-dev") cls.module_type = ModuleType.objects.create(manufacturer=manufacturer, model="SigX-SFP", part_number="SigX-SFP") ModuleBayTemplate.objects.create(device_type=cls.device_type, name="SigXBay 0", position="0") - role = DeviceRole.objects.create(name="SigXRole", slug="sigxrole") - site = Site.objects.create(name="SigXSite", slug="sigxsite") + placement = make_placement("SigX") cls.vc = VirtualChassis.objects.create(name="sigx-vc") cls.device = Device.objects.create( name="sigx-sw1", device_type=cls.device_type, - role=role, - site=site, + role=placement.role, + site=placement.site, virtual_chassis=cls.vc, vc_position=1, ) @@ -401,14 +395,13 @@ def setUpTestData(cls): cls.module_type = ModuleType.objects.create( manufacturer=manufacturer, model="NullBay-SFP", part_number="NullBay-SFP" ) - role = DeviceRole.objects.create(name="NullBayRole", slug="nullbayrole") - site = Site.objects.create(name="NullBaySite", slug="nullbaysite") + placement = make_placement("NullBay") vc = VirtualChassis.objects.create(name="nullbay-vc") cls.device = Device.objects.create( name="nullbay-sw1", device_type=device_type, - role=role, - site=site, + role=placement.role, + site=placement.site, virtual_chassis=vc, vc_position=1, ) @@ -465,14 +458,13 @@ def setUpTestData(cls): cls.module_type = ModuleType.objects.create( manufacturer=manufacturer, model="OuterX-SFP", part_number="OuterX-SFP" ) - role = DeviceRole.objects.create(name="OuterXRole", slug="outerxrole") - site = Site.objects.create(name="OuterXSite", slug="outerxsite") + placement = make_placement("OuterX") vc = VirtualChassis.objects.create(name="outerx-vc") cls.device = Device.objects.create( name="outerx-sw1", device_type=device_type, - role=role, - site=site, + role=placement.role, + site=placement.site, virtual_chassis=vc, vc_position=1, ) @@ -518,9 +510,7 @@ def setUpTestData(cls): cls.module_type = ModuleType.objects.create( manufacturer=manufacturer, model="DelCas-SFP", part_number="DelCas-SFP" ) - role = DeviceRole.objects.create(name="DelCasRole", slug="delcasrole") - site = Site.objects.create(name="DelCasSite", slug="delcassite") - cls.device = Device.objects.create(name="delcas-sw1", device_type=device_type, role=role, site=site) + cls.device = make_device("DelCas", device_type, name="delcas-sw1") cls.bay = ModuleBay.objects.get(device=cls.device, name="DCBay 0") def test_interfaces_deleted_when_module_removed(self): @@ -556,9 +546,7 @@ def setUpTestData(cls): cls.device_type = DeviceType.objects.create(manufacturer=mfg, model="PSL-Dev", slug="psl-dev") cls.module_type = ModuleType.objects.create(manufacturer=mfg, model="PSL-SFP", part_number="PSL-SFP") ModuleBayTemplate.objects.create(device_type=cls.device_type, name="Bay 0", position="0") - role = DeviceRole.objects.create(name="PSLRole", slug="pslrole") - site = Site.objects.create(name="PSLSite", slug="pslsite") - cls.device = Device.objects.create(name="psl-dev-01", device_type=cls.device_type, role=role, site=site) + cls.device = make_device("PSL", cls.device_type, name="psl-dev-01") cls.bay = ModuleBay.objects.get(device=cls.device, name="Bay 0") def test_pre_save_db_error_logs_warning(self): @@ -594,9 +582,7 @@ def setUpTestData(cls): cls.device_type = DeviceType.objects.create(manufacturer=manufacturer, model="PRED-Dev", slug="pred-dev") cls.module_type = ModuleType.objects.create(manufacturer=manufacturer, model="PRED-SFP", part_number="PRED-SFP") ModuleBayTemplate.objects.create(device_type=cls.device_type, name="PredBay 0", position="c9") - role = DeviceRole.objects.create(name="PredRole", slug="predrole") - site = Site.objects.create(name="PredSite", slug="predsite") - cls.device = Device.objects.create(name="pred-test-01", device_type=cls.device_type, role=role, site=site) + cls.device = make_device("Pred", cls.device_type, name="pred-test-01") cls.bay = ModuleBay.objects.get(device=cls.device, name="PredBay 0") @skipUnless(_librenms_available, "netbox_librenms_plugin not installed") diff --git a/netbox_interface_name_rules/tests/test_structural_families.py b/netbox_interface_name_rules/tests/test_structural_families.py index dce9670b..bfac21a8 100644 --- a/netbox_interface_name_rules/tests/test_structural_families.py +++ b/netbox_interface_name_rules/tests/test_structural_families.py @@ -10,7 +10,7 @@ from unittest import skipIf, skipUnless -from dcim.models import Device, DeviceRole, DeviceType, Interface, Manufacturer, Site +from dcim.models import DeviceType, Interface, Manufacturer from django.db import IntegrityError, connection from django.test import TestCase from django.test.utils import CaptureQueriesContext @@ -30,6 +30,7 @@ ) from netbox_interface_name_rules.family import names as family_names from netbox_interface_name_rules.models import InterfaceNameRule +from netbox_interface_name_rules.tests.helpers import make_device from netbox_interface_name_rules.tests.out_of_band import rename_out_of_band from netbox_interface_name_rules.tests.test_breakout_mode import CHANNELIZED, _plain_module_type from netbox_interface_name_rules.tests.test_channelization import ( @@ -134,9 +135,7 @@ class DeferredChannelNameReconciliationTest(TestCase): def setUpTestData(cls): manufacturer = Manufacturer.objects.create(name="ReconMfg", slug="recon-mfg") device_type = DeviceType.objects.create(manufacturer=manufacturer, model="RECON-DEVICE", slug="recon-device") - role = DeviceRole.objects.create(name="ReconRole", slug="recon-role") - site = Site.objects.create(name="ReconSite", slug="recon-site") - cls.device = Device.objects.create(name="recon-device-01", device_type=device_type, role=role, site=site) + cls.device = make_device("Recon", device_type, name="recon-device-01") def _interface(self, name): """Create one plain interface on the shared device.""" @@ -346,7 +345,7 @@ def test_the_scan_reports_the_first_taken_name_in_plan_order(self): def test_the_base_row_never_counts_as_a_collision(self): _module, _bay, plan = self._plan() - Interface.objects.filter(pk=plan.base.pk).update(name=plan.target_names[0]) + rename_out_of_band(Interface.objects.get(pk=plan.base.pk), plan.target_names[0]) self.assertIsNone(structural._first_taken_name(plan)) diff --git a/netbox_interface_name_rules/tests/test_text_encoding.py b/netbox_interface_name_rules/tests/test_text_encoding.py new file mode 100644 index 00000000..bdf79eaf --- /dev/null +++ b/netbox_interface_name_rules/tests/test_text_encoding.py @@ -0,0 +1,56 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (C) 2025 Marcin Zieba +"""Require explicit encodings for text file access in the package.""" + +import ast +import pathlib + +from django.test import SimpleTestCase + +PACKAGE = pathlib.Path(__file__).resolve().parents[1] + + +def _missing_encodings(tree): + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + function = node.func + name = function.attr if isinstance(function, ast.Attribute) else getattr(function, "id", None) + if name not in {"open", "read_text", "write_text"}: + continue + if any(keyword.arg == "encoding" for keyword in node.keywords): + continue + if name == "open": + module_open = ( + isinstance(function, ast.Attribute) + and isinstance(function.value, ast.Name) + and function.value.id in {"io", "builtins"} + ) + index = 0 if isinstance(function, ast.Attribute) and not module_open else 1 + mode = next((keyword.value for keyword in node.keywords if keyword.arg == "mode"), None) + if mode is None and len(node.args) > index: + mode = node.args[index] + if isinstance(mode, ast.Constant) and isinstance(mode.value, str) and "b" in mode.value: + continue + yield node.lineno + + +class TextEncodingTest(SimpleTestCase): + def test_text_file_calls_declare_an_encoding(self): + violations = [] + for path in sorted(PACKAGE.rglob("*.py")): + tree = ast.parse(path.read_text(encoding="utf-8")) + violations.extend(f"{path.relative_to(PACKAGE)}:{line}" for line in _missing_encodings(tree)) + self.assertEqual(violations, [], "Missing encoding= at:\n" + "\n".join(violations)) + + def test_binary_modes_do_not_need_an_encoding(self): + tree = ast.parse('open("data", "rb"); path.open("wb"); open("data", mode="ab"); path.open(mode="rb")') + self.assertEqual(list(_missing_encodings(tree)), []) + + def test_text_access_requires_an_encoding_keyword(self): + tree = ast.parse('open("data")\npath.open("r")\npath.read_text()\npath.write_text("text")') + self.assertEqual(list(_missing_encodings(tree)), [1, 2, 3, 4]) + + def test_module_open_uses_the_second_argument_as_mode(self): + tree = ast.parse('io.open("blob.txt", "r")\nbuiltins.open("blob.txt")\nio.open("data.txt", "rb")') + self.assertEqual(list(_missing_encodings(tree)), [1, 2]) diff --git a/netbox_interface_name_rules/tests/test_views.py b/netbox_interface_name_rules/tests/test_views.py index 25d095c0..25e7544b 100644 --- a/netbox_interface_name_rules/tests/test_views.py +++ b/netbox_interface_name_rules/tests/test_views.py @@ -5,21 +5,19 @@ from unittest.mock import ANY, MagicMock, patch from dcim.models import ( - Device, - DeviceRole, DeviceType, Interface, Manufacturer, Module, ModuleBay, ModuleType, - Site, ) from django.contrib.auth import get_user_model from django.test import TestCase from django.urls import reverse from netbox_interface_name_rules.models import InterfaceNameRule +from netbox_interface_name_rules.tests.helpers import make_device User = get_user_model() @@ -187,9 +185,7 @@ def test_a_conversion_scan_failure_keeps_the_apply_preview(self): """A bug in the conversion scan must not blank the unrelated apply preview.""" from django.contrib.messages import get_messages - role = DeviceRole.objects.create(name="SplitRole", slug="splitrole") - site = Site.objects.create(name="SplitSite", slug="splitsite") - device = Device.objects.create(name="split-dev-01", device_type=self.device_type, role=role, site=site) + device = make_device("Split", self.device_type, name="split-dev-01") bay = ModuleBay.objects.create(device=device, name="Bay 0", position="0") module = Module.objects.create(device=device, module_bay=bay, module_type=self.module_type) Interface.objects.create(device=device, module=module, name="0", type="10gbase-x-sfpp") @@ -539,9 +535,7 @@ def test_post_apply_with_interface_ids_renames_real_interfaces(self): from django.contrib.messages import get_messages # cls.rule matches module_type VIEW-SFP on device_type VIEW-Dev → "et-0/0/{bay_position}". - role = DeviceRole.objects.create(name="ApplyRole", slug="applyrole") - site = Site.objects.create(name="ApplySite", slug="applysite") - device = Device.objects.create(name="apply-dev-01", device_type=self.device_type, role=role, site=site) + device = make_device("Apply", self.device_type, name="apply-dev-01") bay = ModuleBay.objects.create(device=device, name="Bay 0", position="0") module = Module.objects.create(device=device, module_bay=bay, module_type=self.module_type) iface = Interface.objects.create(device=device, module=module, name="0", type="10gbase-x-sfpp") diff --git a/netbox_interface_name_rules/tests/test_workflows.py b/netbox_interface_name_rules/tests/test_workflows.py new file mode 100644 index 00000000..96bc50e4 --- /dev/null +++ b/netbox_interface_name_rules/tests/test_workflows.py @@ -0,0 +1,283 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (C) 2025 Marcin Zieba +"""Every workflow that runs pytest must install the plugins `addopts` makes mandatory. + +pytest fails during argument parsing when `addopts` names an option no installed plugin registers, +so a missing distribution breaks the job before a single test runs. +""" + +import pathlib +import re +import shlex +import tempfile +import tomllib + +import tree_sitter_bash +import yaml +from django.test import SimpleTestCase +from tree_sitter import Language, Parser + +_PROJECT_ROOT = pathlib.Path(__file__).resolve().parents[2] +_WORKFLOWS = _PROJECT_ROOT / ".github" / "workflows" +_SHELL_LANGUAGE = Language(tree_sitter_bash.language()) + +# The option each pytest plugin registers. `--no-cov` needs pytest-cov for the same reason `--cov` +# does: an unregistered option is a parse error, whether it turns coverage on or off. +_OPTION_OWNERS = { + "--cov": "pytest-cov", + "--no-cov": "pytest-cov", + "-n": "pytest-xdist", + "--dist": "pytest-xdist", +} + + +def _configured_addopts(): + """Return the `addopts` string pytest applies to every invocation in this repository.""" + with (_PROJECT_ROOT / "pyproject.toml").open("rb") as handle: + return tomllib.load(handle)["tool"]["pytest"]["ini_options"]["addopts"] + + +def _required_distributions(*command_lines): + """Return the pytest plugins the options in *command_lines* require.""" + text = " ".join(command_lines) + # An option ends at whitespace, at `=`, or at end of input; `--cov-report` is not `--cov`. + return { + owner for option, owner in _OPTION_OWNERS.items() if re.search(rf"(?=0.16", @@ -62,6 +63,10 @@ dev = [ "python-semantic-release", "django>=5.1,<7.0", ] +workflow-tests = [ + "tree-sitter==0.26.0", + "tree-sitter-bash==0.25.1", +] docs = [ "mkdocs>=1,<2", "mkdocs-material>=9,<10", @@ -69,8 +74,8 @@ docs = [ [tool.pytest.ini_options] pythonpath = ["/opt/netbox/netbox", "."] -DJANGO_SETTINGS_MODULE = "netbox.settings" -addopts = "-n auto --dist loadscope" +DJANGO_SETTINGS_MODULE = "netbox_interface_name_rules.tests.isolated_settings" +addopts = "-n auto --dist loadscope --reuse-db --cov=netbox_interface_name_rules --cov-report=term-missing" [tool.ruff] line-length = 120