import asyncio
import time
from urllib.parse import parse_qs, urlsplit

import httpx
import pytest

from src import main
from src.session_store import SessionStoreError
from src.views import FRONTEND_BUILD


@pytest.fixture
def access_control(monkeypatch, tmp_path):
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "correct horse battery staple")
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "a-separate-session-signing-secret-with-enough-entropy")
    monkeypatch.setenv("STACKCHAIN_SESSION_DB", str(tmp_path / "sessions.sqlite3"))
    monkeypatch.setenv("STACKCHAIN_LOGIN_ATTEMPT_DB", str(tmp_path / "login-attempts.sqlite3"))
    monkeypatch.setenv("STACKCHAIN_LOGIN_MAX_FAILURES", "3")
    monkeypatch.setenv("STACKCHAIN_LOGIN_WINDOW_SECONDS", "60")


async def fresh_grant(client, action: str, target: str) -> str:
    response = await client.post(
        "/api/v1/fresh-authorization",
        json={
            "access_token": "correct horse battery staple",
            "action": action,
            "target": target,
        },
        headers={
            "Origin": "https://test",
            "X-CSRF-Token": client.cookies["stackchain_csrf"],
        },
    )
    assert response.status_code == 201
    return response.json()["grant"]


@pytest.mark.anyio
async def test_default_operator_mode_fails_closed_before_gitea_when_secrets_are_missing(monkeypatch):
    monkeypatch.delenv("STACKCHAIN_DASHBOARD_AUTH_MODE", raising=False)
    called = False

    async def user():
        nonlocal called
        called = True
        return {"id": 1, "login": "timmy"}

    monkeypatch.setattr(main, "current_user", user)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        health = await client.get("/healthz")
        responses = [
            await client.get("/readyz"),
            await client.get("/login"),
            await client.get("/"),
            await client.get("/api/v1/context"),
        ]

    assert health.status_code == 200
    assert [response.status_code for response in responses] == [503, 503, 503, 503]
    assert all(response.headers["cache-control"] == "no-store" for response in responses)
    assert all(response.json() == {
        "detail": "Dashboard authentication is not configured"
    } for response in responses)
    assert called is False


@pytest.mark.anyio
async def test_operator_mode_rejects_reused_or_incomplete_secrets(monkeypatch):
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "same-secret-with-at-least-thirty-two-characters")
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "same-secret-with-at-least-thirty-two-characters")
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        reused = await client.get("/login")
        monkeypatch.setenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "")
        incomplete = await client.get("/login")

    assert reused.status_code == 503
    assert incomplete.status_code == 503
    assert "same-secret" not in reused.text


@pytest.mark.anyio
async def test_insecure_local_mode_is_restricted_to_loopback(monkeypatch):
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "insecure-local")
    local_transport = httpx.ASGITransport(app=main.app, client=("127.0.0.1", 1234))
    remote_transport = httpx.ASGITransport(app=main.app, client=("203.0.113.9", 1234))
    async with httpx.AsyncClient(transport=local_transport, base_url="http://test") as client:
        local = await client.get("/login")
    async with httpx.AsyncClient(transport=remote_transport, base_url="http://test") as client:
        remote = await client.get("/login")

    assert local.status_code == 200
    assert remote.status_code == 403
    assert remote.json() == {"detail": "Insecure local mode requires a loopback client"}


@pytest.mark.anyio
async def test_anonymous_private_request_is_rejected_before_gitea(access_control, monkeypatch):
    called = False

    async def user():
        nonlocal called
        called = True
        return {"id": 1, "login": "timmy"}

    monkeypatch.setattr(main, "current_user", user)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        response = await client.get("/api/v1/context")

    assert response.status_code == 401
    assert response.json() == {"detail": "Authentication required"}
    assert response.headers["cache-control"] == "no-store"
    assert called is False


