"""Tests for the multi-realm machinery: registry resolution, config
validation, ASGI dispatch, realm binding of auth codes, legacy database
conversion, log censoring and bootstrap caveats.
"""

from __future__ import annotations

import asyncio
import os
from datetime import UTC, datetime
from uuid import UUID

import httpx
import pytest
from kanta import Kanta

from paskia import authcode, realms
from paskia.bootstrap import check_admin_credentials
from paskia.db import create_credential
from paskia.db.legacy import (
    LegacyConfig,
    LegacyCredential,
    LegacyDB,
    LegacySession,
    convert_legacy_database,
)
from paskia.db.lifecycle import format_log_uuid
from paskia.db.operations import DB
from paskia.db.structs import Client, Config, Credential, RealmConfig
from paskia.fastapi.dispatch import DispatchMiddleware

# -------------------------------------------------------------------------
# Registry construction helpers
# -------------------------------------------------------------------------


def build_registry(*realm_configs: RealmConfig) -> realms.RealmRegistry:
    """Build and install a registry from realm configs (listen unset)."""
    realms.configure(listen=None)
    return realms.init_registry(Config(realms=list(realm_configs)))


ROR_CONFIG = Config(
    realms=[
        RealmConfig(
            rp_id="company.com",
            auth_host="https://auth.company.com",
            origins=["https://auth.company.com", "https://app.com"],
        ),
        RealmConfig(rp_id="pro.com"),
    ]
)


class StubApp:
    """ASGI app recording the scope it was called with."""

    def __init__(self):
        self.scope = None

    async def __call__(self, scope, receive, send):
        self.scope = scope


async def drive_ws(middleware: DispatchMiddleware, headers: list[tuple[bytes, bytes]]):
    """Run a websocket scope through the middleware, capturing sent messages."""

    async def receive():
        return {"type": "websocket.connect"}

    sent = []

    async def send(message):
        sent.append(message)

    stub = middleware.app
    await middleware(
        {"type": "websocket", "headers": headers, "path": "/"}, receive, send
    )
    return stub, sent


async def drive_http(
    middleware: DispatchMiddleware, headers: list[tuple[bytes, bytes]]
):
    """Run an http scope through the middleware, capturing sent messages."""

    async def receive():
        return {"type": "http.request", "body": b""}

    sent = []

    async def send(message):
        sent.append(message)

    stub = middleware.app
    await middleware(
        {
            "type": "http",
            "headers": headers,
            "method": "GET",
            "path": "/",
            "query_string": b"",
        },
        receive,
        send,
    )
    return stub, sent


# -------------------------------------------------------------------------
# Host resolution
# -------------------------------------------------------------------------


class TestResolve:
    def test_exact_rp_id(self):
        reg = build_registry(*ROR_CONFIG.realms)
        assert reg.resolve("pro.com").rp_id == "pro.com"
        assert reg.resolve("company.com").rp_id == "company.com"

    def test_auth_host_and_related_origin(self):
        reg = build_registry(*ROR_CONFIG.realms)
        assert reg.resolve("auth.company.com").rp_id == "company.com"
        assert reg.resolve("app.com").rp_id == "company.com"

    def test_subdomain_suffix_longest_match(self):
        reg = build_registry(
            RealmConfig(rp_id="example.com"), RealmConfig(rp_id="sub.example.com")
        )
        assert reg.resolve("www.example.com").rp_id == "example.com"
        assert reg.resolve("api.sub.example.com").rp_id == "sub.example.com"

    def test_port_and_trailing_dot_normalized(self):
        reg = build_registry(*ROR_CONFIG.realms)
        assert reg.resolve("pro.com:8443").rp_id == "pro.com"
        assert reg.resolve("app.com.").rp_id == "company.com"

    def test_unknown_host(self):
        reg = build_registry(*ROR_CONFIG.realms)
        assert reg.resolve("evil.com") is None
        assert reg.resolve("") is None
        assert reg.resolve(None) is None

    def test_effective_auth_host_fallback(self):
        reg = build_registry(*ROR_CONFIG.realms)
        company = reg.get("company.com")
        pro = reg.get("pro.com")
        assert reg.effective_auth_host(company) == "auth.company.com"
        # pro.com has no own auth host: falls back to the first configured one
        assert reg.effective_auth_host(pro) == "auth.company.com"
        # No auth hosts at all: None
        reg2 = build_registry(RealmConfig(rp_id="a.com"), RealmConfig(rp_id="b.com"))
        assert reg2.effective_auth_host(reg2.get("a.com")) is None


