from __future__ import annotations

from dataclasses import replace
from types import SimpleNamespace

import pytest
from gateway.config import GatewayConfig, Platform, PlatformConfig
from gateway.pairing import PairingStore
from gateway.session import SessionSource


class TestWebVerificationAdmin:
    @pytest.fixture(autouse=True)
    def _setup_test_client(self, monkeypatch, _isolate_hermes_home):
        try:
            from starlette.testclient import TestClient
        except ImportError:
            pytest.skip("fastapi/starlette not installed")

        from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
        from gateway.user_verification import GatewayUserStore

        self.store = GatewayUserStore()
        app.state.gateway_user_store = self.store
        self.client = TestClient(app)
        self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN

    def test_force_bind_and_list_identities(self):
        bind = self.client.post(
            "/api/admin/verification/force-bind",
            json={
                "platform": "telegram",
                "external_user_id": "tg-web-1",
                "email": "staff4@bremen.com.tw",
                "user_name": "Web Tester",
            },
        )
        assert bind.status_code == 200
        identity = bind.json()["identity"]
        assert identity["state"] == "verified"
        assert identity["verified_email"] == "staff4@bremen.com.tw"

        listed = self.client.get("/api/admin/verification/identities")
        assert listed.status_code == 200
        items = listed.json()["items"]
        assert any(
            item["platform"] == "telegram"
            and item["external_user_id"] == "tg-web-1"
            and item["verified_email"] == "staff4@bremen.com.tw"
            for item in items
        )

        unbind = self.client.post(
            "/api/admin/verification/unbind",
            json={"platform": "telegram", "external_user_id": "tg-web-1"},
        )
        assert unbind.status_code == 200
        assert unbind.json()["identity"]["state"] == "new"
        assert unbind.json()["identity"]["verified_email"] is None

    def test_unbind_revokes_pairing_approval(self):
        store = PairingStore()
        store._approve_user("telegram", "tg-web-unbind", "Web Tester")
        assert store.is_approved("telegram", "tg-web-unbind") is True

        bind = self.client.post(
            "/api/admin/verification/force-bind",
            json={
                "platform": "telegram",
                "external_user_id": "tg-web-unbind",
                "email": "staff-unbind@bremen.com.tw",
                "user_name": "Web Tester",
            },
        )
        assert bind.status_code == 200

        unbind = self.client.post(
            "/api/admin/verification/unbind",
            json={"platform": "telegram", "external_user_id": "tg-web-unbind"},
        )
        assert unbind.status_code == 200
        assert unbind.json()["identity"]["state"] == "new"
        assert store.is_approved("telegram", "tg-web-unbind") is False

        audit = self.client.get("/api/admin/verification/audit", params={"limit": 20})
        assert audit.status_code == 200
        assert any(
            item["action"] == "unbind_identity"
            and item["platform"] == "telegram"
            and item["external_user_id"] == "tg-web-unbind"
            and item["details"].get("pairing_revoked") is True
            for item in audit.json()["items"]
        )

    def test_unbind_removes_runner_authorization_when_pairing_was_only_grant(self, monkeypatch):
        from gateway.run import GatewayRunner

        for key in (
            "TELEGRAM_ALLOWED_USERS",
            "TELEGRAM_GROUP_ALLOWED_USERS",
            "TELEGRAM_GROUP_ALLOWED_CHATS",
            "TELEGRAM_ALLOW_ALL_USERS",
            "GATEWAY_ALLOWED_USERS",
            "GATEWAY_ALLOW_ALL_USERS",
        ):
            monkeypatch.delenv(key, raising=False)

        external_user_id = "tg-web-authz"
        PairingStore()._approve_user("telegram", external_user_id, "Web Tester")

        bind = self.client.post(
            "/api/admin/verification/force-bind",
            json={
                "platform": "telegram",
                "external_user_id": external_user_id,
                "email": "staff-authz@bremen.com.tw",
                "user_name": "Web Tester",
            },
        )
        assert bind.status_code == 200

        runner = object.__new__(GatewayRunner)
        runner.config = GatewayConfig(platforms={Platform.TELEGRAM: PlatformConfig(enabled=True)})
        runner.adapters = {Platform.TELEGRAM: SimpleNamespace()}
        runner.pairing_store = PairingStore()

        source = SessionSource(
            platform=Platform.TELEGRAM,
            user_id=external_user_id,
            chat_id=external_user_id,
            user_name="Web Tester",
            chat_type="dm",
        )

        assert runner._is_user_authorized(source) is True

        unbind = self.client.post(
            "/api/admin/verification/unbind",
            json={"platform": "telegram", "external_user_id": external_user_id},
        )
        assert unbind.status_code == 200

        assert runner._is_user_authorized(source) is False

    @pytest.mark.parametrize(
        ("platform_enum", "platform_name", "external_user_id", "email"),
        [
            ("TELEGRAM", "telegram", "tg-web-resend", "staff-resend@journeys.com.tw"),
            ("line", "line", "line-web-resend", "line-resend@bremen.com.tw"),
        ],
    )
    def test_resend_endpoint_sends_verification_email(
        self,
        monkeypatch,
        platform_enum,
        platform_name,
        external_user_id,
        email,
    ):
        from gateway.config import Platform
        from gateway.session import SessionSource
        from gateway import user_verification as uv
        from hermes_cli import web_server

        source = SessionSource(
            platform=(getattr(Platform, platform_enum) if platform_enum.isupper() else Platform(platform_enum)),
            chat_id=external_user_id,
            user_id=external_user_id,
            user_name="Resend Tester",
            chat_type="dm",
        )
        decision = self.store.process_inbound_message(
            source,
            email,
            public_base_url="https://agent.example.com",
        )
        assert decision.action == "send_verification_email"

        with self.store._connect() as conn:
            conn.execute(
                "UPDATE verification_requests SET last_sent_at = datetime('now', '-2 minutes') WHERE platform = ? AND external_user_id = ?",
                (platform_name, external_user_id),
            )
            conn.commit()

        calls: list[tuple[str, str, str]] = []

        async def fake_send(to_email: str, *, token: str, verification_url: str, platform: str, external_user_id: str, pconfig=None):
            calls.append((to_email, token, verification_url))
            return {"success": True, "platform": platform, "chat_id": to_email}

        monkeypatch.setattr(uv, "send_verification_email", fake_send)
        monkeypatch.setattr(
            web_server,
            "_gateway_platform_config",
            lambda platform_id: (None, None, SimpleNamespace(extra={})),
        )

        resp = self.client.post(
            "/api/admin/verification/resend",
            json={"platform": platform_name, "external_user_id": external_user_id},
        )
        assert resp.status_code == 200
        body = resp.json()
        assert body["ok"] is True
        assert body["result"]["email"] == email
        assert body["result"]["token"]
        assert calls == [
            (
                email,
                body["result"]["token"],
                body["result"]["verification_url"],
            )
        ]

    def test_public_verify_endpoint_accepts_token_without_session_header(self):
        from gateway.config import Platform
        from gateway.session import SessionSource
        from hermes_cli import web_server

        source = SessionSource(
            platform=Platform.TELEGRAM,
            chat_id="tg-web-2",
            user_id="tg-web-2",
            user_name="Public Verify",
            chat_type="dm",
        )
        decision = self.store.process_inbound_message(
            source,
            "staff5@journeys.com.tw",
            public_base_url="https://agent.example.com",
        )

        prev_host = getattr(web_server.app.state, "bound_host", None)
        prev_port = getattr(web_server.app.state, "bound_port", None)
        prev_required = getattr(web_server.app.state, "auth_required", None)
        web_server.app.state.bound_host = "fly-app.fly.dev"
        web_server.app.state.bound_port = 443
        web_server.app.state.auth_required = True
        try:
            public_client = self.client.__class__(self.client.app, base_url="https://fly-app.fly.dev")
            resp = public_client.get(f"/api/verification/verify?token={decision.token}")
        finally:
            web_server.app.state.bound_host = prev_host
            web_server.app.state.bound_port = prev_port
            web_server.app.state.auth_required = prev_required
        assert resp.status_code == 200
        assert "驗證成功" in resp.text

    @pytest.mark.parametrize(
        ("platform_obj", "platform_name", "external_user_id", "email"),
        [
            (Platform.TELEGRAM, "telegram", "tg-state-machine", "tg-state@journeys.com.tw"),
            (Platform("line"), "line", "UstateMachine", "line-state@bremen.com.tw"),
        ],
    )
    def test_verification_state_machine_bind_handoff_unbind(self, monkeypatch, platform_obj, platform_name, external_user_id, email):
        from gateway.run import GatewayRunner
        from gateway.platform_registry import platform_registry, PlatformEntry
        from tests.gateway._plugin_adapter_loader import load_plugin_adapter

        for key in (
            "TELEGRAM_ALLOWED_USERS",
            "TELEGRAM_GROUP_ALLOWED_USERS",
            "TELEGRAM_GROUP_ALLOWED_CHATS",
            "TELEGRAM_ALLOW_ALL_USERS",
            "GATEWAY_ALLOWED_USERS",
            "GATEWAY_ALLOW_ALL_USERS",
            "LINE_ALLOWED_USERS",
            "LINE_ALLOWED_GROUPS",
            "LINE_ALLOWED_ROOMS",
            "LINE_ALLOW_ALL_USERS",
        ):
            monkeypatch.delenv(key, raising=False)

        if platform_name == "line":
            try:
                platform_registry.register(
                    PlatformEntry(
                        name="line",
                        label="LINE",
                        adapter_factory=lambda cfg: None,
                        check_fn=lambda: True,
                        allowed_users_env="LINE_ALLOWED_USERS",
                        allow_all_env="LINE_ALLOW_ALL_USERS",
                    )
                )
            except Exception:
                pass

        source = SessionSource(
            platform=platform_obj,
            chat_id=external_user_id,
            user_id=external_user_id,
            user_name="State Machine Tester",
            chat_type="dm",
        )

        decision = self.store.process_inbound_message(
            source,
            email,
            public_base_url="https://agent.example.com",
        )
        assert decision.action == "send_verification_email"
        assert decision.token
        assert self.store.is_verified_source(source) is False
        assert self.store.get_verified_email_for_source(source) is None

        verify = self.store.verify_token(decision.token)
        assert verify.success is True
        assert self.store.is_verified_source(source) is True
        assert self.store.get_verified_email_for_source(source) == email
        assert self.store.bound_email_matches_source(source, email) is True
        assert self.store.bound_email_matches_source(source, f"wrong+{email}") is False

        if platform_name == "telegram":
            runner = object.__new__(GatewayRunner)
            runner.config = GatewayConfig(platforms={Platform.TELEGRAM: PlatformConfig(enabled=True)})
            runner.adapters = {Platform.TELEGRAM: SimpleNamespace()}
            runner.pairing_store = PairingStore()
            assert runner._is_user_authorized(source) is False

            PairingStore()._approve_user(platform_name, external_user_id, "State Machine Tester")
            assert runner._is_user_authorized(source) is True
        else:
            line_mod = load_plugin_adapter("line")
            sent = {}

            async def _fake_email_send(_cfg, chat_id, message, **kwargs):
                sent["chat_id"] = chat_id
                sent["message"] = message
                sent["kwargs"] = kwargs
                return {"success": True}

            monkeypatch.setattr(line_mod, "_email_standalone_send", _fake_email_send)
            adapter = line_mod.LineAdapter(
                PlatformConfig(
                    enabled=True,
                    extra={"channel_access_token": "tok", "channel_secret": "sec"},
                )
            )
            adapter._client = SimpleNamespace(push=None, reply=None)

            payload = pytest.importorskip("asyncio").run(adapter._send_email_quota_fallback(external_user_id, "hello", mode="final"))
            assert payload is not None
            assert payload["email"] == email
            assert sent["chat_id"] == email

        unbind = self.client.post(
            "/api/admin/verification/unbind",
            json={"platform": platform_name, "external_user_id": external_user_id},
        )
        assert unbind.status_code == 200
        assert unbind.json()["identity"]["state"] == "new"
        assert self.store.is_verified_source(source) is False
        assert self.store.get_verified_email_for_source(source) is None
        assert self.store.bound_email_matches_source(source, email) is False

        if platform_name == "telegram":
            assert runner._is_user_authorized(source) is False
        else:
            payload_after = pytest.importorskip("asyncio").run(adapter._send_email_quota_fallback(external_user_id, "hello", mode="final"))
            assert payload_after is None

    def test_loopback_role_endpoint_defaults_to_admin(self):
        resp = self.client.get("/api/admin/verification/role")
        assert resp.status_code == 200
        assert resp.json()["role"] == "admin"

    def test_dashboard_role_assignment_round_trip(self):
        set_role = self.client.post(
            "/api/admin/verification/roles",
            json={"email": "viewer@example.com", "role": "viewer"},
        )
        assert set_role.status_code == 200
        assert set_role.json()["assignment"]["role"] == "viewer"

        listed = self.client.get("/api/admin/verification/roles")
        assert listed.status_code == 200
        assert any(item["email"] == "viewer@example.com" and item["role"] == "viewer" for item in listed.json()["items"])

    def test_quota_and_model_policy_round_trip(self):
        bind = self.client.post(
            "/api/admin/verification/force-bind",
            json={
                "platform": "telegram",
                "external_user_id": "tg-web-3",
                "email": "staff6@bremen.com.tw",
            },
        )
        principal_id = bind.json()["identity"]["principal_id"]

        quota = self.client.post(
            "/api/admin/quota-policies",
            json={
                "principal_id": principal_id,
                "daily_message_limit": 30,
                "daily_token_limit": 50000,
                "notes": "vip",
            },
        )
        assert quota.status_code == 200
        assert quota.json()["policy"]["daily_message_limit"] == 30

        model = self.client.post(
            "/api/admin/model-policies",
            json={
                "principal_id": principal_id,
                "allowed_models": ["openai/gpt-5", "anthropic/claude-sonnet-4"],
                "default_model": "openai/gpt-5",
            },
        )
        assert model.status_code == 200
        assert model.json()["policy"]["allowed_models"] == [
            "openai/gpt-5",
            "anthropic/claude-sonnet-4",
        ]

        self.store.increment_principal_usage(
            principal_id,
            input_tokens=210,
            output_tokens=90,
            message_count=3,
            model="openai/gpt-5",
        )

        quota_list = self.client.get("/api/admin/quota-policies")
        model_list = self.client.get("/api/admin/model-policies")
        usage_list = self.client.get("/api/admin/principal-usage")
        assert quota_list.status_code == 200
        assert model_list.status_code == 200
        assert usage_list.status_code == 200
        assert any(item["principal_id"] == principal_id for item in quota_list.json()["items"])
        assert any(item["principal_id"] == principal_id for item in model_list.json()["items"])
        usage_item = next(item for item in usage_list.json()["items"] if item["principal_id"] == principal_id)
        assert usage_item["email"] == "staff6@bremen.com.tw"
        assert usage_item["message_count"] == 3
        assert usage_item["total_tokens"] == 300
        assert usage_item["daily_token_limit"] == 50000

    def test_principals_and_audit_endpoints_round_trip(self):
        bind = self.client.post(
            "/api/admin/verification/force-bind",
            json={
                "platform": "line",
                "external_user_id": "line-web-principal",
                "email": "staff7@journeys.com.tw",
                "user_name": "Principal Tester",
            },
        )
        assert bind.status_code == 200
        principal_id = bind.json()["identity"]["principal_id"]

        principals = self.client.get("/api/admin/verification/principals")
        assert principals.status_code == 200
        principal = next(item for item in principals.json()["items"] if item["principal_id"] == principal_id)
        assert principal["email"] == "staff7@journeys.com.tw"
        assert any(
            identity["platform"] == "line"
            and identity["external_user_id"] == "line-web-principal"
            for identity in principal["identities"]
        )

        audit = self.client.get("/api/admin/verification/audit", params={"limit": 20})
        assert audit.status_code == 200
        assert any(
            item["action"] == "force_bind_identity"
            and item["platform"] == "line"
            and item["external_user_id"] == "line-web-principal"
            for item in audit.json()["items"]
        )