@pytest.mark.anyio
async def test_anonymous_share_target_redirect_preserves_only_bounded_capture_fields(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        response = await client.get(
            "/",
            params={
                "title": "Production crash",
                "text": "Steps from the mobile app",
                "url": "https://example.com/incidents/42",
                "next": "https://evil.example/steal",
            },
        )

    assert response.status_code == 303
    login_query = parse_qs(urlsplit(response.headers["location"]).query)
    assert login_query == {
        "continue": [
            "./?title=Production+crash&text=Steps+from+the+mobile+app&url="
            "https%3A%2F%2Fexample.com%2Fincidents%2F42"
        ]
    }
    assert response.headers["cache-control"] == "no-store"


@pytest.mark.anyio
async def test_oversized_share_target_is_not_carried_through_login(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        response = await client.get(
            "/", params={"title": "x" * 201, "text": "keep me"}
        )

    assert response.status_code == 303
    assert response.headers["location"] == "login"


@pytest.mark.anyio
async def test_sign_in_creates_secure_session_without_echoing_access_token(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        response = await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )

    assert response.status_code == 200
    assert response.json() == {"authenticated": True}
    cookies = response.headers.get_list("set-cookie")
    assert any("stackchain_session=" in value and "HttpOnly" in value and "Secure" in value and "SameSite=strict" in value for value in cookies)
    assert any("stackchain_csrf=" in value and "Secure" in value and "SameSite=strict" in value and "HttpOnly" not in value for value in cookies)
    assert "correct horse battery staple" not in response.text
    assert response.headers["cache-control"] == "no-store"


@pytest.mark.anyio
async def test_sign_in_throttles_repeated_failures_with_retry_guidance(access_control):
    transport = httpx.ASGITransport(app=main.app, client=("203.0.113.7", 1234))
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        failures = [
            await client.post("/api/v1/session", json={"access_token": "wrong"})
            for _ in range(3)
        ]
        blocked = await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )

    assert [response.status_code for response in failures] == [401, 401, 401]
    assert blocked.status_code == 429
    assert blocked.json() == {"detail": "Too many sign-in attempts"}
    assert blocked.headers["retry-after"].isdigit()
    assert blocked.headers["cache-control"] == "no-store"


@pytest.mark.anyio
async def test_successful_sign_in_clears_prior_failures(access_control):
    transport = httpx.ASGITransport(app=main.app, client=("203.0.113.8", 1234))
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        for _ in range(2):
            await client.post("/api/v1/session", json={"access_token": "wrong"})
        success = await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        after_success = [
            await client.post("/api/v1/session", json={"access_token": "wrong"})
            for _ in range(3)
        ]

    assert success.status_code == 200
    assert [response.status_code for response in after_success] == [401, 401, 401]


@pytest.mark.anyio
async def test_authenticated_get_reaches_private_api(access_control, monkeypatch):
    async def user():
        return {"id": 1, "login": "timmy", "full_name": "", "email": ""}

    async def empty():
        return []

    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", empty)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        signed_in = await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        response = await client.get("/api/v1/context")

    assert signed_in.status_code == 200
    assert response.status_code == 200
    assert response.json()["user"]["login"] == "timmy"


@pytest.mark.anyio
async def test_session_status_reuses_the_middleware_validation(access_control, monkeypatch):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        original_store = main.dashboard_auth._session_store()
        lookups = 0

        class CountingStore:
            def is_active(self, session_id, expires_at):
                nonlocal lookups
                lookups += 1
                return original_store.is_active(session_id, expires_at)

        monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: CountingStore())
        response = await client.get("/api/v1/session")

    assert response.status_code == 200
    payload = response.json()
    assert payload["authenticated"] is True
    assert isinstance(payload["expires_at"], int)
    assert payload["expires_at"] > int(time.time())
    assert response.headers["cache-control"] == "no-store"
    assert lookups == 1


