391 lines
14 KiB
Python
391 lines
14 KiB
Python
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()
|