# A router configured by settings.json, changed and rebuilt on the device:
# install it (its configuration directory travels with the secrets), boot it
# without network, then
#   - rebuilding the unchanged settings gives exactly the installed system,
#   - a changed setting is built offline, activated and confirmed,
#   - an unconfirmed change is rolled back,
#   - and the same through the web UI from a LAN client: discard the
#     unapplied change, edit the DHCP pool and add a reservation, apply,
#     confirm.
# The rest of the web UI (dashboard, backup, forms) is tested in webui.nix,
# which needs no installed system.
{ self, pkgs }:
let
  inherit (pkgs) lib;
  common = import ./common.nix;

  settings = pkgs.writeText "settings.json" (
    builtins.toJSON {
      # The schema of this source: the number of migrations.
      version = lib.count (lib.hasSuffix ".jq") (builtins.attrNames (builtins.readDir ../migrations));
      system = {
        hostName = "router";
        stateVersion = "26.11";
        sshKeys = [
          "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIFakeKeyForTestsOnlyXXXXXXXXXXXXXXXXXXXXXXXXXX test"
        ];
      };
      router = {
        update.channel = "26.05";
        wan.interface = "eth1";
        lans.main = {
          interface = "eth2";
          internet = true;
          ipv4.address = "192.168.10.1";
          dhcp.pool = {
            start = "192.168.10.100";
            end = "192.168.10.199";
          };
        };
      };
    }
  );

  # Only so the test driver can talk to the installed system; the channel is a
  # server on the client, and a newer nixpkgs a copy of the installed one (made below).
  local = pkgs.writeText "local.nix" ''
    { modulesPath, ... }:
    {
      imports = [ (modulesPath + "/testing/test-instrumentation.nix") ];
      routerSettings.upgrade = {
        channelsUrl = "http://192.168.10.50:8080";
        nixpkgsRef = "path:/var/lib/newer-nixpkgs?lastModified=${toString self.inputs.nixpkgs.lastModified}&rev=%rev%";
      };
    }
  '';

  configDir = self.lib.mkConfigDir { inherit pkgs settings local; };

  # What the device evaluates from configDir.
  toplevel =
    (self.lib.mkRouterSystem {
      settings = "${configDir}/settings.json";
      modules = [ "${configDir}/local.nix" ];
    }).config.system.build.toplevel;

  channelServer = pkgs.writeScript "channels.py" ''
    #!${pkgs.python3}/bin/python3
    import http.server

    class Handler(http.server.SimpleHTTPRequestHandler):
        def __init__(self, *args, **kwargs):
            super().__init__(*args, directory="/srv/channels", **kwargs)

        def do_GET(self):
            parts = self.path.strip("/").split("/")
            if len(parts) == 2 and parts[1] == "git-revision":
                self.send_response(302)
                self.send_header("Location", "/files/" + self.path.strip("/"))
                self.end_headers()
            else:
                super().do_GET()

    http.server.ThreadingHTTPServer(("0.0.0.0", 8080), Handler).serve_forever()
  '';

  sharedDisk = {
    virtualisation.diskImage = "./target.qcow2";
    virtualisation.diskSize = 8192;
  };
in
pkgs.testers.runNixOSTest {
  name = "router-apply";

  nodes = {
    installer =
      { lib, ... }:
      {
        imports = [
          sharedDisk
          self.nixosModules.installer
          common.tools
        ];
        router.installer.system = toplevel;
        virtualisation.memorySize = 2048;
        virtualisation.fileSystems."/".autoFormat = true;
        virtualisation.emptyDiskImages = [ 1024 ];
        virtualisation.rootDevice = "/dev/vdb";
        nix.settings.substituters = lib.mkForce [ ];
      };

    client =
      { lib, ... }:
      {
        imports = [ common.tools ];
        virtualisation.vlans = [ 2 ];
        networking.interfaces.eth1.ipv4.addresses = lib.mkForce [
          {
            address = "192.168.10.50";
            prefixLength = 24;
          }
        ];
        # channels.nixos.org: <branch>/git-revision is a redirect to the file.
        networking.firewall.allowedTCPPorts = [ 8080 ];
        environment.etc."channels.py".source = channelServer;
      };

    target = {
      imports = [ sharedDisk ];
      # Evaluating a NixOS system takes memory.
      virtualisation.memorySize = 3072;
      virtualisation.vlans = [
        1
        2
      ];
      virtualisation.useBootLoader = true;
      virtualisation.useEFIBoot = true;
      virtualisation.useDefaultFilesystems = false;
      virtualisation.efi.keepVariables = false;
      virtualisation.fileSystems."/" = {
        device = "/dev/disk/by-label/this-is-not-real-and-will-never-be-used";
        fsType = "ext4";
      };
    };
  };

  testScript = ''
    installer.start()
    installer.wait_for_unit("multi-user.target")
    installer.succeed("mkdir -p /root/secrets && cp -r ${configDir} /root/secrets/config")
    installer.succeed("router-install --yes --secrets /root/secrets /dev/vda >&2")
    installer.shutdown()

    target.state_dir = installer.state_dir
    target.start()
    target.wait_for_unit("multi-user.target")

    import json

    def current():
        return target.succeed("readlink -f /run/current-system").strip()

    def pool_end():
        return target.succeed("grep -o '192.168.10.100 - 192.168.10.[0-9]*' /etc/kea/dhcp4-server.conf").split()[-1]

    with subtest("the installed system is the one built from the settings"):
        assert current() == "${toplevel}"
        target.succeed("test -f /var/lib/router/config/settings.json")

    with subtest("rebuilding unchanged settings on the device changes nothing"):
        out = target.succeed("router-apply 2>&1")
        assert "no changes" in out, out

    with subtest("a changed setting is built offline, activated and confirmed"):
        target.succeed("sed -i 's/192.168.10.199/192.168.10.150/' /var/lib/router/config/settings.json")
        target.succeed("router-apply 5min >&2")
        assert current() != "${toplevel}"
        assert pool_end() == "192.168.10.150"
        target.succeed("systemctl is-active router-rollback.timer")
        target.succeed("router-confirm >&2")
        target.fail("systemctl is-active router-rollback.timer")
        confirmed = current()
        assert target.succeed("readlink -f /nix/var/nix/profiles/system").strip() == confirmed

    with subtest("an unconfirmed change is rolled back"):
        target.succeed("sed -i 's/192.168.10.150/192.168.10.120/' /var/lib/router/config/settings.json")
        target.succeed("router-apply 20s >&2")
        assert pool_end() == "192.168.10.120"
        target.wait_until_succeeds(f"test $(readlink -f /run/current-system) = {confirmed}", timeout=120)
        assert pool_end() == "192.168.10.150"

    ui = "https://192.168.10.1"

    def curl(args):
        return client.succeed(f"curl -sk -b /tmp/jar -c /tmp/jar {args}")

    def csrf(path):
        return curl(f"{ui}{path} | grep -o 'name=\"csrf\" value=\"[^\"]*' | head -1 | cut -d'\"' -f4").strip()

    with subtest("the web UI is reachable from the LAN and wants the password"):
        target.succeed("echo hunter2hunter2 | router-webui passwd /var/lib/router/webui-password")
        target.wait_for_unit("router-webui.service")
        client.start()
        client.wait_until_succeeds(f"curl -sk -o /dev/null {ui}/login", timeout=60)
        assert curl(f"-o /dev/null -w '%{{http_code}}' -d password=wrong {ui}/login") == "401"
        assert curl(f"-o /dev/null -w '%{{http_code}}' -d password=hunter2hunter2 {ui}/login") == "303"

    with subtest("the web UI shows the rolled back change as unapplied and discards it"):
        page = curl(f"{ui}/")
        assert "not part of the confirmed system" in page, page
        token = csrf("/")
        curl(f"-d csrf={token} {ui}/discard")
        assert "192.168.10.150" in target.succeed("cat /var/lib/router/config/settings.json")
        assert "match the confirmed system" in curl(f"{ui}/")

    with subtest("a pool change and a reservation made in the web UI are applied and confirmed"):
        token = csrf("/network")
        form = " ".join(f"--data-urlencode '{k}={v}'" for k, v in {
            "csrf": token, "hostName": "router", "wan.interface": "eth1", "wan.type": "dhcp",
            "lans": "1", "lan.0.name": "main", "lan.0.interface": "eth2", "lan.0.address": "192.168.10.1",
            "lan.0.internet": "on", "lan.0.dhcp": "on",
            "lan.0.poolStart": "192.168.10.100", "lan.0.poolEnd": "192.168.10.130",
        }.items())
        assert curl(f"-o /dev/null -w '%{{http_code}}' -X POST {ui}/network") == "403"
        assert curl(f"-o /dev/null -w '%{{http_code}}' {form} {ui}/network") == "303"
        # A reservation too: one apply for both changes.
        reservation = " ".join(f"--data-urlencode '{k}={v}'" for k, v in {
            "csrf": csrf("/dhcp"), "rows": "1", "res.0.lan": "main", "res.0.name": "printer",
            "res.0.mac": "AA:BB:CC:00:11:22", "res.0.address": "192.168.10.60",
        }.items())
        assert curl(f"-o /dev/null -w '%{{http_code}}' {reservation} {ui}/dhcp") == "303"
        curl(f"-d csrf={csrf('/')} {ui}/apply")
        target.wait_until_succeeds("test $(systemctl show -p ActiveState --value router-apply.service) = inactive"
                                   " && test $(systemctl show -p Result --value router-apply.service) = success", timeout=600)
        assert pool_end() == "192.168.10.130"
        target.succeed("grep -q 'aa:bb:cc:00:11:22' /etc/kea/dhcp4-server.conf && grep -q '192.168.10.60' /etc/kea/dhcp4-server.conf")
        assert "Waiting for confirmation" in curl(f"{ui}/")
        curl(f"-d csrf={csrf('/')} {ui}/confirm")
        target.wait_until_fails("systemctl is-active router-rollback.timer", timeout=60)
        target.wait_until_succeeds("test $(systemctl show -p ActiveState --value router-confirm.service) = inactive")
        assert target.succeed("readlink -f /nix/var/nix/profiles/system").strip() == current()
        assert "match the confirmed system" in curl(f"{ui}/")

    with subtest("naming the network cards in the web UI is staged for the next boot"):
        before = current()
        mac = {n: target.succeed(f"cat /sys/class/net/{n}/address").strip() for n in ("eth1", "eth2")}
        names = " ".join(f"--data-urlencode '{k}={v}'" for k, v in {
            "csrf": csrf("/network"), "rows": "2",
            "dev.0": "eth1", "mac.0": mac["eth1"], "from.0": "eth1", "name.0": "wan",
            "dev.1": "eth2", "mac.1": mac["eth2"], "from.1": "eth2", "name.1": "lan",
        }.items())
        assert curl(f"-o /dev/null -w '%{{http_code}}' {names} {ui}/interfaces") == "303"
        # The settings follow the names at once; the system does not.
        saved = target.succeed("cat /var/lib/router/config/settings.json")
        assert '"interface": "lan"' in saved and '"interface": "wan"' in saved and mac["eth2"] in saved, saved
        assert "after the next reboot" in curl(f"{ui}/network")
        curl(f"-d csrf={csrf('/')} {ui}/apply")
        target.wait_until_succeeds("test $(systemctl show -p ActiveState --value router-apply.service) = inactive"
                                   " && test $(systemctl show -p Result --value router-apply.service) = success", timeout=600)
        assert current() == before
        target.fail("systemctl is-active router-rollback.timer")
        staged = target.succeed("readlink -f /nix/var/nix/profiles/system").strip()
        assert staged != before
        target.succeed(f"grep -q '^nic {mac['eth2']} lan$' {staged}/etc/router/interface-names")
        target.succeed("ip link show eth2")
        # The overview samples the system every few seconds.
        client.wait_until_succeeds(f"curl -sk -b /tmp/jar {ui}/ | grep -c 'starts at the next reboot' >/dev/null", timeout=60)

    with subtest("after the reboot the cards have their names and the router works"):
        target.shutdown()
        target.start()
        target.wait_for_unit("multi-user.target")
        target.succeed("ip link show wan && ip link show lan")
        target.fail("ip link show eth1")
        target.wait_until_succeeds("ip -4 addr show dev lan | grep -q 192.168.10.1/24")
        target.wait_for_unit("kea-dhcp4-server.service")
        target.wait_for_unit("router-webui.service")
        client.wait_until_succeeds(f"curl -sk -o /dev/null {ui}/login", timeout=120)
        assert "match the confirmed system" in curl(f"-d password=hunter2hunter2 -L {ui}/login")
        assert "starts at the next reboot" not in curl(f"{ui}/")

    with subtest("an assistant with a change token edits and checks the settings through MCP"):
        secret = curl(f"-d csrf={csrf('/tokens')} -d name=assistant -d scope=change {ui}/tokens | grep -o 'rtr_[A-Za-z0-9_-]*' | sed -n 1p").strip()
        assert secret.startswith("rtr_"), secret
        def mcp(tool, args=None, token=secret):
            body = json.dumps({"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": tool, "arguments": args or {}}})
            out = client.succeed(f"curl -sk -H 'Authorization: Bearer {token}' -d '{body}' {ui}/mcp")
            res = json.loads(out)["result"]
            return res["content"][0]["text"], res.get("isError", False)

        assert client.succeed(f"curl -sk -o /dev/null -w '%{{http_code}}' -d '{{}}' {ui}/mcp").strip() == "401"
        text, failed = mcp("get_settings")
        got = json.loads(text)
        assert not failed and got["settings"]["router"]["lans"]["main"], text
        assert "WAN" in mcp("list_interfaces")[0]
        assert json.loads(mcp("get_status")[0])["uplink"] is not None
        # A bad pool: saved, not applied, and the check reports the router's assertion.
        text, failed = mcp("update_settings", {"etag": got["etag"], "patch": {"router": {"lans": {"main": {"dhcp": {"pool": {"end": "10.9.9.9"}}}}}}})
        assert not failed and '"applied": false' in text, text
        text, failed = mcp("check_settings")
        for _ in range(20):
            if "running" not in text:
                break
            text, failed = mcp("check_settings")
        res = json.loads(text)
        assert res["ok"] is False and "pool is not inside" in res["error"], text
        assert current() == target.succeed("readlink -f /nix/var/nix/profiles/system").strip()
        assert "10.9.9.9" in mcp("get_pending_changes")[0]
        # Discarded, the settings evaluate again.
        assert not mcp("discard_changes")[1]
        text, failed = mcp("check_settings")
        for _ in range(20):
            if "running" not in text:
                break
            text, failed = mcp("check_settings")
        assert json.loads(text)["ok"] is True, text
        # The journal can be read, through MCP and through the REST API (same tokens).
        text, failed = mcp("get_logs", {"unit": "router-webui", "lines": 5})
        assert not failed and "router-webui" in text, text
        text, failed = mcp("get_logs", {"unit": "--vacuum-time=1s"})
        assert failed, text
        rest = f"curl -sk -H 'Authorization: Bearer {secret}'"
        assert "router-webui" in client.succeed(f"{rest} '{ui}/api/v1/logs?unit=router-webui&lines=5'")
        assert json.loads(client.succeed(f"{rest} {ui}/api/v1/status"))["uplink"] is not None
        assert client.succeed(f"{rest} -o /dev/null -w '%{{http_code}}' -X PATCH -d '{{}}' {ui}/api/v1/settings").strip() == "400"
        assert client.succeed(f"curl -sk -o /dev/null -w '%{{http_code}}' {ui}/api/v1/status").strip() == "401"
        spec = json.loads(client.succeed(f"curl -sk {ui}/api/openapi.json"))
        assert spec["openapi"].startswith("3.") and "/api/v1/logs" in spec["paths"]
        # Areas of the settings, typed by the real option documentation.
        area = f"{rest} -H 'Content-Type: application/json'"
        etag = client.succeed(f"{rest} -D - -o /dev/null {ui}/api/v1/settings/router/lans/main | tr -d '\\r' | sed -n 's/^[Ee][Tt]ag: *//p'").strip()
        assert etag, "no ETag from an area"
        assert json.loads(client.succeed(f"{rest} {ui}/api/v1/settings/router/lans/main"))["interface"]
        code = client.succeed(f"{area} -o /tmp/area.json -w '%{{http_code}}' -X PUT -H 'If-Match: {etag}' -d '{{\"domian\": \"x\"}}' {ui}/api/v1/settings/router/dns").strip()
        assert code == "400" and "unknown option" in client.succeed("cat /tmp/area.json"), code
        code = client.succeed(f"{area} -o /tmp/area.json -w '%{{http_code}}' -X PUT -H 'If-Match: {etag}' -d '{{\"domain\": \"area.test\"}}' {ui}/api/v1/settings/router/dns").strip()
        assert code == "200", (code, client.succeed("cat /tmp/area.json"))
        assert "area.test" in mcp("get_pending_changes")[0]
        assert not mcp("discard_changes")[1]
        spec = json.loads(client.succeed(f"curl -sk {ui}/api/openapi.json"))
        assert "/api/v1/settings/router/lans/{name}" in spec["paths"]
        # The log shows what was done, by whom.
        assert "update_settings" in curl(f"{ui}/tokens") and "assistant" in curl(f"{ui}/tokens")

    # "A newer nixpkgs": the installed one, copied, and told it is the revision
    # the channel names (the running one). A different nixpkgs could not be
    # built here (even a changed version rebuilds the NixOS tools), so what is
    # tested is everything around the build.
    target.succeed("cp -r ${self.inputs.nixpkgs} /var/lib/newer-nixpkgs")
    running_version = target.succeed("cat /run/current-system/nixos-version")
    running_rev = "${self.inputs.nixpkgs.rev}"

    client.succeed("mkdir -p /srv/channels/files/nixos-26.05 && systemd-run --unit=channels /etc/channels.py")
    client.wait_for_open_port(8080)

    def channel(rev):
        client.succeed(f"printf %s {rev} > /srv/channels/files/nixos-26.05/git-revision")

    with subtest("router-upgrade check needs a channel and a reachable revision"):
        rc, out = target.execute("router-upgrade check 2>&1")
        assert rc != 0 and "cannot read" in out, out
        channel("not-a-revision")
        rc, out = target.execute("router-upgrade check 2>&1")
        assert rc != 0 and "unexpected answer" in out, out
        target.succeed("jq 'del(.router.update)' /var/lib/router/config/settings.json > /tmp/s.json && cp /var/lib/router/config/settings.json /tmp/settings.orig && cp /tmp/s.json /var/lib/router/config/settings.json")
        rc, out = target.execute("router-upgrade check 2>&1")
        assert rc != 0 and "no channel" in out, out
        target.succeed("cp /tmp/settings.orig /var/lib/router/config/settings.json")
        rc, out = target.execute("router-upgrade apply 2>&1")
        assert rc != 0 and "nothing is previewed" in out, out

    with subtest("a newer revision is reported, the running one is up to date"):
        rev = "0123456789abcdef0123456789abcdef01234567"
        channel(rev)
        out = target.succeed("router-upgrade check 2>&1")
        assert "a newer revision is available" in out, out
        assert json.loads(target.succeed("cat /var/lib/router/upgrade/check.json"))["upToDate"] is False
        channel(running_rev)
        out = target.succeed("router-upgrade check 2>&1")
        assert "up to date" in out, out
        assert json.loads(target.succeed("cat /var/lib/router/upgrade/check.json"))["upToDate"] is True
        out = target.succeed("router-upgrade preview 2>&1")
        assert "up to date" in out, out
        target.succeed("test ! -e /var/lib/router/upgrade/preview.json")

    with subtest("a preview lists the changes and the download without fetching or touching anything"):
        before = current()
        out = target.succeed("router-upgrade preview --force 2>&1")
        assert "no package changes" in out and "to fetch:" in out, out
        preview = json.loads(target.succeed("cat /var/lib/router/upgrade/preview.json"))
        assert preview["rev"] == running_rev and preview["version"] == running_version and preview["changes"] == [], preview
        assert preview["ref"].startswith("path:/var/lib/newer-nixpkgs?") and "rev=" + running_rev in preview["ref"] and "narHash=sha256-" in preview["ref"], preview
        assert preview["rebootNeeded"] is False and preview["builds"] == 0, preview
        target.succeed("test ! -e /var/lib/router/upgrade/result")
        assert current() == before
        target.succeed("test ! -e /var/lib/router/config/nixpkgs-pin.json")

    with subtest("the package list names what changes in version, and only that"):
        target.succeed("printf '%s\\n' linux-6.12.5 openssl-3.4.1 openssl-3.4.1-bin zlib-1.3.1 frr-10.1 ripgrep-14.1 rg-1.0 glibc-2.40-66 glibc-2.40-66-getent > /tmp/old")
        target.succeed("printf '%s\\n' linux-6.12.8 openssl-3.4.2 zlib-1.3.1 zlib-1.3.2 frr-10.2 cmake-4.0 bash-5.3 glibc-2.40-66 foo-1.tar.gz > /tmp/new")
        target.succeed("sed -n '/^changes() {/,/^}/p' /run/current-system/sw/bin/router-upgrade > /tmp/changes.sh && test -s /tmp/changes.sh")
        out = target.succeed("bash -c 'source /tmp/changes.sh; changes /tmp/old /tmp/new' | jq -c 'map(.name + \":\" + .from + \">\" + .to)'")
        assert json.loads(out) == ["frr:10.1>10.2", "linux:6.12.5>6.12.8", "openssl:3.4.1>3.4.2"], out

    with subtest("an upgrade that is not confirmed is rolled back, the lock and the pin included"):
        lock = target.succeed("cat /var/lib/router/config/flake.lock")
        # The packages are the same, so there is nothing to activate unless the
        # settings changed too (an upgrade applies what is saved, as `router-apply` does).
        target.succeed("grep -q 192.168.10.130 /var/lib/router/config/settings.json && sed -i 's/192.168.10.130/192.168.10.140/' /var/lib/router/config/settings.json")
        target.succeed("router-upgrade download >&2")
        target.succeed("test -e /var/lib/router/upgrade/result")
        target.succeed("router-upgrade apply 20s >&2")
        assert current() != before
        target.succeed("systemctl is-active router-rollback.timer")
        assert json.loads(target.succeed("cat /var/lib/router/config/nixpkgs-pin.json"))["rev"] == running_rev
        target.succeed("test ! -e /var/lib/router/upgrade/preview.json && test ! -e /var/lib/router/upgrade/result")
        target.wait_until_succeeds(f"test $(readlink -f /run/current-system) = {before}", timeout=180)
        target.succeed("test ! -e /var/lib/router/config/nixpkgs-pin.json && test ! -e /var/lib/router/config.pre-update")
        assert target.succeed("cat /var/lib/router/config/flake.lock") == lock

    with subtest("a confirmed upgrade stays, pins nixpkgs and rebuilds to exactly itself"):
        target.succeed("router-upgrade preview --force >&2")
        target.succeed("router-upgrade apply 5min >&2")
        target.succeed("router-confirm >&2")
        target.fail("systemctl is-active router-rollback.timer")
        upgraded = current()
        assert upgraded != before
        assert target.succeed("readlink -f /nix/var/nix/profiles/system").strip() == upgraded
        assert json.loads(target.succeed("cat /var/lib/router/config/nixpkgs-pin.json"))["rev"] == running_rev
        assert "newer-nixpkgs" in target.succeed("cat /var/lib/router/config/flake.lock")
        target.succeed("test ! -e /var/lib/router/config.pre-update")
        assert "no changes" in target.succeed("router-apply 2>&1")

    with subtest("an upgrade that fails to build puts the configuration back"):
        channel("fedcba9876543210fedcba9876543210fedcba98")
        target.succeed("router-upgrade preview >&2")
        pin = target.succeed("cat /var/lib/router/config/nixpkgs-pin.json")
        # An unknown option, after the preview: the build of the apply fails.
        target.succeed("cp /var/lib/router/config/settings.json /tmp/settings.good && jq '.router.nonsense = true' /tmp/settings.good > /var/lib/router/config/settings.json")
        rc, out = target.execute("router-upgrade apply 5min 2>&1")
        assert rc != 0 and "the upgrade failed" in out, out
        assert target.succeed("cat /var/lib/router/config/nixpkgs-pin.json") == pin
        target.succeed("test ! -e /var/lib/router/config.pre-update")
        target.succeed("cp /tmp/settings.good /var/lib/router/config/settings.json")
        assert current() == upgraded

    with subtest("the packages page drives the same steps"):
        channel(running_rev)

        def unit_done(unit):
            target.wait_until_succeeds(f"test $(systemctl show -p ActiveState --value {unit}) = inactive"
                                       f" && test $(systemctl show -p Result --value {unit}) = success", timeout=300)

        def step(name, extra=""):
            code = curl(f"-o /dev/null -w '%{{http_code}}' -d csrf={csrf('/upgrade')} {extra} {ui}/upgrade/{name}")
            assert code == "303", (name, code)
            unit_done("router-upgrade.service")

        page = curl(f"{ui}/upgrade")
        assert "Packages" in page and 'value="26.05" selected' in page and target.succeed("cat /run/current-system/nixos-version").strip() in page, page
        assert curl(f"-o /dev/null -w '%{{http_code}}' -d csrf={csrf('/upgrade')} -d 'channel=nixos-26.05%3Bx' {ui}/upgrade/channel") == "303"
        assert "26.05" in target.succeed("jq -r .router.update.channel /var/lib/router/config/settings.json")
        assert curl(f"-o /dev/null -w '%{{http_code}}' -d csrf=wrong {ui}/upgrade/check") == "403"
        step("check")
        assert "Up to date." in curl(f"{ui}/upgrade")
        step("preview", "-d force=on")
        page = curl(f"{ui}/upgrade")
        assert "No package changes" in page and "Upgrade to" in page, page
        step("download")
        step("apply")
        assert "Upgrade to" not in curl(f"{ui}/upgrade")
        assert json.loads(target.succeed("cat /var/lib/router/config/nixpkgs-pin.json"))["rev"] == running_rev
        assert "Last run" in curl(f"{ui}/upgrade")
  '';
}