@pytest.mark.anyio
async def test_authenticated_session_status_exposes_csrf_proof_and_offline_lease(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        response = await client.get("/api/v1/session")

    payload = response.json()
    assert response.status_code == 200
    assert payload["authenticated"] is True
    assert payload["csrf_token"] == client.cookies["stackchain_csrf"]
    assert payload["expires_at"] > int(time.time())
    assert "correct horse battery staple" not in response.text


@pytest.mark.anyio
async def test_anonymous_session_status_discloses_no_offline_lease(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        response = await client.get("/api/v1/session")

    assert response.status_code == 401
    assert response.json() == {"detail": "Authentication required"}
    assert "expires_at" not in response.text
    assert response.headers["cache-control"] == "no-store"


@pytest.mark.anyio
async def test_merge_requires_single_use_fresh_authorization_bound_to_exact_target(
    access_control, monkeypatch
):
    merge_calls = []

    async def assigned(repository, number):
        return True

    async def merge(repository, number, expected_head_sha):
        merge_calls.append((repository, number, expected_head_sha))
        return {"number": number, "merged": True, "state": "closed"}

    monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned)
    monkeypatch.setattr(main.gitea_proxy, "merge_assigned_pull", merge)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        csrf_headers = {
            "Origin": "https://test",
            "X-CSRF-Token": client.cookies["stackchain_csrf"],
        }
        missing = await client.post(
            "/api/v1/repos/stackchain/api/pulls/7/merge",
            json={"expected_head_sha": "abc123"},
            headers=csrf_headers,
        )
        authorized = await client.post(
            "/api/v1/fresh-authorization",
            json={
                "access_token": "correct horse battery staple",
                "action": "merge_pull",
                "target": "stackchain/api#7",
            },
            headers=csrf_headers,
        )
        grant_headers = {**csrf_headers, "X-Step-Up-Grant": authorized.json()["grant"]}
        merged = await client.post(
            "/api/v1/repos/stackchain/api/pulls/7/merge",
            json={"expected_head_sha": "abc123"},
            headers=grant_headers,
        )
        replayed = await client.post(
            "/api/v1/repos/stackchain/api/pulls/7/merge",
            json={"expected_head_sha": "abc123"},
            headers=grant_headers,
        )

    assert missing.status_code == 428
    assert missing.json() == {
        "detail": {
            "detail": "Fresh authorization required",
            "code": "step_up_required",
            "action": "merge_pull",
            "target": "stackchain/api#7",
        }
    }
    assert authorized.status_code == 201
    assert authorized.json()["expires_in"] == 90
    assert merged.status_code == 200
    assert replayed.status_code == 428
    assert merge_calls == [("stackchain/api", 7, "abc123")]


@pytest.mark.anyio
async def test_other_high_impact_routes_require_fresh_authorization_before_mutation(
    access_control, monkeypatch
):
    close_calls = []

    async def assigned(repository, number):
        return True

    async def close(repository, number):
        close_calls.append((repository, number))
        return {"number": number, "state": "closed"}

    monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
    monkeypatch.setattr(main.gitea_proxy, "close_issue", close)
    transport = httpx.ASGITransport(app=main.app)
    async with (
        httpx.AsyncClient(transport=transport, base_url="https://test") as phone,
        httpx.AsyncClient(transport=transport, base_url="https://test") as laptop,
    ):
        await phone.post(
            "/api/v1/session",
            json={"access_token": "correct horse battery staple", "device_label": "Phone"},
        )
        await laptop.post(
            "/api/v1/session",
            json={"access_token": "correct horse battery staple", "device_label": "Laptop"},
        )
        remote = next(
            item
            for item in (await phone.get("/api/v1/sessions")).json()["devices"]
            if not item["current"]
        )
        headers = {
            "Origin": "https://test",
            "X-CSRF-Token": phone.cookies["stackchain_csrf"],
        }
        closed = await phone.patch(
            "/api/v1/repos/stackchain/api/issues/7/close", headers=headers
        )
        revoked = await phone.delete(
            f"/api/v1/sessions/{remote['management_id']}", headers=headers
        )
        revoked_all = await phone.delete("/api/v1/sessions", headers=headers)
        laptop_still_active = await laptop.get("/api/v1/session")

    assert [closed.status_code, revoked.status_code, revoked_all.status_code] == [428, 428, 428]
    assert [closed.json()["detail"]["action"], revoked.json()["detail"]["action"], revoked_all.json()["detail"]["action"]] == [
        "close_issue", "revoke_device", "revoke_all_sessions"
    ]
    assert close_calls == []
    assert laptop_still_active.status_code == 200


@pytest.mark.anyio
async def test_operator_can_review_and_revoke_one_remote_device(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with (
        httpx.AsyncClient(transport=transport, base_url="https://test") as phone,
        httpx.AsyncClient(transport=transport, base_url="https://test") as laptop,
    ):
        await phone.post(
            "/api/v1/session",
            json={
                "access_token": "correct horse battery staple",
                "device_label": "Pixel <script>",
            },
        )
        await laptop.post(
            "/api/v1/session",
            json={
                "access_token": "correct horse battery staple",
                "device_label": "Work laptop",
            },
        )

        listed = await laptop.get("/api/v1/sessions")
        devices = listed.json()["devices"]
        phone_device = next(device for device in devices if device["device_label"] == "Pixel <script>")
        current_device = next(device for device in devices if device["current"])

        missing_csrf = await laptop.delete(
            f"/api/v1/sessions/{phone_device['management_id']}"
        )
        grant = await fresh_grant(
            laptop, "revoke_device", phone_device["management_id"]
        )
        revoked = await laptop.delete(
            f"/api/v1/sessions/{phone_device['management_id']}",
            headers={
                "Origin": "https://test",
                "X-CSRF-Token": laptop.cookies["stackchain_csrf"],
                "X-Step-Up-Grant": grant,
            },
        )
        phone_status = await phone.get("/api/v1/session")
        phone_page = await phone.get("/", follow_redirects=False)
        laptop_status = await laptop.get("/api/v1/session")

    assert listed.status_code == 200
    assert listed.headers["cache-control"] == "no-store"
    assert current_device["device_label"] == "Work laptop"
    assert missing_csrf.status_code == 403
    assert revoked.json() == {"revoked": True, "current_session": False}
    assert phone_status.status_code == 401
    assert phone_status.json() == {
        "detail": "Authentication required",
        "code": "session_revoked",
    }
    assert phone_page.status_code == 303
    assert phone_page.headers["location"] == "login?reason=session-revoked"
    assert laptop_status.status_code == 200
    assert "session_hash" not in listed.text
    assert "csrf" not in listed.text


@pytest.mark.anyio
async def test_active_device_labels_are_bounded_at_sign_in(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        response = await client.post(
            "/api/v1/session",
            json={
                "access_token": "correct horse battery staple",
                "device_label": "x" * 65,
            },
        )

    assert response.status_code == 422


@pytest.mark.anyio
async def test_mutation_requires_same_origin_and_session_csrf(access_control, monkeypatch):
    calls = 0

    async def mark_read(notification_id):
        nonlocal calls
        calls += 1
        return {"id": notification_id, "read": True}

    monkeypatch.setattr(main, "mark_notification_read", mark_read)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        missing = await client.patch("/api/v1/notifications/7/read")
        foreign = await client.patch(
            "/api/v1/notifications/7/read",
            headers={
                "Origin": "https://evil.example",
                "X-CSRF-Token": client.cookies["stackchain_csrf"],
            },
        )
        valid = await client.patch(
            "/api/v1/notifications/7/read",
            headers={
                "Origin": "https://test",
                "X-CSRF-Token": client.cookies["stackchain_csrf"],
            },
        )

    assert missing.status_code == 403
    assert foreign.status_code == 403
    assert valid.status_code == 200
    assert calls == 1


@pytest.mark.anyio
async def test_logout_clears_session_and_blocks_private_routes(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        csrf = client.cookies["stackchain_csrf"]
        response = await client.delete(
            "/api/v1/session",
            headers={"Origin": "https://test", "X-CSRF-Token": csrf},
        )
        private = await client.get("/api/v1/background-identity")

    assert response.status_code == 200
    assert response.json() == {"authenticated": False, "clear_private_device_data": True}
    assert private.status_code == 401
    assert all("Max-Age=0" in value for value in response.headers.get_list("set-cookie"))


@pytest.mark.anyio
async def test_sign_out_all_devices_revokes_every_existing_session(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with (
        httpx.AsyncClient(transport=transport, base_url="https://test") as phone,
        httpx.AsyncClient(transport=transport, base_url="https://test") as laptop,
    ):
        await phone.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        await laptop.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )

        grant = await fresh_grant(phone, "revoke_all_sessions", "all")
        response = await phone.delete(
            "/api/v1/sessions",
            headers={
                "Origin": "https://test",
                "X-CSRF-Token": phone.cookies["stackchain_csrf"],
                "X-Step-Up-Grant": grant,
            },
        )
        phone_private = await phone.get("/api/v1/background-identity")
        laptop_private = await laptop.get("/api/v1/background-identity")

    assert response.status_code == 200
    assert response.json() == {
        "authenticated": False,
        "all_sessions_revoked": True,
        "clear_private_device_data": True,
    }
    assert phone_private.status_code == 401
    assert laptop_private.status_code == 401
    assert all("Max-Age=0" in value for value in response.headers.get_list("set-cookie"))


@pytest.mark.anyio
async def test_sign_out_all_devices_rejects_cross_site_requests_without_revoking(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        response = await client.delete(
            "/api/v1/sessions",
            headers={
                "Origin": "https://evil.example",
                "X-CSRF-Token": client.cookies["stackchain_csrf"],
            },
        )
        session = await client.get("/api/v1/session")

    assert response.status_code == 403
    assert response.headers.get_list("set-cookie") == []
    assert session.status_code == 200
    assert session.json()["authenticated"] is True


@pytest.mark.anyio
async def test_sign_out_all_devices_registry_failure_sets_no_cookies(access_control, monkeypatch):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        csrf = client.cookies["stackchain_csrf"]
        grant = await fresh_grant(client, "revoke_all_sessions", "all")

        class BrokenStore:
            def is_active(self, session_id, expires_at):
                return True

            def consume_step_up(self, grant, session_id, *, action, target):
                return True

            def revoke_all(self):
                raise SessionStoreError("database path and secret details")

        monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
        response = await client.delete(
            "/api/v1/sessions",
            headers={
                "Origin": "https://test",
                "X-CSRF-Token": csrf,
                "X-Step-Up-Grant": grant,
            },
        )

    assert response.status_code == 503
    assert response.json() == {"detail": "Session registry is temporarily unavailable"}
    assert response.headers["cache-control"] == "no-store"
    assert response.headers.get_list("set-cookie") == []
    assert "database path" not in response.text


@pytest.mark.anyio
async def test_logout_revokes_a_captured_cookie_before_gitea(access_control, monkeypatch):
    calls = 0

    async def user():
        nonlocal calls
        calls += 1
        return {"id": 1, "login": "timmy"}

    monkeypatch.setattr(main, "current_user", user)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        captured = client.cookies["stackchain_session"]
        csrf = client.cookies["stackchain_csrf"]

        async with httpx.AsyncClient(
            transport=transport,
            base_url="https://test",
            headers={"Cookie": f"stackchain_session={captured}"},
        ) as replay:
            before = await replay.get("/api/v1/background-identity")

        await client.delete(
            "/api/v1/session",
            headers={"Origin": "https://test", "X-CSRF-Token": csrf},
        )

        async with httpx.AsyncClient(
            transport=transport,
            base_url="https://test",
            headers={"Cookie": f"stackchain_session={captured}"},
        ) as replay:
            after = await replay.get("/api/v1/background-identity")

    assert before.status_code == 200
    assert after.status_code == 401
    assert after.headers["cache-control"] == "no-store"
    assert calls == 1


@pytest.mark.anyio
async def test_session_registry_latency_does_not_block_the_event_loop(access_control, monkeypatch):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )

        class SlowStore:
            def is_active(self, session_id, expires_at):
                time.sleep(0.15)
                return True

        monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: SlowStore())
        private_request = asyncio.create_task(client.get("/api/v1/session"))
        await asyncio.sleep(0)
        started = time.perf_counter()
        await asyncio.sleep(0.01)
        heartbeat_elapsed = time.perf_counter() - started
        response = await private_request

    assert response.status_code == 200
    assert heartbeat_elapsed < 0.08


@pytest.mark.anyio
async def test_sign_in_throttle_lookup_does_not_block_the_event_loop(
    access_control, monkeypatch
):
    class SlowAttempts:
        def retry_after(self, source):
            time.sleep(0.15)
            return 17

    monkeypatch.setattr(main, "_login_attempt_store", lambda: SlowAttempts())
    transport = httpx.ASGITransport(app=main.app, client=("203.0.113.10", 1234))
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        sign_in = asyncio.create_task(
            client.post("/api/v1/session", json={"access_token": "wrong"})
        )
        await asyncio.sleep(0)
        started = time.perf_counter()
        await asyncio.sleep(0.01)
        heartbeat_elapsed = time.perf_counter() - started
        response = await sign_in

    assert response.status_code == 429
    assert response.headers["retry-after"] == "17"
    assert heartbeat_elapsed < 0.08


@pytest.mark.anyio
async def test_failed_sign_in_recording_does_not_block_the_event_loop(
    access_control, monkeypatch
):
    class SlowAttempts:
        def retry_after(self, source):
            time.sleep(0.02)
            return 0

        def record_failure(self, source):
            time.sleep(0.15)

    monkeypatch.setattr(main, "_login_attempt_store", lambda: SlowAttempts())
    transport = httpx.ASGITransport(app=main.app, client=("203.0.113.11", 1234))
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        sign_in = asyncio.create_task(
            client.post("/api/v1/session", json={"access_token": "wrong"})
        )
        await asyncio.sleep(0)
        started = time.perf_counter()
        await asyncio.sleep(0.04)
        heartbeat_elapsed = time.perf_counter() - started
        response = await sign_in

    assert response.status_code == 401
    assert heartbeat_elapsed < 0.08


@pytest.mark.anyio
async def test_successful_sign_in_throttle_clear_does_not_block_the_event_loop(
    access_control, monkeypatch
):
    class SlowAttempts:
        def retry_after(self, source):
            time.sleep(0.02)
            return 0

        def clear(self, source):
            time.sleep(0.15)

    monkeypatch.setattr(main, "_login_attempt_store", lambda: SlowAttempts())
    transport = httpx.ASGITransport(app=main.app, client=("203.0.113.12", 1234))
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        sign_in = asyncio.create_task(
            client.post(
                "/api/v1/session",
                json={"access_token": "correct horse battery staple"},
            )
        )
        await asyncio.sleep(0)
        started = time.perf_counter()
        await asyncio.sleep(0.04)
        heartbeat_elapsed = time.perf_counter() - started
        response = await sign_in

    assert response.status_code == 200
    assert heartbeat_elapsed < 0.08


@pytest.mark.anyio
async def test_session_activation_does_not_block_the_event_loop(
    access_control, monkeypatch
):
    class Attempts:
        def retry_after(self, source):
            time.sleep(0.02)
            return 0

        def clear(self, source):
            return None

    class SlowStore:
        def activate(self, session_id, expires_at, **kwargs):
            time.sleep(0.15)

    monkeypatch.setattr(main, "_login_attempt_store", lambda: Attempts())
    monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: SlowStore())
    transport = httpx.ASGITransport(app=main.app, client=("203.0.113.13", 1234))
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        sign_in = asyncio.create_task(
            client.post(
                "/api/v1/session",
                json={"access_token": "correct horse battery staple"},
            )
        )
        await asyncio.sleep(0)
        started = time.perf_counter()
        await asyncio.sleep(0.04)
        heartbeat_elapsed = time.perf_counter() - started
        response = await sign_in

    assert response.status_code == 200
    assert any(
        "stackchain_session=" in value
        for value in response.headers.get_list("set-cookie")
    )
    assert heartbeat_elapsed < 0.08


@pytest.mark.anyio
async def test_single_session_revocation_does_not_block_the_event_loop(
    access_control, monkeypatch
):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session",
            json={"access_token": "correct horse battery staple"},
        )
        csrf = client.cookies["stackchain_csrf"]

        class SlowStore:
            def is_active(self, session_id, expires_at):
                time.sleep(0.02)
                return True

            def revoke(self, session_id):
                time.sleep(0.15)

        monkeypatch.setattr(
            main.dashboard_auth, "_session_store", lambda now=None: SlowStore()
        )
        sign_out = asyncio.create_task(
            client.delete(
                "/api/v1/session",
                headers={"Origin": "https://test", "X-CSRF-Token": csrf},
            )
        )
        await asyncio.sleep(0)
        started = time.perf_counter()
        await asyncio.sleep(0.04)
        heartbeat_elapsed = time.perf_counter() - started
        response = await sign_out

    assert response.status_code == 200
    assert response.json()["authenticated"] is False
    assert heartbeat_elapsed < 0.08


@pytest.mark.anyio
async def test_session_registry_read_failure_fails_closed_before_gitea(
    access_control, monkeypatch
):
    called = False

    async def user():
        nonlocal called
        called = True
        return {"id": 1, "login": "timmy"}

    monkeypatch.setattr(main, "current_user", user)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )

        class BrokenStore:
            def is_active(self, session_id, expires_at):
                raise SessionStoreError("database path and secret details")

        monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
        response = await client.get("/api/v1/background-identity")
        health = await client.get("/healthz")

    assert response.status_code == 503
    assert response.json() == {"detail": "Session registry is temporarily unavailable"}
    assert response.headers["cache-control"] == "no-store"
    assert "database path" not in response.text
    assert called is False
    assert health.status_code == 200


