from __future__ import annotations

import shutil
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

from fastapi import APIRouter, Request
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

from govoplan_core.audit.logging import audit_event, audit_operation_context
from govoplan_core.core.events import (
    DurableEventConsumer,
    EventActorRef,
    EventBus,
    EventObjectRef,
    EventTenantRef,
    PlatformEvent,
    current_event_trace,
    emit_platform_event,
    event_bus_context,
    event_context,
    normalize_trace_id,
)
from govoplan_core.core.modules import ModuleContext
from govoplan_core.core.registry import PlatformRegistry
from govoplan_core.core.runtime import clear_runtime, configure_runtime
from govoplan_core.db.base import Base
from govoplan_core.server.fastapi import create_govoplan_app
from govoplan_core.server.registry import build_platform_registry
from tests.db_isolation import temporary_database


def _configure_audit_runtime() -> None:
    registry = build_platform_registry(("access", "audit"))
    context = ModuleContext(registry=registry, settings=object())
    registry.configure_capability_context(context)
    configure_runtime(context)


class CoreEventTests(unittest.TestCase):
    def test_durable_consumer_requires_policy_for_classified_events(self) -> None:
        with self.assertRaisesRegex(ValueError, "policy decision"):
            DurableEventConsumer(
                consumer_id="workflow.triggers.v1",
                classifications=frozenset({"restricted"}),
                handler=lambda _event, _delivery_key: None,
            )

        consumer = DurableEventConsumer(
            consumer_id="workflow.triggers.v1",
            event_types=frozenset({"case.changed"}),
            classifications=frozenset(
                {"internal", "confidential"}
            ),
            policy_decision_ref="policy:decision:1",
            handler=lambda _event, _delivery_key: None,
        )
        accepted = PlatformEvent(
            type="case.changed",
            module_id="cases",
            classification="confidential",
            event_id="event-1",
        )
        rejected = PlatformEvent(
            type="case.deleted",
            module_id="cases",
            classification="confidential",
        )

        self.assertTrue(consumer.accepts(accepted))
        self.assertFalse(consumer.accepts(rejected))
        self.assertEqual(
            "event-1:workflow.triggers.v1",
            consumer.delivery_key(accepted),
        )

    def test_fallback_event_waits_for_outer_commit_after_savepoint(self) -> None:
        engine = create_engine("sqlite:///:memory:")
        seen: list[PlatformEvent] = []
        bus = EventBus()
        bus.subscribe("*", seen.append)
        try:
            with Session(engine) as session, event_bus_context(bus):
                with session.begin():
                    with session.begin_nested():
                        emit_platform_event(
                            session,
                            PlatformEvent(type="nested.created", module_id="core"),
                        )
                    self.assertEqual([], seen)
                self.assertEqual(
                    ["nested.created"],
                    [event.type for event in seen],
                )
        finally:
            engine.dispose()

    def test_fallback_event_discards_only_rolled_back_savepoint(self) -> None:
        engine = create_engine("sqlite:///:memory:")
        seen: list[PlatformEvent] = []
        bus = EventBus()
        bus.subscribe("*", seen.append)
        try:
            with Session(engine) as session, event_bus_context(bus):
                with session.begin():
                    emit_platform_event(
                        session,
                        PlatformEvent(type="outer.created", module_id="core"),
                    )
                    savepoint = session.begin_nested()
                    emit_platform_event(
                        session,
                        PlatformEvent(type="nested.created", module_id="core"),
                    )
                    savepoint.rollback()
                    self.assertEqual([], seen)
                self.assertEqual(
                    ["outer.created"],
                    [event.type for event in seen],
                )
        finally:
            engine.dispose()

    def test_event_bus_adds_trace_ids_and_propagates_causation_to_nested_events(self) -> None:
        bus = EventBus()
        seen: list[PlatformEvent] = []

        def parent_handler(event: PlatformEvent) -> None:
            seen.append(event)
            bus.publish(PlatformEvent(type="child.ready", module_id="demo"))

        bus.subscribe("parent.ready", parent_handler)
        bus.subscribe("child.ready", seen.append)

        bus.publish(PlatformEvent(type="parent.ready", module_id="demo", correlation_id="corr-1"))

        self.assertEqual(2, len(seen))
        parent, child = seen
        self.assertEqual("corr-1", parent.correlation_id)
        self.assertIsNone(parent.causation_id)
        self.assertEqual("corr-1", child.correlation_id)
        self.assertEqual(parent.event_id, child.causation_id)

    def test_platform_event_serializes_governed_envelope_fields(self) -> None:
        event = PlatformEvent(
            type="case.permit.updated",
            module_id="cases",
            payload={"status": "approved"},
            event_id="event-1",
            correlation_id="corr-1",
            causation_id="cause-1",
            actor=EventActorRef(type="user", id="user-1", label="Case worker"),
            tenant=EventTenantRef(id="tenant-1", slug="city", label="City Office"),
            subject=EventObjectRef(type="person", id="person-1", label="Applicant"),
            resource=EventObjectRef(type="permit", id="permit-1", label="Permit"),
            classification="confidential",
        )

        payload = event.to_dict()

        self.assertEqual("case.permit.updated", payload["type"])
        self.assertEqual("cases", payload["module_id"])
        self.assertEqual({"status": "approved"}, payload["payload"])
        self.assertEqual("event-1", payload["event_id"])
        self.assertEqual("corr-1", payload["correlation_id"])
        self.assertEqual("cause-1", payload["causation_id"])
        self.assertEqual({"type": "user", "id": "user-1", "label": "Case worker"}, payload["actor"])
        self.assertEqual({"id": "tenant-1", "slug": "city", "label": "City Office"}, payload["tenant"])
        self.assertEqual({"type": "person", "id": "person-1", "label": "Applicant"}, payload["subject"])
        self.assertEqual({"type": "permit", "id": "permit-1", "label": "Permit"}, payload["resource"])
        self.assertEqual("confidential", payload["classification"])

    def test_event_bus_preserves_envelope_fields_when_adding_trace(self) -> None:
        bus = EventBus()
        seen: list[PlatformEvent] = []
        event = PlatformEvent(
            type="files.file.uploaded",
            module_id="files",
            actor=EventActorRef(type="api_key", id="key-1"),
            tenant=EventTenantRef(id="tenant-1"),
            subject=EventObjectRef(type="file", id="file-1"),
            resource=EventObjectRef(type="folder", id="folder-1"),
            classification="restricted",
        )

        bus.subscribe("files.file.uploaded", seen.append)
        bus.publish(event)

        self.assertEqual(1, len(seen))
        traced = seen[0]
        self.assertIsNotNone(traced.correlation_id)
        self.assertEqual(event.actor, traced.actor)
        self.assertEqual(event.tenant, traced.tenant)
        self.assertEqual(event.subject, traced.subject)
        self.assertEqual(event.resource, traced.resource)
        self.assertEqual("restricted", traced.classification)

    def test_request_correlation_middleware_sets_event_context_and_response_header(self) -> None:
        router = APIRouter()

        @router.get("/trace")
        def trace() -> dict[str, str | None]:
            active = current_event_trace()
            return {"correlation_id": active.correlation_id if active else None}

        app = create_govoplan_app(
            title="event test",
            version="test",
            registry=PlatformRegistry(),
            api_router=router,
        )

        with TestClient(app) as client:
            response = client.get("/trace", headers={"X-Request-ID": "request-123"})

        self.assertEqual(200, response.status_code, response.text)
        self.assertEqual("request-123", response.json()["correlation_id"])
        self.assertEqual("request-123", response.headers["X-Correlation-ID"])

    def test_app_rejects_wildcard_cors_with_credentials(self) -> None:
        with self.assertRaisesRegex(ValueError, "wildcard origin"):
            create_govoplan_app(
                title="event test",
                version="test",
                registry=PlatformRegistry(),
                cors_origins=("*",),
            )

    def test_app_sets_safe_default_browser_headers_and_https_hsts(self) -> None:
        with patch.dict(
            "os.environ",
            {"APP_ENV": "test", "GOVOPLAN_HTTP_HSTS_SECONDS": "86400"},
        ):
            app = create_govoplan_app(
                title="security header test",
                version="test",
                registry=PlatformRegistry(),
            )

        with TestClient(app, base_url="https://govoplan.example.test") as client:
            response = client.get("/health")

        self.assertEqual("nosniff", response.headers["X-Content-Type-Options"])
        self.assertEqual("DENY", response.headers["X-Frame-Options"])
        self.assertEqual("strict-origin-when-cross-origin", response.headers["Referrer-Policy"])
        self.assertIn("frame-ancestors 'none'", response.headers["Content-Security-Policy"])
        self.assertEqual("max-age=86400", response.headers["Strict-Transport-Security"])

    def test_slow_request_log_excludes_query_headers_and_cookies(self) -> None:
        router = APIRouter()

        @router.get("/slow")
        def slow() -> dict[str, bool]:
            return {"ok": True}

        with patch("govoplan_core.server.fastapi._slow_request_threshold_ms", return_value=0.0001):
            app = create_govoplan_app(
                title="request log redaction test",
                version="test",
                registry=PlatformRegistry(),
                api_router=router,
            )

        with self.assertLogs("govoplan.request", level="WARNING") as captured:
            with TestClient(app) as client:
                response = client.get(
                    "/slow?token=query-secret",
                    headers={
                        "Authorization": "Bearer header-secret",
                        "Cookie": "govoplan_session=cookie-secret",
                    },
                )

        self.assertEqual(200, response.status_code, response.text)
        rendered = "\n".join(captured.output)
        self.assertIn("path=/slow", rendered)
        self.assertNotIn("query-secret", rendered)
        self.assertNotIn("header-secret", rendered)
        self.assertNotIn("cookie-secret", rendered)

    def test_production_app_startup_rejects_an_unvalidated_environment(self) -> None:
        with patch.dict("os.environ", {"APP_ENV": "production"}, clear=True), self.assertRaisesRegex(
            RuntimeError,
            "GovOPlaN configuration validation: FAILED",
        ):
            create_govoplan_app(
                title="unsafe production test",
                version="test",
                registry=PlatformRegistry(),
            )

    def test_app_rejects_request_bodies_above_deployment_limit(self) -> None:
        router = APIRouter()

        @router.post("/body")
        async def body(request: Request):
            return {"size": len(await request.body())}

        with patch.dict("os.environ", {"GOVOPLAN_HTTP_MAX_REQUEST_BODY_BYTES": "10"}):
            app = create_govoplan_app(
                title="request body limit test",
                version="test",
                registry=PlatformRegistry(),
                api_router=router,
            )

        with TestClient(app) as client:
            response = client.post("/body", content=b"12345678901")
            streamed = client.post("/body", content=(chunk for chunk in (b"123456", b"78901")))

        self.assertEqual(413, response.status_code, response.text)
        self.assertIn("10 bytes", response.json()["detail"])
        self.assertEqual(413, streamed.status_code, streamed.text)

    def test_app_enforces_configured_trusted_hosts(self) -> None:
        with patch.dict("os.environ", {"GOVOPLAN_TRUSTED_HOSTS": "govoplan.example.test,*.internal.test"}):
            app = create_govoplan_app(
                title="trusted host test",
                version="test",
                registry=PlatformRegistry(),
            )

        with TestClient(app, base_url="https://govoplan.example.test") as client:
            allowed = client.get("/health")
            rejected = client.get("https://attacker.example.test/health")

        self.assertEqual(200, allowed.status_code, allowed.text)
        self.assertEqual(400, rejected.status_code, rejected.text)

    def test_audit_event_persists_trace_details_from_event_context(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-audit-trace-"))
        try:
            with temporary_database(f"sqlite:///{root / 'audit.db'}") as database:
                _configure_audit_runtime()
                from govoplan_access.backend.db import models as access_models  # noqa: F401
                from govoplan_admin.backend.db import models as admin_models  # noqa: F401
                from govoplan_audit.backend.db import models as audit_models  # noqa: F401
                from govoplan_tenancy.backend.db import models as tenancy_models  # noqa: F401

                Base.metadata.create_all(bind=database.engine)
                with database.session() as session:
                    with event_context(correlation_id="corr-1", causation_id="cause-1"):
                        item = audit_event(
                            session,
                            tenant_id=None,
                            scope="system",
                            action="demo.changed",
                            object_type="demo",
                            object_id="demo-1",
                            details={"password": "secret"},
                        )

                    self.assertEqual({"correlation_id": "corr-1", "causation_id": "cause-1"}, item.details["_trace"])
                    self.assertEqual("<redacted>", item.details["password"])
        finally:
            clear_runtime()
            shutil.rmtree(root, ignore_errors=True)

    def test_audit_event_publishes_governed_platform_event(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-audit-event-"))
        try:
            with temporary_database(f"sqlite:///{root / 'audit.db'}") as database:
                _configure_audit_runtime()
                from govoplan_access.backend.db import models as access_models  # noqa: F401
                from govoplan_admin.backend.db import models as admin_models  # noqa: F401
                from govoplan_audit.backend.db import models as audit_models  # noqa: F401
                from govoplan_tenancy.backend.db import models as tenancy_models  # noqa: F401

                Base.metadata.create_all(bind=database.engine)
                seen: list[PlatformEvent] = []
                bus = EventBus()
                bus.subscribe("*", seen.append)
                with database.session() as session:
                    with event_bus_context(bus), event_context(correlation_id="corr-1"):
                        item = audit_event(
                            session,
                            tenant_id="tenant-1",
                            user_id="user-1",
                            scope="tenant",
                            action="user.updated",
                            object_type="user",
                            object_id="user-2",
                            details={"password": "secret", "field": "display_name"},
                        )
                        session.commit()
                        from govoplan_audit.backend.outbox import SqlAuditOutbox

                        SqlAuditOutbox().dispatch_pending(session)
                        session.commit()

                action_events = [event for event in seen if event.type == "user.updated"]
                self.assertEqual(1, len(action_events))
                event = action_events[0]
                self.assertEqual("access", event.module_id)
                self.assertEqual("corr-1", event.correlation_id)
                self.assertEqual(EventActorRef(type="user", id="user-1"), event.actor)
                self.assertEqual(EventTenantRef(id="tenant-1"), event.tenant)
                self.assertEqual(EventObjectRef(type="user", id="user-2"), event.resource)
                self.assertEqual(item.id, event.payload["audit_log_id"])
                self.assertEqual("<redacted>", event.payload["details"]["password"])
        finally:
            clear_runtime()
            shutil.rmtree(root, ignore_errors=True)

    def test_audit_events_map_existing_module_action_prefixes(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-audit-module-events-"))
        try:
            with temporary_database(f"sqlite:///{root / 'audit.db'}") as database:
                _configure_audit_runtime()
                from govoplan_access.backend.db import models as access_models  # noqa: F401
                from govoplan_admin.backend.db import models as admin_models  # noqa: F401
                from govoplan_audit.backend.db import models as audit_models  # noqa: F401
                from govoplan_tenancy.backend.db import models as tenancy_models  # noqa: F401

                Base.metadata.create_all(bind=database.engine)
                seen: list[PlatformEvent] = []
                bus = EventBus()
                bus.subscribe("*", seen.append)
                cases = {
                    "user.created": "access",
                    "tenant.created": "tenancy",
                    "privacy_retention.policy_updated": "policy",
                    "files.connector.imported": "files",
                    "campaign.created": "campaigns",
                    "report.email_sent": "campaigns",
                }
                with database.session() as session:
                    with event_bus_context(bus):
                        for action in cases:
                            audit_event(
                                session,
                                tenant_id="tenant-1",
                                user_id="user-1",
                                action=action,
                                object_type="demo",
                                object_id=action,
                            )
                        session.commit()
                        from govoplan_audit.backend.outbox import SqlAuditOutbox

                        SqlAuditOutbox().dispatch_pending(session)
                        session.commit()

                by_type = {event.type: event.module_id for event in seen if event.type in cases}
                self.assertEqual(cases, by_type)
        finally:
            clear_runtime()
            shutil.rmtree(root, ignore_errors=True)

    def test_audit_operation_context_keeps_lifecycle_ids_and_sanitizes_details(self) -> None:
        payload = audit_operation_context(
            module_id="mail",
            request_id="request-1",
            run_id="run-1",
            outcome="queued",
            trace={"correlation_id": " corr-1 ", "causation_id": "cause-1", "token": "secret"},
            options={"migrate_database": True, "api_key": "secret"},
            empty=None,
        )

        self.assertEqual("mail", payload["module_id"])
        self.assertEqual("request-1", payload["request_id"])
        self.assertEqual("run-1", payload["run_id"])
        self.assertEqual("queued", payload["outcome"])
        self.assertEqual({"correlation_id": "corr-1", "causation_id": "cause-1"}, payload["_trace"])
        self.assertEqual("<redacted>", payload["options"]["api_key"])
        self.assertEqual("<redacted>", audit_operation_context(options={"bearer_token": "secret"})["options"]["bearer_token"])
        self.assertEqual("<redacted>", audit_operation_context(options={"clientSecret": "secret"})["options"]["clientSecret"])
        self.assertNotIn("token", payload["_trace"])
        self.assertNotIn("empty", payload)

    def test_trace_ids_are_normalized_for_headers_and_audit_input(self) -> None:
        self.assertEqual("abc-123", normalize_trace_id(" abc-123 "))
        self.assertIsNone(normalize_trace_id("../bad"))
        self.assertIsNone(normalize_trace_id("x" * 129))


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