#!/bin/bash
# The Wi-Fi hand-over, driven end to end against stubbed tools.
#
# tests/test-classroom-network-adopt.sh covers the parsing. This covers
# the part that actually failed on the prayer-room laptop during the
# 0.2.5 setup: the ORDER of the hand-over, and what is carried across
# with the network.
#
# The laptop's own configuration was
#
#     allow-hotplug wlp2s0
#     iface wlp2s0 inet dhcp
#         wpa-ssid "ROGC Classroom"
#         wpa-psk  <secret>
#         wpa-scan-ssid 1
#
# and the migration produced a NetworkManager connection with
# 802-11-wireless.hidden: no, which can never associate with an access
# point that does not broadcast its name. It also ran `ifdown` after the
# stanza had been commented out, where ifdown has nothing left to read
# and does nothing, so ifupdown's own supplicant was still holding the
# radio. Then the rollback left the adapter unmanaged.
#
# nmcli, ifup, ifdown and curl are stubs that record what they were
# asked and can be told to fail, so the sequence is asserted rather than
# described. No real interface, no real NetworkManager, no laptop.
set -uo pipefail

ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"

python3 - "$ROOT" <<'PY'
import importlib.machinery
import importlib.util
import os
import shutil
import sys
import tempfile
from pathlib import Path

root = Path(sys.argv[1])

passed = failed = 0


def check(label, condition, detail=""):
    global passed, failed
    if condition:
        print(f"  ok   {label}")
        passed += 1
    else:
        print(f"  FAIL {label}")
        if detail:
            print(f"       {detail}")
        failed += 1


def load():
    """classroom-wifi as a module. It has no .py, so name the loader."""
    path = root / "scripts" / "classroom-wifi"
    loader = importlib.machinery.SourceFileLoader("classroom_wifi", str(path))
    spec = importlib.util.spec_from_loader("classroom_wifi", loader)
    module = importlib.util.module_from_spec(spec)
    loader.exec_module(module)
    return module


# The laptop's file, with the hidden network and an SSID with a space in
# it. Also a wired stanza, which nothing here may touch.
LAPTOP = """# This file describes the network interfaces available on your system
source /etc/network/interfaces.d/*

auto lo
iface lo inet loopback

auto enp3s0
iface enp3s0 inet static
\taddress 10.40.0.90/24
\tgateway 10.40.0.1

allow-hotplug wlp2s0
iface wlp2s0 inet dhcp
\twpa-ssid "ROGC Classroom"
\twpa-psk  s#cret-with-a-hash
\twpa-scan-ssid 1
"""

BROADCAST = """allow-hotplug wlp2s0
iface wlp2s0 inet dhcp
\twpa-ssid Sanctuary
\twpa-psk  ordinary-secret
"""

ENTERPRISE = """allow-hotplug wlp2s0
iface wlp2s0 inet dhcp
\twpa-ssid Campus
\twpa-key-mgmt WPA-EAP
"""

STUB = r"""#!/bin/bash
# Records the call, and the state of the interfaces file at the moment of
# it, so the test can assert ordering rather than trust it.
printf '%s' "$(basename "$0")" >> "$LOG"
for argument in "$@"; do printf ' %s' "$argument" >> "$LOG"; done
printf '\n' >> "$LOG"

if [[ -f "$INTERFACES" ]]; then
    cp "$INTERFACES" "$SNAPSHOTS/$(basename "$0").$$.interfaces"
fi

case "$(basename "$0")" in
nmcli)
    case "$*" in
    *"-f NAME,UUID,TYPE connection show"*) cat "$FAKE_CONNECTIONS" ;;
    *"-f NAME,FILENAME connection show"*)  cat "$FAKE_FILENAMES" ;;
    *"connection show --active"*)          cat "$FAKE_ACTIVE" ;;
    *"device status"*)                     cat "$FAKE_DEVICES" ;;
    *"networking connectivity"*)           cat "$FAKE_CONNECTIVITY" ;;
    *"connection up"*)                     exit "${UP_STATUS:-0}" ;;
    esac
    exit 0
    ;;
ifdown) exit "${IFDOWN_STATUS:-0}" ;;
ifup)   exit "${IFUP_STATUS:-0}" ;;
curl)   exit "${CURL_STATUS:-0}" ;;
esac
exit 0
"""


