#!/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

import faulthandler

# Watchdog: self-abort WITH a traceback after 30s instead of hanging the whole
# run-tests-all.sh gauntlet if any test ever blocks. This is exactly how the
# carve_gua /40 subnet-enumeration hang (DOCFIX-181) was diagnosed; now a future
# blocking regression fails loudly (non-zero exit + stack dump) rather than
# stalling the gauntlet indefinitely.
faulthandler.dump_traceback_later(30, exit=True)

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:
        result = fn(*a, **kw)
    except SystemExit as exc:
        no(label, f"unexpectedly raised SystemExit({exc})")
        return None
    ok(label)
    return result


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_dc1_supernet
# -----------------------------------------------------------------------------
dc2_super = expect_ok("second-DC supernet valid /19 no overlap", T.validate_vr1_dc1_supernet, "10.12.64.0/19")
expect_dies("second-DC supernet too small (/20) rejected", T.validate_vr1_dc1_supernet, "10.12.64.0/20")
expect_dies("second-DC supernet overlapping VR1 DC0 rejected", T.validate_vr1_dc1_supernet, "10.12.0.0/19")
expect_dies("second-DC supernet IPv6 input rejected", T.validate_vr1_dc1_supernet, "fd00::/19")

# -----------------------------------------------------------------------------
# carve_dc1_v4
# -----------------------------------------------------------------------------
if dc2_super is not None:
    v4 = T.carve_vr1_dc1_v4(dc2_super)
    check(set(v4.keys()) == set(T.PLANE_ORDER), "carve_vr1_dc1_v4 covers all six planes")
    all_22 = all(ipaddress.ip_network(c).prefixlen == 22 for c in v4.values())
    check(all_22, "carve_vr1_dc1_v4 all /22")
    within = all(ipaddress.ip_network(c).subnet_of(dc2_super) for c in v4.values())
    check(within, "carve_vr1_dc1_v4 all within supernet")
    distinct = len(set(v4.values())) == len(v4)
    check(distinct, "carve_vr1_dc1_v4 all distinct")

# -----------------------------------------------------------------------------
# carve_v6 (D-111 NN layout: /60+/64 per plane, deployed net-byte mnemonic)
# -----------------------------------------------------------------------------
org_ula = ipaddress.ip_network("fd12:3456:789a::/48")
gua_dc0 = ipaddress.ip_network("2602:f3e2:f02::/48")
recs1 = T.carve_v6("vr1-dc0", org_ula, gua_dc0)
recs2 = T.carve_v6("vr1-dc1", org_ula, ipaddress.ip_network("2602:f3e2:f03::/48"))

# structure: 12 records (provider 3, metal-admin 2, metal-internal 1, data/storage/repl 2 each)
check(len(recs1) == 12, "carve_v6 VR1-DC0 yields 12 records", str(len(recs1)))
roles1 = [r for _, r, _ in recs1]
check(roles1.count("provider-public") == 3, "carve_v6 provider-public has 3 (/60,/64,VIP)")
check(roles1.count("metal-admin") == 2, "carve_v6 metal-admin has 2 (/60,/64)")
check(roles1.count("metal-internal") == 1, "carve_v6 metal-internal has 1 (/64 in metal /60)")
for role in ("data-tenant", "storage", "replication"):
    check(roles1.count(role) == 2, f"carve_v6 {role} has 2 (/60,/64)")

# family matrix: provider-public is GUA (within its /48); the rest are ULA (fd..)
addrs1 = {c for c, _, _ in recs1}
prov = [ipaddress.ip_network(c) for c, r, _ in recs1 if r == "provider-public"]
check(all(n.subnet_of(gua_dc0) for n in prov), "carve_v6 provider-public within the GUA /48")
nonprov = [ipaddress.ip_network(c) for c, r, _ in recs1 if r != "provider-public"]
check(all(str(n.network_address).startswith("fd") for n in nonprov), "carve_v6 non-provider planes are ULA")

# NN offsets land on the deployed mnemonic; DC.NN reads in the 4th hextet
check("2602:f3e2:f02:10::/60" in addrs1, "carve_v6 VR1-DC0 provider GUA /60 at :10")
check("2602:f3e2:f02:11::/64" in addrs1, "carve_v6 VR1-DC0 provider API-VIP /64 at :11")
check("fd12:3456:789a:220::/64" in addrs1, "carve_v6 VR1-DC0 metal-admin ULA /64 at DC.NN :220")
check("fd12:3456:789a:221::/64" in addrs1, "carve_v6 VR1-DC0 metal-internal ULA /64 at :221")
check("fd12:3456:789a:230::/64" in addrs1, "carve_v6 VR1-DC0 data-tenant ULA /64 at :230")
check("fd12:3456:789a:250::/64" in addrs1, "carve_v6 VR1-DC0 replication ULA /64 at :250")
m60 = ipaddress.ip_network("fd12:3456:789a:220::/60")
check(ipaddress.ip_network("fd12:3456:789a:221::/64").subnet_of(m60), "carve_v6 metal-internal /64 sits inside metal /60")

