from __future__ import annotations import unittest from sqlalchemy import create_engine from sqlalchemy.orm import Session from govoplan_core.auth import ApiPrincipal from govoplan_core.core.access import PrincipalRef from govoplan_core.core.semantic_documentation import ( SemanticDocumentationSubjectQuery, ) from govoplan_core.db.base import Base from govoplan_workflow.backend.semantic_subjects import ( WorkflowSemanticDocumentationSubjectProvider, ) from govoplan_workflow_engine.backend.db.models import ( WorkflowDefinition, WorkflowDefinitionRevision, ) from govoplan_workflow_engine.backend.schemas import ( WorkflowDefinitionCreateRequest, WorkflowDefinitionUpdateRequest, WorkflowEdge, WorkflowGraph, WorkflowNode, WorkflowPosition, ) from govoplan_workflow_engine.backend.service import ( create_definition, delete_definition, update_definition, ) def principal( *, tenant_id: str = "tenant-1", scopes: frozenset[str] = frozenset({"workflow:definition:read"}), ) -> ApiPrincipal: return ApiPrincipal( principal=PrincipalRef( account_id="author-1", membership_id="membership-1", tenant_id=tenant_id, scopes=scopes, ), account=object(), user=object(), ) def graph( *, activity_label: str = "Review", activity_x: float = 280, include_activity: bool = True, ) -> WorkflowGraph: nodes = [ WorkflowNode( id="start", type="workflow.start.manual", label="Start", position=WorkflowPosition(x=40, y=100), ), ] if include_activity: nodes.append( WorkflowNode( id="activity", type="workflow.activity", label=activity_label, position=WorkflowPosition(x=activity_x, y=100), config={ "instructions": "Check the submitted evidence.", "api_secret": "must-not-leak", }, ) ) nodes.append( WorkflowNode( id="complete", type="workflow.end.completed", label="Completed", position=WorkflowPosition(x=520, y=100), ) ) edges = [] if include_activity: edges.extend( ( WorkflowEdge(id="start-activity", source="start", target="activity"), WorkflowEdge( id="activity-complete", source="activity", target="complete", ), ) ) else: edges.append( WorkflowEdge(id="start-complete", source="start", target="complete") ) return WorkflowGraph(nodes=nodes, edges=edges) class WorkflowSemanticSubjectTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite+pysqlite:///:memory:") Base.metadata.create_all( self.engine, tables=( WorkflowDefinition.__table__, WorkflowDefinitionRevision.__table__, ), ) self.session = Session(self.engine) self.principal = principal() self.provider = WorkflowSemanticDocumentationSubjectProvider(None) self.definition = create_definition( self.session, tenant_id="tenant-1", actor_id="author-1", payload=WorkflowDefinitionCreateRequest( name="Resident permit review", description="Review an application before issuing the permit.", graph=graph(), ), ) self.session.commit() def tearDown(self) -> None: self.session.close() self.engine.dispose() def subjects(self): return self.provider.list_subjects( self.session, self.principal, request=SemanticDocumentationSubjectQuery( tenant_id="tenant-1", limit=200, ), ).subjects def activity(self): return next( item for item in self.subjects() if item.route_anchor == "workflow-node-activity" ) def revise(self, updated_graph: WorkflowGraph) -> None: update_definition( self.session, tenant_id="tenant-1", definition_id=self.definition.id, actor_id="author-1", payload=WorkflowDefinitionUpdateRequest( name=self.definition.name, description=self.definition.description, graph=updated_graph, expected_revision=self.definition.current_revision, ), ) self.session.commit() def test_exposes_safe_definition_and_step_descriptors(self) -> None: subjects = self.subjects() self.assertEqual(4, len(subjects)) definition = next(item for item in subjects if item.reference.anchor is None) activity = self.activity() self.assertEqual("Resident permit review", definition.labels["en"]) self.assertEqual("step", activity.reference.anchor.kind) self.assertTrue(activity.reference.anchor.id.startswith("step-")) self.assertEqual("/workflow?definition=" + self.definition.id, activity.route) payload = activity.to_dict() self.assertNotIn("config", payload) self.assertNotIn("must-not-leak", str(payload)) def test_lineage_survives_layout_and_label_changes_with_review_fingerprint(self) -> None: before = self.activity().reference self.revise(graph(activity_x=900)) after_layout = self.activity().reference self.assertEqual(before.stable_key, after_layout.stable_key) self.assertEqual( before.observed_fingerprint, after_layout.observed_fingerprint, ) self.revise(graph(activity_x=900, activity_label="Assess evidence")) after_label = self.activity().reference self.assertEqual(before.stable_key, after_label.stable_key) self.assertNotEqual( before.observed_fingerprint, after_label.observed_fingerprint, ) resolution = self.provider.resolve_subject( self.session, self.principal, reference=before, ) self.assertEqual("changed", resolution.availability) def test_deleted_and_recreated_step_id_gets_new_lineage(self) -> None: old = self.activity().reference self.revise(graph(include_activity=False)) deleted = self.provider.resolve_subject( self.session, self.principal, reference=old, ) self.assertEqual("missing", deleted.availability) self.assertEqual("step_deleted", deleted.reason_code) self.revise(graph(activity_label="Recreated review")) recreated = self.activity().reference self.assertNotEqual(old.stable_key, recreated.stable_key) still_deleted = self.provider.resolve_subject( self.session, self.principal, reference=old, ) self.assertEqual("step_deleted", still_deleted.reason_code) def test_resolution_rechecks_scope_tenant_governance_and_deletion(self) -> None: reference = self.activity().reference self.assertIsNone( self.provider.resolve_subject( self.session, principal(scopes=frozenset()), reference=reference, ) ) self.assertIsNone( self.provider.resolve_subject( self.session, principal(tenant_id="tenant-2"), reference=reference, ) ) admin = principal(scopes=frozenset({"workflow:instance:admin"})) self.assertIsNotNone( self.provider.resolve_subject( self.session, admin, reference=reference, ) ) delete_definition( self.session, tenant_id="tenant-1", definition_id=self.definition.id, actor_id="author-1", ) self.session.commit() deleted = self.provider.resolve_subject( self.session, self.principal, reference=reference, ) self.assertEqual("missing", deleted.availability) self.assertEqual("definition_deleted", deleted.reason_code) def test_query_filters_and_pages_without_docs_runtime(self) -> None: page = self.provider.list_subjects( self.session, self.principal, request=SemanticDocumentationSubjectQuery( tenant_id="tenant-1", query="review", limit=1, ), ) self.assertEqual(1, len(page.subjects)) self.assertTrue(page.has_more) second = self.provider.list_subjects( self.session, self.principal, request=SemanticDocumentationSubjectQuery( tenant_id="tenant-1", query="review", limit=1, cursor=page.next_cursor, ), ) self.assertEqual(1, len(second.subjects)) if __name__ == "__main__": unittest.main()