class Rig:
    """A machine mid-migration, with every external tool stubbed."""

    def __init__(self, interfaces_text, **environment):
        self.directory = Path(tempfile.mkdtemp(prefix="classroom-wifi-"))
        self.bin = self.directory / "bin"
        self.snapshots = self.directory / "snapshots"
        self.bin.mkdir()
        self.snapshots.mkdir()

        self.interfaces = self.directory / "interfaces"
        self.interfaces.write_text(interfaces_text)

        self.log = self.directory / "calls.log"
        self.log.write_text("")

        for tool in ("nmcli", "ifdown", "ifup", "curl"):
            target = self.bin / tool
            target.write_text(STUB)
            target.chmod(0o755)

        # What the stubbed nmcli answers with. Written as files so a test
        # can change an answer partway through if it needs to.
        self.answers = {
            "FAKE_CONNECTIONS": "",
            "FAKE_FILENAMES": "",
            "FAKE_ACTIVE": "ROGC Classroom:wlp2s0\n",
            "FAKE_DEVICES": "wlp2s0:connected\n",
            "FAKE_CONNECTIVITY": "full\n",
        }

        self.environment = {
            "LOG": str(self.log),
            "INTERFACES": str(self.interfaces),
            "SNAPSHOTS": str(self.snapshots),
        }

        for key, value in self.answers.items():
            path = self.directory / key
            path.write_text(value)
            self.environment[key] = str(path)

        self.environment.update({k: str(v) for k, v in environment.items()})

    def answer(self, key, value):
        Path(self.environment[key]).write_text(value)

    def run(self, action, **kwargs):
        saved_path = os.environ.get("PATH", "")
        saved = {k: os.environ.get(k) for k in self.environment}
        os.environ["PATH"] = f"{self.bin}:{saved_path}"
        os.environ.update(self.environment)

        module = load()
        module.INTERFACES = str(self.interfaces)
        # The poll loops are real time; nothing here needs to wait.
        module.time.sleep = lambda _seconds: None
        module.ADOPT_TIMEOUT = 0.2

        try:
            return getattr(module, action)(**kwargs), module
        finally:
            os.environ["PATH"] = saved_path
            for key, value in saved.items():
                if value is None:
                    os.environ.pop(key, None)
                else:
                    os.environ[key] = value

    @property
    def calls(self):
        return self.log.read_text().splitlines()

    def snapshot_at(self, tool):
        """The interfaces file as it was when `tool` ran."""
        for path in sorted(self.snapshots.glob(f"{tool}.*.interfaces")):
            return path.read_text()
        return None

    def clean(self):
        shutil.rmtree(self.directory, ignore_errors=True)


print("1. the hidden network survives the hand-over")
# The 0.2.5 regression itself. wpa-scan-ssid 1 was sitting in the parsed
# options and the only accessor threw it away.
rig = Rig(LAPTOP)
status, module = rig.run("adopt")
added = [c for c in rig.calls if c.startswith("nmcli connection add")]
check("the migration reports success", status == 0, str(status))
check("a connection is created", len(added) == 1, str(rig.calls))
check("and it is told the access point does not broadcast its name",
      added and "802-11-wireless.hidden yes" in added[0],
      added[0] if added else "no connection was added")
check("the SSID keeps its space",
      added and "ssid ROGC Classroom" in added[0], added[0] if added else "")
check("it is still a system connection, with no permissions",
      added and "connection.permissions" not in added[0])
check("the secret is stored in the connection file",
      added and "802-11-wireless-security.psk-flags 0" in added[0])
