528 lines
17 KiB
Python
528 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine, select
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.automation import AutomationPrincipalResolution
|
|
from govoplan_core.core.policy import PolicyDecision
|
|
from govoplan_core.db.base import Base, utcnow
|
|
from govoplan_dataflow.backend.db.models import (
|
|
DataflowPipeline,
|
|
DataflowPipelineRevision,
|
|
DataflowRun,
|
|
DataflowTrigger,
|
|
DataflowTriggerDelivery,
|
|
)
|
|
from govoplan_dataflow.backend.schemas import (
|
|
DataflowEventDeliveryRequest,
|
|
DataflowTriggerCreateRequest,
|
|
DataflowTriggerEvent,
|
|
DataflowTriggerSchedule,
|
|
PipelineCreateRequest,
|
|
PipelineDeriveRequest,
|
|
PipelineUpdateRequest,
|
|
)
|
|
from govoplan_dataflow.backend.service import (
|
|
DataflowConflictError,
|
|
create_pipeline,
|
|
derive_pipeline,
|
|
start_pipeline_run,
|
|
update_pipeline,
|
|
)
|
|
from govoplan_core.core.dataflows import DataflowRunRequest
|
|
from govoplan_dataflow.backend.triggers import (
|
|
create_trigger,
|
|
dispatch_due_triggers,
|
|
ingest_event_delivery,
|
|
)
|
|
|
|
|
|
POLICY_CAPABILITY = "policy.definitionGovernance"
|
|
AUTOMATION_CAPABILITY = "auth.automationPrincipalProvider"
|
|
|
|
|
|
def sample_graph():
|
|
from test_service import sample_graph as build_graph
|
|
|
|
return build_graph()
|
|
|
|
|
|
def principal() -> ApiPrincipal:
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id="account-1",
|
|
membership_id="membership-1",
|
|
tenant_id="tenant-1",
|
|
scopes=frozenset(
|
|
{
|
|
"dataflow:pipeline:run",
|
|
"dataflow:trigger:read",
|
|
"dataflow:trigger:write",
|
|
}
|
|
),
|
|
),
|
|
account=object(),
|
|
user=object(),
|
|
)
|
|
|
|
|
|
class AllowDefinitionPolicy:
|
|
def resolve_definition_action(self, *, request):
|
|
allowed = not (
|
|
request.definition_kind == "template"
|
|
and request.action in {"run", "automate"}
|
|
)
|
|
if request.action == "automate":
|
|
allowed = allowed and request.allow_automation
|
|
if request.action == "run":
|
|
allowed = (
|
|
allowed
|
|
and request.allow_run
|
|
and request.status == "active"
|
|
)
|
|
if request.action in {"reuse", "derive"}:
|
|
allowed = allowed and request.allow_reuse
|
|
return PolicyDecision(
|
|
allowed=allowed,
|
|
reason=None if allowed else "Definition action denied.",
|
|
)
|
|
|
|
|
|
class CurrentPrincipalProvider:
|
|
def __init__(self, actor: ApiPrincipal, *, allowed: bool = True) -> None:
|
|
self.actor = actor
|
|
self.allowed = allowed
|
|
|
|
def resolve_automation_principal(self, _session, *, request):
|
|
return AutomationPrincipalResolution(
|
|
allowed=self.allowed,
|
|
principal=self.actor if self.allowed else None,
|
|
reason=None if self.allowed else "Owner authorization revoked.",
|
|
granted_scopes=tuple(request.grant_scopes) if self.allowed else (),
|
|
missing_scopes=() if self.allowed else tuple(request.grant_scopes),
|
|
provenance={"status": "current" if self.allowed else "revoked"},
|
|
)
|
|
|
|
|
|
class Registry:
|
|
def __init__(
|
|
self,
|
|
actor: ApiPrincipal,
|
|
*,
|
|
automation_allowed: bool = True,
|
|
) -> None:
|
|
self.capabilities = {
|
|
POLICY_CAPABILITY: AllowDefinitionPolicy(),
|
|
AUTOMATION_CAPABILITY: CurrentPrincipalProvider(
|
|
actor,
|
|
allowed=automation_allowed,
|
|
),
|
|
}
|
|
|
|
def has_capability(self, name: str) -> bool:
|
|
return name in self.capabilities
|
|
|
|
def capability(self, name: str):
|
|
return self.capabilities[name]
|
|
|
|
|
|
class DataflowTriggerTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(
|
|
self.engine,
|
|
tables=[
|
|
DataflowPipeline.__table__,
|
|
DataflowPipelineRevision.__table__,
|
|
DataflowRun.__table__,
|
|
DataflowTrigger.__table__,
|
|
DataflowTriggerDelivery.__table__,
|
|
],
|
|
)
|
|
self.Session = sessionmaker(bind=self.engine)
|
|
self.session: Session = self.Session()
|
|
self.actor = principal()
|
|
self.registry = Registry(self.actor)
|
|
self.pipeline = create_pipeline(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
actor_id="account-1",
|
|
payload=PipelineCreateRequest(
|
|
name="Automated comparison",
|
|
status="active",
|
|
allow_automation=True,
|
|
graph=sample_graph(),
|
|
),
|
|
)
|
|
self.session.commit()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
Base.metadata.drop_all(
|
|
self.engine,
|
|
tables=[
|
|
DataflowTriggerDelivery.__table__,
|
|
DataflowTrigger.__table__,
|
|
DataflowRun.__table__,
|
|
DataflowPipelineRevision.__table__,
|
|
DataflowPipeline.__table__,
|
|
],
|
|
)
|
|
self.engine.dispose()
|
|
|
|
def _event_trigger(self) -> DataflowTrigger:
|
|
trigger = create_trigger(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
pipeline_id=self.pipeline.id,
|
|
actor_id="account-1",
|
|
principal=self.actor,
|
|
registry=self.registry,
|
|
payload=DataflowTriggerCreateRequest(
|
|
name="Monthly file arrived",
|
|
kind="event",
|
|
enabled=True,
|
|
event=DataflowTriggerEvent(
|
|
event_type="files.upload.completed",
|
|
module_id="files",
|
|
filters={"folder_id": "incoming"},
|
|
),
|
|
),
|
|
)
|
|
self.session.commit()
|
|
return trigger
|
|
|
|
def test_interval_delivery_runs_pinned_revision_with_provenance(self) -> None:
|
|
trigger = create_trigger(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
pipeline_id=self.pipeline.id,
|
|
actor_id="account-1",
|
|
principal=self.actor,
|
|
registry=self.registry,
|
|
payload=DataflowTriggerCreateRequest(
|
|
name="Hourly",
|
|
kind="interval",
|
|
enabled=True,
|
|
schedule=DataflowTriggerSchedule(
|
|
interval_seconds=3600,
|
|
timezone="Europe/Berlin",
|
|
),
|
|
),
|
|
)
|
|
now = utcnow()
|
|
trigger.next_fire_at = now - timedelta(minutes=1)
|
|
self.session.commit()
|
|
|
|
result = dispatch_due_triggers(
|
|
self.session,
|
|
registry=self.registry,
|
|
now=now,
|
|
)
|
|
self.session.commit()
|
|
|
|
delivery = self.session.scalar(
|
|
select(DataflowTriggerDelivery).where(
|
|
DataflowTriggerDelivery.trigger_id == trigger.id
|
|
)
|
|
)
|
|
run = self.session.get(DataflowRun, delivery.run_id)
|
|
self.assertEqual(1, result.queued)
|
|
self.assertEqual(1, result.succeeded)
|
|
self.assertEqual("succeeded", delivery.status)
|
|
self.assertEqual("schedule", run.invocation_kind)
|
|
self.assertEqual(trigger.id, run.trigger_id)
|
|
self.assertEqual(delivery.id, run.trigger_delivery_id)
|
|
self.assertEqual("current", delivery.authorization_provenance["status"])
|
|
|
|
def test_skip_catch_up_records_missed_interval_without_running(self) -> None:
|
|
trigger = create_trigger(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
pipeline_id=self.pipeline.id,
|
|
actor_id="account-1",
|
|
principal=self.actor,
|
|
registry=self.registry,
|
|
payload=DataflowTriggerCreateRequest(
|
|
name="Skip stale intervals",
|
|
kind="interval",
|
|
enabled=True,
|
|
catch_up_policy="skip",
|
|
schedule=DataflowTriggerSchedule(
|
|
anchor_at=datetime(
|
|
2000,
|
|
1,
|
|
1,
|
|
tzinfo=timezone.utc,
|
|
),
|
|
interval_seconds=60,
|
|
),
|
|
),
|
|
)
|
|
now = utcnow()
|
|
trigger.next_fire_at = now - timedelta(minutes=3)
|
|
self.session.commit()
|
|
|
|
result = dispatch_due_triggers(
|
|
self.session,
|
|
registry=self.registry,
|
|
now=now,
|
|
)
|
|
self.session.commit()
|
|
|
|
delivery = self.session.scalar(
|
|
select(DataflowTriggerDelivery).where(
|
|
DataflowTriggerDelivery.trigger_id == trigger.id
|
|
)
|
|
)
|
|
self.assertEqual(0, result.queued)
|
|
self.assertEqual(1, result.skipped)
|
|
self.assertEqual("skipped", delivery.status)
|
|
self.assertIsNone(delivery.run_id)
|
|
next_fire_at = trigger.next_fire_at
|
|
self.assertIsNotNone(next_fire_at)
|
|
if next_fire_at.tzinfo is None:
|
|
next_fire_at = next_fire_at.replace(tzinfo=timezone.utc)
|
|
self.assertGreater(next_fire_at, now)
|
|
|
|
def test_disabled_trigger_can_be_prepared_for_a_draft(self) -> None:
|
|
self.pipeline.status = "draft"
|
|
self.session.flush()
|
|
|
|
trigger = create_trigger(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
pipeline_id=self.pipeline.id,
|
|
actor_id="account-1",
|
|
principal=self.actor,
|
|
registry=self.registry,
|
|
payload=DataflowTriggerCreateRequest(
|
|
name="Prepared event",
|
|
kind="event",
|
|
enabled=False,
|
|
event=DataflowTriggerEvent(
|
|
event_type="files.upload.completed",
|
|
),
|
|
),
|
|
)
|
|
|
|
self.assertEqual("disabled", trigger.status)
|
|
self.assertIsNone(trigger.next_fire_at)
|
|
|
|
def test_event_ingress_is_filtered_and_idempotent(self) -> None:
|
|
trigger = self._event_trigger()
|
|
event = DataflowEventDeliveryRequest(
|
|
event_id="event-1",
|
|
type="files.upload.completed",
|
|
module_id="files",
|
|
payload={"folder_id": "incoming", "file_id": "file-1"},
|
|
occurred_at=utcnow(),
|
|
)
|
|
|
|
self.assertEqual(
|
|
1,
|
|
ingest_event_delivery(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
event=event,
|
|
),
|
|
)
|
|
self.assertEqual(
|
|
0,
|
|
ingest_event_delivery(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
event=event,
|
|
),
|
|
)
|
|
result = dispatch_due_triggers(
|
|
self.session,
|
|
registry=self.registry,
|
|
)
|
|
self.session.commit()
|
|
|
|
delivery = self.session.scalar(
|
|
select(DataflowTriggerDelivery).where(
|
|
DataflowTriggerDelivery.trigger_id == trigger.id
|
|
)
|
|
)
|
|
run = self.session.get(DataflowRun, delivery.run_id)
|
|
self.assertEqual(1, result.succeeded)
|
|
self.assertEqual("event", run.invocation_kind)
|
|
self.assertEqual("event-1", run.request_["invocation"]["event_id"])
|
|
|
|
def test_revoked_owner_blocks_delivery_before_execution(self) -> None:
|
|
trigger = self._event_trigger()
|
|
ingest_event_delivery(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
event=DataflowEventDeliveryRequest(
|
|
event_id="event-revoked",
|
|
type="files.upload.completed",
|
|
module_id="files",
|
|
payload={"folder_id": "incoming"},
|
|
occurred_at=utcnow(),
|
|
),
|
|
)
|
|
|
|
result = dispatch_due_triggers(
|
|
self.session,
|
|
registry=Registry(self.actor, automation_allowed=False),
|
|
)
|
|
self.session.commit()
|
|
|
|
delivery = self.session.scalar(
|
|
select(DataflowTriggerDelivery).where(
|
|
DataflowTriggerDelivery.trigger_id == trigger.id
|
|
)
|
|
)
|
|
self.assertEqual(1, result.blocked)
|
|
self.assertEqual("blocked", delivery.status)
|
|
self.assertIn("revoked", delivery.error)
|
|
self.assertIsNone(delivery.run_id)
|
|
|
|
def test_restricted_event_requires_durable_bridge(self) -> None:
|
|
self._event_trigger()
|
|
with self.assertRaises(DataflowConflictError):
|
|
ingest_event_delivery(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
event=DataflowEventDeliveryRequest(
|
|
event_id="event-secret",
|
|
type="files.upload.completed",
|
|
module_id="files",
|
|
payload={"folder_id": "incoming"},
|
|
occurred_at=utcnow(),
|
|
classification="restricted",
|
|
),
|
|
)
|
|
|
|
def test_template_derivation_pins_source_and_template_cannot_run(self) -> None:
|
|
template = create_pipeline(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
actor_id="account-1",
|
|
payload=PipelineCreateRequest(
|
|
name="System import template",
|
|
graph=sample_graph(),
|
|
scope_type="system",
|
|
definition_kind="template",
|
|
inherit_to_lower_scopes=True,
|
|
allow_reuse=True,
|
|
),
|
|
)
|
|
derived = derive_pipeline(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
actor_id="account-1",
|
|
principal=self.actor,
|
|
registry=self.registry,
|
|
source_pipeline_id=template.id,
|
|
payload=PipelineDeriveRequest(
|
|
name="Tenant monthly import",
|
|
definition_kind="flow",
|
|
allow_run=True,
|
|
),
|
|
)
|
|
self.session.commit()
|
|
|
|
self.assertEqual(template.id, derived.derived_from_pipeline_id)
|
|
self.assertEqual(1, derived.derived_from_revision)
|
|
self.assertEqual(
|
|
template.revisions[0].content_hash,
|
|
derived.derived_from_hash,
|
|
)
|
|
self.assertEqual(
|
|
"system",
|
|
derived.derivation_provenance["source_scope"]["scope_type"],
|
|
)
|
|
with self.assertRaises(DataflowConflictError):
|
|
start_pipeline_run(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
actor_id="account-1",
|
|
principal=self.actor,
|
|
registry=self.registry,
|
|
request=DataflowRunRequest(
|
|
pipeline_ref=f"pipeline:{template.id}",
|
|
revision=1,
|
|
idempotency_key="template-must-not-run",
|
|
),
|
|
)
|
|
|
|
def test_derived_limits_cannot_be_broadened_transitively(self) -> None:
|
|
template = create_pipeline(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
actor_id="account-1",
|
|
payload=PipelineCreateRequest(
|
|
name="Restricted template",
|
|
graph=sample_graph(),
|
|
definition_kind="template",
|
|
allow_reuse=True,
|
|
allow_automation=False,
|
|
inherit_to_lower_scopes=False,
|
|
),
|
|
)
|
|
derived = derive_pipeline(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
actor_id="account-1",
|
|
principal=self.actor,
|
|
registry=self.registry,
|
|
source_pipeline_id=template.id,
|
|
payload=PipelineDeriveRequest(
|
|
name="Tenant flow",
|
|
allow_reuse=True,
|
|
allow_automation=True,
|
|
inherit_to_lower_scopes=True,
|
|
),
|
|
)
|
|
self.assertFalse(derived.allow_automation)
|
|
self.assertFalse(derived.inherit_to_lower_scopes)
|
|
|
|
update_pipeline(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
pipeline_id=derived.id,
|
|
actor_id="account-1",
|
|
payload=PipelineUpdateRequest(
|
|
name=derived.name,
|
|
graph=sample_graph(),
|
|
status="draft",
|
|
expected_revision=1,
|
|
allow_reuse=True,
|
|
allow_automation=True,
|
|
inherit_to_lower_scopes=True,
|
|
),
|
|
)
|
|
grandchild = derive_pipeline(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
actor_id="account-1",
|
|
principal=self.actor,
|
|
registry=self.registry,
|
|
source_pipeline_id=derived.id,
|
|
payload=PipelineDeriveRequest(
|
|
name="User flow",
|
|
scope_type="user",
|
|
scope_id="membership-1",
|
|
allow_automation=True,
|
|
inherit_to_lower_scopes=True,
|
|
),
|
|
)
|
|
|
|
self.assertFalse(derived.allow_automation)
|
|
self.assertFalse(derived.inherit_to_lower_scopes)
|
|
self.assertFalse(grandchild.allow_automation)
|
|
self.assertFalse(grandchild.inherit_to_lower_scopes)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|