feat: implement governed wiki vertical slice

tests/test_manifest.py

@@ -2,37 +2,45 @@ from __future__ import annotations
 
 import unittest
 
-from govoplan_wiki.backend.manifest import (
+from govoplan_wiki.backend.dsar_provider import WIKI_DSAR_CAPABILITY
+from govoplan_wiki.backend.manifest import get_manifest
+from govoplan_wiki.backend.service import (
     ADMIN_SCOPE,
+    CAPABILITY_WIKI_REGISTRY,
+    COMMENT_SCOPE,
+    PUBLISH_SCOPE,
     READ_SCOPE,
     WRITE_SCOPE,
-    get_manifest,
 )
 
 
 class WikiManifestTests(unittest.TestCase):
-    def test_manifest_registers_domain_seed(self) -> None:
+    def test_manifest_registers_complete_vertical_slice(self) -> None:
         manifest = get_manifest()
-
         self.assertEqual("wiki", manifest.id)
+        self.assertEqual("0.1.20", manifest.version)
         self.assertEqual(("access",), manifest.dependencies)
         self.assertEqual(
-            {READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE},
-            {permission.scope for permission in manifest.permissions},
+            {READ_SCOPE, WRITE_SCOPE, COMMENT_SCOPE, PUBLISH_SCOPE, ADMIN_SCOPE},
+            {item.scope for item in manifest.permissions},
         )
-        self.assertIn("search", manifest.optional_dependencies)
-        self.assertTrue(manifest.documentation)
-        topic = manifest.documentation[0]
-        self.assertEqual("reference", topic.metadata["kind"])
-        self.assertIn("seed_boundary", topic.metadata["consequence_classes"])
-        self.assertTrue(
-            all(
-                topic.translations.get("de", {}).get(field)
-                for field in ("title", "summary", "body")
-            )
+        self.assertEqual(
+            {CAPABILITY_WIKI_REGISTRY, WIKI_DSAR_CAPABILITY},
+            {item.name for item in manifest.provides_interfaces},
         )
-        self.assertIsNone(manifest.route_factory)
-        self.assertIsNone(manifest.frontend)
+        self.assertIsNotNone(manifest.route_factory)
+        self.assertEqual("@govoplan/wiki-webui", manifest.frontend.package_name)
+        self.assertEqual(
+            ("wiki.pages",), tuple(item.id for item in manifest.search_sources)
+        )
+        self.assertIsNotNone(manifest.migration_spec)
+        self.assertTrue(manifest.migration_spec.retirement_supported)
+        self.assertTrue(manifest.uninstall_guard_providers)
+        self.assertEqual("vertical_slice", manifest.architecture.maturity)
+        self.assertGreaterEqual(len(manifest.documentation), 5)
+        for topic in manifest.documentation:
+            self.assertIn("de", topic.translations)
+            self.assertTrue(topic.metadata.get("kind"))
 
 
 if __name__ == "__main__":

tests/test_wiki_service.py

@@ -0,0 +1,390 @@
+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()