nixos/scripts/router-ui-check.py

"""Self-check for router-ui.py: TOML round-trips and the refusal rules.

Run it against the same python the service uses:

    nix shell .#nixosConfigurations.FredOS-Mediaserver.pkgs.python3Packages.tomlkit \
      -c python3 scripts/router-ui-check.py

The TOML round-trip cases are the ones that matter: a save must leave an
unedited file byte-identical, and must never eat the parked/commented forwards
in ports.toml.
"""
import difflib
import importlib.util
import os
import sys
import tempfile

import tomlkit

HERE = os.path.dirname(os.path.abspath(__file__))
REPO = os.path.join(HERE, "..")

spec = importlib.util.spec_from_file_location("routerui", os.path.join(HERE, "router-ui.py"))
r = importlib.util.module_from_spec(spec)
spec.loader.exec_module(r)


def refuses(fn, *a):
    try:
        fn(*a)
    except r.Refused as e:
        return str(e)
    raise AssertionError(f"{fn.__name__} accepted what it should have refused")


def same(before, after, what):
    assert before == after, what + ":\n" + "".join(
        difflib.unified_diff(before.splitlines(1), after.splitlines(1), "before", "after"))


# --- ports.toml round-trip ---------------------------------------------------
src = open(os.path.join(REPO, "ports.toml")).read()
doc = tomlkit.parse(src)
items = r.forwards_from(doc)
assert len(items) == 4, items
assert items[3]["port"] == "25565-25600", items[3]

# Saving without editing anything must not touch a single byte.
r.check_forwards(items)
r.edit_aot(doc, "forward", items)
same(src, tomlkit.dumps(doc), "no-op forward save rewrote the file")

# Add one, drop the Pelican range; every comment block must survive.
doc2 = tomlkit.parse(src)
items2 = r.forwards_from(doc2)
items2.pop(3)
items2.append({"_i": None, "name": "Test", "port": "9999", "protocol": "udp",
               "dest": "10.0.0.5"})
r.check_forwards(items2)
r.edit_aot(doc2, "forward", items2)
out = tomlkit.dumps(doc2)
assert "7DTD voice/dynamic" in out, "parked blocks were lost"
assert "WoW Classic realmd" in out, "parked blocks were lost"
assert "Pelican game-server allocation range" in out, "inline comment was lost"
assert "Pelican game servers" not in out.replace("# Pelican", ""), "deletion did not happen"
assert 'name = "Test"' in out and "port = 9999" in out, out[-300:]
back = tomlkit.parse(out)["forward"]
assert len(back) == 4 and back[3]["protocol"] == "udp" and back[3]["dest"] == "10.0.0.5"
assert [f["name"] for f in back[:3]] == ["HTTP", "HTTPS", "SSH"]

# --- forward validation ------------------------------------------------------
mk = lambda: r.forwards_from(tomlkit.parse(src))  # noqa: E731 - fresh fixture per case
print(refuses(r.check_forwards, [f for f in mk() if f["name"] != "SSH"]))
print(refuses(r.check_forwards, [f for f in mk() if f["name"] != "HTTPS"]))
print(refuses(r.check_forwards, mk() + [{"_i": None, "name": "x", "port": "70000",
                                         "protocol": "tcp", "dest": ""}]))
print(refuses(r.check_forwards, mk() + [{"_i": None, "name": "x", "port": "9-8",
                                         "protocol": "tcp", "dest": ""}]))
print(refuses(r.check_forwards, mk() + [{"_i": None, "name": "x", "port": "80",
                                         "protocol": "sctp", "dest": ""}]))
print(refuses(r.check_forwards, mk() + [{"_i": None, "name": "", "port": "80",
                                         "protocol": "tcp", "dest": ""}]))
# A range that covers 22 and 443 satisfies the keep-me-reachable rule.
r.check_forwards([{"_i": None, "name": "wide", "port": "20-500", "protocol": "both",
                   "dest": ""}])

# --- device validation -------------------------------------------------------
def dev(**kw):
    base = {"_i": None, "mac": "aa:bb:cc:dd:ee:01", "name": "thing", "ip": "",
            "blocked": False, "note": ""}
    return {**base, **kw}


me, my_ip = "aa:bb:cc:dd:ee:99", "10.0.0.161"

print(refuses(r.check_devices, [dev(ip="10.0.0.161")], me, my_ip))   # inside the pool
print(refuses(r.check_devices, [dev(ip="10.0.0.1")], me, my_ip))     # the router itself
print(refuses(r.check_devices, [dev(ip="192.168.1.5")], me, my_ip))  # off-LAN
print(refuses(r.check_devices, [dev(mac="nope")], me, my_ip))
print(refuses(r.check_devices, [dev(name="bad name")], me, my_ip))
print(refuses(r.check_devices, [dev(), dev()], me, my_ip))           # duplicate MAC
print(refuses(r.check_devices, [dev(ip="10.0.0.20"),
                                dev(mac="aa:bb:cc:dd:ee:02", ip="10.0.0.20")], me, my_ip))
print(refuses(r.check_devices, [dev(mac=me, blocked=True)], me, my_ip))
print(refuses(r.check_devices, [dev(ip="10.0.0.30", blocked=True)], me, "10.0.0.30"))

ok = [dev(ip="10.0.0.39", name="camera-bedroom")]
r.check_devices(ok, me, my_ip)
assert ok[0]["ip"] == "10.0.0.39" and "blocked" not in ok[0] and "note" not in ok[0]
cleared = [dev(ip="")]
r.check_devices(cleared, me, my_ip)
assert "ip" not in cleared[0], "an empty reservation should drop the key entirely"

# --- devices.toml round-trip -------------------------------------------------
dsrc = open(os.path.join(REPO, "devices.toml")).read()
ddoc = tomlkit.parse(dsrc)
ditems = r.devices_from(ddoc)
assert len(ditems) == 1 and ditems[0]["ip"] == "10.0.0.39"
r.check_devices(ditems, me, my_ip)
r.edit_aot(ddoc, "device", ditems)
same(dsrc, tomlkit.dumps(ddoc), "no-op device save rewrote the file")

ditems = r.devices_from(ddoc)
ditems[0]["ip"] = ""
ditems[0]["blocked"] = True
r.check_devices(ditems, me, my_ip)
r.edit_aot(ddoc, "device", ditems)
entry = tomlkit.dumps(ddoc).split("[[device]]")[1]
assert "ip = " not in entry, entry
assert "blocked = true" in entry, entry

# --- speedtest parsing (bytes/s -> Mbps, ns -> ms) ---------------------------
sample = ('{"timestamp":"2026-08-15 12:00:55","servers":[{"name":"Preston",'
          '"latency":18796260,"dl_speed":63607890.17,"ul_speed":8839605.08}]}\n')
tmp = os.path.join(tempfile.mkdtemp(), "st.jsonl")
open(tmp, "w").write(sample + "not json at all\n" + sample)
r.SPEEDTEST_LOG = tmp
got = r.speedtests()
assert len(got) == 2, got
assert got[0] == {"ts": "2026-08-15 12:00:55", "down": 508.9, "up": 70.7,
                  "ping": 18.8, "server": "Preston"}, got[0]

print("\nall checks passed", file=sys.stderr)