from __future__ import annotations import json import re from datetime import UTC, datetime, timedelta from pathlib import Path import polars as pl import pytest from minio import Minio from sqlalchemy import create_engine, text from app.core.common.identifiers import new_governance_uid from app.core.data_rules.contracts import rule_spec_hash, validate_rule_spec from app.core.data_rules.execution_contracts import canonical_schema_hash COMPOSE = ( Path(__file__).resolve().parents[2] / "deploy" / "docker" / "docker-compose.yml" ) def _compose_value(pattern): source = COMPOSE.read_text(encoding="utf-8") match = re.search(pattern, source, flags=re.DOTALL) assert match is not None return match.group(1) def _schema(schema_ref, fields): normalized = [ {"name": name, "type": field_type, "nullable": nullable} for name, field_type, nullable in fields ] return { "id": new_governance_uid(), "schema_ref": schema_ref, "schema_hash": canonical_schema_hash(normalized), "fields": normalized, "source_revision": "task5:real-cross-source", } def _binding(schema, *, source_uid, access_mode, object_ref): return { "id": new_governance_uid(), "data_source_uid": source_uid, "object_kind": "parquet_artifact", "object_ref": object_ref, "schema_snapshot_id": schema["id"], "access_mode": access_mode, "dialect": "parquet", "write_mode": "append", } def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path): from app.core.data_rules.compilers.polars import PolarsRuleCompiler from app.runner.artifacts import ArtifactStore, PostgresArtifactResolver from app.runner.rule_polars import PolarsRulePlanAdapter from app.runner.rules import PostgresRulePlanRepository, RulePlanExecutor source_user = _compose_value( r"source-postgres:.*?POSTGRES_USER:\s*([^\s]+)" ) source_password = _compose_value( r"source-postgres:.*?POSTGRES_PASSWORD:\s*([^\s]+)" ) platform_user = _compose_value( r"\n postgres:.*?POSTGRES_USER:\s*([^\s]+)" ) platform_password = _compose_value( r"\n postgres:.*?POSTGRES_PASSWORD:\s*([^\s]+)" ) postgres_port = _compose_value(r'"(25432):5432"') mysql_port = _compose_value(r'"(23306):3306"') minio_user = _compose_value(r"MINIO_ROOT_USER:\s*([^\s]+)") minio_password = _compose_value(r"MINIO_ROOT_PASSWORD:\s*([^\s]+)") minio_port = _compose_value(r'"(19000):9000"') platform_port = _compose_value(r'"(15432):5432"') bucket = _compose_value(r"mc mb --ignore-existing local/([^\s]+)") postgres = create_engine( f"postgresql+psycopg2://{source_user}:{source_password}" f"@127.0.0.1:{postgres_port}/acceptance", pool_pre_ping=True, ) mysql = create_engine( f"mysql+pymysql://{source_user}:{source_password}" f"@127.0.0.1:{mysql_port}/acceptance", pool_pre_ping=True, ) platform = create_engine( f"postgresql+psycopg2://{platform_user}:{platform_password}" f"@127.0.0.1:{platform_port}/dataops", pool_pre_ping=True, ) minio = Minio( f"127.0.0.1:{minio_port}", access_key=minio_user, secret_key=minio_password, secure=False, ) store = ArtifactStore( minio, bucket=bucket, max_artifact_bytes=4 * 1024 * 1024, max_rows=1_000, memory_limit_bytes=256 * 1024 * 1024, max_ttl_seconds=3600, ) correlation_id = new_governance_uid() failure_correlation_id = new_governance_uid() unknown_correlation_id = new_governance_uid() lease_correlation_id = new_governance_uid() sample_crash_correlation_id = new_governance_uid() receipt_correlation_id = new_governance_uid() failed_receipt_correlation_id = new_governance_uid() sql_binding_id = new_governance_uid() prefix = f"rules/{correlation_id}/" customer_table = "task5_polars_customers" segment_table = "task5_polars_segments" rule_uid = new_governance_uid() rule_id = new_governance_uid() downstream_rule_uid = new_governance_uid() downstream_rule_id = new_governance_uid() dataflow_uid = new_governance_uid() dataflow_version_id = new_governance_uid() deployment_id = new_governance_uid() component_binding_id = new_governance_uid() plan_id = new_governance_uid() downstream_component_binding_id = new_governance_uid() downstream_plan_id = new_governance_uid() ledger_jti = None retry_ledger_jti = None try: with postgres.begin() as connection: connection.execute(text(f"DROP TABLE IF EXISTS {customer_table}")) connection.execute( text( f"CREATE TABLE {customer_table} (" "customer_id BIGINT NOT NULL, " "name VARCHAR(100), mobile VARCHAR(30), " "segment_code VARCHAR(20), version_no BIGINT NOT NULL)" ) ) connection.execute( text( f"INSERT INTO {customer_table} " "(customer_id, name, mobile, segment_code, version_no) " "VALUES " "(1, ' Alice ', '13800138000', 'A', 1), " "(1, ' Alice Updated ', '13800138000', 'A', 2), " "(2, ' Bad ', 'invalid', 'B', 1), " "(3, ' Carol ', '13900139000', 'C', 1)" ) ) with mysql.begin() as connection: connection.execute(text(f"DROP TABLE IF EXISTS {segment_table}")) connection.execute( text( f"CREATE TABLE {segment_table} (" "code VARCHAR(20) PRIMARY KEY, " "segment_name VARCHAR(100) NOT NULL)" ) ) connection.execute( text( f"INSERT INTO {segment_table} (code, segment_name) " "VALUES ('A', 'Gold'), ('B', 'Basic'), ('C', 'Silver')" ) ) with postgres.connect() as connection: customer_rows = [ dict(row) for row in connection.execute( text( f"SELECT customer_id, name, mobile, " f"segment_code, version_no FROM {customer_table}" ) ).mappings() ] with mysql.connect() as connection: segment_rows = [ dict(row) for row in connection.execute( text( f"SELECT code, segment_name FROM {segment_table}" ) ).mappings() ] input_schema = _schema( "bd:task5:customer:raw", [ ("customer_id", "integer", False), ("name", "string", True), ("mobile", "string", True), ("segment_code", "string", True), ("version_no", "integer", False), ], ) lookup_schema = _schema( "bd:task5:segment:lookup", [ ("code", "string", False), ("segment_name", "string", False), ], ) output_schema = _schema( "bd:task5:customer:enriched", [ ("customer_id", "integer", False), ("name", "string", True), ("mobile", "string", True), ("segment_code", "string", True), ("version_no", "integer", False), ("segment_name", "string", True), ], ) input_binding = _binding( input_schema, source_uid=new_governance_uid(), access_mode="read", object_ref="postgres-customer-artifact", ) lookup_binding = _binding( lookup_schema, source_uid=new_governance_uid(), access_mode="read", object_ref="mysql-segment-artifact", ) output_binding = _binding( output_schema, source_uid=new_governance_uid(), access_mode="read_write", object_ref="polars-output-artifact", ) downstream_output_binding = _binding( output_schema, source_uid=new_governance_uid(), access_mode="write", object_ref="polars-downstream-output-artifact", ) spec = validate_rule_spec( { "schema_version": "2.0", "rule_uid": new_governance_uid(), "name": "task5_real_cross_source", "input_schema_ref": input_schema["schema_ref"], "output_schema_ref": output_schema["schema_ref"], "steps": [ { "id": "normalize_name", "op": "normalize_text", "column": "name", "trim": True, }, { "id": "join_segment", "op": "lookup_join", "lookup": { "binding_id": lookup_binding["id"], "left_on": ["segment_code"], "right_on": ["code"], "select": { "segment_name": "segment_name" }, "how": "left", }, }, { "id": "valid_mobile", "op": "assert", "expression": "matches(mobile, '^[0-9]{11}$')", "on_failure": "reject", "severity": "error", }, { "id": "latest_customer", "op": "deduplicate", "keys": ["customer_id"], "order_by": ["version_no"], "keep": "last", }, ], "null_policy": "explicit", "timezone": "Asia/Shanghai", } ) rule = { "id": rule_id, "status": "published", "rule_spec": spec, "spec_hash": rule_spec_hash(spec), } compiled = PolarsRuleCompiler().compile( rule_version=rule, input_schema=input_schema, output_schema=output_schema, input_binding=input_binding, output_binding=output_binding, backend={ "max_rows": 1_000, "max_artifact_bytes": 4 * 1024 * 1024, "memory_limit_bytes": 256 * 1024 * 1024, "masking_policies": {}, "lookup_bindings": { lookup_binding["id"]: { "binding": lookup_binding, "schema": lookup_schema, } }, }, ) downstream_spec = validate_rule_spec( { "schema_version": "2.0", "rule_uid": new_governance_uid(), "name": "task6_real_artifact_handoff", "input_schema_ref": output_schema["schema_ref"], "output_schema_ref": output_schema["schema_ref"], "steps": [ { "id": "normalize_downstream_name", "op": "normalize_text", "column": "name", "trim": True, } ], "null_policy": "explicit", "timezone": "Asia/Shanghai", } ) downstream_rule = { "id": downstream_rule_id, "status": "published", "rule_spec": downstream_spec, "spec_hash": rule_spec_hash(downstream_spec), } downstream_compiled = PolarsRuleCompiler().compile( rule_version=downstream_rule, input_schema=output_schema, output_schema=output_schema, input_binding=output_binding, output_binding=downstream_output_binding, backend={ "max_rows": 1_000, "max_artifact_bytes": 4 * 1024 * 1024, "memory_limit_bytes": 256 * 1024 * 1024, "masking_policies": {}, "lookup_bindings": {}, }, ) lookup_operation = compiled["plan"]["operations"][1] schema_hashes = { "rule_spec_hash": compiled["plan"]["rule_spec_hash"], "input_schema_snapshot_id": input_schema["id"], "input_schema_hash": input_schema["schema_hash"], "output_schema_snapshot_id": output_schema["id"], "output_schema_hash": output_schema["schema_hash"], } with platform.begin() as connection: for schema in (input_schema, lookup_schema, output_schema): connection.execute( text( """ INSERT INTO public.data_schema_snapshots (id, schema_ref, schema_hash, fields, source_revision) VALUES (CAST(:id AS uuid), :schema_ref, :schema_hash, CAST(:fields AS jsonb), :source_revision) """ ), {**schema, "fields": json.dumps(schema["fields"])}, ) connection.execute( text( """ INSERT INTO public.data_rules (id, rule_uid, name, category, status) VALUES (CAST(:id AS uuid), CAST(:rule_uid AS uuid), :name, 'general', 'active') """ ), { "id": new_governance_uid(), "rule_uid": rule_uid, "name": spec["name"], }, ) connection.execute( text( """ INSERT INTO public.data_rules (id, rule_uid, name, category, status) VALUES (CAST(:id AS uuid), CAST(:rule_uid AS uuid), :name, 'general', 'active') """ ), { "id": new_governance_uid(), "rule_uid": downstream_rule_uid, "name": downstream_spec["name"], }, ) connection.execute( text( """ INSERT INTO public.data_rule_versions (id, rule_uid, version_no, source_text, source_language, rule_spec, spec_hash, generated_kind, status, published_at) VALUES (CAST(:id AS uuid), CAST(:rule_uid AS uuid), 1, :source_text, 'en', CAST(:rule_spec AS jsonb), :spec_hash, 'polars', 'published', CURRENT_TIMESTAMP) """ ), { "id": rule_id, "rule_uid": rule_uid, "source_text": "Task 5 real cross-source integration", "rule_spec": json.dumps(spec), "spec_hash": rule["spec_hash"], }, ) connection.execute( text( """ INSERT INTO public.data_rule_versions (id, rule_uid, version_no, source_text, source_language, rule_spec, spec_hash, generated_kind, status, published_at) VALUES ( CAST(:id AS uuid), CAST(:rule_uid AS uuid), 1, :source_text, 'en', CAST(:rule_spec AS jsonb), :spec_hash, 'polars', 'published', CURRENT_TIMESTAMP ) """ ), { "id": downstream_rule_id, "rule_uid": downstream_rule_uid, "source_text": ( "Task 6 real two-node artifact handoff" ), "rule_spec": json.dumps(downstream_spec), "spec_hash": downstream_rule["spec_hash"], }, ) connection.execute( text( """ INSERT INTO public.dataflow_versions (id, dataflow_uid, version_no, name, dataflow_spec, input_schema_hashes, output_schema_hash, status, released_at) VALUES (CAST(:id AS uuid), CAST(:dataflow_uid AS uuid), 1, :name, '{}'::jsonb, CAST(:input_schema_hashes AS jsonb), :output_schema_hash, 'released', CURRENT_TIMESTAMP) """ ), { "id": dataflow_version_id, "dataflow_uid": dataflow_uid, "name": "Task 5 real cross-source integration", "input_schema_hashes": json.dumps( [ input_schema["schema_hash"], lookup_schema["schema_hash"], ] ), "output_schema_hash": output_schema["schema_hash"], }, ) connection.execute( text( """ INSERT INTO public.dataflow_deployments (id, dataflow_version_id, environment, deployment_config, status, activated_at) VALUES (CAST(:id AS uuid), CAST(:dataflow_version_id AS uuid), 'test', '{}'::jsonb, 'active', CURRENT_TIMESTAMP) """ ), { "id": deployment_id, "dataflow_version_id": dataflow_version_id, }, ) for logical_ref, binding, binding_hash in ( ( "customers", input_binding, compiled["plan"]["input_binding_hash"], ), ( "segments", lookup_binding, lookup_operation["lookup_binding_hash"], ), ( "enriched", output_binding, compiled["plan"]["output_binding_hash"], ), ( "downstream", downstream_output_binding, downstream_compiled["plan"][ "output_binding_hash" ], ), ): connection.execute( text( """ INSERT INTO public.dataflow_dataset_bindings (id, dataflow_deployment_id, logical_ref, data_source_uid, object_kind, object_ref, schema_snapshot_id, dialect, access_mode, write_mode, binding_hash) VALUES (CAST(:id AS uuid), CAST(:deployment_id AS uuid), :logical_ref, CAST(:source_uid AS uuid), 'parquet_artifact', :object_ref, CAST(:schema_snapshot_id AS uuid), 'parquet', :access_mode, 'append', :binding_hash) """ ), { "id": binding["id"], "deployment_id": deployment_id, "logical_ref": logical_ref, "source_uid": binding["data_source_uid"], "object_ref": binding["object_ref"], "schema_snapshot_id": binding["schema_snapshot_id"], "access_mode": binding["access_mode"], "binding_hash": binding_hash, }, ) connection.execute( text( """ INSERT INTO public.dataflow_component_bindings (id, dataflow_version_id, component_id, component_kind, rule_version_id, stage, order_no, idempotency, provenance) VALUES (CAST(:id AS uuid), CAST(:dataflow_version_id AS uuid), 'task5_real_polars', 'rule.apply', CAST(:rule_version_id AS uuid), 'transform', 0, CAST(:idempotency AS jsonb), '{}'::jsonb) """ ), { "id": component_binding_id, "dataflow_version_id": dataflow_version_id, "rule_version_id": rule_id, "idempotency": json.dumps( { "strategy": "deduplication_key", "key": "customer_id", } ), }, ) connection.execute( text( """ INSERT INTO public.dataflow_component_bindings (id, dataflow_version_id, component_id, component_kind, rule_version_id, stage, order_no, idempotency, provenance) VALUES ( CAST(:id AS uuid), CAST(:dataflow_version_id AS uuid), 'task6_real_handoff', 'rule.apply', CAST(:rule_version_id AS uuid), 'transform', 1, CAST(:idempotency AS jsonb), '{}'::jsonb ) """ ), { "id": downstream_component_binding_id, "dataflow_version_id": dataflow_version_id, "rule_version_id": downstream_rule_id, "idempotency": json.dumps( { "strategy": "deduplication_key", "key": "customer_id", } ), }, ) connection.execute( text( """ INSERT INTO public.rule_execution_plans (id, component_binding_id, backend, compiler_version, plan, plan_hash, schema_hashes, status) VALUES (CAST(:id AS uuid), CAST(:component_binding_id AS uuid), 'polars_batch', :compiler_version, CAST(:plan AS jsonb), :plan_hash, CAST(:schema_hashes AS jsonb), 'published') """ ), { "id": plan_id, "component_binding_id": component_binding_id, "compiler_version": compiled["compiler_version"], "plan": json.dumps(compiled["plan"]), "plan_hash": compiled["plan_hash"], "schema_hashes": json.dumps(schema_hashes), }, ) connection.execute( text( """ INSERT INTO public.rule_execution_plans (id, component_binding_id, backend, compiler_version, plan, plan_hash, schema_hashes, status) VALUES ( CAST(:id AS uuid), CAST(:component_binding_id AS uuid), 'polars_batch', :compiler_version, CAST(:plan AS jsonb), :plan_hash, CAST(:schema_hashes AS jsonb), 'published' ) """ ), { "id": downstream_plan_id, "component_binding_id": ( downstream_component_binding_id ), "compiler_version": downstream_compiled[ "compiler_version" ], "plan": json.dumps(downstream_compiled["plan"]), "plan_hash": downstream_compiled["plan_hash"], "schema_hashes": json.dumps( { "rule_spec_hash": downstream_compiled[ "plan" ]["rule_spec_hash"], "input_schema_snapshot_id": output_schema[ "id" ], "input_schema_hash": output_schema[ "schema_hash" ], "output_schema_snapshot_id": output_schema[ "id" ], "output_schema_hash": output_schema[ "schema_hash" ], } ), }, ) customer_path = tmp_path / "customers.parquet" segment_path = tmp_path / "segments.parquet" pl.DataFrame(customer_rows).write_parquet(customer_path) pl.DataFrame(segment_rows).write_parquet(segment_path) resolver = PostgresArtifactResolver(platform, store) customer_artifact = resolver.publish_path( str(customer_path), binding_id=input_binding["id"], binding_hash=compiled["plan"]["input_binding_hash"], correlation_id=correlation_id, kind="input", ttl_seconds=900, schema_fields=input_schema["fields"], limits=compiled["plan"]["resource_limits"], ) segment_artifact = resolver.publish_path( str(segment_path), binding_id=lookup_binding["id"], binding_hash=lookup_operation["lookup_binding_hash"], correlation_id=correlation_id, kind="lookup", ttl_seconds=900, schema_fields=lookup_schema["fields"], limits=compiled["plan"]["resource_limits"], ) node = { "id": "task5_real_polars", "type": "rule.apply", "purpose": "write", "idempotency": { "strategy": "deduplication_key", "key": "customer_id", }, "config": { "component_binding_id": component_binding_id, "rule_version_id": rule["id"], "execution_plan_hash": compiled["plan_hash"], }, } executor = RulePlanExecutor( PostgresRulePlanRepository(platform), adapters={ "polars_batch": PolarsRulePlanAdapter( artifact_store=store, artifact_resolver=resolver, artifact_ttl_seconds=900, ) }, ) result = executor.execute( node, {}, write_authorized=True, correlation_id=correlation_id, ) repeated = executor.execute( node, {}, write_authorized=True, correlation_id=correlation_id, ) assert result["rows_in"] == 4 assert result["rows_out"] == 2 assert result["rows_rejected"] == 1 assert result["rows_deduplicated"] == 1 assert result["rows_filtered"] == 0 assert result["rows_join_dropped"] == 0 assert result["rows_aggregated"] == 0 assert result["violation_count"] == 1 assert result["violations"] == [ {"step_id": "valid_mobile", "count": 1} ] output = store.read( result["artifact_ref"], result["digest"], expected_schema_fields=output_schema["fields"], limits=compiled["plan"]["resource_limits"], ).collect() assert output.sort("customer_id").to_dicts() == [ { "customer_id": 1, "mobile": "13800138000", "name": "Alice Updated", "segment_code": "A", "segment_name": "Gold", "version_no": 2, }, { "customer_id": 3, "mobile": "13900139000", "name": "Carol", "segment_code": "C", "segment_name": "Silver", "version_no": 1, }, ] assert all( item.object_name.startswith(prefix) for item in minio.list_objects( bucket, prefix=prefix, recursive=True ) ) assert repeated["rows_out"] == 2 assert repeated["artifact_ref"] == result["artifact_ref"] assert "schema_fields" not in result from app.runner.api import create_runner_app from app.runner.auth import TaskTokenIssuer, TaskTokenVerifier from app.runner.ledger import PostgresTaskLedger from app.runner.nodes import NodeRegistry from app.runner.rule_evidence import PostgresRuleEvidenceWriter task_secret = "task5-real-http-secret-value-32-bytes" verifier = TaskTokenVerifier(task_secret) task_token = TaskTokenIssuer(task_secret).issue( task_uid=new_governance_uid(), dataflow_uid=dataflow_uid, deployment_id=deployment_id, environment="test", workflow_version=1, correlation_id=correlation_id, node=node, write_authorized=True, ) ledger_jti = verifier.verify(task_token, node=node).jti retry_token = TaskTokenIssuer(task_secret).issue( task_uid=new_governance_uid(), dataflow_uid=dataflow_uid, deployment_id=deployment_id, environment="test", workflow_version=1, correlation_id=correlation_id, node=node, write_authorized=True, ) retry_ledger_jti = verifier.verify( retry_token, node=node ).jti evidenced_executor = RulePlanExecutor( PostgresRulePlanRepository(platform), adapters={ "polars_batch": PolarsRulePlanAdapter( artifact_store=store, artifact_resolver=resolver, artifact_ttl_seconds=900, ) }, evidence_writer=PostgresRuleEvidenceWriter( platform, store, sample_ttl_seconds=900, ), ) runner_app = create_runner_app( verifier=verifier, ledger=PostgresTaskLedger(platform), registry=NodeRegistry({"rule.apply": evidenced_executor}), ) with runner_app.test_client() as client: http_result = client.post( "/v1/tasks/execute", json={ "task_token": task_token, "node": node, "parameters": {}, }, ) replay = client.post( "/v1/tasks/execute", json={ "task_token": task_token, "node": node, "parameters": {}, }, ) retried = client.post( "/v1/tasks/execute", json={ "task_token": retry_token, "node": node, "parameters": {}, }, ) assert http_result.status_code == 200, http_result.get_json() assert http_result.get_json()["output_artifact"] == result[ "artifact_ref" ] assert http_result.get_json()["result"]["artifact_ref"] == result[ "artifact_ref" ] assert replay.status_code == 200 assert replay.headers["X-Idempotent-Replay"] == "true" assert replay.get_json() == http_result.get_json() assert retried.status_code == 200 assert retried.get_json()["output_artifact"] == result[ "artifact_ref" ] ledger_record = PostgresTaskLedger(platform).get(ledger_jti) assert ledger_record is not None assert ledger_record.status == "success" assert ledger_record.commit_outcome == "committed" with platform.connect() as connection: run_evidence = connection.execute( text( """ SELECT status, commit_outcome, rows_in, rows_out, rows_rejected, rows_quarantined, public_result FROM public.rule_runs WHERE correlation_id = CAST(:correlation_id AS uuid) AND component_binding_id = CAST(:component_binding_id AS uuid) """ ), { "correlation_id": correlation_id, "component_binding_id": component_binding_id, }, ).mappings().one() sample_evidence = connection.execute( text( """ SELECT s.artifact_ref, s.artifact_digest, s.sample_count, s.redaction_policy, s.expires_at, s.handoff_status FROM public.rule_violation_samples s JOIN public.rule_runs r ON r.id = s.rule_run_id WHERE r.correlation_id = CAST(:correlation_id AS uuid) """ ), {"correlation_id": correlation_id}, ).mappings().one() assert run_evidence["status"] == "success" assert run_evidence["commit_outcome"] == "committed" assert run_evidence["rows_in"] == 4 assert run_evidence["rows_out"] == 2 assert run_evidence["rows_rejected"] == 1 assert run_evidence["rows_quarantined"] == 0 assert run_evidence["public_result"]["output_artifact"] == result[ "artifact_ref" ] assert sample_evidence["artifact_digest"] assert sample_evidence["sample_count"] == 1 assert ( sample_evidence["redaction_policy"] == "rule-violation-default-v1" ) assert sample_evidence["handoff_status"] == "ready" assert store.read( sample_evidence["artifact_ref"], sample_evidence["artifact_digest"], expected_schema_fields=[ { "name": name, "type": "string", "nullable": True, } for name in ( "customer_id", "mobile", "name", "segment_code", "segment_name", "version_no", ) ], ).collect().to_dicts() == [ { "customer_id": "[REDACTED]", "mobile": "[REDACTED]", "name": "[REDACTED]", "segment_code": "[REDACTED]", "segment_name": "[REDACTED]", "version_no": "[REDACTED]", } ] orphan = store.write( pl.DataFrame({"value": ["orphan"]}), correlation_id, 900, schema_fields=[ { "name": "value", "type": "string", "nullable": True, } ], ) original_clock = store.clock store.clock = lambda: datetime.now(UTC) + timedelta(seconds=60) try: reconciliation = resolver.reconcile( limit=20, grace_seconds=30, ) finally: store.clock = original_clock assert reconciliation["orphans_deleted"] >= 1 assert store.describe_optional(orphan["artifact_ref"]) is None assert store.describe(sample_evidence["artifact_ref"])[ "digest" ] == sample_evidence["artifact_digest"] downstream_node = { "id": "task6_real_handoff", "type": "rule.apply", "purpose": "write", "idempotency": { "strategy": "deduplication_key", "key": "customer_id", }, "config": { "component_binding_id": ( downstream_component_binding_id ), "rule_version_id": downstream_rule_id, "execution_plan_hash": downstream_compiled[ "plan_hash" ], }, } downstream_result = evidenced_executor.execute( downstream_node, {"input_artifact": result["artifact_ref"]}, write_authorized=True, correlation_id=correlation_id, dataflow_uid=dataflow_uid, deployment_id=deployment_id, environment="test", workflow_version=1, node_id="task6_real_handoff", task_jti=new_governance_uid(), ) assert downstream_result["rows_in"] == 2 assert downstream_result["rows_out"] == 2 assert downstream_result["output_artifact"] != result[ "artifact_ref" ] assert store.read( downstream_result["artifact_ref"], downstream_result["digest"], expected_schema_fields=output_schema["fields"], limits=downstream_compiled["plan"]["resource_limits"], ).collect().sort("customer_id").to_dicts() == ( output.sort("customer_id").to_dicts() ) replayed_downstream = evidenced_executor.execute( downstream_node, {"input_artifact": result["artifact_ref"]}, write_authorized=True, correlation_id=correlation_id, dataflow_uid=dataflow_uid, deployment_id=deployment_id, environment="test", workflow_version=1, node_id="task6_real_handoff", task_jti=new_governance_uid(), ) assert replayed_downstream["artifact_ref"] == downstream_result[ "artifact_ref" ] with platform.connect() as connection: assert connection.execute( text( """ SELECT COUNT(*) FROM public.rule_runs WHERE correlation_id = CAST(:correlation_id AS uuid) """ ), {"correlation_id": correlation_id}, ).scalar_one() == 2 from app.runner.nodes import NodeExecutionError class FailingAdapter: def execute(self, **_kwargs): raise NodeExecutionError( "safe downstream failure", commit_outcome="not_committed", ) failed_executor = RulePlanExecutor( PostgresRulePlanRepository(platform), adapters={"polars_batch": FailingAdapter()}, evidence_writer=PostgresRuleEvidenceWriter( platform, store, sample_ttl_seconds=900, ), ) with pytest.raises(NodeExecutionError, match="safe downstream"): failed_executor.execute( node, {}, write_authorized=True, correlation_id=failure_correlation_id, dataflow_uid=dataflow_uid, deployment_id=deployment_id, environment="test", workflow_version=1, node_id="task5_real_polars", task_jti=new_governance_uid(), ) with platform.connect() as connection: failed_evidence = connection.execute( text( """ SELECT status, commit_outcome FROM public.rule_runs WHERE correlation_id = CAST(:correlation_id AS uuid) """ ), {"correlation_id": failure_correlation_id}, ).mappings().one() assert dict(failed_evidence) == { "status": "failed", "commit_outcome": "not_committed", } class UnknownAdapter: def execute(self, **_kwargs): raise NodeExecutionError( "safe uncertain commit", commit_outcome="unknown", ) unknown_executor = RulePlanExecutor( PostgresRulePlanRepository(platform), adapters={"polars_batch": UnknownAdapter()}, evidence_writer=PostgresRuleEvidenceWriter( platform, store, sample_ttl_seconds=900, ), ) with pytest.raises(NodeExecutionError, match="uncertain commit"): unknown_executor.execute( node, {}, write_authorized=True, correlation_id=unknown_correlation_id, dataflow_uid=dataflow_uid, deployment_id=deployment_id, environment="test", workflow_version=1, node_id="task5_real_polars", task_jti=new_governance_uid(), ) with platform.connect() as connection: unknown_evidence = connection.execute( text( """ SELECT status, commit_outcome FROM public.rule_runs WHERE correlation_id = CAST(:correlation_id AS uuid) """ ), {"correlation_id": unknown_correlation_id}, ).mappings().one() assert dict(unknown_evidence) == { "status": "unknown", "commit_outcome": "unknown", } evidence_writer = PostgresRuleEvidenceWriter( platform, store, sample_ttl_seconds=900, lease_seconds=30, ) lease_owner = new_governance_uid() lease_run_id = evidence_writer.start( component_binding_id=component_binding_id, rule_version_id=rule_id, plan_hash=compiled["plan_hash"], correlation_id=lease_correlation_id, dataflow_uid=dataflow_uid, deployment_id=deployment_id, environment="test", workflow_version=1, node_id="task5_real_polars", lease_owner=lease_owner, ) with pytest.raises(ValueError, match="not owned"): evidence_writer.heartbeat( lease_run_id, new_governance_uid(), ) evidence_writer.heartbeat(lease_run_id, lease_owner) with platform.begin() as connection: connection.execute( text( """ UPDATE public.rule_runs SET lease_expires_at = CURRENT_TIMESTAMP - INTERVAL '1 second' WHERE id = CAST(:id AS uuid) """ ), {"id": lease_run_id}, ) assert evidence_writer.start( component_binding_id=component_binding_id, rule_version_id=rule_id, plan_hash=compiled["plan_hash"], correlation_id=lease_correlation_id, dataflow_uid=dataflow_uid, deployment_id=deployment_id, environment="test", workflow_version=1, node_id="task5_real_polars", lease_owner=new_governance_uid(), ) == lease_run_id assert evidence_writer.replay(lease_run_id)["status"] == "unknown" class FailOnceConnection: def __init__(self, connection, state): self.connection = connection self.state = state def execute(self, statement, parameters=None): sql = str(statement) if ( self.state["armed"] and "UPDATE public.rule_runs" in sql and "rows_in = :rows_in" in sql ): self.state["armed"] = False raise RuntimeError("simulated response loss") return self.connection.execute(statement, parameters) class FailOnceBegin: def __init__(self, context, state): self.context = context self.state = state def __enter__(self): return FailOnceConnection( self.context.__enter__(), self.state, ) def __exit__(self, *args): return self.context.__exit__(*args) class FailOnceEngine: def __init__(self, engine): self.engine = engine self.state = {"armed": True} def connect(self): return self.engine.connect() def begin(self): return FailOnceBegin( self.engine.begin(), self.state, ) sample_crash_writer = PostgresRuleEvidenceWriter( FailOnceEngine(platform), store, sample_ttl_seconds=900, ) sample_crash_run_id = sample_crash_writer.start( component_binding_id=component_binding_id, rule_version_id=rule_id, plan_hash=compiled["plan_hash"], correlation_id=sample_crash_correlation_id, dataflow_uid=dataflow_uid, deployment_id=deployment_id, environment="test", workflow_version=1, node_id="task5_real_polars", lease_owner=new_governance_uid(), ) sample_crash_evidence = { "status": "success", "rows_in": 1, "rows_out": 0, "rows_rejected": 1, "rows_quarantined": 0, "commit_outcome": "committed", "timings": {"duration_ms": 1}, "violation_sample": [{"mobile": "[REDACTED]"}], "sample_count": 1, "redaction_policy": "rule-violation-default-v1", } with pytest.raises(RuntimeError, match="outcome is unknown"): sample_crash_writer.finish( sample_crash_run_id, sample_crash_evidence, ) with platform.connect() as connection: assert connection.execute( text( """ SELECT handoff_status FROM public.rule_violation_samples WHERE rule_run_id = CAST(:id AS uuid) """ ), {"id": sample_crash_run_id}, ).scalar_one() == "ready" evidence_writer.finish( sample_crash_run_id, sample_crash_evidence, ) assert evidence_writer.replay(sample_crash_run_id)[ "status" ] == "success" with platform.begin() as connection: connection.execute( text( """ INSERT INTO public.dataflow_dataset_bindings (id, dataflow_deployment_id, logical_ref, data_source_uid, object_kind, object_ref, schema_snapshot_id, dialect, access_mode, write_mode, binding_hash) VALUES ( CAST(:id AS uuid), CAST(:deployment_id AS uuid), 'sql_receipt', CAST(:source_uid AS uuid), 'table', 'public.task6_sql_receipt', CAST(:schema_snapshot_id AS uuid), 'postgresql', 'read_write', 'upsert', :binding_hash ) """ ), { "id": sql_binding_id, "deployment_id": deployment_id, "source_uid": output_binding["data_source_uid"], "schema_snapshot_id": output_schema["id"], "binding_hash": "b" * 64, }, ) receipt_run_id = evidence_writer.start( component_binding_id=component_binding_id, rule_version_id=rule_id, plan_hash=compiled["plan_hash"], correlation_id=receipt_correlation_id, dataflow_uid=dataflow_uid, deployment_id=deployment_id, environment="test", workflow_version=1, node_id="task5_real_polars", lease_owner=new_governance_uid(), ) receipt = evidence_writer.stage_sql_output( receipt_run_id, output_binding_id=sql_binding_id, ) assert re.fullmatch( r"dataops-staging://[0-9a-f-]{36}", receipt, ) with pytest.raises(ValueError, match="not executable"): evidence_writer.resolve_sql_staging( receipt, deployment_id=deployment_id, correlation_id=receipt_correlation_id, input_binding_id=sql_binding_id, ) receipt_evidence = { "status": "success", "rows_in": 2, "rows_out": 2, "rows_rejected": 0, "rows_quarantined": 0, "commit_outcome": "committed", "timings": {"duration_ms": 1}, "public_result": { "rows_in": 2, "rows_out": 2, "rows_rejected": 0, "rows_quarantined": 0, "commit_outcome": "committed", }, } evidence_writer.finish(receipt_run_id, receipt_evidence) evidence_writer.finish(receipt_run_id, receipt_evidence) with pytest.raises(ValueError, match="immutable"): evidence_writer.finish( receipt_run_id, { **receipt_evidence, "rows_out": 1, }, ) resolved_receipt = evidence_writer.resolve_sql_staging( receipt, deployment_id=deployment_id, correlation_id=receipt_correlation_id, input_binding_id=sql_binding_id, ) assert resolved_receipt["relation_ref"] == ( "public.task6_sql_receipt" ) assert len(resolved_receipt["relation_digest"]) == 64 with pytest.raises(ValueError, match="not executable"): evidence_writer.resolve_sql_staging( receipt, deployment_id=deployment_id, correlation_id=new_governance_uid(), input_binding_id=sql_binding_id, ) with pytest.raises(ValueError, match="invalid"): evidence_writer.resolve_sql_staging( "dataops-staging://public.task6_sql_receipt", deployment_id=deployment_id, correlation_id=receipt_correlation_id, input_binding_id=sql_binding_id, ) failed_receipt_run_id = evidence_writer.start( component_binding_id=component_binding_id, rule_version_id=rule_id, plan_hash=compiled["plan_hash"], correlation_id=failed_receipt_correlation_id, dataflow_uid=dataflow_uid, deployment_id=deployment_id, environment="test", workflow_version=1, node_id="task5_real_polars", lease_owner=new_governance_uid(), ) failed_receipt = evidence_writer.stage_sql_output( failed_receipt_run_id, output_binding_id=sql_binding_id, ) evidence_writer.finish( failed_receipt_run_id, { "status": "failed", "commit_outcome": "not_committed", "timings": {"duration_ms": 1}, }, ) with pytest.raises(ValueError, match="not executable"): evidence_writer.resolve_sql_staging( failed_receipt, deployment_id=deployment_id, correlation_id=failed_receipt_correlation_id, input_binding_id=sql_binding_id, ) conflict_path = tmp_path / "conflict.parquet" pl.DataFrame( { "customer_id": [99], "mobile": ["13800138000"], "name": ["Conflict"], "segment_code": ["A"], "segment_name": ["Gold"], "version_no": [1], } ).write_parquet(conflict_path) object_count_before_conflict = len( list(minio.list_objects(bucket, prefix=prefix, recursive=True)) ) with pytest.raises(ValueError, match="immutable|digest"): resolver.publish_path( str(conflict_path), binding_id=output_binding["id"], binding_hash=compiled["plan"]["output_binding_hash"], correlation_id=correlation_id, kind="output", ttl_seconds=900, schema_fields=output_schema["fields"], limits=compiled["plan"]["resource_limits"], ) assert len( list(minio.list_objects(bucket, prefix=prefix, recursive=True)) ) == object_count_before_conflict assert customer_artifact["digest"] assert segment_artifact["digest"] with platform.connect() as connection: catalog_rows = connection.execute( text( """ SELECT artifact_kind, artifact_ref, handoff_status, binding_hash FROM public.rule_run_artifacts WHERE correlation_id = CAST(:correlation_id AS uuid) ORDER BY artifact_kind """ ), {"correlation_id": correlation_id}, ).mappings().all() # The repeated deterministic output has the same digest and is # idempotently retained as one stable catalog handoff. assert len(catalog_rows) == 4 assert all(row["handoff_status"] == "ready" for row in catalog_rows) assert all(len(row["binding_hash"]) == 64 for row in catalog_rows) assert { row["artifact_ref"] for row in catalog_rows if row["artifact_kind"] == "output" } == { result["artifact_ref"], downstream_result["artifact_ref"], } assert len( list(minio.list_objects(bucket, prefix=prefix, recursive=True)) ) == 5 finally: for item in list( minio.list_objects(bucket, prefix=prefix, recursive=True) ): minio.remove_object(bucket, item.object_name) assert list( minio.list_objects(bucket, prefix=prefix, recursive=True) ) == [] with postgres.begin() as connection: connection.execute(text(f"DROP TABLE IF EXISTS {customer_table}")) with mysql.begin() as connection: connection.execute(text(f"DROP TABLE IF EXISTS {segment_table}")) with platform.begin() as connection: if ledger_jti is not None: connection.execute( text( "DELETE FROM public.runner_task_executions " "WHERE token_jti = CAST(:jti AS uuid)" ), {"jti": ledger_jti}, ) if retry_ledger_jti is not None: connection.execute( text( "DELETE FROM public.runner_task_executions " "WHERE token_jti = CAST(:jti AS uuid)" ), {"jti": retry_ledger_jti}, ) connection.execute( text( """ DELETE FROM public.rule_sql_staging_receipts WHERE correlation_id IN ( CAST(:receipt_correlation_id AS uuid), CAST(:failed_receipt_correlation_id AS uuid) ) """ ), { "receipt_correlation_id": receipt_correlation_id, "failed_receipt_correlation_id": ( failed_receipt_correlation_id ), }, ) connection.execute( text( """ DELETE FROM public.rule_violation_samples s USING public.rule_runs r WHERE s.rule_run_id = r.id AND r.correlation_id IN ( CAST(:correlation_id AS uuid), CAST(:sample_crash_correlation_id AS uuid) ) """ ), { "correlation_id": correlation_id, "sample_crash_correlation_id": ( sample_crash_correlation_id ), }, ) connection.execute( text( "DELETE FROM public.rule_runs " "WHERE correlation_id IN (" "CAST(:correlation_id AS uuid), " "CAST(:failure_correlation_id AS uuid), " "CAST(:unknown_correlation_id AS uuid), " "CAST(:lease_correlation_id AS uuid), " "CAST(:sample_crash_correlation_id AS uuid), " "CAST(:receipt_correlation_id AS uuid), " "CAST(:failed_receipt_correlation_id AS uuid))" ), { "correlation_id": correlation_id, "failure_correlation_id": failure_correlation_id, "unknown_correlation_id": unknown_correlation_id, "lease_correlation_id": lease_correlation_id, "sample_crash_correlation_id": ( sample_crash_correlation_id ), "receipt_correlation_id": receipt_correlation_id, "failed_receipt_correlation_id": ( failed_receipt_correlation_id ), }, ) connection.execute( text( "DELETE FROM public.dataflow_dataset_bindings " "WHERE id = CAST(:id AS uuid)" ), {"id": sql_binding_id}, ) connection.execute( text( "DELETE FROM public.rule_run_artifacts " "WHERE correlation_id = CAST(:correlation_id AS uuid)" ), {"correlation_id": correlation_id}, ) connection.execute( text( "DELETE FROM public.rule_execution_plans " "WHERE id IN (CAST(:id AS uuid), " "CAST(:downstream_id AS uuid))" ), { "id": plan_id, "downstream_id": downstream_plan_id, }, ) connection.execute( text( "DELETE FROM public.dataflow_dataset_bindings " "WHERE dataflow_deployment_id = CAST(:id AS uuid)" ), {"id": deployment_id}, ) connection.execute( text( "DELETE FROM public.dataflow_component_bindings " "WHERE id IN (CAST(:id AS uuid), " "CAST(:downstream_id AS uuid))" ), { "id": component_binding_id, "downstream_id": downstream_component_binding_id, }, ) connection.execute( text( "DELETE FROM public.dataflow_deployments " "WHERE id = CAST(:id AS uuid)" ), {"id": deployment_id}, ) connection.execute( text( "DELETE FROM public.dataflow_versions " "WHERE id = CAST(:id AS uuid)" ), {"id": dataflow_version_id}, ) connection.execute( text( "DELETE FROM public.data_rule_versions " "WHERE id IN (CAST(:id AS uuid), " "CAST(:downstream_id AS uuid))" ), { "id": rule_id, "downstream_id": downstream_rule_id, }, ) connection.execute( text( "DELETE FROM public.data_rules " "WHERE rule_uid IN (CAST(:rule_uid AS uuid), " "CAST(:downstream_rule_uid AS uuid))" ), { "rule_uid": rule_uid, "downstream_rule_uid": downstream_rule_uid, }, ) for schema in (input_schema, lookup_schema, output_schema): connection.execute( text( "DELETE FROM public.data_schema_snapshots " "WHERE id = CAST(:id AS uuid)" ), {"id": schema["id"]}, ) postgres.dispose() mysql.dispose() platform.dispose()