# -------------------------------------------------------------------------
# Cross-realm configuration validation
# -------------------------------------------------------------------------


class TestValidateConfig:
    def test_valid(self):
        realms.validate_config(ROR_CONFIG)

    def test_related_origin_cap(self):
        realms.validate_config(
            Config(
                realms=[
                    RealmConfig(
                        rp_id="company.com",
                        origins=[f"https://app{i}.com" for i in range(5)],
                    )
                ]
            )
        )
        with pytest.raises(ValueError, match="related origins"):
            realms.validate_config(
                Config(
                    realms=[
                        RealmConfig(
                            rp_id="company.com",
                            origins=[f"https://app{i}.com" for i in range(6)],
                        )
                    ]
                )
            )

    def test_auth_host_collision(self):
        with pytest.raises(ValueError, match="collides with a related origin"):
            realms.validate_config(
                Config(
                    realms=[
                        RealmConfig(rp_id="a.com", auth_host="https://auth.a.com"),
                        RealmConfig(
                            rp_id="b.com",
                            origins=["https://auth.a.com"],
                        ),
                    ]
                )
            )

    def test_related_origin_inside_other_realm(self):
        with pytest.raises(ValueError, match="falls inside realm"):
            realms.validate_config(
                Config(
                    realms=[
                        RealmConfig(rp_id="a.com", origins=["https://app.b.com"]),
                        RealmConfig(rp_id="b.com"),
                    ]
                )
            )

    def test_auth_host_must_not_collide_with_rp_id(self):
        with pytest.raises(ValueError, match="collides with an rp-id"):
            realms.validate_config(
                Config(
                    realms=[
                        RealmConfig(rp_id="a.com", auth_host="https://b.a.com"),
                        RealmConfig(rp_id="b.a.com"),
                    ]
                )
            )


# -------------------------------------------------------------------------
# ASGI dispatch
# -------------------------------------------------------------------------