check("the password is not truncated at the # inside it",
      added and "wifi-sec.psk s#cret-with-a-hash" in added[0],
      added[0] if added else "")
rig.clean()

print("2. the interface is released BEFORE its stanza is taken away")
# ifdown reads /etc/network/interfaces to know how to bring an interface
# down. Run after the stanza is commented out it reports "Ignoring
# unknown interface" and does nothing, so ifupdown's wpa_supplicant kept
# the radio and NetworkManager could not have it.
rig = Rig(LAPTOP)
rig.run("adopt")
order = [c.split()[0] for c in rig.calls]
at_ifdown = rig.snapshot_at("ifdown")
check("ifdown is called at all", "ifdown" in order, str(order))
check("the stanza is still live in the file when ifdown runs",
      at_ifdown is not None
      and "\niface wlp2s0 inet dhcp\n" in at_ifdown,
      "ifdown ran against a file whose stanza was already commented out")
check("and the file is commented out afterwards",
      "# ekklesia-classroom: iface wlp2s0" in rig.interfaces.read_text())
check("NetworkManager is only told to manage it after the file changed",
      order.index("ifdown") < len(order) - 1)
managed = [c for c in rig.calls if "device set" in c]
check("the device is explicitly handed to NetworkManager",
      any("managed yes" in c for c in managed), str(managed))
up = [c for c in rig.calls if "connection up" in c]
check("the connection is brought up on that interface by name",
      up and "ifname wlp2s0" in up[0], str(up))
rig.clean()

print("3. an ordinary broadcast network is not marked hidden")
# The flag has to be an answer in both directions, not something only
# ever switched on.
rig = Rig(BROADCAST)
rig.run("adopt")
added = [c for c in rig.calls if c.startswith("nmcli connection add")]
check("hidden is written as no", added and "802-11-wireless.hidden no" in added[0],
      added[0] if added else "")
rig.clean()

print("4. a failed hand-over puts the file back, and only the file")
# The backup is written once and never refreshed, so a whole-file restore
# would silently revert a wired change made in between. The rollback
# un-comments its own marker instead.
rig = Rig(LAPTOP, UP_STATUS="1", FAKE_CONNECTIVITY="none\n", CURL_STATUS="1")
status, _ = rig.run("adopt")
after = rig.interfaces.read_text()
check("the migration reports failure", status == 1, str(status))
check("the wireless stanza is live again",
      "\niface wlp2s0 inet dhcp\n" in after, after)
check("the allow-hotplug line came back too",
      "\nallow-hotplug wlp2s0\n" in after)
check("nothing of the marker is left behind",
      "ekklesia-classroom:" not in after, after)
check("the file is byte for byte what it was", after == LAPTOP,
      "restored file differs from the original")
check("the wired stanza was never touched",
      "\taddress 10.40.0.90/24\n" in after)
rig.clean()

print("5. a rollback preserves an unrelated change made since")
# The case a whole-file copy gets wrong: the wired address is corrected
# after the migration, and then the migration fails.
rig = Rig(LAPTOP, UP_STATUS="1", FAKE_CONNECTIVITY="none\n", CURL_STATUS="1")


def edit_between(module):
    original = module.put_interfaces_back

    def wrapped():
        text = rig.interfaces.read_text()
        rig.interfaces.write_text(
            text.replace("10.40.0.90/24", "10.40.0.91/24"))
        return original()

    module.put_interfaces_back = wrapped


saved_run = Rig.run


def run_with_edit(self, action, **kwargs):
    saved_path = os.environ.get("PATH", "")
    os.environ["PATH"] = f"{self.bin}:{saved_path}"
    os.environ.update(self.environment)
    module = load()
    module.INTERFACES = str(self.interfaces)
    module.time.sleep = lambda _seconds: None
    module.ADOPT_TIMEOUT = 0.2
    edit_between(module)
    try:
        return getattr(module, action)(**kwargs), module
    finally:
        os.environ["PATH"] = saved_path