@pytest.mark.anyio
async def test_sign_in_registry_write_failure_issues_no_cookie(access_control, monkeypatch):
    class BrokenStore:
        def activate(self, session_id, expires_at, **kwargs):
            raise SessionStoreError("database path and secret details")

    monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        response = await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )

    assert response.status_code == 503
    assert response.json() == {"detail": "Session registry is temporarily unavailable"}
    assert response.headers["cache-control"] == "no-store"
    assert response.headers.get_list("set-cookie") == []
    assert "database path" not in response.text


@pytest.mark.anyio
async def test_logout_registry_failure_does_not_claim_revocation(access_control, monkeypatch):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        csrf = client.cookies["stackchain_csrf"]

        class BrokenStore:
            def is_active(self, session_id, expires_at):
                return True

            def revoke(self, session_id):
                raise SessionStoreError("database path and secret details")

        monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
        response = await client.delete(
            "/api/v1/session",
            headers={"Origin": "https://test", "X-CSRF-Token": csrf},
        )

    assert response.status_code == 503
    assert response.json() == {"detail": "Session registry is temporarily unavailable"}
    assert response.headers["cache-control"] == "no-store"
    assert response.headers.get_list("set-cookie") == []
    assert "database path" not in response.text


@pytest.mark.anyio
async def test_public_routes_skip_session_registry_validation(access_control, monkeypatch):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        signed_in = await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )

        class BrokenStore:
            def is_active(self, session_id, expires_at):
                raise SessionStoreError("public routes must not read the registry")

        monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
        login = await client.get("/login")
        manifest = await client.get("/manifest.webmanifest")
        static = await client.get("/static/session.js")
        runtime = await client.get("/" + FRONTEND_BUILD.runtime_name)

    assert signed_in.status_code == 200
    assert [login.status_code, manifest.status_code, static.status_code, runtime.status_code] == [
        200,
        200,
        200,
        200,
    ]
    assert runtime.headers["cache-control"] == "public, max-age=31536000, immutable"


