"""Tests for DMS Phase 14 — Workflow Engine Integration + Event Bus. Covers: * transitions.py — TRANSITIONS dict, WorkflowState constants, WorkflowError * state_machine.py — can_transition, validate, available_actions * services.py — trigger_workflow_transition: updates workflow_status / Document.status * services.py — invalid action / state raises WorkflowError * events/definitions.py — all domain events registered in bus * API POST /documents/{id}/transition/ — happy path, 401, 403, 400 (bad action), 400 (bad state) * Full workflow flow: initial → draft → under_review → approved → published * Sync between workflow_status and Document.status on publish / archive * Celery tasks — importable and callable (smoke test with mocks) * release_expired_locks task — releases locks and dispatches event * cleanup_expired_share_links task — deactivates expired share links """ from __future__ import annotations import uuid from datetime import timedelta from unittest.mock import patch import pytest from django.utils import timezone from simorgh.apps.dms.workflow.transitions import ( TRANSITIONS, WorkflowError, WorkflowState, ) from simorgh.apps.dms.workflow.state_machine import DocumentStateMachine # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def alice_membership(alice, tenant_acme, acme_tree, role_admin): """alice gets role_admin (all permissions including workflow perms).""" from simorgh.apps.memberships.models import Membership m = Membership.objects.create( tenant=tenant_acme, organization_node=acme_tree["root"], role=role_admin, ) m.users.add(alice) return m @pytest.fixture def bob(db): """User with no permissions.""" from django.contrib.auth import get_user_model User = get_user_model() return User.objects.create_user( "+989000000097", password="x", email="bob_wf14@example.com", ) @pytest.fixture def repo(tenant_acme, acme_tree): from simorgh.apps.dms.repositories.models import Repository return Repository.objects.create( name="WF Phase 14 Repo", slug="wf-ph14-repo", tenant=tenant_acme, organization_node=acme_tree["root"], ) @pytest.fixture def doc(repo, tenant_acme, acme_tree): """A bare document with no version (workflow_status="").""" from simorgh.apps.dms.documents.models import Document return Document.objects.create( title="Phase 14 Test Document", repository=repo, tenant=tenant_acme, organization_node=acme_tree["root"], workflow_status="", ) @pytest.fixture def doc_with_version(doc, tenant_acme, acme_tree): """Document with one DocumentVersion (so validators pass).""" from simorgh.apps.dms.documents.models import DocumentVersion DocumentVersion.objects.create( document=doc, tenant=tenant_acme, organization_node=acme_tree["root"], version_major=1, version_minor=0, ) doc.refresh_from_db() return doc def _url(path: str) -> str: return f"/api/v1/dms/{path}" # --------------------------------------------------------------------------- # 1. Transitions module — structural tests # --------------------------------------------------------------------------- def test_all_workflow_actions_present(): """All 7 workflow actions are defined in TRANSITIONS.""" expected = { "submit_for_review", "approve", "reject", "publish", "archive", "resubmit", "retract", } assert expected == set(TRANSITIONS.keys()) def test_workflow_state_constants(): """WorkflowState constants have the expected string values.""" assert WorkflowState.INITIAL == "" assert WorkflowState.DRAFT == "draft" assert WorkflowState.UNDER_REVIEW == "under_review" assert WorkflowState.APPROVED == "approved" assert WorkflowState.REJECTED == "rejected" assert WorkflowState.PUBLISHED == "published" assert WorkflowState.ARCHIVED == "archived" def test_transition_spec_from_states(): """submit_for_review is valid from initial and draft only.""" spec = TRANSITIONS["submit_for_review"] assert WorkflowState.INITIAL in spec.from_states assert WorkflowState.DRAFT in spec.from_states assert WorkflowState.UNDER_REVIEW not in spec.from_states def test_publish_transition_sets_to_published(): assert TRANSITIONS["publish"].to_state == WorkflowState.PUBLISHED def test_archive_transition_sets_to_archived(): assert TRANSITIONS["archive"].to_state == WorkflowState.ARCHIVED # --------------------------------------------------------------------------- # 2. DocumentStateMachine — unit tests (no DB required) # --------------------------------------------------------------------------- class _FakeDoc: """Minimal stub for DocumentStateMachine unit tests.""" def __init__(self, workflow_status="", versions_exist=True): self.workflow_status = workflow_status self._versions_exist = versions_exist def refresh_from_db(self): pass def save(self, update_fields=None): pass def test_can_transition_initial_to_under_review(): machine = DocumentStateMachine(_FakeDoc("")) assert machine.can_transition("submit_for_review") is True def test_can_transition_draft_to_under_review(): machine = DocumentStateMachine(_FakeDoc("draft")) assert machine.can_transition("submit_for_review") is True def test_can_transition_invalid(): machine = DocumentStateMachine(_FakeDoc("")) assert machine.can_transition("approve") is False def test_can_transition_unknown_action(): machine = DocumentStateMachine(_FakeDoc("")) assert machine.can_transition("nonexistent_action") is False def test_available_actions_initial_state(): machine = DocumentStateMachine(_FakeDoc("")) # From initial: only submit_for_review actions = machine.available_actions() assert "submit_for_review" in actions assert "approve" not in actions def test_available_actions_under_review(): machine = DocumentStateMachine(_FakeDoc("under_review")) actions = machine.available_actions() assert "approve" in actions assert "reject" in actions assert "retract" in actions assert "submit_for_review" not in actions def test_available_actions_with_permissions_filter(): """available_actions filters by required_permission.""" from simorgh.apps.dms.documents.permissions import PERM_DOCUMENT_MANAGE from simorgh.apps.dms.workflow.transitions import PERM_DOCUMENT_APPROVE machine = DocumentStateMachine(_FakeDoc("under_review")) # Only manage perm — approve/reject (require APPROVE perm) are excluded. actions = machine.available_actions(permissions={PERM_DOCUMENT_MANAGE}) assert "approve" not in actions assert "reject" not in actions assert "retract" in actions # retract requires manage def test_validate_raises_for_wrong_state(): machine = DocumentStateMachine(_FakeDoc("published")) with pytest.raises(WorkflowError, match="Cannot perform"): machine.validate("submit_for_review") def test_validate_raises_for_unknown_action(): machine = DocumentStateMachine(_FakeDoc("draft")) with pytest.raises(WorkflowError, match="Unknown workflow action"): machine.validate("bogus") # --------------------------------------------------------------------------- # 3. Service layer — trigger_workflow_transition # --------------------------------------------------------------------------- @pytest.mark.django_db def test_service_submit_for_review(doc_with_version, alice): from simorgh.apps.dms.workflow.services import trigger_workflow_transition doc = doc_with_version assert doc.workflow_status == "" result = trigger_workflow_transition(doc, "submit_for_review", actor_id=alice.pk) assert result.workflow_status == WorkflowState.UNDER_REVIEW doc.refresh_from_db() assert doc.workflow_status == WorkflowState.UNDER_REVIEW @pytest.mark.django_db def test_service_full_approval_flow(doc_with_version, alice): """draft/initial → under_review → approved → published (Document.status updated).""" from simorgh.apps.dms.documents.models import DocumentStatus from simorgh.apps.dms.workflow.services import trigger_workflow_transition doc = doc_with_version trigger_workflow_transition(doc, "submit_for_review", actor_id=alice.pk) assert doc.workflow_status == WorkflowState.UNDER_REVIEW trigger_workflow_transition(doc, "approve", actor_id=alice.pk) assert doc.workflow_status == WorkflowState.APPROVED trigger_workflow_transition(doc, "publish", actor_id=alice.pk) assert doc.workflow_status == WorkflowState.PUBLISHED assert doc.status == DocumentStatus.PUBLISHED @pytest.mark.django_db def test_service_archive_syncs_document_status(doc_with_version, alice): """archive transition also sets Document.status = ARCHIVED.""" from simorgh.apps.dms.documents.models import DocumentStatus from simorgh.apps.dms.workflow.services import trigger_workflow_transition doc = doc_with_version # Get to published first. trigger_workflow_transition(doc, "submit_for_review", actor_id=alice.pk) trigger_workflow_transition(doc, "approve", actor_id=alice.pk) trigger_workflow_transition(doc, "publish", actor_id=alice.pk) trigger_workflow_transition(doc, "archive", actor_id=alice.pk) assert doc.workflow_status == WorkflowState.ARCHIVED assert doc.status == DocumentStatus.ARCHIVED @pytest.mark.django_db def test_service_reject_and_resubmit(doc_with_version, alice): """under_review → rejected → under_review (resubmit).""" from simorgh.apps.dms.workflow.services import trigger_workflow_transition doc = doc_with_version trigger_workflow_transition(doc, "submit_for_review", actor_id=alice.pk) trigger_workflow_transition(doc, "reject", actor_id=alice.pk) assert doc.workflow_status == WorkflowState.REJECTED trigger_workflow_transition(doc, "resubmit", actor_id=alice.pk) assert doc.workflow_status == WorkflowState.UNDER_REVIEW @pytest.mark.django_db def test_service_retract(doc_with_version, alice): """under_review → draft (retract).""" from simorgh.apps.dms.workflow.services import trigger_workflow_transition doc = doc_with_version trigger_workflow_transition(doc, "submit_for_review", actor_id=alice.pk) trigger_workflow_transition(doc, "retract", actor_id=alice.pk) assert doc.workflow_status == WorkflowState.DRAFT @pytest.mark.django_db def test_service_invalid_action_raises(doc, alice): from simorgh.apps.dms.workflow.services import trigger_workflow_transition with pytest.raises(WorkflowError, match="Unknown workflow action"): trigger_workflow_transition(doc, "bogus_action", actor_id=alice.pk) @pytest.mark.django_db def test_service_invalid_state_transition_raises(doc, alice): """Attempting approve on a doc in initial state raises WorkflowError.""" from simorgh.apps.dms.workflow.services import trigger_workflow_transition with pytest.raises(WorkflowError, match="Cannot perform"): trigger_workflow_transition(doc, "approve", actor_id=alice.pk) @pytest.mark.django_db def test_service_validator_blocks_no_version(doc, alice): """submit_for_review requires at least one version.""" from simorgh.apps.dms.workflow.services import trigger_workflow_transition # doc has no versions with pytest.raises(WorkflowError, match="version"): trigger_workflow_transition(doc, "submit_for_review", actor_id=alice.pk) @pytest.mark.django_db def test_service_dispatches_event_on_publish(doc_with_version, alice): """trigger_workflow_transition dispatches dms.document.published on publish.""" from simorgh.apps.events.bus import subscribe, clear_subscribers from simorgh.apps.dms.workflow.services import trigger_workflow_transition received = [] @subscribe("dms.document.published") def _h(payload): received.append(payload) try: doc = doc_with_version trigger_workflow_transition(doc, "submit_for_review", actor_id=alice.pk) trigger_workflow_transition(doc, "approve", actor_id=alice.pk) trigger_workflow_transition(doc, "publish", actor_id=alice.pk) finally: clear_subscribers("dms.document.published") assert len(received) == 1 assert received[0]["workflow_status"] == WorkflowState.PUBLISHED @pytest.mark.django_db def test_service_dispatches_event_on_reject(doc_with_version, alice): """trigger_workflow_transition dispatches dms.document.rejected on reject.""" from simorgh.apps.events.bus import subscribe, clear_subscribers from simorgh.apps.dms.workflow.services import trigger_workflow_transition received = [] @subscribe("dms.document.rejected") def _h(payload): received.append(payload) try: doc = doc_with_version trigger_workflow_transition(doc, "submit_for_review", actor_id=alice.pk) trigger_workflow_transition(doc, "reject", actor_id=alice.pk) finally: clear_subscribers("dms.document.rejected") assert len(received) == 1 assert received[0]["workflow_status"] == WorkflowState.REJECTED # --------------------------------------------------------------------------- # 4. Event definitions — all events registered in bus # --------------------------------------------------------------------------- def test_dms_domain_events_registered(): """All Phase 14 DMS events are present in the event registry.""" from simorgh.apps.events.bus import _REGISTRY expected_events = { "dms.document.created", "dms.document.submitted_for_review", "dms.document.approved", "dms.document.rejected", "dms.document.published", "dms.document.archived", "dms.document.retracted", "dms.version.added", "dms.checkout.expired", "dms.hold.placed", "dms.ocr.completed", } with _REGISTRY.lock: registered = set(_REGISTRY.events.keys()) missing = expected_events - registered assert not missing, f"Events not registered: {missing}" # --------------------------------------------------------------------------- # 5. API — POST /documents/{id}/transition/ # --------------------------------------------------------------------------- @pytest.mark.django_db def test_api_transition_requires_auth(client, doc_with_version, tenant_acme): doc = doc_with_version resp = client.post( _url(f"documents/{doc.public_id}/transition/"), {"action": "submit_for_review"}, content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 401 @pytest.mark.django_db def test_api_transition_unknown_action_returns_400( client, alice, alice_membership, tenant_acme, doc_with_version ): client.force_login(alice) doc = doc_with_version resp = client.post( _url(f"documents/{doc.public_id}/transition/"), {"action": "fly_to_moon"}, content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 400 assert "Unknown action" in resp.json()["detail"] @pytest.mark.django_db def test_api_transition_missing_action_returns_400( client, alice, alice_membership, tenant_acme, doc_with_version ): client.force_login(alice) doc = doc_with_version resp = client.post( _url(f"documents/{doc.public_id}/transition/"), {}, content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 400 assert "action" in resp.json()["detail"].lower() @pytest.mark.django_db def test_api_transition_no_perm_returns_403( client, bob, tenant_acme, acme_tree, role_viewer, doc_with_version ): """bob has only viewer role — cannot perform submit_for_review.""" from simorgh.apps.memberships.models import Membership m = Membership.objects.create( tenant=tenant_acme, organization_node=acme_tree["root"], role=role_viewer, ) m.users.add(bob) client.force_login(bob) doc = doc_with_version resp = client.post( _url(f"documents/{doc.public_id}/transition/"), {"action": "submit_for_review"}, content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 403 @pytest.mark.django_db def test_api_transition_invalid_state_returns_400( client, alice, alice_membership, tenant_acme, doc_with_version ): """approve from initial state is invalid → 400.""" client.force_login(alice) doc = doc_with_version resp = client.post( _url(f"documents/{doc.public_id}/transition/"), {"action": "approve"}, content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 400 assert "Cannot perform" in resp.json()["detail"] @pytest.mark.django_db def test_api_transition_document_not_found( client, alice, alice_membership, tenant_acme ): client.force_login(alice) fake_id = str(uuid.uuid4()) resp = client.post( _url(f"documents/{fake_id}/transition/"), {"action": "submit_for_review"}, content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 404 @pytest.mark.django_db def test_api_transition_submit_for_review( client, alice, alice_membership, tenant_acme, doc_with_version ): """Happy path: submit_for_review → 200, workflow_status updated.""" client.force_login(alice) doc = doc_with_version resp = client.post( _url(f"documents/{doc.public_id}/transition/"), {"action": "submit_for_review", "comment": "ready for review"}, content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 200 data = resp.json() assert data["workflow_status"] == WorkflowState.UNDER_REVIEW assert data["action"] == "submit_for_review" doc.refresh_from_db() assert doc.workflow_status == WorkflowState.UNDER_REVIEW @pytest.mark.django_db def test_api_transition_full_publish_flow( client, alice, alice_membership, tenant_acme, doc_with_version ): """End-to-end API: draft → under_review → approved → published.""" from simorgh.apps.dms.documents.models import DocumentStatus client.force_login(alice) doc = doc_with_version headers = {"HTTP_X_TENANT": tenant_acme.slug} def _transition(action, comment=""): return client.post( _url(f"documents/{doc.public_id}/transition/"), {"action": action, "comment": comment}, content_type="application/json", **headers, ) assert _transition("submit_for_review").status_code == 200 assert _transition("approve").status_code == 200 resp = _transition("publish") assert resp.status_code == 200 doc.refresh_from_db() assert doc.workflow_status == WorkflowState.PUBLISHED assert doc.status == DocumentStatus.PUBLISHED @pytest.mark.django_db def test_api_no_version_blocks_submit( client, alice, alice_membership, tenant_acme, doc ): """submit_for_review on a doc without versions → 400.""" client.force_login(alice) resp = client.post( _url(f"documents/{doc.public_id}/transition/"), {"action": "submit_for_review"}, content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 400 assert "version" in resp.json()["detail"].lower() # --------------------------------------------------------------------------- # 6. Celery tasks — smoke tests # --------------------------------------------------------------------------- def test_celery_tasks_importable(): from simorgh.apps.dms.tasks import ( dms_release_expired_locks, dms_generate_preview_async, dms_process_ocr_async, dms_cleanup_expired_share_links, ) assert callable(dms_release_expired_locks) assert callable(dms_generate_preview_async) assert callable(dms_process_ocr_async) assert callable(dms_cleanup_expired_share_links) @pytest.mark.django_db def test_release_expired_locks_returns_count(tenant_acme, acme_tree, repo, doc): """dms_release_expired_locks releases locks whose expires_at is in the past.""" from simorgh.apps.dms.versioning.models import DocumentLock, LockStatus from simorgh.apps.dms.tasks import dms_release_expired_locks past = timezone.now() - timedelta(minutes=10) lock = DocumentLock.objects.create( document=doc, tenant=tenant_acme, organization_node=acme_tree["root"], expires_at=past, is_active=True, status=LockStatus.ACTIVE, ) count = dms_release_expired_locks() assert count == 1 lock.refresh_from_db() assert lock.is_active is False assert lock.status == LockStatus.EXPIRED @pytest.mark.django_db def test_release_expired_locks_ignores_future_locks(tenant_acme, acme_tree, doc): """Locks expiring in the future are not released.""" from simorgh.apps.dms.versioning.models import DocumentLock, LockStatus from simorgh.apps.dms.tasks import dms_release_expired_locks future = timezone.now() + timedelta(hours=1) DocumentLock.objects.create( document=doc, tenant=tenant_acme, organization_node=acme_tree["root"], expires_at=future, is_active=True, status=LockStatus.ACTIVE, ) count = dms_release_expired_locks() assert count == 0 @pytest.mark.django_db def test_cleanup_expired_share_links(tenant_acme, acme_tree, repo, doc): """dms_cleanup_expired_share_links deactivates expired links.""" from simorgh.apps.dms.permissions.models import ShareLink from simorgh.apps.dms.tasks import dms_cleanup_expired_share_links past = timezone.now() - timedelta(hours=2) link = ShareLink.objects.create( document=doc, tenant=tenant_acme, organization_node=acme_tree["root"], expires_at=past, is_active=True, ) count = dms_cleanup_expired_share_links() assert count == 1 link.refresh_from_db() assert link.is_active is False @pytest.mark.django_db def test_cleanup_skips_valid_share_links(tenant_acme, acme_tree, doc): """Share links without expiry are not deactivated.""" from simorgh.apps.dms.permissions.models import ShareLink from simorgh.apps.dms.tasks import dms_cleanup_expired_share_links ShareLink.objects.create( document=doc, tenant=tenant_acme, organization_node=acme_tree["root"], expires_at=None, is_active=True, ) count = dms_cleanup_expired_share_links() assert count == 0 # --------------------------------------------------------------------------- # 7. IAM permission registered # --------------------------------------------------------------------------- @pytest.mark.django_db def test_workflow_approve_permission_registered(perms): """dms.document.approve is registered in the IAM permission registry.""" from simorgh.apps.dms.workflow.transitions import PERM_DOCUMENT_APPROVE assert PERM_DOCUMENT_APPROVE in perms, ( f"Permission {PERM_DOCUMENT_APPROVE!r} not found in registry" )