"""WorkItem platform service — test suite. Coverage -------- [models] 1 WorkItem model fields and __str__ 2 WorkItemTimeline model creation 3 WorkItem soft-delete (no SoftDelete mixin, but verify cascade) [services — WorkItemService] 4 create_work_item — happy path 5 create_work_item — creates timeline entry 6 complete_work_item — happy path 7 complete_work_item — raises on completed 8 complete_work_item — raises on cancelled 9 cancel_work_item — happy path 10 cancel_work_item — raises on completed 11 reassign_work_item — happy path 12 reassign_work_item — raises on completed 13 reassign_work_item — updates assigned_to_user 14 get_timeline — returns entries in reverse order 15 expire_work_items — marks past-due items as expired [services — backward compat] 16 upsert_workbox_item — creates new item 17 upsert_workbox_item — idempotent update 18 remove_workbox_item — removes matching items 19 get_dashboard_counts — returns expected keys 20 refresh_user_workbox — returns stats dict [API — auth & tenant] 21 GET /workitems/ returns 401 without auth 22 GET /workitems/ returns 400 without X-Tenant header [API — CRUD] 23 POST /workitems/ creates WorkItem 24 POST /workitems/ returns 201 on create 25 GET /workitems/{public_id}/ returns detail with timeline 26 POST /workitems/{public_id}/complete/ 27 POST /workitems/{public_id}/cancel/ 28 POST /workitems/{public_id}/reassign/ [permissions] 29 WorkItem permissions registered [events] 30 workitem_created event registered 31 workitem_completed event registered """ from __future__ import annotations import pytest from django.utils import timezone from simorgh.apps.workbox.models import ( WorkboxItemStatus, WorkboxItemType, WorkItem, WorkItemTimeline, ) from simorgh.apps.workbox.services import ( WorkItemService, get_dashboard_counts, refresh_user_workbox, remove_workbox_item, upsert_workbox_item, ) # ── Models ───────────────────────────────────────────────────────────────── @pytest.mark.django_db class TestWorkItemModel: """Model fields, __str__, and basic persistence.""" def test_create_basic(self, tenant_acme, acme_tree): wi = WorkItem.objects.create( tenant=tenant_acme, organization_node=acme_tree["root"], item_type=WorkboxItemType.TASK, title="Review Q4 report", ) assert wi.public_id is not None assert str(wi) == "Review Q4 report" assert wi.status == WorkboxItemStatus.PENDING assert wi.priority == "medium" def test_all_item_types(self, tenant_acme, acme_tree): for t in WorkboxItemType.values: wi = WorkItem.objects.create( tenant=tenant_acme, organization_node=acme_tree["root"], item_type=t, title=f"Item of type {t}", ) assert wi.item_type == t def test_all_statuses(self, tenant_acme, acme_tree): for s in WorkboxItemStatus.values: wi = WorkItem.objects.create( tenant=tenant_acme, organization_node=acme_tree["root"], item_type=WorkboxItemType.CUSTOM, title=f"Status {s}", status=s, ) assert wi.status == s def test_assigned_fields(self, tenant_acme, acme_tree, alice): wi = WorkItem.objects.create( tenant=tenant_acme, organization_node=acme_tree["root"], item_type=WorkboxItemType.ASSIGNMENT, title="Assignment test", assigned_to_user=alice, assigned_to_role_id=None, assigned_to_unit=acme_tree["eu"], ) assert wi.assigned_to_user == alice assert wi.assigned_to_unit == acme_tree["eu"] def test_source_fields(self, tenant_acme, acme_tree): wi = WorkItem.objects.create( tenant=tenant_acme, organization_node=acme_tree["root"], item_type=WorkboxItemType.APPROVAL, title="Approval test", source_type="approval", source_entity="approval_engine.request", source_id="abc-123", source_url="/approvals/abc-123/", ) assert wi.source_type == "approval" assert wi.source_entity == "approval_engine.request" @pytest.mark.django_db class TestWorkItemTimelineModel: """Timeline records.""" def test_create_timeline_entry(self, tenant_acme, acme_tree, alice): wi = WorkItem.objects.create( tenant=tenant_acme, organization_node=acme_tree["root"], item_type=WorkboxItemType.TASK, title="T1", ) entry = WorkItemTimeline.objects.create( work_item=wi, event="created", to_status=wi.status, actor=alice, note="Initial creation", ) assert entry.event == "created" assert entry.actor == alice assert str(entry).startswith(f"WorkItem {wi.pk}") # ── WorkItemService ───────────────────────────────────────────────────────── @pytest.mark.django_db class TestWorkItemService: """Canonical WorkItem CRUD operations.""" def test_create_work_item(self, tenant_acme, acme_tree, alice): wi = WorkItemService.create_work_item( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, item_type="task", title="Deploy v2.1", summary="Deploy the next release", description="Full deployment procedure", assigned_to_user_id=alice.pk, priority="high", source_type="task", source_entity="tasks.task", source_id="uuid-1", ) assert wi.title == "Deploy v2.1" assert wi.summary == "Deploy the next release" assert wi.assigned_to_user == alice assert wi.priority == "high" assert wi.status == WorkboxItemStatus.PENDING assert wi.created_at is not None def test_create_work_item_creates_timeline(self, tenant_acme, acme_tree): wi = WorkItemService.create_work_item( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, item_type="task", title="T1", ) assert wi.timeline.count() == 1 entry = wi.timeline.first() assert entry.event == "created" assert entry.to_status == "pending" def test_complete_work_item(self, tenant_acme, acme_tree, alice): wi = WorkItemService.create_work_item( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, item_type="task", title="To complete", assigned_to_user_id=alice.pk, ) wi = WorkItemService.complete_work_item(wi, actor_id=alice.pk, note="Done!") assert wi.status == WorkboxItemStatus.COMPLETED assert wi.completed_at is not None assert wi.timeline.count() == 2 assert wi.timeline.first().event == "completed" def test_complete_raises_on_completed(self, tenant_acme, acme_tree, alice): wi = WorkItemService.create_work_item( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, item_type="task", title="Completed", ) WorkItemService.complete_work_item(wi, actor_id=alice.pk) with pytest.raises(ValueError, match="Cannot complete"): WorkItemService.complete_work_item(wi, actor_id=alice.pk) def test_complete_raises_on_cancelled(self, tenant_acme, acme_tree, alice): wi = WorkItemService.create_work_item( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, item_type="task", title="Cancelled", ) WorkItemService.cancel_work_item(wi, actor_id=alice.pk) with pytest.raises(ValueError, match="Cannot complete"): WorkItemService.complete_work_item(wi, actor_id=alice.pk) def test_cancel_work_item(self, tenant_acme, acme_tree, alice): wi = WorkItemService.create_work_item( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, item_type="task", title="To cancel", ) wi = WorkItemService.cancel_work_item(wi, actor_id=alice.pk, note="No longer needed") assert wi.status == WorkboxItemStatus.CANCELLED assert wi.timeline.first().event == "cancelled" def test_cancel_raises_on_completed(self, tenant_acme, acme_tree, alice): wi = WorkItemService.create_work_item( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, item_type="task", title="Done", ) WorkItemService.complete_work_item(wi, actor_id=alice.pk) with pytest.raises(ValueError, match="Cannot cancel"): WorkItemService.cancel_work_item(wi, actor_id=alice.pk) def test_reassign_work_item(self, tenant_acme, acme_tree, alice, tenant_globex): bob = UserFactory(mobile="+989000000002") wi = WorkItemService.create_work_item( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, item_type="task", title="Reassign me", assigned_to_user_id=alice.pk, ) wi = WorkItemService.reassign_work_item( wi, to_user_id=bob.pk, actor_id=alice.pk, note="Bob, please take over", ) assert wi.assigned_to_user == bob assert wi.timeline.first().event == "reassigned" def test_reassign_raises_on_completed(self, tenant_acme, acme_tree, alice): wi = WorkItemService.create_work_item( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, item_type="task", title="Done", ) WorkItemService.complete_work_item(wi, actor_id=alice.pk) with pytest.raises(ValueError, match="Cannot reassign"): WorkItemService.reassign_work_item(wi, to_user_id=alice.pk, actor_id=alice.pk) def test_reassign_with_role_and_unit(self, tenant_acme, acme_tree, alice): wi = WorkItemService.create_work_item( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, item_type="task", title="Complex reassign", ) wi = WorkItemService.reassign_work_item( wi, to_unit_id=acme_tree["eu"].pk, actor_id=alice.pk, ) assert wi.assigned_to_unit == acme_tree["eu"] def test_get_timeline(self, tenant_acme, acme_tree, alice): wi = WorkItemService.create_work_item( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, item_type="task", title="Timeline test", ) WorkItemService.complete_work_item(wi, actor_id=alice.pk, note="First") # Re-fetch after updates wi.refresh_from_db() timeline = WorkItemService.get_timeline(wi) assert len(timeline) == 2 assert timeline[0]["event"] == "completed" assert timeline[1]["event"] == "created" def test_expire_work_items(self, tenant_acme, acme_tree, alice): past = timezone.now() - timezone.timedelta(days=1) wi = WorkItemService.create_work_item( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, item_type="task", title="Expired task", assigned_to_user_id=alice.pk, ) # Force the due_date to be in the past wi.due_date = past wi.save(update_fields=["due_date"]) count = WorkItemService.expire_work_items() assert count == 1 wi.refresh_from_db() assert wi.status == WorkboxItemStatus.EXPIRED def test_create_with_all_optional_fields(self, tenant_acme, acme_tree, alice): wi = WorkItemService.create_work_item( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, item_type="custom", title="Full custom", summary="summary", description="desc", assigned_to_user_id=alice.pk, assigned_to_role_id=None, assigned_to_unit_id=acme_tree["eu"].pk, priority="urgent", source_type="custom", source_entity="my.custom", source_id="123", source_url="/custom/123/", is_delegated=True, metadata={"key": "value"}, ) assert wi.title == "Full custom" assert wi.priority == "urgent" assert wi.is_delegated is True assert wi.metadata == {"key": "value"} # ── Backward-compatible services ──────────────────────────────────────────── @pytest.mark.django_db class TestBackwardCompatServices: """Ensure legacy workbox functions still work with the new model.""" def test_upsert_creates(self, tenant_acme, acme_tree, alice): item = upsert_workbox_item( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, user_id=alice.pk, item_type="assignment", source_entity="assignments.assignment", source_id="pub-1", title="Test assignment", ) assert item.title == "Test assignment" assert item.assigned_to_user_id == alice.pk def test_upsert_idempotent(self, tenant_acme, acme_tree, alice): kwargs = { "tenant_id": tenant_acme.pk, "organization_node_id": acme_tree["root"].pk, "user_id": alice.pk, "item_type": "assignment", "source_entity": "assignments.assignment", "source_id": "pub-2", "title": "First title", } item1 = upsert_workbox_item(**kwargs) kwargs["title"] = "Updated title" item2 = upsert_workbox_item(**kwargs) assert item1.pk == item2.pk assert item2.title == "Updated title" def test_remove_workbox_item(self, tenant_acme, acme_tree, alice): upsert_workbox_item( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, user_id=alice.pk, item_type="task", source_entity="tasks.task", source_id="pub-3", title="Removable", ) deleted = remove_workbox_item( tenant_id=tenant_acme.pk, source_entity="tasks.task", source_id="pub-3", ) assert deleted == 1 def test_get_dashboard_counts(self, tenant_acme, alice): assert WorkItem.objects.count() == 0 counts = get_dashboard_counts(tenant_id=tenant_acme.pk, user_id=alice.pk) assert counts["total"] == 0 assert "pending" in counts assert "cancelled" in counts assert "expired" in counts assert "delegated" in counts def test_refresh_user_workbox(self, tenant_acme, acme_tree, alice): stats = refresh_user_workbox( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, user_id=alice.pk, ) assert "assignments" in stats assert "approvals" in stats assert "tasks" in stats assert "workflow" in stats # ── Helpers ───────────────────────────────────────────────────────────────── from tests.factories import UserFactory # noqa: E402