from __future__ import annotations from dataclasses import replace from datetime import UTC, datetime from types import SimpleNamespace import unittest from sqlalchemy import create_engine from sqlalchemy.orm import Session from govoplan_core.core.external_references import ExternalObjectReference from govoplan_core.core.search import ( SearchAuthorizationRequest, SearchBackfillRequest, ) from govoplan_core.db.base import Base from govoplan_core.security.module_permissions import scopes_grant_compatible from govoplan_projects.backend.domain import ( ProjectDomainError, ProjectMembership, ProjectRecord, ProjectSubjectRef, ) from govoplan_projects.backend.search_source import ( PROVIDER_ID, ProjectsSearchSource, ) from govoplan_projects.backend.service import ( ADMIN_SCOPE, READ_SCOPE, WRITE_SCOPE, ProjectStoreError, create_project_object, get_project_object, list_project_objects, project_object_history, update_project_object, ) NOW = datetime(2026, 8, 1, 10, 0, tzinfo=UTC) class Principal: def __init__( self, tenant_id: str, account_id: str, *, scopes: tuple[str, ...] = (READ_SCOPE, WRITE_SCOPE), group_ids: tuple[str, ...] = (), ) -> None: self.tenant_id = tenant_id self.account_id = account_id self.identity_id = f"identity-{account_id}" self.membership_id = f"membership-{account_id}" self.group_ids = frozenset(group_ids) self.role_ids = frozenset() self.function_assignment_ids = frozenset() self.acting_assignment_id = None self.service_account_id = None self.scopes = frozenset(scopes) self.user = SimpleNamespace(id=f"user-{account_id}") def has(self, scope: str) -> bool: return scopes_grant_compatible(self.scopes, scope) class ProjectServiceTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite+pysqlite:///:memory:") Base.metadata.create_all(self.engine) self.session = Session(self.engine, expire_on_commit=False) self.owner = Principal("tenant-1", "owner-1") def tearDown(self) -> None: self.session.close() self.engine.dispose() def test_lifecycle_is_versioned_idempotent_and_occ_guarded(self) -> None: record = project_record() created = create_project_object( self.session, self.owner, record=record, idempotency_key="create-project-1", ) replay = create_project_object( self.session, self.owner, record=record, idempotency_key="create-project-1", ) self.assertEqual(created.to_dict(), replay.to_dict()) updated = update_project_object( self.session, self.owner, object_kind="project", object_id="project-1", expected_revision=1, changes={"state": "proposed", "title": "Permit modernization"}, recorded_at=NOW.replace(hour=11), change_reason="Submitted for approval.", idempotency_key="update-project-1", ) self.assertEqual(2, updated.revision) self.assertEqual("proposed", updated.state) self.assertEqual( [2, 1], [ item.revision for item in project_object_history( self.session, self.owner, object_kind="project", object_id="project-1", ) ], ) with self.assertRaisesRegex(ProjectStoreError, "stale"): update_project_object( self.session, self.owner, object_kind="project", object_id="project-1", expected_revision=1, changes={"title": "Stale title"}, recorded_at=NOW.replace(hour=12), change_reason="Stale update.", idempotency_key="update-project-stale", ) def test_state_machine_rejects_invalid_transition(self) -> None: create_project_object( self.session, self.owner, record=project_record(), idempotency_key="create-transition-project", ) with self.assertRaisesRegex(ProjectDomainError, "Cannot move"): update_project_object( self.session, self.owner, object_kind="project", object_id="project-1", expected_revision=1, changes={"state": "completed"}, recorded_at=NOW.replace(hour=11), change_reason="Invalid shortcut.", idempotency_key="invalid-transition", ) def test_parent_hierarchy_and_tenant_boundary_are_enforced(self) -> None: with self.assertRaisesRegex(ProjectStoreError, "parent"): create_project_object( self.session, self.owner, record=replace( project_record(), object_kind="milestone", object_id="milestone-1", object_key="milestone-1", state="planned", parent_kind="project", parent_id="missing-project", ), idempotency_key="orphan-milestone", ) with self.assertRaisesRegex(ProjectStoreError, "cross tenants"): create_project_object( self.session, Principal("tenant-2", "owner-2"), record=project_record(), idempotency_key="cross-tenant-project", ) def test_restricted_memberships_filter_lists_and_reads(self) -> None: record = replace( project_record(), visibility="restricted", memberships=( ProjectMembership( subject=ProjectSubjectRef(kind="group", id="group-reviewers"), role="reviewer", permissions=("read",), ), ), ) create_project_object( self.session, self.owner, record=record, idempotency_key="create-restricted-project", ) outsider = Principal("tenant-1", "outsider") reviewer = Principal( "tenant-1", "reviewer", group_ids=("group-reviewers",), ) self.assertIsNone( get_project_object( self.session, outsider, object_kind="project", object_id="project-1", ) ) visible, total = list_project_objects(self.session, outsider) self.assertEqual(((), 0), (visible, total)) self.assertIsNotNone( get_project_object( self.session, reviewer, object_kind="project", object_id="project-1", ) ) admin = Principal( "tenant-1", "admin", scopes=(READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE), ) self.assertEqual(1, list_project_objects(self.session, admin)[1]) def test_search_backfill_has_acl_tokens_and_rechecks_access(self) -> None: create_project_object( self.session, self.owner, record=replace(project_record(), visibility="restricted"), idempotency_key="create-search-project", ) source = ProjectsSearchSource() page = source.backfill( self.session, request=SearchBackfillRequest( tenant_id="tenant-1", provider_id=PROVIDER_ID, resource_type="project", rebuild_id="rebuild-1", ), ) self.assertTrue(page.complete) self.assertEqual(1, len(page.documents)) document = page.documents[0] self.assertIn("account:owner-1", document.acl_tokens) request = SearchAuthorizationRequest( reference=document.reference, source_revision=document.source_revision, ) owner_decision = source.authorize( self.session, self.owner, requests=(request,), ) outsider_decision = source.authorize( self.session, Principal("tenant-1", "outsider"), requests=(request,), ) self.assertTrue(owner_decision[document.reference.key]) self.assertFalse(outsider_decision[document.reference.key]) def project_record( *, object_kind: str = "project", ) -> ProjectRecord: return ProjectRecord.from_mapping( { "tenant_id": "tenant-1", "object_kind": object_kind, "object_id": "project-1", "object_key": "permit-modernization", "revision": 1, "title": "Permit modernization", "state": "draft", "description": "Replace a fragmented permit process.", "visibility": "tenant", "recorded_at": NOW.isoformat(), "change_reason": "Initial planning baseline.", "owner": { "kind": "account", "id": "owner-1", "label": "Project owner", }, "outcomes": [ { "key": "faster-decisions", "title": "Faster permit decisions", "success_indicators": ["Median processing time"], } ], "benefits": [ { "key": "less-rework", "title": "Less manual rework", "target": "Reduce rework by 30 percent", } ], "external_references": [ ExternalObjectReference( system="openproject", object_type="project", object_id="42", ).to_dict() ], } ) if __name__ == "__main__": unittest.main()