118 lines
7.0 KiB
Python
118 lines
7.0 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import date, datetime, timezone
|
|
from decimal import Decimal
|
|
import time
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from govoplan_core.security.bounded_process import ProcessBudgetError
|
|
from govoplan_dataflow.backend.backends import (
|
|
BackendExecutionError, BackendSource, ExecutionBudget, execute_typed_graph,
|
|
)
|
|
from govoplan_dataflow.backend.backends.reference import ReferenceExecutionBackend
|
|
from govoplan_dataflow.backend.batches import TypedBatch
|
|
from govoplan_dataflow.backend.executor import PipelineExecutionError, ResolvedSource
|
|
from govoplan_dataflow.backend.manifest import get_manifest
|
|
from govoplan_dataflow.backend.schemas import GraphEdge, GraphNode, GraphPosition, PipelineGraph
|
|
from govoplan_dataflow.backend.service import _execute_pipeline_preview, _typed_backend_sources
|
|
|
|
|
|
def graph_for(expression: str = "value", rows: list | None = None) -> PipelineGraph:
|
|
nodes = [
|
|
GraphNode(id="source", type="source.inline", label="Source", position=GraphPosition(x=0, y=0),
|
|
config={"source_name": "records", "rows": rows or [{"value": "normal"}]}),
|
|
GraphNode(id="expression", type="expression", label="Expression", position=GraphPosition(x=100, y=0),
|
|
config={"target_column": "result", "expression": expression, "result_type": "unknown"}),
|
|
GraphNode(id="output", type="output", label="Output", position=GraphPosition(x=200, y=0), config={}),
|
|
]
|
|
return PipelineGraph(nodes=nodes, edges=[
|
|
GraphEdge(id="first", source="source", target="expression"),
|
|
GraphEdge(id="second", source="expression", target="output"),
|
|
])
|
|
|
|
|
|
class ReferenceProcessTests(unittest.TestCase):
|
|
def test_non_finite_deadlines_are_rejected_before_registry_wait(self) -> None:
|
|
for value in (float("nan"), float("inf")):
|
|
with self.subTest(value=value), self.assertRaisesRegex(ValueError, "finite"):
|
|
ExecutionBudget(max_wall_seconds=value)
|
|
|
|
def test_real_child_not_parent_helper_and_typed_result_preserved(self) -> None:
|
|
row = {"value": Decimal("1.20"), "date": date(2026, 9, 8),
|
|
"when": datetime(2026, 9, 8, tzinfo=timezone.utc), "binary": b"\x00\xff"}
|
|
graph = graph_for()
|
|
graph.nodes[0] = graph.nodes[0].model_copy(update={"type": "source.reference", "config": {
|
|
"source_ref": "datasource:fixture", "source_name": "records",
|
|
}})
|
|
source = BackendSource(node_id="source", batch=TypedBatch.from_rows([row]),
|
|
source_ref="datasource:fixture", provider="test", fingerprint="pinned", total_rows=1)
|
|
with patch.object(ReferenceExecutionBackend, "_execute_in_process", side_effect=AssertionError("parent evaluation")):
|
|
result = execute_typed_graph(graph, backend="reference", sources={"source": source})
|
|
self.assertEqual(result.rows, [{**row, "result": Decimal("1.20")}])
|
|
self.assertEqual(result.contract.lineage.source_fingerprints[0]["fingerprint"], "pinned")
|
|
|
|
def test_real_pathological_regex_is_stopped_with_structured_failure(self) -> None:
|
|
graph = graph_for("regexp_full_match(value, '(a+)+$')", [{"value": "a" * 100 + "!"}])
|
|
started = time.monotonic()
|
|
with self.assertRaises(BackendExecutionError) as caught:
|
|
execute_typed_graph(graph, backend="reference", budget=ExecutionBudget(max_wall_seconds=2))
|
|
self.assertIn(caught.exception.code, {"backend.process.timeout", "backend.process.cpu_limit"})
|
|
self.assertLess(time.monotonic() - started, 4)
|
|
|
|
def test_aggregate_padding_allocation_is_contained_by_child_memory_limit(self) -> None:
|
|
graph = graph_for("lpad(value, 900000, '0')", [{"value": "x"} for _ in range(250)])
|
|
with self.assertRaises(BackendExecutionError) as caught:
|
|
execute_typed_graph(graph, backend="reference", budget=ExecutionBudget(
|
|
max_wall_seconds=5, max_memory_bytes=128 * 1024 * 1024,
|
|
))
|
|
self.assertEqual(caught.exception.code, "backend.process.memory_limit")
|
|
|
|
def test_busy_preview_is_retryable_and_never_evaluates_inline(self) -> None:
|
|
with patch("govoplan_dataflow.backend.backends.reference.run_bounded_operation", side_effect=ProcessBudgetError("busy")):
|
|
with self.assertRaises(PipelineExecutionError) as caught:
|
|
_execute_pipeline_preview(graph_for(), session=None, principal=None, registry=None,
|
|
backend="reference", row_limit=10, preview_node_id=None)
|
|
self.assertTrue(caught.exception.retryable)
|
|
self.assertEqual(caught.exception.diagnostics[-1].code, "backend.process.busy")
|
|
|
|
def test_nested_source_ids_do_not_alias_different_authorized_data(self) -> None:
|
|
outer = graph_for()
|
|
inner = graph_for()
|
|
for graph, ref in ((outer, "datasource:outer"), (inner, "datasource:inner")):
|
|
graph.nodes[0] = graph.nodes[0].model_copy(update={"type": "source.reference", "config": {
|
|
"source_ref": ref, "source_name": "records",
|
|
}})
|
|
outer.nodes[1] = outer.nodes[1].model_copy(update={"type": "subflow", "config": {
|
|
"graph": inner.model_dump(mode="python"), "parameters": {},
|
|
}})
|
|
def resolve(node, limit):
|
|
return ResolvedSource(rows=({"value": node.config["source_ref"]},),
|
|
source_ref=node.config["source_ref"], provider="test",
|
|
fingerprint=node.config["source_ref"], total_rows=1)
|
|
sources = _typed_backend_sources(outer, source_resolver=resolve, include_subflows=True)
|
|
self.assertEqual({source.source_ref for source in sources.values()}, {"datasource:outer", "datasource:inner"})
|
|
self.assertEqual(len(sources), 2)
|
|
|
|
def test_static_worker_documentation_is_bilingual(self) -> None:
|
|
topic = next(item for item in get_manifest().documentation if item.id == "dataflow.reference-worker-limits")
|
|
for body in (topic.body, topic.translations["de"]["body"]):
|
|
self.assertIn("GOVOPLAN_ISOLATED_PROCESS_CONCURRENCY", body)
|
|
self.assertIn("32 MiB", body)
|
|
self.assertEqual(set(topic.documentation_types), {"user", "admin"})
|
|
|
|
def test_cumulative_source_budget_stops_before_reading_further_sources(self) -> None:
|
|
graph = graph_for()
|
|
graph.nodes = [graph.nodes[0].model_copy(update={
|
|
"id": f"source-{index}", "type": "source.reference", "config": {"source_ref": f"fixture:{index}"},
|
|
}) for index in range(3)]
|
|
calls = []
|
|
def resolve(node, limit):
|
|
calls.append(node.id)
|
|
return ResolvedSource(rows=({"value": "x" * 100},), source_ref=node.config["source_ref"],
|
|
provider="test", fingerprint="fixed", total_rows=1)
|
|
with self.assertRaisesRegex(PipelineExecutionError, "Combined source data"):
|
|
_typed_backend_sources(graph, source_resolver=resolve, include_subflows=True,
|
|
_remaining_source_bytes=[200])
|
|
self.assertEqual(calls, ["source-0", "source-1"])
|