from __future__ import annotations

from contextlib import contextmanager
import unittest
from unittest.mock import patch

from sqlalchemy import create_engine, event
from sqlalchemy.orm import Session

from govoplan_core.core.events import EventObjectRef, EventTenantRef, PlatformEvent
from govoplan_core.core.search import (
    SearchAuthorizationRequest,
    SearchBackfillRequest,
    SearchResourceReference,
)
from govoplan_core.db.base import Base
from govoplan_docs.backend.db.models import (
    SemanticDocumentationEntry,
    SemanticDocumentationRevision,
)
from govoplan_docs.backend.search_source import (
    PROVIDER_ID,
    RESOURCE_TYPE,
    SemanticDocumentationSearchSource,
)
from govoplan_docs.backend.semantic_service import (
    SemanticDocumentationError,
    current_revision,
    list_semantic_entries,
    prefetch_semantic_revisions,
    semantic_entry_payload,
)
from test_semantic_documentation import (
    _content,
    _principal,
    _Registry,
    _SubjectProvider,
)


@contextmanager
def select_queries(engine):
    statements: list[str] = []

    def observe(_connection, _cursor, statement, _parameters, _context, _many):
        if statement.lstrip().upper().startswith("SELECT"):
            statements.append(statement)

    event.listen(engine, "before_cursor_execute", observe)
    try:
        yield statements
    finally:
        event.remove(engine, "before_cursor_execute", observe)


class SemanticReadEfficiencyTests(unittest.TestCase):
    def setUp(self) -> None:
        self.engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(
            self.engine,
            tables=[
                SemanticDocumentationEntry.__table__,
                SemanticDocumentationRevision.__table__,
            ],
        )
        self.provider = _SubjectProvider()
        self.registry = _Registry(self.provider)
        self.principal = _principal("author")

    def tearDown(self) -> None:
        self.engine.dispose()

    def seed(self, count: int) -> None:
        with Session(self.engine) as session:
            for index in range(count):
                identifier = f"entry-{index:04}"
                session.add(
                    SemanticDocumentationEntry(
                        id=identifier,
                        tenant_id="tenant-1",
                        subject_stable_key="same-subject",
                        subject_module_id="forms",
                        subject_kind="form",
                        subject_id="form-1",
                        locale=f"de-{index:03}",
                        lifecycle_state="draft",
                        current_revision=2,
                        current_revision_id=f"draft-{index:04}",
                        published_revision_id=f"published-{index:04}",
                        created_by="author",
                        updated_by="author",
                    )
                )
                session.flush()
                for revision, state, title in (
                    (1, "published", f"Published {index}"),
                    (2, "draft", f"Unpublished secret {index}"),
                ):
                    session.add(
                        SemanticDocumentationRevision(
                            id=f"{state}-{index:04}",
                            tenant_id="tenant-1",
                            entry_id=identifier,
                            revision=revision,
                            lifecycle_state=state,
                            action="publish" if revision == 1 else "save",
                            change_reason="Fixture",
                            content=_content(title),
                            content_hash=f"hash-{state}-{index}",
                            authored_by="author",
                            search_text=title,
                        )
                    )
            session.commit()

    def test_reader_and_editor_batching_preserve_payloads_with_two_queries_for_forty_entries(
        self,
    ) -> None:
        self.seed(40)
        for editor in (False, True):
            with self.subTest(editor=editor):
                with (
                    Session(self.engine) as session,
                    select_queries(self.engine) as ordinary_queries,
                ):
                    entries = list_semantic_entries(session, self.principal)
                    ordinary = [
                        semantic_entry_payload(
                            session,
                            self.registry,
                            self.principal,
                            entry=entry,
                            editor=editor,
                        )
                        for entry in entries
                    ]
                with (
                    Session(self.engine) as session,
                    select_queries(self.engine) as batch_queries,
                ):
                    entries = list_semantic_entries(session, self.principal)
                    revisions = prefetch_semantic_revisions(
                        session, self.principal, entries=entries, editor=editor
                    )
                    batched = [
                        semantic_entry_payload(
                            session,
                            self.registry,
                            self.principal,
                            entry=entry,
                            editor=editor,
                            revisions=revisions,
                        )
                        for entry in entries
                    ]
                    if not editor:
                        self.assertTrue(
                            all(
                                item.lifecycle_state == "published"
                                for item in revisions.values()
                            )
                        )
                        self.assertTrue(all(item["pending_draft"] for item in batched))
                        self.assertTrue(
                            all(
                                "Unpublished" not in item["content"]["title"]
                                for item in batched
                            )
                        )
                self.assertEqual(ordinary, batched)
                self.assertEqual(81 if editor else 41, len(ordinary_queries))
                self.assertEqual(2, len(batch_queries))

    def test_revision_prefetch_chunks_large_collections(self) -> None:
        self.seed(401)
        with Session(self.engine) as session, select_queries(self.engine) as queries:
            entries = list_semantic_entries(session, self.principal)
            revisions = prefetch_semantic_revisions(
                session, self.principal, entries=entries, editor=False
            )
        self.assertEqual(401, len(revisions))
        self.assertEqual(
            3, len(queries), "One entry query and two bounded revision batches."
        )

    def test_search_authorization_batches_rows_without_weakening_owner_audience_or_revision_checks(
        self,
    ) -> None:
        self.seed(40)
        requests = tuple(
            SearchAuthorizationRequest(
                reference=SearchResourceReference(
                    tenant_id="tenant-1",
                    module_id="docs",
                    resource_type=RESOURCE_TYPE,
                    resource_id=f"entry-{index:04}",
                ),
                source_revision=f"published-{index:04}",
            )
            for index in range(40)
        )
        source = SemanticDocumentationSearchSource(self.registry)
        with Session(self.engine) as session, select_queries(self.engine) as queries:
            decisions = source.authorize(session, self.principal, requests=requests)
        self.assertTrue(all(decisions.values()))
        self.assertEqual(2, len(queries))
        self.provider.denied_accounts.add(self.principal.account_id)
        with Session(self.engine) as session:
            self.assertFalse(
                any(
                    source.authorize(
                        session, self.principal, requests=requests
                    ).values()
                )
            )
        self.provider.denied_accounts.clear()
        stale = SearchAuthorizationRequest(
            reference=requests[0].reference, source_revision="old-publication"
        )
        with Session(self.engine) as session:
            self.assertFalse(
                source.authorize(session, self.principal, requests=(stale,))[
                    stale.reference.key
                ]
            )

    def test_audience_denial_precedes_provider_work_and_foreign_entries_do_not_load_revisions(
        self,
    ) -> None:
        self.seed(1)
        with Session(self.engine) as session:
            revision = session.get(SemanticDocumentationRevision, "published-0000")
            revision.content = {
                **revision.content,
                "classification": "restricted",
                "audience": ["account:other"],
            }
            session.commit()
        with (
            Session(self.engine) as session,
            patch.object(
                self.provider, "resolve_subject", wraps=self.provider.resolve_subject
            ) as resolve,
        ):
            entry = session.get(SemanticDocumentationEntry, "entry-0000")
            self.assertIsNone(
                semantic_entry_payload(
                    session, self.registry, self.principal, entry=entry, editor=False
                )
            )
            resolve.assert_not_called()
            with select_queries(self.engine) as queries:
                self.assertIsNone(
                    semantic_entry_payload(
                        session,
                        self.registry,
                        _principal("other", tenant_id="tenant-other"),
                        entry=entry,
                        editor=False,
                    )
                )
            self.assertEqual([], queries)

    def test_cross_tenant_or_cross_entry_revision_references_fail_closed_in_reading_and_indexing(
        self,
    ) -> None:
        self.seed(2)
        source = SemanticDocumentationSearchSource(self.registry)
        publication_event = PlatformEvent(
            type="docs.semantic.published",
            module_id="docs",
            tenant=EventTenantRef(id="tenant-1"),
            resource=EventObjectRef(type=RESOURCE_TYPE, id="entry-0000"),
        )
        with Session(self.engine) as session:
            changes = source.index_changes_for_event(
                session, event=publication_event, delivery_key="valid-fixture"
            )
            self.assertEqual("upsert", changes[0].kind)
            self.assertEqual("published-0000", changes[0].document.source_revision)
        for corruption in ("tenant", "entry"):
            with self.subTest(corruption=corruption), Session(self.engine) as session:
                entry = session.get(SemanticDocumentationEntry, "entry-0000")
                revision = session.get(SemanticDocumentationRevision, "published-0000")
                if corruption == "tenant":
                    revision.tenant_id = "tenant-other"
                else:
                    revision.tenant_id = "tenant-1"
                    revision.entry_id = "entry-0001"
                    revision.revision = 3
                session.commit()
                self.assertIsNone(
                    semantic_entry_payload(
                        session,
                        self.registry,
                        self.principal,
                        entry=entry,
                        editor=False,
                    )
                )
                page = SemanticDocumentationSearchSource(self.registry).backfill(
                    session,
                    request=SearchBackfillRequest(
                        tenant_id="tenant-1",
                        provider_id=PROVIDER_ID,
                        resource_type=RESOURCE_TYPE,
                        rebuild_id="fixture",
                        limit=100,
                    ),
                )
                self.assertNotIn(
                    "entry-0000", {document.resource_id for document in page.documents}
                )
                changes = source.index_changes_for_event(
                    session,
                    event=publication_event,
                    delivery_key=f"invalid-{corruption}",
                )
                self.assertEqual("delete", changes[0].kind)
                self.assertIsNone(changes[0].document)

    def test_published_reader_does_not_depend_on_unavailable_draft_and_editor_still_validates_it(
        self,
    ) -> None:
        self.seed(1)
        with Session(self.engine) as session:
            entry = session.get(SemanticDocumentationEntry, "entry-0000")
            draft = session.get(
                SemanticDocumentationRevision, entry.current_revision_id
            )
            draft.tenant_id = "tenant-other"
            session.commit()
            self.assertEqual(
                "Published 0",
                semantic_entry_payload(
                    session, self.registry, self.principal, entry=entry, editor=False
                )["content"]["title"],
            )
            with self.assertRaises(SemanticDocumentationError):
                current_revision(session, entry)

    def test_unpublished_revision_pointer_cannot_promote_a_draft_into_reading_or_search(
        self,
    ) -> None:
        self.seed(1)
        with Session(self.engine) as session:
            entry = session.get(SemanticDocumentationEntry, "entry-0000")
            entry.published_revision_id = entry.current_revision_id
            session.commit()
            revisions = prefetch_semantic_revisions(
                session, self.principal, entries=(entry,), editor=False
            )
            self.assertEqual({}, revisions)
            self.assertIsNone(
                semantic_entry_payload(
                    session,
                    self.registry,
                    self.principal,
                    entry=entry,
                    editor=False,
                    revisions=revisions,
                )
            )
            self.assertIsNone(
                semantic_entry_payload(
                    session, self.registry, self.principal, entry=entry, editor=False
                )
            )
            page = SemanticDocumentationSearchSource(self.registry).backfill(
                session,
                request=SearchBackfillRequest(
                    tenant_id="tenant-1",
                    provider_id=PROVIDER_ID,
                    resource_type=RESOURCE_TYPE,
                    rebuild_id="fixture",
                    limit=100,
                ),
            )
            self.assertEqual((), page.documents)
            changes = SemanticDocumentationSearchSource(
                self.registry
            ).index_changes_for_event(
                session,
                delivery_key="draft-fixture",
                event=PlatformEvent(
                    type="docs.semantic.published",
                    module_id="docs",
                    tenant=EventTenantRef(id="tenant-1"),
                    resource=EventObjectRef(type=RESOURCE_TYPE, id=entry.id),
                ),
            )
            self.assertEqual("delete", changes[0].kind)
            self.assertIsNone(changes[0].document)


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