GovOPlaN/govoplan
2
0
Fork 0
Code Issues 18 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Exercise recovery in datasource composition check
Dependency Audit / dependency-audit (push) Successful in 1m40s
Details
Deployment Installer / deployment-installer (push) Successful in 6s
Details
Security Audit / security-audit (push) Successful in 10m14s
Details

Browse Source
This commit is contained in:
zemion
2026-08-03 07:46:57 +02:00
parent ac80d7e4e3
commit c7d1cd0e8f
1 changed files with 174 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/checks/check-datasource-composition.py
+174 -139
View File
@@ -25,6 +25,12 @@ from govoplan_core.core.datasources import (
datasource_publication, datasource_publication,
) )
from govoplan_core.core.modules import ModuleContext from govoplan_core.core.modules import ModuleContext
from govoplan_core.core.recovery import RecoveryCheckpoint, RecoveryOperation
from govoplan_core.core.runtime_coordination import (
DistributedLease,
RuntimeIdentity,
bind_process_runtime_identity,
)
from govoplan_core.core.tabular_sources import ( from govoplan_core.core.tabular_sources import (
TabularSnapshotInput, TabularSnapshotInput,
tabular_snapshot_writer, tabular_snapshot_writer,
@@ -73,6 +79,9 @@ def main() -> int:
Base.metadata.create_all( Base.metadata.create_all(
engine, engine,
tables=[ tables=[
DistributedLease.__table__,
RecoveryOperation.__table__,
RecoveryCheckpoint.__table__,
ConnectorTabularSource.__table__, ConnectorTabularSource.__table__,
DatasourceRecord.__table__, DatasourceRecord.__table__,
DatasourcePayloadRecord.__table__, DatasourcePayloadRecord.__table__,
@@ -86,153 +95,168 @@ def main() -> int:
], ],
) )
session_factory = sessionmaker(bind=engine) session_factory = sessionmaker(bind=engine)
with session_factory() as session: bind_process_runtime_identity(_runtime_identity())
principal = _principal() try:
writer = tabular_snapshot_writer(registry) with session_factory() as session:
lifecycle = datasource_lifecycle(registry) principal = _principal()
catalogue = datasource_catalogue(registry) writer = tabular_snapshot_writer(registry)
publisher = datasource_publication(registry) lifecycle = datasource_lifecycle(registry)
runner = dataflow_run_lifecycle(registry) catalogue = datasource_catalogue(registry)
if ( publisher = datasource_publication(registry)
writer is None runner = dataflow_run_lifecycle(registry)
or lifecycle is None if (
or catalogue is None writer is None
or publisher is None or lifecycle is None
or runner is None or catalogue is None
): or publisher is None
raise RuntimeError("Datasource composition capabilities are incomplete.") or runner is None
):
raise RuntimeError(
"Datasource composition capabilities are incomplete."
)
origin = writer.create_snapshot( origin = writer.create_snapshot(
session, session,
principal, principal,
snapshot=TabularSnapshotInput( snapshot=TabularSnapshotInput(
name="Monthly cases", name="Monthly cases",
source_name="connector_monthly_cases", source_name="connector_monthly_cases",
rows=( rows=(
{"id": 1, "amount": 5}, {"id": 1, "amount": 5},
{"id": 2, "amount": 15}, {"id": 2, "amount": 15},
),
), ),
), )
) datasource = lifecycle.register_origin(
datasource = lifecycle.register_origin( session,
session, principal,
principal, origin_ref=origin.ref,
origin_ref=origin.ref, name="Monthly cases cache",
name="Monthly cases cache", source_name="monthly_cases",
source_name="monthly_cases", mode="cached",
mode="cached", )
) result = preview_pipeline(
result = preview_pipeline( session,
session, tenant_id="tenant-1",
tenant_id="tenant-1", actor_id="account-1",
actor_id="account-1", payload=PipelinePreviewRequest(
payload=PipelinePreviewRequest( graph=_graph(
graph=_graph( datasource_ref=datasource.ref,
datasource_ref=datasource.ref, fingerprint=datasource.fingerprint,
fingerprint=datasource.fingerprint, ),
row_limit=100,
), ),
row_limit=100, principal=principal,
), registry=registry,
principal=principal, )
registry=registry, expected_rows = [
) {"id": 1, "amount": 5},
expected_rows = [ {"id": 2, "amount": 15},
{"id": 1, "amount": 5}, ]
{"id": 2, "amount": 15}, if result.status != "succeeded":
] raise RuntimeError(
if result.status != "succeeded": f"Dataflow preview failed: {result.diagnostics}"
raise RuntimeError(f"Dataflow preview failed: {result.diagnostics}") )
if result.rows != expected_rows: if result.rows != expected_rows:
raise RuntimeError(f"Unexpected Dataflow rows: {result.rows!r}") raise RuntimeError(f"Unexpected Dataflow rows: {result.rows!r}")
if result.source_fingerprints[0]["source_ref"] != datasource.ref: if result.source_fingerprints[0]["source_ref"] != datasource.ref:
raise RuntimeError("Dataflow lineage did not retain the datasource reference.") raise RuntimeError(
pipeline = create_pipeline( "Dataflow lineage did not retain the datasource reference."
session, )
tenant_id="tenant-1", pipeline = create_pipeline(
actor_id="account-1", session,
payload=PipelineCreateRequest( tenant_id="tenant-1",
name="Monthly case output", actor_id="account-1",
status="active", payload=PipelineCreateRequest(
graph=_graph( name="Monthly case output",
datasource_ref=datasource.ref, status="active",
fingerprint=datasource.fingerprint, graph=_graph(
datasource_ref=datasource.ref,
fingerprint=datasource.fingerprint,
),
editor_mode="graph",
), ),
editor_mode="graph",
),
)
run_request = DataflowRunRequest(
pipeline_ref=f"pipeline:{pipeline.id}",
revision=1,
idempotency_key="composition-run-1",
publication=DataflowPublicationTarget(
name="Monthly case result",
source_name="monthly_case_result",
freeze=True,
frozen_label="Composition evidence",
),
)
published = runner.start_run(
session,
principal,
request=run_request,
)
replayed = runner.start_run(
session,
principal,
request=run_request,
)
if published.status != "queued":
raise RuntimeError(
f"Dataflow run was not queued: {published.status}"
) )
worker = SqlDataflowRunWorker( run_request = DataflowRunRequest(
registry=_AutomationRegistry(registry, principal) pipeline_ref=f"pipeline:{pipeline.id}",
) revision=1,
worker_result = worker.dispatch_pending( idempotency_key="composition-run-1",
session, publication=DataflowPublicationTarget(
worker_id="composition-worker", name="Monthly case result",
) source_name="monthly_case_result",
if worker_result["succeeded"] != 1: freeze=True,
raise RuntimeError( frozen_label="Composition evidence",
f"Dataflow worker failed: {worker_result!r}" ),
) )
completed = runner.get_run( published = runner.start_run(
session, session,
principal, principal,
run_ref=published.ref, request=run_request,
)
if completed is None:
raise RuntimeError("Dataflow run evidence disappeared.")
published = completed
if published.status != "succeeded":
raise RuntimeError(f"Dataflow publication failed: {published.error}")
if replayed.ref != published.ref or not replayed.replayed:
raise RuntimeError("Dataflow run idempotency did not replay the prior run.")
if (
not published.output_datasource_ref
or not published.output_materialization_ref
):
raise RuntimeError("Dataflow publication did not retain output references.")
output = catalogue.read_datasource(
session,
principal,
request=DatasourceReadRequest(
datasource_ref=published.output_datasource_ref,
),
)
if list(output.rows) != expected_rows:
raise RuntimeError(
f"Unexpected published Dataflow rows: {list(output.rows)!r}"
) )
if ( replayed = runner.start_run(
output.materialization is None session,
or output.materialization.ref != published.output_materialization_ref principal,
or output.materialization.frozen_at is None request=run_request,
):
raise RuntimeError(
"Published Datasource materialization is not pinned and frozen."
) )
engine.dispose() if published.status != "queued":
raise RuntimeError(
f"Dataflow run was not queued: {published.status}"
)
worker = SqlDataflowRunWorker(
registry=_AutomationRegistry(registry, principal)
)
worker_result = worker.dispatch_pending(
session,
worker_id="composition-worker",
)
if worker_result["succeeded"] != 1:
raise RuntimeError(f"Dataflow worker failed: {worker_result!r}")
completed = runner.get_run(
session,
principal,
run_ref=published.ref,
)
if completed is None:
raise RuntimeError("Dataflow run evidence disappeared.")
published = completed
if published.status != "succeeded":
raise RuntimeError(
f"Dataflow publication failed: {published.error}"
)
if replayed.ref != published.ref or not replayed.replayed:
raise RuntimeError(
"Dataflow run idempotency did not replay the prior run."
)
if (
not published.output_datasource_ref
or not published.output_materialization_ref
):
raise RuntimeError(
"Dataflow publication did not retain output references."
)
output = catalogue.read_datasource(
session,
principal,
request=DatasourceReadRequest(
datasource_ref=published.output_datasource_ref,
),
)
if list(output.rows) != expected_rows:
raise RuntimeError(
f"Unexpected published Dataflow rows: {list(output.rows)!r}"
)
if (
output.materialization is None
or output.materialization.ref
!= published.output_materialization_ref
or output.materialization.frozen_at is None
):
raise RuntimeError(
"Published Datasource materialization is not pinned and frozen."
)
finally:
bind_process_runtime_identity(None)
engine.dispose()
print( print(
"Connector -> Datasources -> pinned Dataflow publication composition passed." "Connector -> Datasources -> pinned Dataflow publication composition passed."
) )
@@ -252,6 +276,17 @@ class _AutomationProvider:
) )
def _runtime_identity() -> RuntimeIdentity:
return RuntimeIdentity(
installation_id="datasource-composition-check",
node_id="composition-worker",
incarnation="composition-worker-incarnation",
role="worker",
software_version="test",
composition_hash="c" * 64,
)
class _AutomationRegistry: class _AutomationRegistry:
def __init__(self, registry, principal: ApiPrincipal) -> None: def __init__(self, registry, principal: ApiPrincipal) -> None:
self.registry = registry self.registry = registry