class TestDispatchMiddleware:
    @pytest.mark.asyncio
    async def test_http_unknown_host_421(self):
        build_registry(*ROR_CONFIG.realms)
        stub, sent = await drive_http(
            DispatchMiddleware(StubApp()), [(b"host", b"evil.com")]
        )
        assert stub.scope is None  # Inner app not called
        assert sent[0]["type"] == "http.response.start"
        assert sent[0]["status"] == 421

    @pytest.mark.asyncio
    async def test_http_dispatches_realm(self):
        build_registry(*ROR_CONFIG.realms)
        stub, _sent = await drive_http(
            DispatchMiddleware(StubApp()), [(b"host", b"app.com.")]
        )
        assert stub.scope is not None
        assert stub.scope["state"]["realm"].rp_id == "company.com"

    @pytest.mark.asyncio
    async def test_http_current_realm_set_inside_request(self):
        reg = build_registry(*ROR_CONFIG.realms)
        seen = {}

        async def app(scope, receive, send):
            seen["realm"] = realms.current_realm()

        await drive_http(DispatchMiddleware(app), [(b"host", b"pro.com")])
        assert seen["realm"].rp_id == "pro.com"
        # Contextvar is reset after the request
        assert realms.current_realm() is reg.default

    @pytest.mark.asyncio
    async def test_ws_unknown_host_closed(self):
        build_registry(*ROR_CONFIG.realms)
        stub, sent = await drive_ws(
            DispatchMiddleware(StubApp()), [(b"host", b"evil.com")]
        )
        assert stub.scope is None
        assert sent == [{"type": "websocket.close", "code": 1008}]

    @pytest.mark.asyncio
    async def test_ws_same_realm_origin(self):
        build_registry(*ROR_CONFIG.realms)
        stub, sent = await drive_ws(
            DispatchMiddleware(StubApp()),
            [(b"host", b"auth.company.com"), (b"origin", b"https://app.com")],
        )
        assert sent == []
        assert stub.scope["state"]["realm"].rp_id == "company.com"

    @pytest.mark.asyncio
    async def test_ws_cross_realm_requires_effective_auth_host(self):
        build_registry(*ROR_CONFIG.realms)
        # pro.com page connecting to the shared auth host: allowed, pro realm
        stub, sent = await drive_ws(
            DispatchMiddleware(StubApp()),
            [(b"host", b"auth.company.com"), (b"origin", b"https://pro.com")],
        )
        assert sent == []
        assert stub.scope["state"]["realm"].rp_id == "pro.com"

        # pro.com page connecting to some other host: closed pre-accept
        stub, sent = await drive_ws(
            DispatchMiddleware(StubApp()),
            [(b"host", b"company.com"), (b"origin", b"https://pro.com")],
        )
        assert stub.scope is None
        assert sent == [{"type": "websocket.close", "code": 1008}]

    @pytest.mark.asyncio
    async def test_ws_unknown_origin_uses_host_realm(self):
        build_registry(*ROR_CONFIG.realms)
        # Missing origin
        stub, _ = await drive_ws(DispatchMiddleware(StubApp()), [(b"host", b"pro.com")])
        assert stub.scope["state"]["realm"].rp_id == "pro.com"
        # Unknown origin: host realm applies (endpoint-side validation decides)
        stub, _ = await drive_ws(
            DispatchMiddleware(StubApp()),
            [(b"host", b"pro.com"), (b"origin", b"https://evil.com")],
        )
        assert stub.scope["state"]["realm"].rp_id == "pro.com"


# -------------------------------------------------------------------------
# Realm binding of auth codes
# -------------------------------------------------------------------------


class TestAuthCodeRealmBinding:
    @pytest.mark.asyncio
    async def test_cookie_code_rejected_on_other_realm(
        self, client: httpx.AsyncClient, session_token: str
    ):
        code = authcode.store_cookie(
            authcode.CookieCode(
                session_key=session_token,
                created=datetime.now(UTC),
                rp_id="other.com",
            )
        )
        response = await client.post(
            "/auth/api/set-session",
            headers={
                "Authorization": f"Bearer {code}",
                "Host": "localhost:4401",
            },
        )
        assert response.status_code == 401

    @pytest.mark.asyncio
    async def test_oidc_code_rejected_on_other_realm(
        self, client: httpx.AsyncClient, test_db: DB
    ):
        oidc_client, secret = Client.create(
            name="Test Client",
            redirect_uris=["https://client.example/callback"],
            client_secret="topsecret",
        )
        store = test_db._store
        with store.transaction("create_test_oidc_client"):
            test_db.oidc["localhost"].clients[oidc_client.uuid] = oidc_client

        code = authcode.store_oidc(
            authcode.OIDCCode(
                session_key="doesnotmatter1234",
                created=datetime.now(UTC),
                redirect_uri="https://client.example/callback",
                scope="openid",
                rp_id="other.com",
            )
        )
        response = await client.post(
            "/auth/oidc/token",
            data={
                "grant_type": "authorization_code",
                "code": code,
                "redirect_uri": "https://client.example/callback",
                "client_id": str(oidc_client.uuid),
                "client_secret": secret,
            },
            headers={"Host": "localhost:4401"},
        )
        assert response.status_code == 400
        assert response.json()["error"] == "invalid_grant"
        assert "realm" in response.json()["error_description"]


