from __future__ import annotations

import unittest
from datetime import UTC, datetime, timedelta
from types import SimpleNamespace

from sqlalchemy import create_engine, select
from sqlalchemy.orm import sessionmaker

from govoplan_core.auth import ApiPrincipal
from govoplan_core.core.access import PrincipalRef
from govoplan_core.core.change_sequence import ChangeSequenceEntry
from govoplan_core.core.concurrency import RevisionConflictError
from govoplan_core.core.tasks import (
    TaskCreateCommand,
    WorkAssignmentRef,
    WorkItemQuery,
    WorkSourceRef,
)
from govoplan_core.db.base import Base
from govoplan_tasks.backend.db.models import TaskAssignment, TaskItem
from govoplan_tasks.backend.manifest import get_manifest
from govoplan_tasks.backend.service import (
    ADMIN_SCOPE,
    READ_SCOPE,
    WRITE_SCOPE,
    SqlTaskService,
    TaskConflict,
    TaskForbidden,
    TaskNotFound,
)


class TaskServiceTests(unittest.TestCase):
    def setUp(self) -> None:
        self.engine = create_engine("sqlite:///:memory:")
        self.tables = (
            ChangeSequenceEntry.__table__,
            TaskItem.__table__,
            TaskAssignment.__table__,
        )
        Base.metadata.create_all(self.engine, tables=self.tables)
        self.Session = sessionmaker(bind=self.engine)
        self.service = SqlTaskService()

    def tearDown(self) -> None:
        Base.metadata.drop_all(self.engine, tables=reversed(self.tables))
        self.engine.dispose()

    @staticmethod
    def principal(
        account_id: str,
        *,
        tenant_id: str = "tenant-1",
        scopes: set[str] | None = None,
        group_ids: set[str] | None = None,
        role_ids: set[str] | None = None,
        function_assignment_ids: set[str] | None = None,
    ) -> ApiPrincipal:
        return ApiPrincipal(
            principal=PrincipalRef(
                account_id=account_id,
                membership_id=f"membership-{account_id}",
                tenant_id=tenant_id,
                scopes=frozenset(
                    {READ_SCOPE, WRITE_SCOPE} if scopes is None else scopes
                ),
                group_ids=frozenset(group_ids or ()),
                role_ids=frozenset(role_ids or ()),
                function_assignment_ids=frozenset(function_assignment_ids or ()),
            ),
            account=SimpleNamespace(id=account_id),
            user=SimpleNamespace(id=f"membership-{account_id}"),
        )

    @staticmethod
    def command(
        *,
        idempotency_key: str = "request-1",
        title: str = "Review the submission",
        assignments: tuple[WorkAssignmentRef, ...] | None = None,
    ) -> TaskCreateCommand:
        return TaskCreateCommand(
            tenant_id="tenant-1",
            title=title,
            summary="Resolve the governed handoff.",
            priority="high",
            due_at=datetime.now(UTC) + timedelta(days=2),
            required_action="Review and decide",
            action_url="/cases/case-1",
            assignments=assignments
            or (WorkAssignmentRef(kind="account", id="account-1"),),
            sources=(
                WorkSourceRef(
                    module_id="cases",
                    resource_type="case",
                    resource_id="case-1",
                    revision="3",
                    url="/cases/case-1",
                ),
            ),
            provenance={"workflow_instance_id": "workflow-1"},
            idempotency_key=idempotency_key,
        )

    def test_replayed_create_is_idempotent_and_mismatch_is_rejected(self) -> None:
        principal = self.principal("account-1")
        with self.Session() as session:
            command = self.command()
            first = self.service.create_task(session, principal, command=command)
            replay = self.service.create_task(session, principal, command=command)
            self.assertEqual(first.id, replay.id)
            self.assertEqual(1, len(session.scalars(select(TaskItem)).all()))
            self.assertEqual(
                1,
                len(session.scalars(select(ChangeSequenceEntry)).all()),
            )

            with self.assertRaises(TaskConflict):
                self.service.create_task(
                    session,
                    principal,
                    command=self.command(title="A different task"),
                )

    def test_visibility_uses_typed_assignments_and_fails_closed(self) -> None:
        creator = self.principal(
            "account-1",
            scopes={READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE},
        )
        with self.Session() as session:
            account_task = self.service.create_task(
                session,
                creator,
                command=self.command(idempotency_key="account"),
            )
            group_task = self.service.create_task(
                session,
                creator,
                command=self.command(
                    idempotency_key="group",
                    assignments=(WorkAssignmentRef(kind="group", id="group-1"),),
                ),
            )
            function_assignment_task = self.service.create_task(
                session,
                creator,
                command=self.command(
                    idempotency_key="function-assignment",
                    assignments=(
                        WorkAssignmentRef(
                            kind="function_assignment",
                            id="assignment-1",
                        ),
                    ),
                ),
            )
            session.flush()

            group_reader = self.principal(
                "account-2",
                group_ids={"group-1"},
                function_assignment_ids={"assignment-1"},
            )
            page = self.service.list_items(
                session,
                group_reader,
                query=WorkItemQuery(tenant_id="tenant-1"),
            )
            self.assertEqual(
                {group_task.id, function_assignment_task.id},
                {item.id for item in page.items},
            )
            self.assertNotIn(account_task.id, {item.id for item in page.items})

            hidden = self.principal("account-3")
            empty = self.service.list_items(
                session,
                hidden,
                query=WorkItemQuery(tenant_id="tenant-1"),
            )
            self.assertEqual(0, empty.total)

    def test_admin_can_read_all_tenant_work_but_not_another_tenant(self) -> None:
        admin = self.principal(
            "admin",
            scopes={READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE},
        )
        with self.Session() as session:
            task = self.service.create_task(
                session,
                admin,
                command=self.command(
                    assignments=(WorkAssignmentRef(kind="account", id="someone-else"),),
                ),
            )
            page = self.service.list_items(
                session,
                admin,
                query=WorkItemQuery(tenant_id="tenant-1"),
            )
            self.assertEqual([task.id], [item.id for item in page.items])

            with self.assertRaises(TaskForbidden):
                self.service.list_items(
                    session,
                    admin,
                    query=WorkItemQuery(tenant_id="tenant-2"),
                )

    def test_capability_enforces_read_and_write_scopes(self) -> None:
        without_scopes = self.principal("account-1", scopes=set())
        with self.Session() as session:
            with self.assertRaises(TaskForbidden):
                self.service.create_task(
                    session,
                    without_scopes,
                    command=self.command(),
                )
            with self.assertRaises(TaskForbidden):
                self.service.list_items(
                    session,
                    without_scopes,
                    query=WorkItemQuery(tenant_id="tenant-1"),
                )

    def test_transition_uses_revision_and_records_exact_history(self) -> None:
        principal = self.principal("account-1")
        with self.Session() as session:
            created = self.service.create_task(
                session,
                principal,
                command=self.command(),
            )
            started = self.service.transition_task(
                session,
                principal,
                tenant_id="tenant-1",
                task_id=created.id,
                action="start",
                expected_revision=1,
                comment="Taking responsibility",
            )
            self.assertEqual("in_progress", started.status)
            self.assertEqual("2", started.revision)
            history = started.metadata["transition_history"]
            self.assertEqual("open", history[0]["from_status"])
            self.assertEqual("in_progress", history[0]["to_status"])

            with self.assertRaises(RevisionConflictError):
                self.service.transition_task(
                    session,
                    principal,
                    tenant_id="tenant-1",
                    task_id=created.id,
                    action="complete",
                    expected_revision=1,
                )

    def test_invalid_transition_and_hidden_task_are_indistinguishable(self) -> None:
        owner = self.principal("account-1")
        other = self.principal("account-2")
        with self.Session() as session:
            task = self.service.create_task(session, owner, command=self.command())
            with self.assertRaises(TaskConflict):
                self.service.transition_task(
                    session,
                    owner,
                    tenant_id="tenant-1",
                    task_id=task.id,
                    action="reopen",
                    expected_revision=1,
                )
            with self.assertRaises(TaskNotFound):
                self.service.get_task(
                    session,
                    other,
                    tenant_id="tenant-1",
                    task_id=task.id,
                )


class TaskManifestTests(unittest.TestCase):
    def test_manifest_exposes_static_docs_and_work_provider(self) -> None:
        manifest = get_manifest()
        self.assertEqual("tasks", manifest.id)
        self.assertEqual("tasks.explicit", manifest.work_item_providers[0].id)
        self.assertTrue(manifest.documentation)
        self.assertEqual("@govoplan/tasks-webui", manifest.frontend.package_name)


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