"""One live replay per public IP on the CPU server.

The AWS video workers enforce this in their Python signaling server. On the
CPU server every seat is its own relay process and the page picks the seat,
so the rule lives in nginx, which fronts every seat's /signal socket. These
tests pin the contract between the nginx config, the deploy script and the
live page without starting nginx.
"""
import pathlib
import re
import unittest

ROOT = pathlib.Path(__file__).resolve().parent
NGINX = (ROOT / "deploy" / "nginx-w3cs.conf").read_text(encoding="utf-8")
DEPLOY = (ROOT / "deploy" / "deploy-cpu-server.sh").read_text(encoding="utf-8")
PAGE = (ROOT / "poc" / "interactive-live.html").read_text(encoding="utf-8")


def location_blocks(config):
    """Yield (matcher, body) for every location block, in file order."""
    for match in re.finditer(
            r"location\s+(.+?)\s*\{(.*?)\n    \}", config, re.S):
        yield match.group(1).strip(), match.group(2)


class NginxViewerGateTests(unittest.TestCase):
    def test_zone_keys_on_client_address_with_trusted_exemptions(self):
        self.assertIn("geo $w3cs_viewer_key {", NGINX)
        self.assertIn("default $binary_remote_addr;", NGINX)
        self.assertIn('127.0.0.1 "";', NGINX)
        self.assertIn("include /etc/nginx/w3cs-trusted-ips.conf;", NGINX)
        self.assertIn(
            "limit_conn_zone $w3cs_viewer_key zone=w3cs_viewer:1m;", NGINX)
        self.assertIn("limit_conn_status 429;", NGINX)

    def test_only_seat_holding_sockets_and_the_probe_are_limited(self):
        limited = {matcher for matcher, body in location_blocks(NGINX)
                   if "limit_conn w3cs_viewer 1;" in body}
        self.assertEqual({
            "~ ^/seat(?<seat_number>[1-5])/signal$",
            "= /signal",
            "= /viewer-gate",
        }, limited)
        # Data planes, status probes and bundles of the session that holds
        # the seat must keep flowing while its socket is counted.
        for matcher, body in location_blocks(NGINX):
            if matcher not in limited:
                self.assertNotIn("limit_conn", body, matcher)

    def test_signal_route_is_matched_before_the_generic_seat_route(self):
        matchers = [matcher for matcher, _ in location_blocks(NGINX)]
        signal = matchers.index("~ ^/seat(?<seat_number>[1-5])/signal$")
        generic = matchers.index(
            "~ ^/seat(?<seat_number>[1-5])(?<seat_path>/.*)$")
        self.assertLess(signal, generic)
        # The legacy origin planes no longer swallow /signal unlimited.
        self.assertIn("~ ^/(reliable|frame|control)(/|$)", matchers)
        self.assertNotIn("~ ^/(signal|reliable|frame|control)(/|$)", matchers)

    def test_signal_route_keeps_the_websocket_proxy_contract(self):
        body = dict(location_blocks(NGINX))[
            "~ ^/seat(?<seat_number>[1-5])/signal$"]
        self.assertIn(
            "proxy_pass http://127.0.0.1:$seat_port/signal$is_args$args;",
            body)
        self.assertIn("proxy_set_header Upgrade $http_upgrade;", body)
        self.assertIn("proxy_read_timeout 3600s;", body)

    def test_probe_reaches_the_preaccess_limiter(self):
        body = dict(location_blocks(NGINX))["= /viewer-gate"]
        # `return` runs in the rewrite phase, before limit_conn (preaccess),
        # and would answer 204 beside a held socket. A content handler does
        # not.
        self.assertIn("empty_gif;", body)
        self.assertNotIn("return", body)
        self.assertIn("add_header Cache-Control no-store always;", body)


class DeployTests(unittest.TestCase):
    def test_deploy_guarantees_the_allowlist_include_before_nginx_t(self):
        create = DEPLOY.index(
            "install -m 0644 /dev/null /etc/nginx/w3cs-trusted-ips.conf")
        self.assertIn("[ -f /etc/nginx/w3cs-trusted-ips.conf ] ||", DEPLOY)
        self.assertLess(create, DEPLOY.index("\nnginx -t\n"))
        self.assertTrue(
            (ROOT / "deploy" / "w3cs-trusted-ips.conf.example").exists())


class PageTests(unittest.TestCase):
    def test_page_asks_the_gate_before_and_after_a_failed_connect(self):
        self.assertIn('fetch("/viewer-gate",{cache:"no-store"})', PAGE)
        self.assertIn("return response.status===429;", PAGE)
        connect = PAGE[PAGE.index("const connectWithSeat=async()=>{"):
                       PAGE.index("throw NO_FREE_SEATS;\n};")]
        self.assertIn("if(await viewerGateBusy())throw VIEWER_BUSY;", connect)
        # Pre-connect check comes before the seat search; the retry after a
        # refused socket comes after the 4001 seat-busy branch.
        self.assertLess(connect.index("if(await viewerGateBusy()){"),
                        connect.index("resolveSeat()"))
        self.assertLess(connect.index("lastSignalingClose?.code===4001"),
                        connect.rindex("throw VIEWER_BUSY"))

    def test_page_never_gates_lab_http_origins(self):
        self.assertIn(
            'if(location.protocol!=="https:"||explicitBridgePort)return false;',
            PAGE)

    def test_busy_card_waits_and_restarts_without_a_seat_search(self):
        self.assertIn("}else if(error===VIEWER_BUSY){", PAGE)
        self.assertIn(
            'textContent="Please watch one replay at a time";', PAGE)
        busy = PAGE[PAGE.index("}else if(error===VIEWER_BUSY){"):]
        busy = busy[:busy.index("}else{")]
        self.assertIn("if(!await viewerGateBusy()){", busy)
        self.assertIn("location.reload();", busy)
        self.assertNotIn("probeSeats", busy)


if __name__ == "__main__":
    unittest.main()