294 lines
10 KiB
Python
294 lines
10 KiB
Python
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()
|