3 Commits
Author SHA1 Message Date
zemion a2fc5639db perf(workflow): batch revision reads and add bounded histories 2026-09-08 07:47:18 +02:00
zemion 4c447671b0 docs: complete German structured documentation
Module Package Release / publish-packages (push) Successful in 11s
2026-08-24 01:15:42 +02:00
zemion d338403dbd docs(workflow-engine): complete German reference coverage
Module Package Release / publish-packages (push) Successful in 11s
2026-08-23 20:27:27 +02:00
14 changed files with 1351 additions and 50 deletions
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "govoplan-workflow-engine"
version = "0.1.19"
version = "0.1.21"
description = "Headless, versioned workflow definition and execution engine for GovOPlaN."
readme = "README.md"
requires-python = ">=3.12"
+1 -1
View File
@@ -1,3 +1,3 @@
"""Headless GovOPlaN Workflow Engine module."""
__version__ = "0.1.18"
__version__ = "0.1.21"
@@ -298,6 +298,19 @@ class WorkflowInstance(Base, TimestampMixin):
"status",
"updated_at",
),
Index(
"ix_workflow_instances_tenant_created_id",
"tenant_id",
"created_at",
"id",
),
Index(
"ix_workflow_instances_tenant_definition_created_id",
"tenant_id",
"definition_id",
"created_at",
"id",
),
)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
@@ -383,6 +396,9 @@ class WorkflowInstance(Base, TimestampMixin):
)
definition: Mapped[WorkflowDefinition] = relationship(back_populates="instances")
definition_revision: Mapped[WorkflowDefinitionRevision] = relationship(
viewonly=True,
)
steps: Mapped[list["WorkflowInstanceStep"]] = relationship(
back_populates="instance",
cascade="all, delete-orphan",
@@ -0,0 +1,14 @@
"""German translations for public structured documentation metadata."""
from __future__ import annotations
from typing import Any
GERMAN_STRUCTURED_TRANSLATIONS: dict[str, dict[str, Any]] = {'workflow.external-campaign-handoffs': {'limitations': ['Der Kampagnenanbieter muss installiert '
'sein, um eine Kampagnenübergabe zu '
'starten oder wieder aufzunehmen.',
'Die fokussierte View-Projektion ist '
'optional; der sichere '
'Kampagnen-Aktionslink bleibt ohne sie '
'verfügbar.']}}
@@ -0,0 +1,305 @@
"""Authorized, bounded projections for workflow progress and retained history."""
from __future__ import annotations
import base64
import binascii
from datetime import datetime
import json
from sqlalchemy import and_, func, or_, select
from sqlalchemy.orm import Session, joinedload, load_only, raiseload
from govoplan_core.auth import ApiPrincipal, has_scope
from govoplan_workflow_engine.backend.db.models import (
WorkflowDefinitionRevision,
WorkflowInstance,
WorkflowInstanceEvent,
WorkflowInstanceStep,
)
from govoplan_workflow_engine.backend.governance import (
definition_decision,
require_definition_action,
)
from govoplan_workflow_engine.backend.instance_service import (
event_response,
step_response,
)
from govoplan_workflow_engine.backend.schemas import (
WorkflowInstanceEventPageResponse,
WorkflowInstanceStepPageResponse,
WorkflowInstanceSummaryListResponse,
WorkflowInstanceSummaryResponse,
)
from govoplan_workflow_engine.backend.service import WorkflowNotFoundError
MAX_PAGE_SIZE = 200
MAX_HISTORY_SEQUENCE = 2_147_483_647
_CANDIDATE_BATCH_SIZE = 250
def _require_read(principal: ApiPrincipal) -> None:
if not any(
has_scope(principal, scope)
for scope in ("workflow:instance:read", "workflow:instance:admin")
):
raise PermissionError("Workflow instance read permission is required.")
def _summary_statement():
# Raiseload makes accidental history/payload expansion fail visibly. Only the
# pinned revision metadata is needed, never its graph or BPMN document.
return (
select(WorkflowInstance)
.options(
load_only(
WorkflowInstance.tenant_id,
WorkflowInstance.definition_id,
WorkflowInstance.definition_revision_id,
WorkflowInstance.status,
WorkflowInstance.start_origin,
WorkflowInstance.current_step_id,
WorkflowInstance.started_at,
WorkflowInstance.finished_at,
WorkflowInstance.created_at,
WorkflowInstance.updated_at,
raiseload=True,
),
joinedload(WorkflowInstance.definition),
joinedload(WorkflowInstance.definition_revision).load_only(
WorkflowDefinitionRevision.tenant_id,
WorkflowDefinitionRevision.definition_id,
WorkflowDefinitionRevision.revision,
WorkflowDefinitionRevision.content_hash,
WorkflowDefinitionRevision.execution_mode,
raiseload=True,
),
raiseload("*"),
)
.execution_options(populate_existing=True)
)
def _require_evidence(instance: WorkflowInstance) -> None:
definition = instance.definition
revision = instance.definition_revision
if (
definition is None
or revision is None
or revision.id != instance.definition_revision_id
or definition.id != instance.definition_id
or definition.tenant_id not in {None, instance.tenant_id}
or revision.definition_id != definition.id
or revision.tenant_id != definition.tenant_id
):
raise WorkflowNotFoundError(
"Workflow instance definition evidence is incomplete."
)
def _summary(instance: WorkflowInstance) -> WorkflowInstanceSummaryResponse:
_require_evidence(instance)
return WorkflowInstanceSummaryResponse(
id=instance.id,
definition_id=instance.definition_id,
definition_name=instance.definition.name,
definition_revision=instance.definition_revision.revision,
definition_hash=instance.definition_revision.content_hash,
execution_mode=instance.definition_revision.execution_mode,
start_origin=instance.start_origin,
status=instance.status,
current_step_id=instance.current_step_id,
started_at=instance.started_at,
finished_at=instance.finished_at,
created_at=instance.created_at,
updated_at=instance.updated_at,
)
def _cursor(instance: WorkflowInstance) -> str:
return base64.urlsafe_b64encode(
json.dumps([instance.created_at.isoformat(), instance.id]).encode("utf-8")
).decode("ascii")
def _cursor_boundary(cursor: str) -> tuple[datetime, str]:
try:
if len(cursor) > 256:
raise ValueError
payload = json.loads(base64.b64decode(cursor, altchars=b"-_", validate=True))
if (
not isinstance(payload, list)
or len(payload) != 2
or not all(isinstance(value, str) for value in payload)
or not 1 <= len(payload[1]) <= 36
):
raise ValueError
return datetime.fromisoformat(payload[0]), payload[1]
except (ValueError, TypeError, UnicodeError, binascii.Error) as exc:
raise ValueError("Invalid workflow instance cursor.") from exc
def list_instance_summaries(
session: Session,
*,
principal: ApiPrincipal,
registry: object | None,
definition_id: str | None = None,
limit: int = 100,
cursor: str | None = None,
) -> WorkflowInstanceSummaryListResponse:
_require_read(principal)
if not 1 <= limit <= MAX_PAGE_SIZE:
raise ValueError("Workflow page size must be between 1 and 200.")
boundary = _cursor_boundary(cursor) if cursor is not None else None
statement = _summary_statement().where(
WorkflowInstance.tenant_id == principal.tenant_id,
)
if definition_id is not None:
statement = statement.where(WorkflowInstance.definition_id == definition_id)
statement = statement.order_by(
WorkflowInstance.created_at.desc(), WorkflowInstance.id.desc()
)
items: list[WorkflowInstanceSummaryResponse] = []
last_cursor: str | None = None
# Authorization can depend on a Core policy provider, so a raw SQL count or
# SQL limit before authorization cannot define an authorized page. Scan in
# bounded batches until the page and one authorized lookahead are found.
while True:
decisions: dict[str, bool] = {}
batch_statement = statement
if boundary is not None:
created_at, instance_id = boundary
batch_statement = batch_statement.where(
or_(
WorkflowInstance.created_at < created_at,
and_(
WorkflowInstance.created_at == created_at,
WorkflowInstance.id < instance_id,
),
)
)
candidates = list(session.scalars(batch_statement.limit(_CANDIDATE_BATCH_SIZE)))
for instance in candidates:
_require_evidence(instance)
allowed = decisions.get(instance.definition_id)
if allowed is None:
allowed = definition_decision(
instance.definition,
principal=principal,
registry=registry,
action="view",
).allowed
decisions[instance.definition_id] = allowed
if not allowed:
continue
if len(items) == limit:
return WorkflowInstanceSummaryListResponse(
instances=items, next_cursor=last_cursor
)
items.append(_summary(instance))
last_cursor = _cursor(instance)
if len(candidates) < _CANDIDATE_BATCH_SIZE:
return WorkflowInstanceSummaryListResponse(instances=items)
last = candidates[-1]
boundary = (last.created_at, last.id)
def _authorized_instance(
session: Session,
*,
principal: ApiPrincipal,
registry: object | None,
instance_id: str,
) -> WorkflowInstance:
_require_read(principal)
instance = session.scalar(
_summary_statement().where(
WorkflowInstance.tenant_id == principal.tenant_id,
WorkflowInstance.id == instance_id,
)
)
if instance is None:
raise WorkflowNotFoundError("Workflow instance not found.")
_require_evidence(instance)
require_definition_action(
instance.definition, principal=principal, registry=registry, action="view"
)
return instance
def get_instance_summary(
session: Session,
*,
principal: ApiPrincipal,
registry: object | None,
instance_id: str,
) -> WorkflowInstanceSummaryResponse:
return _summary(
_authorized_instance(
session, principal=principal, registry=registry, instance_id=instance_id
)
)
def instance_history_page(
session: Session,
*,
principal: ApiPrincipal,
registry: object | None,
instance_id: str,
kind: str,
limit: int = 100,
after_sequence: int = 0,
through_sequence: int | None = None,
) -> WorkflowInstanceStepPageResponse | WorkflowInstanceEventPageResponse:
if kind not in {"steps", "events"}:
raise ValueError("Workflow history kind must be steps or events.")
if (
not 1 <= limit <= MAX_PAGE_SIZE
or not 0 <= after_sequence <= MAX_HISTORY_SEQUENCE
or (
through_sequence is not None
and not 0 <= through_sequence <= MAX_HISTORY_SEQUENCE
)
):
raise ValueError("Invalid workflow history page bounds.")
_authorized_instance(
session, principal=principal, registry=registry, instance_id=instance_id
)
model = WorkflowInstanceStep if kind == "steps" else WorkflowInstanceEvent
predicates = [
model.tenant_id == principal.tenant_id,
model.instance_id == instance_id,
]
if through_sequence is not None:
predicates.append(model.sequence <= through_sequence)
total, last_sequence = session.execute(
select(func.count(), func.max(model.sequence)).where(*predicates)
).one()
upper = int(last_sequence or 0)
rows = list(
session.scalars(
select(model)
.where(
*predicates, model.sequence > after_sequence, model.sequence <= upper
)
.order_by(model.sequence.asc())
.limit(limit + 1)
.execution_options(populate_existing=True)
)
)
next_after = rows[limit - 1].sequence if len(rows) > limit else None
common = {
"total": int(total),
"through_sequence": upper,
"next_after_sequence": next_after,
}
if kind == "steps":
return WorkflowInstanceStepPageResponse(
steps=[step_response(row) for row in rows[:limit]], **common
)
return WorkflowInstanceEventPageResponse(
events=[event_response(row) for row in rows[:limit]], **common
)
@@ -38,7 +38,6 @@ from govoplan_core.core.events import (
)
from govoplan_core.db.base import utcnow
from govoplan_workflow_engine.backend.db.models import (
WorkflowDefinition,
WorkflowDefinitionRevision,
WorkflowInstance,
WorkflowInstanceEvent,
@@ -100,6 +99,7 @@ def list_instances(
selectinload(WorkflowInstance.steps),
selectinload(WorkflowInstance.events),
selectinload(WorkflowInstance.definition),
selectinload(WorkflowInstance.definition_revision),
)
.order_by(
WorkflowInstance.updated_at.desc(),
@@ -129,6 +129,7 @@ def get_instance(
selectinload(WorkflowInstance.steps),
selectinload(WorkflowInstance.events),
selectinload(WorkflowInstance.definition),
selectinload(WorkflowInstance.definition_revision),
)
)
if for_update:
@@ -861,12 +862,17 @@ def instance_response(
*,
replayed: bool = False,
) -> WorkflowInstanceResponse:
revision = session.get(
WorkflowDefinitionRevision,
instance.definition_revision_id,
)
definition = session.get(WorkflowDefinition, instance.definition_id)
if revision is None or definition is None:
revision = instance.definition_revision
definition = instance.definition
if (
revision is None
or definition is None
or revision.id != instance.definition_revision_id
or definition.id != instance.definition_id
or definition.tenant_id not in {None, instance.tenant_id}
or revision.definition_id != definition.id
or revision.tenant_id != definition.tenant_id
):
raise WorkflowNotFoundError(
"Workflow instance definition evidence is incomplete."
)
@@ -894,43 +900,45 @@ def instance_response(
created_by=instance.created_by,
created_at=instance.created_at,
updated_at=instance.updated_at,
steps=[
WorkflowInstanceStepResponse(
id=step.id,
sequence=step.sequence,
node_id=step.node_id,
node_type=step.node_type,
status=step.status, # type: ignore[arg-type]
attempt=step.attempt,
input=dict(step.input_),
output=dict(step.output_),
handoff=dict(step.handoff),
external_ref=step.external_ref,
started_at=step.started_at,
finished_at=step.finished_at,
error=step.error,
completed_by=step.completed_by,
created_at=step.created_at,
updated_at=step.updated_at,
)
for step in instance.steps
],
events=[
WorkflowInstanceEventResponse(
id=event.id,
sequence=event.sequence,
step_id=event.step_id,
kind=event.kind,
actor_id=event.actor_id,
payload=dict(event.payload),
created_at=event.created_at,
)
for event in instance.events
],
steps=[step_response(step) for step in instance.steps],
events=[event_response(event) for event in instance.events],
replayed=replayed,
)
def step_response(step: WorkflowInstanceStep) -> WorkflowInstanceStepResponse:
return WorkflowInstanceStepResponse(
id=step.id,
sequence=step.sequence,
node_id=step.node_id,
node_type=step.node_type,
status=step.status, # type: ignore[arg-type]
attempt=step.attempt,
input=dict(step.input_),
output=dict(step.output_),
handoff=dict(step.handoff),
external_ref=step.external_ref,
started_at=step.started_at,
finished_at=step.finished_at,
error=step.error,
completed_by=step.completed_by,
created_at=step.created_at,
updated_at=step.updated_at,
)
def event_response(event: WorkflowInstanceEvent) -> WorkflowInstanceEventResponse:
return WorkflowInstanceEventResponse(
id=event.id,
sequence=event.sequence,
step_id=event.step_id,
kind=event.kind,
actor_id=event.actor_id,
payload=dict(event.payload),
created_at=event.created_at,
)
def _drive_instance(
session: Session,
*,
@@ -1,5 +1,8 @@
from __future__ import annotations
from govoplan_core.core.modules import with_documentation_structured_translations
from govoplan_workflow_engine.backend.german_structured_documentation import GERMAN_STRUCTURED_TRANSLATIONS
from pathlib import Path
from govoplan_core.core.access import (
@@ -60,7 +63,7 @@ from govoplan_workflow_engine.backend.service_launcher import (
MODULE_ID = "workflow_engine"
MODULE_NAME = "Workflow Engine"
MODULE_VERSION = "0.1.19"
MODULE_VERSION = "0.1.21"
DEFINITION_READ_SCOPE = "workflow:definition:read"
DEFINITION_WRITE_SCOPE = "workflow:definition:write"
@@ -386,6 +389,99 @@ manifest = ModuleManifest(
),
),
documentation=(
DocumentationTopic(
id="workflow.instance-history",
title="Read workflow progress and paged history",
summary="Inspect pinned execution evidence with explicit summary and history pages.",
body=(
"The Workflow Engine API keeps full steps and events on the existing "
"GET /instances and GET /instances/{id} responses; no history is silently "
"truncated. Integrators handling long-running processes can use "
"GET /instances/summaries and GET /instances/{id}/summary for identity, "
"pinned definition revision and hash, execution mode, status, current step "
"identifier, and timestamps without runtime payloads or history. Summary "
"lists accept definition_id, limit (default 100, maximum 200), and cursor. "
"They order by creation time and identifier, newest first. Continue with "
"next_cursor until it is null, keeping the same filter. Pages include only "
"currently visible definitions and do not expose a global total. Newer "
"instances require a fresh first page; updates do not reorder older pages. "
"GET /instances/{id}/steps and /events accept limit (default 100, maximum "
"200), after_sequence (initially 0), and through_sequence (sequence bounds "
"range from 0 to 2147483647). Continue using "
"next_after_sequence and the returned through_sequence until the next "
"position is null. Rows are ordered by increasing sequence; total counts "
"all retained rows up to that boundary, including earlier pages. Keeping "
"the boundary excludes subsequently appended history; omit it for a fresh "
"pass. Step fields reflect their current state, not a historical database "
"snapshot. Each request requires workflow:instance:read or "
"workflow:instance:admin and current definition view authorization in "
"the active tenant. A cursor and the authorization recorded at start "
"never grant access after permission changes. Administrators should use "
"summary and history pages for large runs: revision metadata is loaded "
"in batches, history rows are bounded, and summary discovery checks "
"candidates in batches of 250 until an authorized page is filled. Highly "
"restricted catalogues can still require scanning many candidates. "
"The module migration adds tenant and definition creation-order indexes; "
"building them on a large installation needs disk capacity and may "
"temporarily block writes during the normal database upgrade. "
"Work inbox totals remain exact after current assignment, governance, "
"and search filters; counting them still streams all matching candidates "
"in batches of 250 and can take longer for large inboxes. These read "
"contracts require no editor, Tasks installation, or new permission."
),
layer="available",
documentation_types=("admin", "user"),
audience=("operator", "module_admin", "integrator", "auditor"),
order=74,
translations={
"de": {
"title": "Workflow-Fortschritt und paginierten Verlauf lesen",
"summary": "Revisionsgebundene Ausführungsnachweise mit expliziten Übersichts- und Verlaufsseiten prüfen.",
"body": (
"Die Workflow-Engine-API liefert bei GET /instances und GET "
"/instances/{id} weiterhin vollständige Schritte und Ereignisse; "
"Verläufe werden nicht stillschweigend gekürzt. Für lange Prozesse "
"liefern GET /instances/summaries und GET /instances/{id}/summary "
"Identität, gebundene Definitionsrevision und Hash, Ausführungsmodus, "
"Status, aktuelle Schrittkennung und Zeitstempel ohne Laufzeitnutzdaten "
"oder Verlauf. Übersichtslisten akzeptieren definition_id, limit "
"(Standard 100, höchstens 200) und cursor. Sie sortieren nach "
"Erstellungszeit und Kennung, neueste zuerst. Mit unverändertem Filter "
"und next_cursor fortfahren, bis dieser null ist. Seiten enthalten nur "
"aktuell sichtbare Definitionen und keine globale Gesamtzahl. Neue "
"Instanzen erfordern eine neue erste Seite; Aktualisierungen verschieben "
"ältere Seiten nicht. GET /instances/{id}/steps und /events akzeptieren "
"limit (Standard 100, höchstens 200), after_sequence (anfangs 0) und "
"through_sequence (Sequenzgrenzen von 0 bis 2147483647). Mit "
"next_after_sequence und der zurückgegebenen "
"through_sequence fortfahren, bis die nächste Position null ist. Zeilen "
"sind nach aufsteigender Sequenz sortiert; total zählt alle erhaltenen "
"Zeilen bis zur Grenze einschließlich vorheriger Seiten. Eine feste "
"Grenze schließt später angehängte Einträge aus; für einen neuen "
"Durchlauf wird sie weggelassen. Schrittfelder zeigen ihren aktuellen "
"Zustand, keinen historischen Datenbank-Snapshot. Jede Anfrage benötigt "
"workflow:instance:read oder workflow:instance:admin und die aktuelle "
"Sichtberechtigung für die Definition im aktiven Mandanten. Weder ein "
"Cursor noch die beim Start gespeicherte Autorisierung gewährt Zugriff "
"nach Berechtigungsänderungen. Administratoren sollten für große Läufe "
"Übersichts- und Verlaufsseiten verwenden: Revisionsmetadaten werden "
"gebündelt geladen, Verlaufszeilen sind begrenzt, und die Übersicht "
"prüft Kandidaten in Gruppen von 250, bis eine berechtigte Seite gefüllt "
"ist. Stark eingeschränkte Kataloge können weiterhin viele "
"Kandidatenprüfungen erfordern. Die Modulmigration ergänzt nach "
"Erstellungszeit geordnete Mandanten- und Definitionsindizes. Deren "
"Aufbau benötigt bei großen Installationen Speicherplatz und kann "
"während der regulären Datenbankaktualisierung Schreibzugriffe "
"vorübergehend blockieren. Gesamtzahlen im Arbeitskorb bleiben "
"nach aktuellen Zuweisungs-, Governance- und Suchfiltern exakt. Dafür "
"werden weiterhin alle passenden Kandidaten in Gruppen von 250 "
"durchlaufen; große Arbeitskörbe können länger dauern. Diese "
"Leseverträge benötigen weder Editor noch Tasks-Installation noch "
"eine neue Berechtigung."
),
},
},
),
DocumentationTopic(
id="workflow.external-campaign-handoffs",
title="Resume Workflow from accountable Campaign work",
@@ -410,7 +506,12 @@ manifest = ModuleManifest(
),
layer="available",
documentation_types=("admin", "user"),
audience=("operator", "workflow_designer", "campaign_manager", "module_admin"),
audience=(
"operator",
"workflow_designer",
"campaign_manager",
"module_admin",
),
related_modules=("campaigns", "views", "tasks", "notifications", "audit"),
conditions=(
DocumentationCondition(
@@ -418,6 +519,26 @@ manifest = ModuleManifest(
),
),
order=75,
translations={
"de": {
"title": "Workflow aus verantwortlicher Campaign-Arbeit fortsetzen",
"summary": (
"Revisionsgebundene Campaign-Arbeit öffnen, die relevante Oberfläche fokussieren und ohne Polling aus deklarierten "
"Lebenszyklusereignissen fortfahren."
),
"body": (
"Der externe Übergabebaustein folgt einer Modulaktion, die eine Campaign anlegt oder referenziert und genau eine "
"verantwortliche Arbeitszuweisung öffnet. Workflow speichert exakte Campaign-Version und Zuweisungsrevision, einen "
"sicheren Aktionslink, Korrelations- und Idempotenzherkunft sowie ein dauerhaftes Ereignisabonnement. Zugewiesen-, "
"angenommen-, begonnen- und neu-zugewiesen-Ereignisse aktualisieren den Wartezustand; abgeschlossene, abgelehnte, "
"abgebrochene und abgelaufene Ergebnisse nehmen getrennte Graphpfade. Doppelte Ereignisse werden ignoriert. Vor der "
"Fortsetzung durch ein Abschlussereignis löst Workflow seine Befugnis erneut auf und Campaign prüft Zugriff und "
"Ereignisrevision. Eine Zuweisung gewährt niemals Campaign-Zugriff. Fehlende Campaign- oder Views-Fähigkeiten halten die "
"Instanz prüfbar, aber blockiert. Fehlende Tasks- oder Notifications-Integration wird als eingeschränkt gemeldet und "
"ändert das Campaign-Ergebnis nicht."
),
}
},
metadata={
"kind": "workflow",
"help_contexts": [
@@ -443,6 +564,27 @@ manifest = ModuleManifest(
audience=("user", "operator", "module_admin", "auditor"),
related_modules=("core", "cases", "forms_runtime", "services", "tasks"),
order=75,
translations={
"de": {
"title": "Betroffenenanfragen für Workflows",
"summary": (
"Strukturierte Betroffenenverknüpfungen prüfen, ohne beliebige Prozessnutzdaten offenzulegen oder zu erraten."
),
"body": (
"Workflow Engine gleicht exakte mandantenbezogene Kennungen für Definition, Revision, Instanz, Schritt, Ereignis, "
"Auslöser, Lieferung und Wartezustand ab. Zusätzlich werden strukturierte Instanzautorisierung, Arbeitszuweisung, "
"Automatisierungsbefugnis und minimierte Personalzuordnung für Konto-, Identitäts- und Mitgliedschaftsselektoren "
"abgeglichen. DSAR-Ergebnisse kopieren niemals Graphen oder BPMN, Eingaben, Kontext, Ausgaben, Übergaben, Ereignis-/"
"Konfigurationsnutzdaten, Autorisierungs-Snapshots, Fehler, Wiederholungsschlüssel, externe Verweise, Hashes oder "
"Zugangsdaten. Beliebige Laufzeitnutzdaten werden nicht nach Kennungen durchsucht, weil der zuständige Service, Vorgang, "
"das Formular oder ein anderes Quellmodul maßgeblich bleibt. Betroffenenbezogene Automatisierung kann deaktiviert und "
"widerrufen werden; ausdrücklich ausgewählte Details abgeschlossener Auslöserlieferungen können idempotent minimiert "
"werden, während ihr Wiederholungsschlüssel erhalten bleibt. Definitionen, Instanzen, laufende Arbeit, Zuweisungen, "
"unveränderliche Übergangs-/Entscheidungsereignisse, Wartezustände und Betriebszuordnung erfordern die Prüfung durch "
"Workflow-Owner sowie Zuständige für Aufbewahrung, Dritte und Quellhoheit."
),
}
},
metadata={
"help_contexts": [
"workflow.data-subject-requests",
@@ -478,6 +620,21 @@ manifest = ModuleManifest(
"audit",
),
order=76,
translations={
"de": {
"title": "Workflow-Definitionsgraphen",
"summary": "Gesteuerte Prozessgraphen unabhängig von einem Editor bereitstellen.",
"body": (
"Workflow Engine stellt eine Bibliothek aus Auslöser-, Aktivitäts-, Entscheidungs-, Warte-, Integrations- und "
"Ergebnisknoten auf Cores domänenneutralem Graphvertrag bereit. Anders als Dataflow erlaubt Workflow Zyklen für "
"Korrektur- und Wiederholungspfade. Modulaktionen werden über versionierte Fähigkeiten statt Implementierungsimporte "
"adressiert. Definitionen werden als unveränderliche Graphrevisionen gespeichert; die Aktivierung legt die exakte "
"Revision für zukünftige Instanzen fest. Die optionale Service-Startfähigkeit startet eine berechtigte aktive Revision "
"aus einer exakten Portal-Servicebindung und zeichnet diese Herkunft auf. Ist Tasks aktiviert, werden aktuelle menschliche "
"Übergaben mit typisierter Verantwortung, Fälligkeit und fortsetzbarem Quelllink in den gemeinsamen Arbeitskorb projiziert."
),
}
},
),
DocumentationTopic(
id="workflow.bpmn-interchange",
@@ -505,6 +662,21 @@ manifest = ModuleManifest(
audience=("operator", "module_admin", "power_user", "product_owner"),
related_modules=("audit", "policy", "views"),
order=77,
translations={
"de": {
"title": "BPMN-Modellierungs- und Ausführungsprofile",
"summary": "Verlustfreie BPMN-2.0-Revisionen mit ausdrücklicher, sicher schließender Ausführungskonformität.",
"body": (
"BPMN 2.0 ist die kanonische native Graphsprache von Workflow. Der gemeinsame Grapheditor modelliert BPMN-Knoten, "
"Flüsse, Verschachtelung und Diagrammgeometrie direkt, ohne getrennten browserseitigen Modeler. Importiertes XML wird in "
"den Graphen normalisiert und jede unveränderliche Revision legt ein deterministisches XML-Artefakt fest. Die Aktivierung "
"verlangt einen festgelegten Ausführungsadapter und eine Version, deren erklärtes Profil jede modellierte "
"Laufzeitsemantik akzeptiert. Nicht unterstützte Laufzeitkonstrukte bleiben bearbeitbar und exportierbar. Adapterpakete "
"integrieren über die Entry-Point-Gruppe govoplan.workflow.bpmn_adapters und müssen kanonischen Workflow-Laufzeitzustand "
"materialisieren; Workflow Engine importiert niemals ein konkretes Engine-Modul."
),
}
},
),
DocumentationTopic(
id="workflow.runtime-recovery",
@@ -529,6 +701,49 @@ manifest = ModuleManifest(
audience=("operator", "module_admin", "power_user"),
related_modules=("core", "dataflow", "audit"),
order=78,
translations={
"de": {
"title": "Workflow-Laufzeit wiederherstellen",
"summary": (
"Abgesicherte Modulaktionen, Zeitgeber und nachweisbasierte Wiederherstellung bei unbekannten Provider-Ergebnissen."
),
"body": (
"Workflow Engine zeichnet vor jeder folgenreichen Modulaktion einen Core-Wiederherstellungsvorgang auf und schreibt ein "
"abschließendes Provider-Ergebnis gemeinsam mit der lokalen Workflow-Projektion fest. Ein Timeout oder verlorene "
"Bestätigung nach nicht atomarer Auslieferung wird zu einem unbekannten Ergebnis und deaktiviert Wiederholen. "
"Betriebspersonen prüfen den Provider, halten Nachweise fest und wählen Wirkung bestätigt, um ohne Wiederholung fortzufahren, "
"oder Wirkung ausgeblieben, um eine bewusste Wiederholung zu erlauben. Instanz-Worker, Auslöserlieferungen und Zeitgeber "
"verwenden verteilte Sperren und werden vor der Zustandsübernahme nach Mandanten-Modulberechtigung partitioniert. Das "
"Deaktivieren von Workflow Engine bewahrt angenommene Instanzen, Wartezustände und Auslöserlieferungen für die betriebliche "
"Klärung. Verknüpfte Dataflow-Wiederherstellung bleibt blockiert, bis ihr Ergebnis eindeutig ist."
),
}
},
metadata={
"kind": "reference",
"help_contexts": ["workflow.instances", "workflow.runtime-recovery"],
"consequence_classes": {
"effect_confirmed": "Continues from retained evidence without replaying the provider effect.",
"effect_absent": "Enables one deliberate retry after evidence confirms that the provider effect did not occur.",
"disable_module": "Preserves accepted instances, waits, and trigger deliveries for operator resolution.",
},
},
structured_translation_version="1",
structured_translations={
"de": {
"consequence_classes": {
"effect_confirmed": (
"Setzt den Ablauf anhand aufbewahrter Nachweise fort, ohne die Provider-Wirkung zu wiederholen."
),
"effect_absent": (
"Erlaubt genau eine bewusste Wiederholung, nachdem Nachweise bestätigen, dass die Provider-Wirkung ausblieb."
),
"disable_module": (
"Bewahrt angenommene Instanzen, Wartezustände und Auslöserlieferungen für die betriebliche Klärung."
),
}
}
},
),
),
architecture=declared_module_architecture(
@@ -562,6 +777,11 @@ manifest = ModuleManifest(
)
manifest = with_documentation_structured_translations(
manifest, locale="de", translations=GERMAN_STRUCTURED_TRANSLATIONS
)
def get_manifest() -> ModuleManifest:
return manifest
@@ -0,0 +1,39 @@
"""Index tenant-scoped workflow summary pagination.
Revision ID: 9e6b3f8a2c7d
Revises: 8d5a2f7c1b4e
"""
from __future__ import annotations
from alembic import op
revision = "9e6b3f8a2c7d"
down_revision = "8d5a2f7c1b4e"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_index(
"ix_workflow_instances_tenant_created_id",
"workflow_instances",
["tenant_id", "created_at", "id"],
)
op.create_index(
"ix_workflow_instances_tenant_definition_created_id",
"workflow_instances",
["tenant_id", "definition_id", "created_at", "id"],
)
def downgrade() -> None:
op.drop_index(
"ix_workflow_instances_tenant_definition_created_id",
table_name="workflow_instances",
)
op.drop_index(
"ix_workflow_instances_tenant_created_id",
table_name="workflow_instances",
)
@@ -83,8 +83,12 @@ from govoplan_workflow_engine.backend.schemas import (
WorkflowGraphValidationRequest,
WorkflowGraphValidationResponse,
WorkflowInstanceListResponse,
WorkflowInstanceEventPageResponse,
WorkflowInstanceResponse,
WorkflowInstanceStartRequest,
WorkflowInstanceStepPageResponse,
WorkflowInstanceSummaryListResponse,
WorkflowInstanceSummaryResponse,
WorkflowNodeLibraryResponse,
WorkflowNodeTypeResponse,
WorkflowPortResponse,
@@ -102,6 +106,12 @@ from govoplan_workflow_engine.backend.instance_service import (
resolve_step,
start_instance,
)
from govoplan_workflow_engine.backend.instance_reads import (
MAX_HISTORY_SEQUENCE,
get_instance_summary,
instance_history_page,
list_instance_summaries,
)
from govoplan_workflow_engine.backend.runtime import get_registry
from govoplan_workflow_engine.backend.service import (
WorkflowBpmnValidationError,
@@ -699,6 +709,31 @@ def api_list_instances(
raise _http_error(exc) from exc
@router.get("/instances/summaries", response_model=WorkflowInstanceSummaryListResponse)
def api_list_instance_summaries(
definition_id: str | None = None,
limit: int = Query(default=100, ge=1, le=200),
cursor: str | None = Query(default=None, max_length=256),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> WorkflowInstanceSummaryListResponse:
try:
return list_instance_summaries(
session,
principal=principal,
registry=get_registry(),
definition_id=definition_id,
limit=limit,
cursor=cursor,
)
except PermissionError as exc:
raise _governance_http_error(exc) from exc
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
except WorkflowError as exc:
raise _http_error(exc) from exc
@router.post("/standards/reconcile", response_model=dict[str, object])
def api_reconcile_standards(
session: Session = Depends(get_session),
@@ -875,6 +910,83 @@ def api_get_instance(
raise _http_error(exc) from exc
@router.get(
"/instances/{instance_id}/summary", response_model=WorkflowInstanceSummaryResponse
)
def api_get_instance_summary(
instance_id: str,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> WorkflowInstanceSummaryResponse:
try:
return get_instance_summary(
session,
principal=principal,
registry=get_registry(),
instance_id=instance_id,
)
except PermissionError as exc:
raise _governance_http_error(exc) from exc
except WorkflowError as exc:
raise _http_error(exc) from exc
@router.get(
"/instances/{instance_id}/steps", response_model=WorkflowInstanceStepPageResponse
)
def api_list_instance_steps(
instance_id: str,
limit: int = Query(default=100, ge=1, le=200),
after_sequence: int = Query(default=0, ge=0, le=MAX_HISTORY_SEQUENCE),
through_sequence: int | None = Query(default=None, ge=0, le=MAX_HISTORY_SEQUENCE),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> WorkflowInstanceStepPageResponse:
try:
return instance_history_page(
session,
principal=principal,
registry=get_registry(),
instance_id=instance_id,
kind="steps",
limit=limit,
after_sequence=after_sequence,
through_sequence=through_sequence,
)
except PermissionError as exc:
raise _governance_http_error(exc) from exc
except WorkflowError as exc:
raise _http_error(exc) from exc
@router.get(
"/instances/{instance_id}/events", response_model=WorkflowInstanceEventPageResponse
)
def api_list_instance_events(
instance_id: str,
limit: int = Query(default=100, ge=1, le=200),
after_sequence: int = Query(default=0, ge=0, le=MAX_HISTORY_SEQUENCE),
through_sequence: int | None = Query(default=None, ge=0, le=MAX_HISTORY_SEQUENCE),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> WorkflowInstanceEventPageResponse:
try:
return instance_history_page(
session,
principal=principal,
registry=get_registry(),
instance_id=instance_id,
kind="events",
limit=limit,
after_sequence=after_sequence,
through_sequence=through_sequence,
)
except PermissionError as exc:
raise _governance_http_error(exc) from exc
except WorkflowError as exc:
raise _http_error(exc) from exc
@router.post(
"/instances/{instance_id}/reconcile",
response_model=WorkflowInstanceResponse,
@@ -628,3 +628,40 @@ class WorkflowInstanceResponse(BaseModel):
class WorkflowInstanceListResponse(BaseModel):
instances: list[WorkflowInstanceResponse]
class WorkflowInstanceSummaryResponse(BaseModel):
"""Instance identity and progress without runtime payloads or history."""
id: str
definition_id: str
definition_name: str
definition_revision: int
definition_hash: str
execution_mode: WorkflowExecutionMode
start_origin: WorkflowStartOrigin
status: WorkflowInstanceStatus
current_step_id: str | None
started_at: datetime
finished_at: datetime | None
created_at: datetime
updated_at: datetime
class WorkflowInstanceSummaryListResponse(BaseModel):
instances: list[WorkflowInstanceSummaryResponse]
next_cursor: str | None = None
class WorkflowInstanceStepPageResponse(BaseModel):
steps: list[WorkflowInstanceStepResponse]
total: int
through_sequence: int
next_after_sequence: int | None = None
class WorkflowInstanceEventPageResponse(BaseModel):
events: list[WorkflowInstanceEventResponse]
total: int
through_sequence: int
next_after_sequence: int | None = None
@@ -87,8 +87,8 @@ class WorkflowWorkItemProvider:
WorkflowInstanceStep.id.desc(),
)
)
targets = self._targets(principal, query.tenant_id)
if not administrative:
targets = self._targets(principal, query.tenant_id)
conditions = [
and_(
WorkflowInstanceStep.work_assignment_kind == kind,
@@ -108,7 +108,6 @@ class WorkflowWorkItemProvider:
items: list[WorkItem] = []
total = 0
decisions: dict[str, bool] = {}
targets = self._targets(principal, query.tenant_id)
for instance, step, definition in session.execute(statement).yield_per(250):
assignment = _step_assignment(instance, step)
if not administrative and not _assignment_matches(assignment, targets):
+519
View File
@@ -0,0 +1,519 @@
from __future__ import annotations
from contextlib import contextmanager
from dataclasses import replace
from datetime import UTC, datetime, timedelta
import unittest
from unittest.mock import patch
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, event
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool
from govoplan_core.auth import ApiPrincipal, get_api_principal
from govoplan_core.core.access import PrincipalRef
from govoplan_core.core.tasks import WorkItemQuery
from govoplan_core.db.base import Base
from govoplan_core.db.session import get_session
from govoplan_workflow_engine.backend.db.models import (
WorkflowDefinition,
WorkflowDefinitionRevision,
WorkflowInstance,
WorkflowInstanceEvent,
WorkflowInstanceStep,
)
from govoplan_workflow_engine.backend.instance_reads import (
get_instance_summary,
instance_history_page,
list_instance_summaries,
)
from govoplan_workflow_engine.backend.instance_service import (
instance_response,
list_instances,
)
from govoplan_workflow_engine.backend.router import router
from govoplan_workflow_engine.backend.service import WorkflowNotFoundError
from govoplan_workflow_engine.backend.work_items import WorkflowWorkItemProvider
def principal() -> ApiPrincipal:
return ApiPrincipal(
principal=PrincipalRef(
account_id="account-1",
membership_id="membership-1",
tenant_id="tenant-1",
scopes=frozenset({"workflow:instance:read"}),
),
account=object(),
user=object(),
)
class WorkflowInstanceReadTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine(
"sqlite://",
poolclass=StaticPool,
connect_args={"check_same_thread": False},
)
Base.metadata.create_all(
self.engine,
tables=[
WorkflowDefinition.__table__,
WorkflowDefinitionRevision.__table__,
WorkflowInstance.__table__,
WorkflowInstanceStep.__table__,
WorkflowInstanceEvent.__table__,
],
)
self.session = Session(self.engine)
self.actor = principal()
def tearDown(self) -> None:
self.session.close()
self.engine.dispose()
@contextmanager
def statements(self):
statements = []
def capture(_connection, _cursor, statement, _parameters, _context, _many):
statements.append(statement)
event.listen(self.engine, "before_cursor_execute", capture)
try:
yield statements
finally:
event.remove(self.engine, "before_cursor_execute", capture)
def seed(self, index: int, *, tenant_id="tenant-1", scope_type="tenant", rows=3):
now = datetime(2026, 1, 1, tzinfo=UTC)
definition = WorkflowDefinition(
id=f"definition-{index}",
tenant_id=tenant_id,
scope_type=scope_type,
scope_id=tenant_id if scope_type == "tenant" else "group-other",
scope_key=f"{scope_type}:{tenant_id}:{index}",
definition_key=f"workflow-{index}",
name=f"Workflow {index}",
status="active",
current_revision=index + 1,
active_revision=index + 1,
)
revision = WorkflowDefinitionRevision(
id=f"revision-{index}",
tenant_id=tenant_id,
definition=definition,
revision=index + 1,
graph={"sensitive_graph": "not needed for summary"},
bpmn_xml="sensitive BPMN",
content_hash=f"{index:064x}",
library_id="workflow",
library_version="1",
)
instance = WorkflowInstance(
id=f"instance-{index:05d}",
tenant_id=tenant_id,
definition=definition,
definition_revision_id=revision.id,
status="waiting",
idempotency_key=f"key-{index}",
authorization_={"account_id": "account-1"},
input_={"sensitive": "input"},
context_={"sensitive": "context"},
output_={"sensitive": "output"},
started_at=now,
created_at=now,
updated_at=now,
)
self.session.add_all([definition, revision, instance])
for sequence in range(1, rows + 1):
step = WorkflowInstanceStep(
id=f"step-{index}-{sequence}",
tenant_id=tenant_id,
instance=instance,
sequence=sequence,
node_id="activity",
node_type="workflow.activity",
status="waiting",
idempotency_key=f"step-key-{index}-{sequence}",
handoff={"kind": "activity", "title": "Review application"},
work_assignment_kind="account",
work_assignment_id="account-1",
)
self.session.add(step)
self.session.add(
WorkflowInstanceEvent(
id=f"event-{index}-{sequence}",
tenant_id=tenant_id,
instance=instance,
step_id=step.id,
sequence=sequence,
kind="workflow.step.started",
payload={"sequence": sequence},
created_at=now,
)
)
instance.current_step_id = step.id
return instance
def summaries(self, **kwargs):
return list_instance_summaries(
self.session, principal=self.actor, registry=None, **kwargs
)
def history(self, instance_id="instance-00001", **kwargs):
return instance_history_page(
self.session,
principal=self.actor,
registry=None,
instance_id=instance_id,
**kwargs,
)
def test_cold_40_distinct_revisions_have_constant_query_count_and_full_history(
self,
):
for index in range(40):
self.seed(index)
self.seed(100, tenant_id="other-tenant")
self.session.commit()
self.session.expunge_all()
with self.statements() as queries:
instances = list_instances(self.session, tenant_id="tenant-1", limit=40)
responses = [instance_response(self.session, item) for item in instances]
self.assertEqual(5, len(queries))
self.assertEqual(40, len(responses))
self.assertEqual(
set(range(1, 41)), {item.definition_revision for item in responses}
)
for item in responses:
self.assertEqual(
f"{item.definition_revision - 1:064x}", item.definition_hash
)
self.assertEqual([1, 2, 3], [step.sequence for step in item.steps])
self.assertEqual([1, 2, 3], [entry.sequence for entry in item.events])
def test_summary_query_omits_histories_graphs_and_runtime_payloads(self):
for index in range(40):
self.seed(index)
self.session.commit()
self.session.expunge_all()
with self.statements() as queries:
page = self.summaries(limit=40)
self.assertEqual(1, len(queries))
self.assertEqual(40, len(page.instances))
self.assertIsNone(page.next_cursor)
sql = queries[0]
for forbidden in (
"workflow_instance_steps",
"workflow_instance_events",
".graph",
".bpmn_xml",
"workflow_instances.input",
"workflow_instances.context",
"workflow_instances.output",
"workflow_instances.authorization",
):
self.assertNotIn(forbidden, sql)
self.assertNotIn("steps", page.instances[0].model_dump())
self.assertNotIn("events", page.instances[0].model_dump())
self.assertNotIn("total", page.model_dump())
def test_summary_queries_use_ordered_tenant_indexes(self):
self.seed(1, rows=0)
self.session.commit()
for definition_id, index_name in (
(None, "ix_workflow_instances_tenant_created_id"),
("definition-1", "ix_workflow_instances_tenant_definition_created_id"),
):
captured = []
def capture(_connection, _cursor, sql, parameters, _context, _many):
captured.append((sql, parameters))
event.listen(self.engine, "before_cursor_execute", capture)
try:
self.summaries(definition_id=definition_id)
finally:
event.remove(self.engine, "before_cursor_execute", capture)
sql, parameters = captured[0]
plan = " ".join(
row[3]
for row in self.session.connection().exec_driver_sql(
"EXPLAIN QUERY PLAN " + sql, parameters
)
)
self.assertIn(index_name, plan)
self.assertNotIn("TEMP B-TREE", plan)
def test_summary_pages_fill_after_current_governance_and_preserve_ties(self):
for index in range(5):
self.seed(index, rows=0)
# More inaccessible candidates than one SQL batch precede the visible rows.
for index in range(10, 270):
self.seed(index, scope_type="group", rows=0)
self.seed(300, tenant_id="other-tenant", rows=0)
self.session.commit()
self.session.expunge_all()
with self.statements() as queries:
first = self.summaries(limit=2)
self.assertEqual(2, len(queries))
self.assertEqual(
["instance-00004", "instance-00003"], [i.id for i in first.instances]
)
self.assertIsNotNone(first.next_cursor)
# Updating progress must not move an item across the creation-order cursor.
self.session.get(WorkflowInstance, "instance-00002").updated_at += timedelta(
days=2
)
second = self.summaries(limit=2, cursor=first.next_cursor)
third = self.summaries(limit=2, cursor=second.next_cursor)
self.assertEqual(
["instance-00002", "instance-00001"], [i.id for i in second.instances]
)
self.assertEqual(["instance-00000"], [i.id for i in third.instances])
self.assertIsNone(third.next_cursor)
filtered = self.summaries(definition_id="definition-2", limit=1)
self.assertEqual(["instance-00002"], [i.id for i in filtered.instances])
self.assertIsNone(filtered.next_cursor)
def test_summary_and_history_recheck_governance_in_the_same_session(self):
self.seed(1)
self.session.commit()
first = self.summaries(limit=1)
self.assertEqual(1, len(first.instances))
initial = self.history(kind="events", limit=1)
# Simulate governance changing after a prior request, including a warm
# identity map. Retained instance authorization is not a read grant.
with Session(self.engine) as writer:
definition = writer.get(WorkflowDefinition, "definition-1")
definition.scope_type = "group"
definition.scope_id = "group-other"
writer.commit()
self.assertEqual([], self.summaries(limit=1).instances)
with self.assertRaises(PermissionError):
self.history(kind="events", after_sequence=initial.next_after_sequence)
with self.assertRaises(PermissionError):
get_instance_summary(
self.session,
principal=self.actor,
registry=None,
instance_id="instance-00001",
)
def test_history_has_exact_total_and_fixed_boundary_without_silent_truncation(self):
self.seed(1, rows=405)
self.seed(2, tenant_id="other-tenant")
self.session.commit()
self.session.expunge_all()
for kind in ("steps", "events"):
with self.subTest(kind=kind):
model = (
WorkflowInstanceStep if kind == "steps" else WorkflowInstanceEvent
)
loaded = []
def record_load(row, _context):
loaded.append(row.id)
event.listen(model, "load", record_load)
try:
with self.statements() as queries:
first = self.history(kind=kind, limit=200)
finally:
event.remove(model, "load", record_load)
self.assertEqual(3, len(queries))
self.assertEqual(201, len(loaded))
self.assertEqual(405, first.total)
self.assertEqual(405, first.through_sequence)
self.assertEqual(200, first.next_after_sequence)
self.assertEqual(200, len(getattr(first, kind)))
second = self.history(
kind=kind,
limit=200,
after_sequence=first.next_after_sequence,
through_sequence=first.through_sequence,
)
third = self.history(
kind=kind,
limit=200,
after_sequence=second.next_after_sequence,
through_sequence=first.through_sequence,
)
self.assertEqual(405, third.total)
self.assertIsNone(third.next_after_sequence)
sequences = [
row.sequence
for page in (first, second, third)
for row in getattr(page, kind)
]
self.assertEqual(list(range(1, 406)), sequences)
empty = self.history(kind=kind, after_sequence=1000)
self.assertEqual([], getattr(empty, kind))
self.assertEqual(405, empty.total)
self.session.add(
WorkflowInstanceEvent(
id="later",
tenant_id="tenant-1",
instance_id="instance-00001",
sequence=406,
kind="workflow.step.completed",
payload={},
created_at=datetime.now(UTC),
)
)
self.session.commit()
fixed = self.history(kind="events", after_sequence=400, through_sequence=405)
fresh = self.history(kind="events", after_sequence=400)
self.assertEqual(405, fixed.total)
self.assertEqual(406, fresh.total)
self.assertEqual([401, 402, 403, 404, 405], [i.sequence for i in fixed.events])
def test_pages_enforce_tenant_scope_and_read_permission(self):
self.seed(1)
self.seed(2, tenant_id="other-tenant")
self.session.commit()
with self.assertRaises(WorkflowNotFoundError):
self.history(instance_id="instance-00002", kind="events")
with self.assertRaises(WorkflowNotFoundError):
get_instance_summary(
self.session,
principal=self.actor,
registry=None,
instance_id="instance-00002",
)
self.actor = replace(
self.actor, principal=replace(self.actor.principal, scopes=frozenset())
)
with self.assertRaises(PermissionError):
self.summaries()
with self.assertRaises(PermissionError):
self.history(kind="steps")
def test_history_filters_child_tenant_and_empty_history(self):
self.seed(1, rows=0)
self.session.add(
WorkflowInstanceEvent(
tenant_id="other-tenant",
instance_id="instance-00001",
sequence=1,
kind="should-not-be-disclosed",
payload={},
created_at=datetime.now(UTC),
)
)
self.session.commit()
empty = self.history(kind="events")
self.assertEqual(
(0, 0, None, []),
(
empty.total,
empty.through_sequence,
empty.next_after_sequence,
empty.events,
),
)
def test_inconsistent_revision_evidence_fails_closed(self):
instance = self.seed(1)
self.seed(2, tenant_id="other-tenant")
instance.definition_revision_id = "revision-2"
self.session.commit()
self.session.expunge_all()
with self.assertRaises(WorkflowNotFoundError):
self.summaries()
with self.assertRaises(WorkflowNotFoundError):
self.history(kind="events")
loaded = list_instances(self.session, tenant_id="tenant-1")
with self.assertRaises(WorkflowNotFoundError):
instance_response(self.session, loaded[0])
def test_work_item_total_remains_exact_after_authorization_and_filters(self):
for index in range(40):
self.seed(index, rows=1)
self.seed(50, scope_type="group", rows=1)
self.seed(51, tenant_id="other-tenant", rows=1)
self.seed(52, rows=1)
self.session.flush()
self.session.get(
WorkflowInstanceStep, "step-52-1"
).work_assignment_id = "someone-else"
self.session.commit()
self.session.expunge_all()
provider = WorkflowWorkItemProvider()
with (
self.statements() as queries,
patch.object(provider, "_targets", wraps=provider._targets) as targets,
):
page = provider.list_items(
self.session,
self.actor,
query=WorkItemQuery(tenant_id="tenant-1", limit=2),
)
self.assertEqual(1, targets.call_count)
self.assertEqual(1, len(queries))
self.assertEqual((40, 2, True), (page.total, len(page.items), page.truncated))
filtered = provider.list_items(
self.session,
self.actor,
query=WorkItemQuery(tenant_id="tenant-1", text="not present", limit=2),
)
self.assertEqual((0, False), (filtered.total, filtered.truncated))
def test_http_routes_validate_bounds_and_keep_legacy_contract(self):
self.seed(1)
self.session.commit()
app = FastAPI()
app.include_router(router)
app.dependency_overrides[get_session] = lambda: self.session
app.dependency_overrides[get_api_principal] = lambda: self.actor
with (
TestClient(app) as client,
patch(
"govoplan_workflow_engine.backend.router.get_registry",
return_value=None,
),
):
prefix = router.prefix
summary = client.get(f"{prefix}/instances/summaries")
self.assertEqual(200, summary.status_code, summary.text)
self.assertNotIn("steps", summary.json()["instances"][0])
detail = client.get(f"{prefix}/instances/instance-00001/summary")
self.assertEqual(200, detail.status_code, detail.text)
legacy = client.get(f"{prefix}/instances/instance-00001")
self.assertEqual(200, legacy.status_code, legacy.text)
self.assertEqual(3, len(legacy.json()["events"]))
for kind in ("steps", "events"):
page = client.get(f"{prefix}/instances/instance-00001/{kind}?limit=2")
self.assertEqual(200, page.status_code, page.text)
self.assertEqual(3, page.json()["total"])
self.assertEqual(2, page.json()["next_after_sequence"])
self.assertEqual(
422,
client.get(
f"{prefix}/instances/instance-00001/{kind}?limit=201"
).status_code,
)
self.assertEqual(
422,
client.get(
f"{prefix}/instances/instance-00001/{kind}?after_sequence=-1"
).status_code,
)
self.assertEqual(
422,
client.get(
f"{prefix}/instances/instance-00001/{kind}?through_sequence=999999999999999999999"
).status_code,
)
self.assertEqual(
400,
client.get(f"{prefix}/instances/summaries?cursor=invalid").status_code,
)
if __name__ == "__main__":
unittest.main()
+17
View File
@@ -17,6 +17,23 @@ from govoplan_core.core.workflows import (
class WorkflowManifestTests(unittest.TestCase):
def test_all_static_topics_have_complete_german_content(self) -> None:
manifest = get_manifest()
for topic in manifest.documentation:
german = (topic.translations or {}).get("de", {})
self.assertEqual({"title", "summary", "body"}, set(german), topic.id)
self.assertTrue(
all(str(value).strip() for value in german.values()), topic.id
)
recovery = next(
topic
for topic in manifest.documentation
if topic.id == "workflow.runtime-recovery"
)
self.assertEqual("reference", recovery.metadata["kind"])
self.assertTrue(recovery.metadata["consequence_classes"])
def test_manifest_exposes_definition_contracts(self) -> None:
manifest = get_manifest()
+19 -4
View File
@@ -29,7 +29,7 @@ class WorkflowMigrationTests(unittest.TestCase):
try:
with engine.connect() as connection:
self.assertIn(
"8d5a2f7c1b4e",
"9e6b3f8a2c7d",
set(MigrationContext.configure(connection).get_current_heads()),
)
self.assertEqual(
@@ -75,6 +75,18 @@ class WorkflowMigrationTests(unittest.TestCase):
)
}
self.assertIn("start_origin", instance_columns)
indexes = {
item["name"]: item["column_names"]
for item in inspect(connection).get_indexes("workflow_instances")
}
self.assertEqual(
["tenant_id", "created_at", "id"],
indexes["ix_workflow_instances_tenant_created_id"],
)
self.assertEqual(
["tenant_id", "definition_id", "created_at", "id"],
indexes["ix_workflow_instances_tenant_definition_created_id"],
)
definition_columns = {
item["name"]
for item in inspect(connection).get_columns(
@@ -107,6 +119,7 @@ class WorkflowMigrationTests(unittest.TestCase):
"b2e4f6a8c0d1_",
"e4a1f8c2d7b6_",
"8d5a2f7c1b4e_",
"9e6b3f8a2c7d_",
)
):
continue
@@ -151,7 +164,7 @@ class WorkflowMigrationTests(unittest.TestCase):
manifest_factories=(get_manifest,),
)
self.assertIn("8d5a2f7c1b4e", result.current_revision or "")
self.assertIn("9e6b3f8a2c7d", result.current_revision or "")
engine = create_engine(url)
try:
upgraded_tables = set(inspect(engine).get_table_names())
@@ -187,6 +200,8 @@ class WorkflowMigrationTests(unittest.TestCase):
try:
with engine.begin() as connection:
for index_name in (
"ix_workflow_instances_tenant_created_id",
"ix_workflow_instances_tenant_definition_created_id",
"ix_workflow_instance_steps_work_assignment",
"ix_workflow_instance_steps_work_due_at",
"ix_workflow_instance_steps_work_assignment_id",
@@ -220,7 +235,7 @@ class WorkflowMigrationTests(unittest.TestCase):
connection.execute(
text(
"UPDATE alembic_version SET version_num = "
"'b2e4f6a8c0d1' WHERE version_num = '8d5a2f7c1b4e'"
"'b2e4f6a8c0d1' WHERE version_num = '9e6b3f8a2c7d'"
)
)
finally:
@@ -258,7 +273,7 @@ class WorkflowMigrationTests(unittest.TestCase):
)
with engine.connect() as connection:
self.assertIn(
"8d5a2f7c1b4e",
"9e6b3f8a2c7d",
set(MigrationContext.configure(connection).get_current_heads()),
)
finally: