"""Web lifecycle, sign-in, and cross-release HTML caching through the real CLI/daemon."""

import http.cookiejar
import json
import os
import re
import socket
import subprocess
import sys
import tempfile
import unittest
import urllib.error
import urllib.request
from pathlib import Path

from harness import Albedo, exclusive_class

# A tailscale CLI on PATH that keeps its serve config in $FAKE_TAILSCALE, so
# --tailnet-https runs end to end without touching a real tailnet.
FAKE_TAILSCALE = """
import json, os, sys
root = os.environ["FAKE_TAILSCALE"]
path = os.path.join(root, "state.json")
with open(path) as file:
    state = json.load(file)
args = sys.argv[1:]
lost = False
with open(os.path.join(root, "calls"), "a") as file:
    file.write(" ".join(args) + "\\n")
if args[:2] == ["status", "--json"]:
    print(json.dumps({"BackendState": "Running", "Self": {"DNSName": "albedo.example.ts.net."}, "CertDomains": state["certs"]}))
elif args == ["serve", "status", "--json"]:
    lost = state.pop("lose_status", False)
    if not lost:
        print(json.dumps(state["serve"]))
elif args[:3] == ["serve", "--bg", "--https=443"]:
    root = {"Handlers": {"/": {"Proxy": args[3]}}}
    state["serve"] = {"TCP": {"443": {"HTTPS": True}}, "Web": {"albedo.example.ts.net:443": root}}
    state["lose_status"] = state.pop("flaky", False)
elif args == ["serve", "--https=443", "--set-path", "/", "off"]:
    state["serve"] = {}
else:
    sys.exit("unexpected tailscale " + " ".join(args))
with open(path, "w") as file:
    json.dump(state, file)
if lost:
    sys.exit("tailscaled went away")
"""