# -------------------------------------------------------------------------
# Legacy database conversion
# -------------------------------------------------------------------------


def _read_db(path) -> DB:
    async def _read() -> DB:
        new_db = DB()
        kanta = Kanta(str(path), new_db)
        await kanta.open(readonly=True)
        return kanta.data

    return asyncio.run(_read())


class TestLegacyConversion:
    def test_convert_stamps_realm_everywhere(self, tmp_path):
        src = tmp_path / "example.com.paskiadb"
        src.mkdir()
        src_file = src / "main.db"

        cred_uuid = UUID("019c6831-84cf-7b88-b66c-c8165890b7c5")
        user_uuid = UUID("019c6831-84cf-7b88-b66c-c8165890b7c6")

        async def _write() -> None:
            kanta = Kanta(str(src_file), LegacyDB())
            await kanta.open()
            with kanta.transaction("test:seed"):
                kanta.data.config = LegacyConfig(
                    rp_id="example.com",
                    rp_name="Example",
                    origins=["https://app.example.com"],
                )
                kanta.data.credentials[cred_uuid] = LegacyCredential(
                    credential_id=b"credential-id",
                    user_uuid=user_uuid,
                    aaguid=UUID(int=0),
                    public_key=b"public-key",
                    sign_count=3,
                    created_at=datetime.now(UTC),
                )
                kanta.data.sessions["session-key"] = LegacySession(
                    user_uuid=user_uuid,
                    credential_uuid=cred_uuid,
                    host="example.com",
                    ip="127.0.0.1",
                    user_agent="pytest",
                    validated=datetime.now(UTC),
                )
            await kanta.close()

        asyncio.run(_write())

        config = convert_legacy_database(src_file, tmp_path / "paskia.kantadb")
        assert config.default_realm.rp_id == "example.com"
        assert config.default_realm.rp_name == "Example"

        converted = _read_db(tmp_path / "paskia.kantadb")
        assert converted.credentials[cred_uuid].rp_id == "example.com"
        assert converted.sessions["session-key"].rp_id == "example.com"
        assert set(converted.oidc.keys()) == {"example.com"}


# -------------------------------------------------------------------------
# Transaction log censoring
# -------------------------------------------------------------------------


class TestLogCensoring:
    def test_oidc_key_values_hidden(self):
        assert format_log_uuid(b"raw-key-material", "oidc.localhost.key") == "<hidden>"
        assert format_log_uuid("secret", "oidc.example.com.key") == "<hidden>"

    def test_oidc_key_path_component_visible(self):
        # The path component itself must stay visible ("oidc.<rp-id>.key = <hidden>")
        assert format_log_uuid("key", "oidc.localhost.key") is None

    def test_other_paths_unaffected(self):
        assert format_log_uuid("not-a-uuid", "oidc.localhost.clients") is None
        assert format_log_uuid("not-a-uuid", "config.realms") is None


# -------------------------------------------------------------------------
# Bootstrap caveat: admin credential is checked on the default realm
# -------------------------------------------------------------------------


class TestBootstrapCaveat:
    @pytest.mark.asyncio
    async def test_admin_without_credentials_gets_link(
        self, test_db: DB, realm_registry
    ):
        assert await check_admin_credentials() is True

    @pytest.mark.asyncio
    async def test_admin_with_default_realm_credential_ok(
        self, test_db: DB, realm_registry, test_user, test_credential
    ):
        assert await check_admin_credentials() is False

    @pytest.mark.asyncio
    async def test_admin_with_only_other_realm_credential_gets_link(
        self, test_db: DB, realm_registry, test_user
    ):
        """A passkey under a non-default realm does not satisfy the check."""
        cred = Credential.create(
            credential_id=os.urandom(32),
            user=test_user.uuid,
            aaguid=UUID(int=0),
            public_key=os.urandom(64),
            sign_count=0,
            rp_id="example.com",
        )
        create_credential(cred)
        assert await check_admin_credentials() is True