294 lines
9.2 KiB
Python
294 lines
9.2 KiB
Python
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()
|