from __future__ import annotations from datetime import UTC, datetime, timedelta from types import SimpleNamespace import unittest from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from govoplan_core.core.dsar import DsarSubjectRef from govoplan_core.core.search import SearchBackfillRequest from govoplan_core.db.base import Base from govoplan_wiki.backend.db.models import ( WikiComment, WikiPageRevision, WikiSpaceHistory, ) from govoplan_wiki.backend.dsar_provider import WikiDsarProvider from govoplan_wiki.backend.search_source import ( PROVIDER_ID, RESOURCE_TYPE, WikiSearchSource, ) from govoplan_wiki.backend.service import ( ADMIN_SCOPE, COMMENT_SCOPE, PUBLISH_SCOPE, READ_SCOPE, WRITE_SCOPE, WikiConflictError, add_comment, archive_page, compare_revisions, create_page, create_space, get_page, list_comments, list_pages, page_revisions, publish_page, update_page, ) NOW = datetime(2026, 8, 22, 10, 0, tzinfo=UTC) class _Principal: def __init__( self, account_id: str, scopes: set[str], tenant_id: str = "tenant-1", groups: tuple[str, ...] = (), ) -> None: self.account_id = account_id self.identity_id = None self.membership_id = f"membership-{account_id}" self.tenant_id = tenant_id self.scopes = frozenset(scopes) self.group_ids = frozenset(groups) self.role_ids = frozenset() self.function_assignment_ids = frozenset() self.acting_assignment_id = None self.user = SimpleNamespace(id=account_id) def has(self, scope: str) -> bool: return scope in self.scopes class WikiServiceTests(unittest.TestCase): def setUp(self) -> None: engine = create_engine("sqlite+pysqlite:///:memory:", future=True) Base.metadata.create_all(engine) self.Session = sessionmaker(bind=engine, expire_on_commit=False) all_scopes = { READ_SCOPE, WRITE_SCOPE, COMMENT_SCOPE, PUBLISH_SCOPE, ADMIN_SCOPE, } self.admin = _Principal("admin-1", all_scopes) self.editor = _Principal( "editor-1", {READ_SCOPE, WRITE_SCOPE, COMMENT_SCOPE}, groups=("knowledge",) ) self.publisher = _Principal( "publisher-1", {READ_SCOPE, PUBLISH_SCOPE}, groups=("knowledge",), ) self.reader = _Principal("reader-1", {READ_SCOPE}, groups=("knowledge",)) self.outsider = _Principal("outsider-1", {READ_SCOPE}) def test_publication_keeps_last_published_revision_visible_during_new_draft( self, ) -> None: with self.Session() as session: self._space(session) created = self._page(session) published = publish_page( session, self.publisher, page_id="page-1", expected_revision=created["revision"], recorded_at=NOW + timedelta(minutes=1), change_reason="Reviewed for publication.", idempotency_key="publish-1", ) publish_replay = publish_page( session, self.publisher, page_id="page-1", expected_revision=created["revision"], recorded_at=NOW + timedelta(minutes=1), change_reason="Reviewed for publication.", idempotency_key="publish-1", ) session.commit() self.assertEqual(published, publish_replay) self.assertEqual(2, published["published_revision"]) self.assertEqual( "Published body", get_page(session, self.reader, page_id="page-1")["body"], ) draft = update_page( session, self.editor, page_id="page-1", expected_revision=2, changes={"body": "Next draft body", "labels": ["Guide", "Updated"]}, recorded_at=NOW + timedelta(minutes=2), change_reason="Prepare the next edition.", idempotency_key="edit-1", ) draft_replay = update_page( session, self.editor, page_id="page-1", expected_revision=2, changes={"body": "Next draft body", "labels": ["Guide", "Updated"]}, recorded_at=NOW + timedelta(minutes=2), change_reason="Prepare the next edition.", idempotency_key="edit-1", ) session.commit() self.assertEqual(draft, draft_replay) self.assertEqual("draft", draft["state"]) self.assertEqual( "Next draft body", get_page(session, self.editor, page_id="page-1")["body"], ) reader_view = get_page(session, self.reader, page_id="page-1") self.assertEqual("Published body", reader_view["body"]) self.assertEqual(2, reader_view["revision"]) source = WikiSearchSource() page = source.backfill( session, request=SearchBackfillRequest( tenant_id="tenant-1", provider_id=PROVIDER_ID, resource_type=RESOURCE_TYPE, rebuild_id="wiki-test", limit=50, ), ) self.assertEqual(1, len(page.documents)) self.assertEqual("Published body", page.documents[0].body) self.assertEqual("2", page.documents[0].source_revision) diff = compare_revisions( session, self.editor, page_id="page-1", from_revision=2, to_revision=3 ) self.assertIn("Next draft body", diff["diff"]) self.assertEqual( 3, len(page_revisions(session, self.editor, page_id="page-1")) ) def test_hierarchy_acl_occ_replay_links_comments_and_archive_are_governed( self, ) -> None: with self.Session() as session: self._space(session) created = self._page(session) replay = self._page(session) self.assertEqual(created, replay) self.assertEqual(1, session.query(WikiPageRevision).count()) self.assertIsNone(get_page(session, self.outsider, page_id="page-1")) self.assertEqual(0, list_pages(session, self.reader, space_id="space-1")[1]) other_tenant = _Principal( "reader-1", {READ_SCOPE}, tenant_id="tenant-2", groups=("knowledge",) ) self.assertIsNone(get_page(session, other_tenant, page_id="page-1")) session.commit() with self.assertRaises(WikiConflictError): update_page( session, self.editor, page_id="page-1", expected_revision=99, changes={"title": "Stale"}, recorded_at=NOW + timedelta(minutes=1), change_reason="Stale edit.", idempotency_key="stale-1", ) session.rollback() comment = add_comment( session, self.editor, page_id="page-1", expected_revision=1, comment_id="comment-1", body="Please verify the source.", recorded_at=NOW + timedelta(minutes=2), idempotency_key="comment-1", ) replay_comment = add_comment( session, self.editor, page_id="page-1", expected_revision=1, comment_id="comment-1", body="Please verify the source.", recorded_at=NOW + timedelta(minutes=2), idempotency_key="comment-1", ) self.assertEqual(comment, replay_comment) self.assertEqual( 1, len(list_comments(session, self.editor, page_id="page-1")) ) self.assertEqual(1, session.query(WikiComment).count()) published = publish_page( session, self.publisher, page_id="page-1", expected_revision=1, recorded_at=NOW + timedelta(minutes=3), change_reason="Approved.", idempotency_key="publish-archive", ) replay_after_revision = add_comment( session, self.editor, page_id="page-1", expected_revision=1, comment_id="comment-1", body="Please verify the source.", recorded_at=NOW + timedelta(minutes=2), idempotency_key="comment-1", ) self.assertEqual(comment, replay_after_revision) archived = archive_page( session, self.publisher, page_id="page-1", expected_revision=published["revision"], recorded_at=NOW + timedelta(minutes=4), change_reason="Superseded.", idempotency_key="archive-1", ) session.commit() self.assertEqual("archived", archived["state"]) self.assertIsNone(get_page(session, self.reader, page_id="page-1")) self.assertEqual( 0, len( WikiSearchSource() .backfill( session, request=SearchBackfillRequest( tenant_id="tenant-1", provider_id=PROVIDER_ID, resource_type=RESOURCE_TYPE, rebuild_id="wiki-archive-test", limit=10, ), ) .documents ), ) def test_dsar_exports_only_exact_subject_participation_and_manual_review( self, ) -> None: with self.Session() as session: self._space(session) self._page(session) add_comment( session, self.editor, page_id="page-1", expected_revision=1, comment_id="comment-dsar", body="My attributable contribution.", recorded_at=NOW + timedelta(minutes=1), idempotency_key="comment-dsar", ) session.commit() provider = WikiDsarProvider() subject = DsarSubjectRef(account_id="editor-1") records = provider.search_subject( session, tenant_id="tenant-1", subject=subject ) self.assertEqual(1, len(records)) self.assertEqual( "My attributable contribution.", records[0].data["subject_comments"][0]["body"], ) self.assertNotIn("body", records[0].data) self.assertEqual( (), provider.search_subject( session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="editor-1", external_references={"wiki.account": "different-account"}, ), ), ) actions = provider.plan_erasure( session, tenant_id="tenant-1", subject=subject, records=records ) self.assertEqual("manual_review", actions[0].kind) self.assertFalse(actions[0].executable) result = provider.execute_erasure( session, tenant_id="tenant-1", subject=subject, actions=actions, request_id="dsar-1", ) self.assertEqual("blocked", result[0].status) def _space(self, session): item = create_space( session, self.admin, space_id="space-1", space_key="knowledge", title="Knowledge", description="Governed knowledge", visibility="restricted", acl_tokens=["group:knowledge"], publish_mode="publishers", recorded_at=NOW, change_reason="Create governed space.", idempotency_key="space-1", ) self.assertEqual(1, session.query(WikiSpaceHistory).count()) return item def _page(self, session): return create_page( session, self.editor, page_id="page-1", space_id="space-1", parent_page_id=None, slug="service-guide", title="Service guide", body="Published body", summary="How to deliver the service.", inherits_access=True, visibility="tenant", acl_tokens=[], labels=["Guide"], links=[ { "link_id": "file-1", "kind": "attachment", "owner_module": "files", "resource_type": "file", "resource_id": "file-1", "label": "Source", "url": "/files/file-1", } ], recorded_at=NOW, change_reason="Create the guide.", idempotency_key="page-1", ) if __name__ == "__main__": unittest.main()