@exclusive_class
class WebTest(unittest.TestCase):
    def setUp(self):
        self.app = Albedo(providers={}).__enter__()
        self.addCleanup(self.app.__exit__, None, None, None)
        self.addCleanup(self.app.cli, "web", "--stop")
        (self.app.workspace / "index.html").write_text("<p>web fixture</p>")
        self.app.env["ALBEDO_WEB_DIST"] = str(self.app.workspace)

    def web(self, *args):
        out = self.app.cli("web", "--local", "--no-open", *args)
        match = re.search(r"http://127\.0\.0\.1:\d+/\?key=\S+", out)
        self.assertIsNotNone(match, out)
        assert match is not None
        return match[0]

    def tailnet(self, serve=None, certs=("albedo.example.ts.net",), **flags):
        root = Path(tempfile.mkdtemp(dir=self.app.root))
        (root / "state.json").write_text(
            json.dumps({"serve": serve or {}, "certs": certs, **flags})
        )
        script = root / "tailscale"
        script.write_text(f"#!{sys.executable}\n{FAKE_TAILSCALE}")
        script.chmod(0o755)
        self.app.env["FAKE_TAILSCALE"] = str(root)
        self.app.env["PATH"] = f"{root}{os.pathsep}{os.environ['PATH']}"
        return root

    def refused(self, *args):
        command = self.app.daemon.start_cli(
            "web", "--no-open", *args, stdout=subprocess.PIPE, stderr=subprocess.PIPE
        )
        out, err = command.communicate(timeout=60)
        self.assertNotEqual(command.returncode, 0, out)
        return err.decode()

    def browser(self, link):
        jar = http.cookiejar.CookieJar()
        client = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
        with client.open(link, timeout=3) as response:
            self.assertIn(b"web fixture", response.read())
        # A browser closing must not sign it out: the cookie has an expiry.
        (cookie,) = list(jar)
        self.assertIsNotNone(cookie.expires)
        return client

    def test_background_lifecycle_and_sign_in(self):
        link = self.web("--port", "0")
        client = self.browser(link)
        base = link.split("/?")[0]
        with client.open(base + "/api/server", timeout=3) as response:
            self.assertEqual(json.load(response)["state"], "ready")
        self.assertEqual(self.web(), link)
        rotated = self.web("--new-link")
        self.assertNotEqual(rotated, link)
        with self.assertRaises(urllib.error.HTTPError) as caught:
            client.open(rotated.split("/?")[0] + "/api/sessions", timeout=3)
        self.assertEqual(caught.exception.code, 401)
        client = self.browser(rotated)
        link = rotated
        base = rotated.split("/?")[0]
        self.app.cli("web", "--stop")
        with self.assertRaises(urllib.error.URLError):
            client.open(base + "/api/sessions", timeout=3)
        restarted = self.web("--port", "0")
        self.assertEqual(restarted.split("key=")[1], link.split("key=")[1])

    def test_stop_stops_a_server_too_busy_to_answer(self):
        # Orphaned, so init reaps it and --stop sees it go.
        pid = int(
            subprocess.check_output(["sh", "-c", "sleep 60 >/dev/null & echo $!"])
        )
        self.addCleanup(
            lambda: subprocess.run(["kill", "-9", str(pid)], capture_output=True)
        )
        with socket.socket() as listener:
            listener.bind(("127.0.0.1", 0))
            listener.listen()
            port = listener.getsockname()[1]
            (self.app.home / "web.json").write_text(
                json.dumps(
                    {"pid": pid, "port": port, "url": f"http://127.0.0.1:{port}"}
                )
            )
            self.app.cli("web", "--stop")
        with self.assertRaises(ProcessLookupError, msg="--stop left a busy server"):
            os.kill(pid, 0)

    def test_html_revalidation_tracks_content_not_packaged_mtime(self):
        index = self.app.workspace / "index.html"
        first = b"<!doctype html><p>web fixture one</p>"
        second = b"<!doctype html><p>web fixture two</p>"
        legacy_date = "Thu, 01 Jan 1970 00:00:01 GMT"
        index.write_bytes(first)
        os.utime(index, (1, 1))
        assets = self.app.workspace / "assets"
        assets.mkdir()
        asset = assets / "index-abc.js"
        asset.write_text("console.log('fixture')")
        os.utime(asset, (1, 1))
        link = self.web("--port", "0")
        client = self.browser(link)
        base = link.split("/?")[0]

        def fetch(path="/", headers=None, method="GET"):
            request = urllib.request.Request(
                base + path, headers=headers or {}, method=method
            )
            try:
                response = client.open(request, timeout=3)
            except urllib.error.HTTPError as error:
                response = error
            with response:
                return response.status, response.headers, response.read()

        status, before, body = fetch()
        self.assertEqual((status, body), (200, first))
        # Nix gives different builds the same timestamp and these pages also
        # have equal sizes. Neither is a valid cross-release HTML validator.
        index.write_bytes(second)
        os.utime(index, (1, 1))
        status, after, body = fetch(headers={"If-Modified-Since": legacy_date})
        self.assertEqual((status, body), (200, second))
        self.assertEqual(after["Cache-Control"], "no-cache")
        self.assertNotIn("Last-Modified", after)
        self.assertIsNotNone(before["ETag"])
        self.assertIsNotNone(after["ETag"])
        self.assertNotEqual(before["ETag"], after["ETag"])

        for method, expected in [("GET", second), ("HEAD", b"")]:
            with self.subTest(stale_etag=method):
                status, headers, body = fetch(
                    headers={
                        "If-None-Match": before["ETag"],
                        "If-Modified-Since": legacy_date,
                    },
                    method=method,
                )
                self.assertEqual((status, body), (200, expected))
                self.assertEqual(headers["ETag"], after["ETag"])
                self.assertEqual(int(headers["Content-Length"]), len(second))

        for etag in [after["ETag"], "W/" + after["ETag"]]:
            with self.subTest(current_etag=etag):
                status, headers, body = fetch(
                    headers={"If-None-Match": etag, "If-Modified-Since": legacy_date}
                )
                self.assertEqual((status, body), (304, b""))
                self.assertEqual(headers["Cache-Control"], "no-cache")

        status, headers, body = fetch("/index.html")
        self.assertEqual((status, body), (200, second))
        self.assertEqual(headers["ETag"], after["ETag"])
        status, headers, _ = fetch("/assets/index-abc.js")
        self.assertEqual(status, 200)
        self.assertEqual(
            headers["Cache-Control"], "public, max-age=31536000, immutable"
        )
        status, _, body = fetch(
            "/assets/index-abc.js", {"If-Modified-Since": legacy_date}
        )
        self.assertEqual((status, body), (304, b""))
        status, _, body = fetch("/api/server")
        self.assertEqual(status, 200)
        self.assertEqual(json.loads(body)["state"], "ready")
        index.unlink()
        status, _, _ = fetch()
        self.assertEqual(status, 404)

    def test_busy_port_explains_startup_failure_without_waiting(self):
        with socket.socket() as listener:
            listener.bind(("127.0.0.1", 0))
            listener.listen()
            command = self.app.daemon.start_cli(
                "web",
                "--local",
                "--no-open",
                "--port",
                str(listener.getsockname()[1]),
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            )
            try:
                _, err = command.communicate(timeout=10)
            finally:
                if command.poll() is None:
                    command.kill()
                    command.communicate()
            self.assertNotEqual(command.returncode, 0)
            self.assertIn("address already in use", err.decode().lower())

    def test_tailnet_https_publishes_and_stop_removes_it(self):
        root = self.tailnet()
        out = self.app.cli("web", "--tailnet-https", "--no-open", "--port", "0")
        self.assertRegex(out, r"serving at https://albedo\.example\.ts\.net/\?key=")
        self.assertNotIn("http://albedo.example.ts.net", out)
        proxy = json.loads((root / "state.json").read_text())["serve"]["Web"][
            "albedo.example.ts.net:443"
        ]["Handlers"]["/"]["Proxy"]
        port = json.loads((self.app.home / "web.json").read_text())["port"]
        self.assertEqual(proxy, f"http://127.0.0.1:{port}")
        self.app.cli("web", "--stop")
        self.assertEqual(json.loads((root / "state.json").read_text())["serve"], {})

    def test_tailnet_https_refuses_without_changing_tailscale(self):
        with self.subTest("no certificate, as on Headscale"):
            root = self.tailnet(certs=None)
            self.assertIn(
                "issues no certificate", self.refused("--tailnet-https", "--port", "0")
            )
            self.assertNotIn("serve", (root / "calls").read_text())
        with socket.socket() as probe:
            probe.bind(("127.0.0.1", 0))
            port = probe.getsockname()[1]
        with self.subTest("our root opened to the internet by Funnel"):
            funnel = {
                "Web": {
                    "albedo.example.ts.net:443": {
                        "Handlers": {"/": {"Proxy": f"http://127.0.0.1:{port}"}}
                    }
                },
                "AllowFunnel": {"albedo.example.ts.net:443": True},
            }
            root = self.tailnet(serve=funnel)
            self.assertIn(
                "AllowFunnel", self.refused("--tailnet-https", "--port", str(port))
            )
            self.assertEqual(
                json.loads((root / "state.json").read_text())["serve"], funnel
            )
        with self.subTest("a handler created but not confirmed"):
            root = self.tailnet(flaky=True)
            self.assertIn(
                "tailscaled went away",
                self.refused("--tailnet-https", "--port", str(port)),
            )
            self.assertEqual(json.loads((root / "state.json").read_text())["serve"], {})
            self.assertFalse((self.app.home / "web.json").exists())