# VR1 DC1 uses the :3NN nibble; the two DCs never overlap
addrs2 = {c for c, _, _ in recs2}
check("fd12:3456:789a:320::/64" in addrs2, "carve_v6 VR1-DC1 metal-admin at :320 (DC nibble 3)")
check("2602:f3e2:f03:10::/64" in addrs2, "carve_v6 VR1-DC1 provider GUA at f03:10")
no_overlap = not any(
    ipaddress.ip_network(a).overlaps(ipaddress.ip_network(b))
    for a, _, _ in recs1 for b, _, _ in recs2
)
check(no_overlap, "carve_v6 vr1-dc0/vr1-dc1 non-overlapping")

# DOCFIX-181 no-hang property preserved: a /40 GUA computes by math, not enumeration
recs_big = T.carve_v6("vr1-dc1", org_ula, ipaddress.ip_network("2602:f3e2:1000::/40"))
check(len(recs_big) == 12, "carve_v6 with a /40 GUA completes fast (no subnet enumeration)")

# -----------------------------------------------------------------------------
# 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:f02::/48"   # apex: vr1-dc0
for _v in ("DC1_V4_SUPERNET", "DC2_V4_SUPERNET", "VR1_DC1_V4_SUPERNET"):
    os.environ.pop(_v, None)   # both old names RETIRED by D-119

# --- D-117: DRY RUN IS THE DEFAULT. This is the guard that matters most: the
# --- tool used to write with no flag at all, straight into the IPAM apex.
with captured_stdout() as out0:
    rc0 = run_main(["--dc", "vr1-dc0"], fake1)
check(rc0 == 0, "main vr1-dc0 DRY RUN rc==0", str(rc0))
check(out0.getvalue().count("CREATED prefix") == 0,
      "D-117: dry run (no --commit) CREATES NOTHING")
check(out0.getvalue().count("[dry-run] would CREATE prefix") == 18,
      "D-117: dry run PLANS all 18 prefixes",
      str(out0.getvalue().count("[dry-run] would CREATE prefix")))
check("[dry-run] would CREATE site" in out0.getvalue(),
      "D-117: dry run PLANS the site too (fresh-NetBox preview must not bail)")
check("DRY RUN" in out0.getvalue(), "D-117: dry run says so, loudly")
check(len(fake1.ipam.prefixes._items) == 0,
      "D-117: dry run left the NetBox EMPTY (no side effects)",
      str(len(fake1.ipam.prefixes._items)))

# --- now commit for real
with captured_stdout() as out1:
    rc1 = run_main(["--dc", "vr1-dc0", "--commit"], fake1)
check(rc1 == 0, "main vr1-dc0 --commit rc==0", str(rc1))
created_count = out1.getvalue().count("CREATED prefix")
check(created_count == 18, "main vr1-dc0 --commit creates 18 prefixes (6 v4 + 12 v6)", str(created_count))
check("vr0-dc0" in out1.getvalue(), "main vr1-dc0 prints the vr0-dc0 (VR0 rehearsal) collision note")
dc0_site = fake1.dcim.sites.get(slug="vr1-dc0")
check(dc0_site is not None, "D-119: --dc vr1-dc0 created the APEX site vr1-dc0 (identity)")
check(fake1.dcim.sites.get(slug="vr1-dc1") is None,
      "D-119: --dc vr1-dc0 did NOT touch vr1-dc1 -- the off-by-one is dead")

with captured_stdout() as out2:
    rc2 = run_main(["--dc", "vr1-dc0", "--commit"], fake1)
check(rc2 == 0, "main vr1-dc0 second run (idempotent) rc==0")
check(out2.getvalue().count("CREATED prefix") == 0, "main vr1-dc0 second run creates nothing new")
check(out2.getvalue().count("EXISTS  prefix") == 18, "main vr1-dc0 second run reports all 18 as EXISTS")

# -----------------------------------------------------------------------------
# main() end-to-end -- VR1 DC1 (the SECOND DC): fresh site, distinct v4/GUA
# -----------------------------------------------------------------------------
fake2 = fake_pynetbox.FakeApi()
fake2.preseed_roles(T.PLANE_ORDER)
os.environ["DC_GUA_PREFIX"] = "2602:f3e2:f03::/48"   # apex: vr1-dc1
os.environ["VR1_DC1_V4_SUPERNET"] = "10.12.64.0/19"  # D-115

