#!/usr/bin/env python3 """ Unit + fake-end-to-end tests for netbox/dc-dc-prefixes-import.py. No real NetBox exists this session (workstation, not the jumphost) -- the end-to-end tests run against tests/dc-dc-prefixes-import/fake_pynetbox.py, an in-memory stand-in implementing only the calls the target script actually makes. Pure-logic tests (validators, carve functions) need no NetBox at all. Run via tests/dc-dc-prefixes-import/run-tests.sh, which just invokes this file with python3 and relays its exit code -- matching this repo's bash harness convention even though the target under test is Python. """ from __future__ import annotations import contextlib import importlib.util import io import ipaddress import os import sys HERE = os.path.dirname(os.path.abspath(__file__)) REPO_ROOT = os.path.dirname(os.path.dirname(HERE)) STUB_SYSPATH = os.path.join(HERE, "stub_syspath") TARGET_PATH = os.path.join(REPO_ROOT, "netbox", "dc-dc-prefixes-import.py") # The stub must resolve BEFORE any real pynetbox (there isn't one installed # on this workstation anyway, but keep the ordering explicit/defensive). sys.path.insert(0, STUB_SYSPATH) sys.path.insert(0, HERE) # so fake_pynetbox itself is importable directly import pynetbox # noqa: E402 -- resolves to stub_syspath/pynetbox.py import fake_pynetbox # noqa: E402 P = 0 F = 0 def ok(label: str) -> None: global P P += 1 print(f"PASS: {label}") def no(label: str, detail: str = "") -> None: global F F += 1 print(f"FAIL: {label}" + (f" ({detail})" if detail else "")) def check(cond: bool, label: str, detail: str = "") -> None: (ok(label) if cond else no(label, detail)) def expect_dies(label: str, fn, *a, **kw) -> None: try: fn(*a, **kw) no(label, "did not raise SystemExit") except SystemExit: ok(label) def expect_ok(label: str, fn, *a, **kw): try: return fn(*a, **kw) except SystemExit as exc: no(label, f"unexpectedly raised SystemExit({exc})") return None else: ok(label) def load_target(): spec = importlib.util.spec_from_file_location("dc_dc_prefixes_import", TARGET_PATH) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod T = load_target() # ----------------------------------------------------------------------------- # validate_ula_48 # ----------------------------------------------------------------------------- net = expect_ok("ula48 valid /48 within fc00::/7", T.validate_ula_48, "fd12:3456:789a::/48") if net is not None: check((net.prefixlen == 48), "ula48 result prefixlen==48") expect_dies("ula48 wrong length /47 rejected", T.validate_ula_48, "fd12:3456:789a::/47") expect_dies("ula48 wrong length /49 rejected", T.validate_ula_48, "fd12:3456:789a::/49") expect_dies("ula48 GUA input rejected (not in fc00::/7)", T.validate_ula_48, "2602:f3e2:1234::/48") expect_dies("ula48 IPv4 input rejected", T.validate_ula_48, "10.0.0.0/8") expect_dies("ula48 malformed input rejected", T.validate_ula_48, "not-a-cidr") # ----------------------------------------------------------------------------- # validate_gua_prefix # ----------------------------------------------------------------------------- gua = expect_ok("gua valid /40 within 2602:f3e2::/32", T.validate_gua_prefix, "2602:f3e2:1000::/40") if gua is not None: check(gua.prefixlen == 40, "gua result prefixlen==40") expect_dies("gua /64 rejected (no room to carve)", T.validate_gua_prefix, "2602:f3e2:1000::/64") expect_dies("gua unrelated GUA block rejected", T.validate_gua_prefix, "2001:db8::/40") # ----------------------------------------------------------------------------- # validate_dc2_supernet # ----------------------------------------------------------------------------- dc2_super = expect_ok("dc2 supernet valid /19 no overlap", T.validate_dc2_supernet, "10.13.0.0/19") expect_dies("dc2 supernet too small (/20) rejected", T.validate_dc2_supernet, "10.13.0.0/20") expect_dies("dc2 supernet overlapping DC1 rejected", T.validate_dc2_supernet, "10.12.0.0/19") expect_dies("dc2 supernet IPv6 input rejected", T.validate_dc2_supernet, "fd00::/19") # ----------------------------------------------------------------------------- # carve_dc2_v4 # ----------------------------------------------------------------------------- if dc2_super is not None: v4 = T.carve_dc2_v4(dc2_super) check(set(v4.keys()) == set(T.PLANE_ORDER), "carve_dc2_v4 covers all six planes") all_22 = all(ipaddress.ip_network(c).prefixlen == 22 for c in v4.values()) check(all_22, "carve_dc2_v4 all /22") within = all(ipaddress.ip_network(c).subnet_of(dc2_super) for c in v4.values()) check(within, "carve_dc2_v4 all within supernet") distinct = len(set(v4.values())) == len(v4) check(distinct, "carve_dc2_v4 all distinct") # ----------------------------------------------------------------------------- # carve_ula # ----------------------------------------------------------------------------- org_ula = ipaddress.ip_network("fd12:3456:789a::/48") ula_dc1 = T.carve_ula(org_ula, "dc1") ula_dc2 = T.carve_ula(org_ula, "dc2") expected_ula_planes = {p for p in T.PLANE_ORDER if T.PLANE_V6_FAMILY[p] in ("ula", "ula-only")} check(set(ula_dc1.keys()) == expected_ula_planes, "carve_ula dc1 covers exactly the ULA-bearing planes") check(set(ula_dc2.keys()) == expected_ula_planes, "carve_ula dc2 covers exactly the ULA-bearing planes") all_64_dc1 = all(ipaddress.ip_network(c).prefixlen == 64 for c in ula_dc1.values()) check(all_64_dc1, "carve_ula dc1 all /64") no_overlap = not any( ipaddress.ip_network(a).overlaps(ipaddress.ip_network(b)) for a in ula_dc1.values() for b in ula_dc2.values() ) check(no_overlap, "carve_ula dc1/dc2 non-overlapping") # ----------------------------------------------------------------------------- # carve_gua # ----------------------------------------------------------------------------- if gua is not None: gua_carved = T.carve_gua(gua) check(set(gua_carved.keys()) == {"provider-public"}, "carve_gua only provider-public") g64 = ipaddress.ip_network(gua_carved["provider-public"]) check(g64.prefixlen == 64, "carve_gua result is /64") check(g64.subnet_of(gua), "carve_gua result within input prefix") # ----------------------------------------------------------------------------- # require_env # ----------------------------------------------------------------------------- os.environ["DC_DC_TEST_VAR"] = "present" check(T.require_env("DC_DC_TEST_VAR", "test") == "present", "require_env returns set value") del os.environ["DC_DC_TEST_VAR"] expect_dies("require_env dies when unset", T.require_env, "DC_DC_TEST_VAR_NOT_SET", "test") # ----------------------------------------------------------------------------- # main() end-to-end against the fake NetBox -- dc1 happy path + idempotency # ----------------------------------------------------------------------------- @contextlib.contextmanager def captured_stdout(): buf = io.StringIO() with contextlib.redirect_stdout(buf): yield buf def run_main(argv, api_instance): pynetbox.api = lambda url=None, token=None: api_instance old_argv = sys.argv sys.argv = ["dc-dc-prefixes-import.py"] + argv try: return T.main() finally: sys.argv = old_argv os.environ["NETBOX_URL"] = "https://fake.example/netbox" os.environ["NETBOX_TOKEN"] = "fake-token" os.environ["ORG_ULA_48"] = "fd12:3456:789a::/48" fake1 = fake_pynetbox.FakeApi() fake1.preseed_roles(T.PLANE_ORDER) os.environ["DC_GUA_PREFIX"] = "2602:f3e2:1000::/40" with captured_stdout() as out1: rc1 = run_main(["--dc", "dc1"], fake1) check(rc1 == 0, "main dc1 first run rc==0", str(rc1)) created_count = out1.getvalue().count("CREATED prefix") check(created_count == 12, "main dc1 first run creates 12 prefixes (6 v4 + 5 ula + 1 gua)", str(created_count)) check("vr0-dc0" in out1.getvalue(), "main dc1 prints the vr0-dc0 collision note") with captured_stdout() as out2: rc2 = run_main(["--dc", "dc1"], fake1) check(rc2 == 0, "main dc1 second run (idempotent) rc==0") check(out2.getvalue().count("CREATED prefix") == 0, "main dc1 second run creates nothing new") check(out2.getvalue().count("EXISTS prefix") == 12, "main dc1 second run reports all 12 as EXISTS") # ----------------------------------------------------------------------------- # main() end-to-end -- dc2 happy path (fresh site, distinct v4/GUA) # ----------------------------------------------------------------------------- fake2 = fake_pynetbox.FakeApi() fake2.preseed_roles(T.PLANE_ORDER) os.environ["DC_GUA_PREFIX"] = "2602:f3e2:2000::/40" os.environ["DC2_V4_SUPERNET"] = "10.13.0.0/19" with captured_stdout() as out3: rc3 = run_main(["--dc", "dc2"], fake2) check(rc3 == 0, "main dc2 run rc==0", str(rc3)) check(out3.getvalue().count("CREATED prefix") == 12, "main dc2 run creates 12 prefixes", str(out3.getvalue().count("CREATED prefix"))) dc2_site = fake2.dcim.sites.get(slug="vr1-dc2") check(dc2_site is not None, "main dc2 creates the vr1-dc2 site") # ----------------------------------------------------------------------------- # main() -- dc2 with DC2_V4_SUPERNET missing FAILS LOUD # ----------------------------------------------------------------------------- del os.environ["DC2_V4_SUPERNET"] fake3 = fake_pynetbox.FakeApi() fake3.preseed_roles(T.PLANE_ORDER) try: run_main(["--dc", "dc2"], fake3) no("main dc2 without DC2_V4_SUPERNET fails loud", "did not raise SystemExit") except SystemExit: ok("main dc2 without DC2_V4_SUPERNET fails loud") # ----------------------------------------------------------------------------- # argparse: missing/invalid --dc # ----------------------------------------------------------------------------- fake4 = fake_pynetbox.FakeApi() fake4.preseed_roles(T.PLANE_ORDER) os.environ["DC2_V4_SUPERNET"] = "10.13.0.0/19" try: run_main([], fake4) no("main missing --dc fails loud") except SystemExit: ok("main missing --dc fails loud") try: run_main(["--dc", "dc3"], fake4) no("main invalid --dc choice fails loud") except SystemExit: ok("main invalid --dc choice fails loud") # ----------------------------------------------------------------------------- # Summary # ----------------------------------------------------------------------------- print() if F == 0: print(f"ALL PASS ({P} checks)") sys.exit(0) else: print(f"FAILURES: {F}") sys.exit(1)