run_with_edit(rig, "adopt")
after = rig.interfaces.read_text()
check("the corrected wired address survived the rollback",
      "10.40.0.91/24" in after, after)
check("and the wireless stanza is live again",
      "\niface wlp2s0 inet dhcp\n" in after)
rig.clean()

print("6. a failed hand-over never leaves the adapter unmanaged")
# The state the prayer-room laptop was found in: ifupdown had been told
# to take the interface back, ifup failed too, and NetworkManager had
# been told to leave it alone. Nobody owned the radio, and recovering it
# needed `nmcli device set wlp2s0 managed yes` typed by hand.
rig = Rig(LAPTOP, UP_STATUS="1", IFUP_STATUS="1",
          FAKE_CONNECTIVITY="none\n", CURL_STATUS="1")
status, _ = rig.run("adopt")
# Only what the rollback did. The hand-over itself sets managed yes on
# the way in, which is not the question here.
rollback_calls = rig.calls[
    max(index for index, call in enumerate(rig.calls)
        if call.startswith("ifup ")):]
sets = [c for c in rollback_calls if "device set" in c]
check("ifupdown is offered the interface first",
      any("managed no" in c for c in rig.calls), str(rig.calls))
check("and when ifup fails too it is given to NetworkManager",
      sets and "managed yes" in sets[-1],
      "the last word on the adapter was 'managed no', which owns nothing")
check("so the adapter is never left belonging to nobody",
      "managed yes" in rig.calls[-1] or any("managed yes" in c
                                            for c in sets),
      str(rollback_calls))
check("and the migration still reports failure", status == 1, str(status))
rig.clean()

print("7. a network this cannot recreate is refused before anything moves")
# WPA-Enterprise has no pre-shared key to carry across. Guessing would
# take the interface away from ifupdown to find that out.
rig = Rig(ENTERPRISE)
status, _ = rig.run("adopt")
check("it refuses", status == 1, str(status))
check("nothing was added", not any("connection add" in c for c in rig.calls),
      str(rig.calls))
check("the file is untouched", rig.interfaces.read_text() == ENTERPRISE)
check("ifdown was never run", not any(c.startswith("ifdown")
                                      for c in rig.calls), str(rig.calls))
rig.clean()

print("8. a machine already migrated is not migrated again")
# The prayer-room laptop, after tomorrow. Its stanza is gone, so there is
# nothing to adopt and --adopt must be a no-op rather than an error.
MIGRATED = LAPTOP.replace("allow-hotplug wlp2s0",
                          "# ekklesia-classroom: allow-hotplug wlp2s0")
MIGRATED = "\n".join(
    ("# ekklesia-classroom: " + line)
    if line.startswith("iface wlp2s0") or line.startswith("\twpa-")
    else line
    for line in MIGRATED.splitlines()
) + "\n"
rig = Rig(MIGRATED)
status, _ = rig.run("adopt")
check("it reports nothing to do, and succeeds", status == 0, str(status))
check("no connection was created",
      not any("connection add" in c for c in rig.calls), str(rig.calls))
check("nothing was brought down", not any(c.startswith("ifdown")
                                          for c in rig.calls))
check("the file is untouched", rig.interfaces.read_text() == MIGRATED)
rig.clean()

print("9. --check changes nothing at all")
rig = Rig(LAPTOP)
status, _ = rig.run("check")
check("it succeeds", status == 0)
check("the file is untouched", rig.interfaces.read_text() == LAPTOP)
check("it runs no command that could change anything",
      not any(c.startswith(("ifup", "ifdown")) for c in rig.calls),
      str(rig.calls))
check("and no nmcli that writes",
      not any("connection add" in c or "connection modify" in c
              or "device set" in c for c in rig.calls), str(rig.calls))
rig.clean()

print()
print(f"{passed} passed, {failed} failed")
raise SystemExit(1 if failed else 0)
PY