@pytest.mark.anyio
async def test_public_routes_remain_available_and_readiness_hides_identity(access_control, monkeypatch):
    async def user():
        return {"id": 1, "login": "timmy"}

    monkeypatch.setattr(main, "current_user", user)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        health = await client.get("/healthz")
        login = await client.get("/login")
        ready = await client.get("/readyz")

    assert health.status_code == 200
    assert login.status_code == 200
    assert 'name="access_token"' in login.text
    assert ready.status_code == 200
    assert ready.json() == {"status": "ready", "service": "stackchain-dashboard"}


@pytest.mark.anyio
async def test_subpath_deployment_scopes_routes_and_session_cookies(access_control):
    transport = httpx.ASGITransport(app=main.app, root_path="/dashboard")
    async with httpx.AsyncClient(
        transport=transport, base_url="https://test/dashboard/"
    ) as client:
        login = await client.get("login")
        signed_in = await client.post(
            "api/v1/session",
            json={"access_token": "correct horse battery staple"},
        )
        status = await client.get("api/v1/session")

    assert login.status_code == 200
    assert signed_in.status_code == 200
    assert status.status_code == 200
    assert any(
        "stackchain_session=" in value and "Path=/dashboard" in value
        for value in signed_in.headers.get_list("set-cookie")
    )