GovOPlaN/govoplan-projects
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: implement governed project portfolio

Browse Source
This commit is contained in:
zemion
2026-08-01 17:48:38 +02:00
parent df8c3907c5
commit 8bfcf2d6b4
23 changed files with 4577 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_manifest.py
+10 -2
View File
@@ -23,8 +23,16 @@ class ProjectsManifestTests(unittest.TestCase):
self.assertIn("connectors", manifest.optional_dependencies)
self.assertIn("tickets", manifest.optional_dependencies)
self.assertTrue(manifest.documentation)
self.assertIsNone(manifest.route_factory)
self.assertIsNone(manifest.frontend)
self.assertEqual("vertical_slice", manifest.architecture.maturity)
self.assertIn(
"project benefit and change-impact review",
manifest.architecture.owned_concepts,
)
self.assertIsNotNone(manifest.route_factory)
self.assertEqual("@govoplan/projects-webui", manifest.frontend.package_name)
self.assertIsNotNone(manifest.migration_spec)
self.assertEqual("projects.registry", manifest.provides_interfaces[0].name)
self.assertEqual(1, len(manifest.search_sources))
if __name__ == "__main__":
tests/test_project_service.py
+314
View File
@@ -0,0 +1,314 @@
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()