with captured_stdout() as out3:
    rc3 = run_main(["--dc", "vr1-dc1", "--commit"], fake2)
check(rc3 == 0, "main vr1-dc1 (second DC) run rc==0", str(rc3))
check(out3.getvalue().count("CREATED prefix") == 18, "main vr1-dc1 creates 18 prefixes",
      str(out3.getvalue().count("CREATED prefix")))
check(fake2.dcim.sites.get(slug="vr1-dc1") is not None, "main vr1-dc1 binds to the APEX site vr1-dc1")
check(fake2.dcim.sites.get(slug="vr1-dc2") is None, "D-119: no phantom vr1-dc2 site is ever created")

# -----------------------------------------------------------------------------
# main() -- second DC with DC1_V4_SUPERNET missing FAILS LOUD
# -----------------------------------------------------------------------------
del os.environ["VR1_DC1_V4_SUPERNET"]
fake3 = fake_pynetbox.FakeApi()
fake3.preseed_roles(T.PLANE_ORDER)
try:
    with captured_stdout():
        run_main(["--dc", "vr1-dc1", "--commit"], fake3)
    no("main vr1-dc1 without VR1_DC1_V4_SUPERNET fails loud", "did not raise SystemExit")
except SystemExit:
    ok("main vr1-dc1 without VR1_DC1_V4_SUPERNET fails loud")

# -----------------------------------------------------------------------------
# D-119: BOTH retired env var names are rejected, not silently ignored
# -----------------------------------------------------------------------------
os.environ["DC2_V4_SUPERNET"] = "10.12.64.0/19"     # muscle memory
os.environ.pop("DC1_V4_SUPERNET", None)
try:
    with captured_stdout():
        run_main(["--dc", "vr1-dc1", "--commit"], fake_pynetbox.FakeApi())
    no("D-119: retired DC2_V4_SUPERNET is rejected by name")
except SystemExit:
    ok("D-119: retired DC2_V4_SUPERNET is rejected by name")
os.environ.pop("DC2_V4_SUPERNET", None)

# -----------------------------------------------------------------------------
# argparse: missing/invalid --dc  (dc2 is now an INVALID choice -- D-117)
# -----------------------------------------------------------------------------
fake4 = fake_pynetbox.FakeApi()
fake4.preseed_roles(T.PLANE_ORDER)
os.environ["VR1_DC1_V4_SUPERNET"] = "10.12.64.0/19"

try:
    with captured_stdout():
        run_main([], fake4)
    no("main missing --dc fails loud")
except SystemExit:
    ok("main missing --dc fails loud")

# D-119: EVERY bare dcN is now an invalid choice. They were ambiguous across
# regions ('dc0' = VR0's live cloud in lib-net.sh, VR1's first DC here).
for _bare in ("dc0", "dc1", "dc2"):
    try:
        with captured_stdout():
            run_main(["--dc", _bare], fake4)
        no(f"D-119: bare --dc {_bare} is REJECTED (retired, cross-region ambiguous)")
    except SystemExit:
        ok(f"D-119: bare --dc {_bare} is REJECTED (retired, cross-region ambiguous)")

try:
    with captured_stdout():
        run_main(["--dc", "dc3"], fake4)
    no("main invalid --dc choice fails loud")
except SystemExit:
    ok("main invalid --dc choice fails loud")

# -----------------------------------------------------------------------------
# D-119 BREAK-2: DC_GUA_PREFIX must AGREE with --dc.
#
# Without this guard, `--dc vr1-dc0 DC_GUA_PREFIX=<f03::/48>` was ACCEPTED: it
# writes the SECOND DC's GUA prefixes, carved with the FIRST DC's ULA nibble
# (DC_V6_INDEX is keyed off --dc, independently), all scoped to the FIRST DC's
# site. A silently mis-bound datacenter assembled from two disagreeing sources.
# Identity-mapping the site slug does NOT close this -- it is one layer down.
# -----------------------------------------------------------------------------
fake5 = fake_pynetbox.FakeApi()
fake5.preseed_roles(T.PLANE_ORDER)
os.environ["ORG_ULA_48"] = "fd50:840e:74e2::/48"
os.environ["VR1_DC1_V4_SUPERNET"] = "10.12.64.0/19"

os.environ["DC_GUA_PREFIX"] = "2602:f3e2:f03::/48"   # the SECOND DC's block...
try:
    with captured_stdout():
        run_main(["--dc", "vr1-dc0", "--commit"], fake5)   # ...but the FIRST DC's site
    no("D-119 BREAK-2: mismatched DC_GUA_PREFIX/--dc is REJECTED")
except SystemExit:
    ok("D-119 BREAK-2: mismatched DC_GUA_PREFIX/--dc is REJECTED")
