from __future__ import annotations import json import re 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() 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, 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, 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 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 == 409 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]", } ] 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, workflow_version=1, node_id="task6_real_handoff", ) 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, workflow_version=1, node_id="task6_real_handoff", ) 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, workflow_version=1, node_id="task5_real_polars", ) 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, workflow_version=1, node_id="task5_real_polars", ) 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", } 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_violation_samples s USING public.rule_runs r WHERE s.rule_run_id = r.id AND r.correlation_id = CAST(:correlation_id AS uuid) """ ), {"correlation_id": 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))" ), { "correlation_id": correlation_id, "failure_correlation_id": failure_correlation_id, "unknown_correlation_id": unknown_correlation_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()