check(fake5.dcim.sites.get(slug="vr1-dc0") is None,
      "D-119 BREAK-2: the mismatched run wrote NOTHING (no site created)")

os.environ["DC_GUA_PREFIX"] = "2602:f3e2:f02::/48"   # the FIRST DC's block...
try:
    with captured_stdout():
        run_main(["--dc", "vr1-dc1", "--commit"], fake_pynetbox.FakeApi())  # ...second DC's site
    no("D-119 BREAK-2: the reverse mismatch is REJECTED too")
except SystemExit:
    ok("D-119 BREAK-2: the reverse mismatch is REJECTED too")

# The MATCHING pair still works -- the guard must not be a blanket refusal.
fake6 = fake_pynetbox.FakeApi()
fake6.preseed_roles(T.PLANE_ORDER)
os.environ["DC_GUA_PREFIX"] = "2602:f3e2:f02::/48"
with captured_stdout():
    rc_ok = run_main(["--dc", "vr1-dc0", "--commit"], fake6)
check(rc_ok == 0, "D-119 BREAK-2: the MATCHING --dc/DC_GUA_PREFIX pair still succeeds")

# -----------------------------------------------------------------------------
# D-119 BREAK-1: descriptions must not munge the selector into the label.
#
# The old code did f"VR1 {dc.upper()} ..." -- fine when dc was "dc0", but under
# D-119 dc is "vr1-dc0", which renders "VR1 VR1-DC0 provider-public". Same defect
# class as the original bug: DERIVING a label from a token instead of looking it
# up. It hid in the description field, where slug-focused review missed it.
# -----------------------------------------------------------------------------
descs = [p.description for p in fake6.ipam.prefixes.all()]
check(any(d.startswith("VR1 DC0 ") for d in descs),
      "D-119 BREAK-1: descriptions read 'VR1 DC0 ...' (from SITES[name], looked up)")
check(not any("VR1-DC0" in d.upper().replace("VR1 DC0", "") for d in descs),
      "D-119 BREAK-1: no description contains the munged 'VR1 VR1-DC0'")

# -----------------------------------------------------------------------------
# Summary
# -----------------------------------------------------------------------------
print()

# ---------------------------------------------------------------------------
# D-117 REGRESSION GUARDS. These are the whole reason this rename happened:
# the tool bound VR1 DC0's prefixes to the site vr1-dc1 (the OTHER DC), and it
# WROTE BY DEFAULT, so it would have landed silently in the IPAM apex.
# The slugs below are the APEX's, measured live 2026-07-13. Do not "fix" them.
# ---------------------------------------------------------------------------
# THE D-119 STRUCTURAL INVARIANT: the selector IS the slug. If a future edit
# reintroduces an offset table, this check fails -- the bug becomes untypeable.
check(all(k == v["slug"] for k, v in T.SITES.items()),
      "D-119: SITES is an IDENTITY map -- the --dc selector IS the apex site slug")
check(set(T.SITES) == {"vr1-dc0", "vr1-dc1"}, "D-119: only the two apex slugs are selectable")
check(not any(k in T.SITES for k in ("dc0", "dc1", "dc2")),
      "D-119: every bare dcN selector is GONE from SITES")
check(T.DC_V6_INDEX["vr1-dc0"] == 0x02, "D-119: vr1-dc0 -> GUA site nibble 02 (== apex f02::/48)")
check(T.DC_V6_INDEX["vr1-dc1"] == 0x03, "D-119: vr1-dc1 -> GUA site nibble 03 (== apex f03::/48)")
check(T.EXPECTED_GUA == {"vr1-dc0": "2602:f3e2:f02::/48", "vr1-dc1": "2602:f3e2:f03::/48"},
      "D-119: EXPECTED_GUA pins the apex's MEASURED DC->GUA binding")

# dry-by-default: --commit must exist, and the writers must take a commit flag.
import inspect
_src = inspect.getsource(T)
check("--commit" in _src, "dry-by-default: --commit flag exists")
check("commit" in inspect.signature(T.create_or_report_prefix).parameters,
      "dry-by-default: create_or_report_prefix is commit-gated")
check("commit" in inspect.signature(T.find_or_create_site).parameters,
      "dry-by-default: find_or_create_site is commit-gated (a site is a WRITE)")
check("DC2_V4_SUPERNET" in _src and "DC1_V4_SUPERNET" in _src and "RETIRED (D-119)" in _src,
      "D-119: BOTH retired supernet env vars are rejected by name, not ignored")

if F == 0:
    print(f"ALL PASS ({P} checks)")
    sys.exit(0)
else:
    print(f"FAILURES: {F}")
    sys.exit(1)