|
|
@@ -0,0 +1,1721 @@
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import copy
|
|
|
+import hashlib
|
|
|
+import json
|
|
|
+import os
|
|
|
+import socket
|
|
|
+import subprocess
|
|
|
+import time
|
|
|
+import urllib.request
|
|
|
+from contextlib import contextmanager, suppress
|
|
|
+from datetime import UTC, datetime, timedelta
|
|
|
+from pathlib import Path
|
|
|
+
|
|
|
+import polars as pl
|
|
|
+import pytest
|
|
|
+import requests
|
|
|
+from minio import Minio
|
|
|
+from openai import OpenAI
|
|
|
+from sqlalchemy import create_engine, text
|
|
|
+from sqlalchemy.orm import Session
|
|
|
+
|
|
|
+from app.core.common.identifiers import new_governance_uid
|
|
|
+from app.core.data_rules.authoring import (
|
|
|
+ OpenAICompatibleRuleModel,
|
|
|
+ RuleAuthoringAgent,
|
|
|
+)
|
|
|
+from app.core.data_rules.compilers import CompilerRegistry
|
|
|
+from app.core.data_rules.compilers.polars import PolarsRuleCompiler
|
|
|
+from app.core.data_rules.deployment import DataFlowDeploymentService
|
|
|
+from app.core.data_rules.execution_contracts import canonical_schema_hash
|
|
|
+from app.core.data_rules.publication import (
|
|
|
+ GenerationReceiptSigner,
|
|
|
+ LogicalRuleCompiler,
|
|
|
+ PhysicalPlanPublicationService,
|
|
|
+ RulePublicationService,
|
|
|
+ ServerOwnedLogicalDryRunRunner,
|
|
|
+ ServerOwnedPhysicalPreflightRunner,
|
|
|
+ generation_receipt_claims,
|
|
|
+)
|
|
|
+from app.core.data_rules.release import (
|
|
|
+ BoundSqlPlanService,
|
|
|
+ ProductionLineReleaseService,
|
|
|
+)
|
|
|
+from app.core.data_rules.repository import DataRuleRepository
|
|
|
+from app.core.orchestration.engines.kestra import KestraAdapter
|
|
|
+from app.runner.artifacts import ArtifactStore, PostgresArtifactResolver
|
|
|
+from app.runner.auth import TaskTokenIssuer
|
|
|
+from app.runner.rule_polars import PolarsRulePlanAdapter
|
|
|
+from app.runner.rules import PostgresRulePlanRepository, RulePlanExecutor
|
|
|
+
|
|
|
+ROOT = Path(__file__).resolve().parents[2]
|
|
|
+SERVER = ROOT / "scripts" / "validation" / "local_openai_rule_model_server.py"
|
|
|
+MODEL_PYTHON = Path(
|
|
|
+ os.environ.get(
|
|
|
+ "DATA_RULE_LOCAL_MODEL_PYTHON",
|
|
|
+ "/private/tmp/dataops-mlx-lm-venv/bin/python",
|
|
|
+ )
|
|
|
+)
|
|
|
+MODEL_NAME = os.environ.get("DATA_RULE_LOCAL_MODEL_NAME", "Qwen/Qwen3-1.7B")
|
|
|
+HF_HOME = os.environ.get("HF_HOME", "/private/tmp/dataops-hf-cache")
|
|
|
+SOURCE_TEXT = (
|
|
|
+ "客户手机号去除空格后必须为11位数字;格式错误进入隔离区;"
|
|
|
+ "按customer_id去重并保留updated_at最新记录。"
|
|
|
+)
|
|
|
+EVIDENCE = (
|
|
|
+ ROOT / "docs" / "acceptance" / "data-rule" / "real-qwen-authoring-evidence.json"
|
|
|
+)
|
|
|
+FULL_CHAIN_EVIDENCE = (
|
|
|
+ ROOT
|
|
|
+ / "docs"
|
|
|
+ / "acceptance"
|
|
|
+ / "data-rule"
|
|
|
+ / "real-qwen-production-line-evidence.json"
|
|
|
+)
|
|
|
+PLATFORM_DATABASE_URL = os.environ.get(
|
|
|
+ "DATA_RULE_PLATFORM_DATABASE_URL",
|
|
|
+ "postgresql+psycopg2://dataops:dataops-test-password@127.0.0.1:15432/dataops",
|
|
|
+)
|
|
|
+
|
|
|
+
|
|
|
+def _canonical_hash(value) -> str:
|
|
|
+ return hashlib.sha256(
|
|
|
+ json.dumps(
|
|
|
+ value,
|
|
|
+ ensure_ascii=False,
|
|
|
+ separators=(",", ":"),
|
|
|
+ sort_keys=True,
|
|
|
+ ).encode("utf-8")
|
|
|
+ ).hexdigest()
|
|
|
+
|
|
|
+
|
|
|
+def _artifact_store() -> ArtifactStore:
|
|
|
+ return ArtifactStore(
|
|
|
+ Minio(
|
|
|
+ os.environ.get("DATA_RULE_MINIO_ENDPOINT", "127.0.0.1:19000"),
|
|
|
+ access_key=os.environ.get("DATA_RULE_MINIO_ACCESS_KEY", "dataops-test"),
|
|
|
+ secret_key=os.environ.get(
|
|
|
+ "DATA_RULE_MINIO_SECRET_KEY", "dataops-test-password"
|
|
|
+ ),
|
|
|
+ secure=False,
|
|
|
+ ),
|
|
|
+ bucket=os.environ.get("DATA_RULE_MINIO_BUCKET", "dataops-bucket"),
|
|
|
+ max_artifact_bytes=32 * 1024 * 1024,
|
|
|
+ max_rows=100_000,
|
|
|
+ memory_limit_bytes=256 * 1024 * 1024,
|
|
|
+ max_ttl_seconds=3600,
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+class _PinnedSchemaResolver:
|
|
|
+ def __init__(self, snapshots):
|
|
|
+ self.snapshots = {snapshot["schema_ref"]: snapshot for snapshot in snapshots}
|
|
|
+
|
|
|
+ def resolve(self, schema_ref):
|
|
|
+ try:
|
|
|
+ return copy.deepcopy(self.snapshots[schema_ref])
|
|
|
+ except KeyError as exc:
|
|
|
+ raise ValueError("unexpected E2E schema reference") from exc
|
|
|
+
|
|
|
+
|
|
|
+class _SharedConnectionEngine:
|
|
|
+ """Expose one rollback-scoped connection to Runner repositories."""
|
|
|
+
|
|
|
+ def __init__(self, connection):
|
|
|
+ self.connection = connection
|
|
|
+
|
|
|
+ @contextmanager
|
|
|
+ def connect(self):
|
|
|
+ yield self.connection
|
|
|
+
|
|
|
+ @contextmanager
|
|
|
+ def begin(self):
|
|
|
+ yield self.connection
|
|
|
+
|
|
|
+
|
|
|
+def _persist_preparatory_deployment(
|
|
|
+ connection,
|
|
|
+ *,
|
|
|
+ deployment_id,
|
|
|
+ version_id,
|
|
|
+ environment="production",
|
|
|
+):
|
|
|
+ connection.execute(
|
|
|
+ text(
|
|
|
+ "INSERT INTO public.dataflow_deployments "
|
|
|
+ "(id, dataflow_version_id, environment, deployment_config, status) "
|
|
|
+ "VALUES (CAST(:id AS uuid), CAST(:version_id AS uuid), "
|
|
|
+ ":environment, '{}'::jsonb, 'disabled')"
|
|
|
+ ),
|
|
|
+ {
|
|
|
+ "id": deployment_id,
|
|
|
+ "version_id": version_id,
|
|
|
+ "environment": environment,
|
|
|
+ },
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def _catalog_input_artifact(
|
|
|
+ connection,
|
|
|
+ *,
|
|
|
+ binding,
|
|
|
+ correlation_id,
|
|
|
+ artifact,
|
|
|
+ schema_fields,
|
|
|
+):
|
|
|
+ connection.execute(
|
|
|
+ text(
|
|
|
+ "INSERT INTO public.rule_run_artifacts "
|
|
|
+ "(id, binding_id, correlation_id, artifact_kind, artifact_ref, "
|
|
|
+ "artifact_digest, row_count, schema_hash, schema_fields, "
|
|
|
+ "binding_hash, handoff_status, expires_at) VALUES "
|
|
|
+ "(CAST(:id AS uuid), CAST(:binding_id AS uuid), "
|
|
|
+ "CAST(:correlation_id AS uuid), 'input', :artifact_ref, "
|
|
|
+ ":artifact_digest, :row_count, :schema_hash, "
|
|
|
+ "CAST(:schema_fields AS jsonb), :binding_hash, 'ready', "
|
|
|
+ "CURRENT_TIMESTAMP + INTERVAL '30 minutes')"
|
|
|
+ ),
|
|
|
+ {
|
|
|
+ "id": new_governance_uid(),
|
|
|
+ "binding_id": binding["id"],
|
|
|
+ "correlation_id": correlation_id,
|
|
|
+ "artifact_ref": artifact["artifact_ref"],
|
|
|
+ "artifact_digest": artifact["digest"],
|
|
|
+ "row_count": artifact["row_count"],
|
|
|
+ "schema_hash": artifact["schema_hash"],
|
|
|
+ "schema_fields": json.dumps(schema_fields),
|
|
|
+ "binding_hash": binding["binding_hash"],
|
|
|
+ },
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def _materialize_physical_release(
|
|
|
+ *,
|
|
|
+ connection,
|
|
|
+ repository,
|
|
|
+ store,
|
|
|
+ released,
|
|
|
+ deployment_id,
|
|
|
+ actor_uid,
|
|
|
+ data_source_uid,
|
|
|
+ input_snapshot,
|
|
|
+ output_snapshot,
|
|
|
+ sample_correlation,
|
|
|
+ sample_artifact,
|
|
|
+ schema_fields,
|
|
|
+):
|
|
|
+ _persist_preparatory_deployment(
|
|
|
+ connection,
|
|
|
+ deployment_id=deployment_id,
|
|
|
+ version_id=released["id"],
|
|
|
+ )
|
|
|
+ bindings = {}
|
|
|
+ suffix = str(released["version_no"])
|
|
|
+ for logical_ref, snapshot, access_mode, write_mode in (
|
|
|
+ (f"raw_customer_v{suffix}", input_snapshot, "read", None),
|
|
|
+ (
|
|
|
+ f"clean_customer_v{suffix}",
|
|
|
+ output_snapshot,
|
|
|
+ "read_write",
|
|
|
+ "append",
|
|
|
+ ),
|
|
|
+ (
|
|
|
+ f"customer_product_v{suffix}",
|
|
|
+ output_snapshot,
|
|
|
+ "write",
|
|
|
+ "append",
|
|
|
+ ),
|
|
|
+ ):
|
|
|
+ binding = {
|
|
|
+ "data_source_uid": data_source_uid,
|
|
|
+ "object_kind": "parquet_artifact",
|
|
|
+ "object_ref": logical_ref,
|
|
|
+ "schema_snapshot_id": snapshot["id"],
|
|
|
+ "access_mode": access_mode,
|
|
|
+ "dialect": "polars",
|
|
|
+ "write_mode": write_mode,
|
|
|
+ }
|
|
|
+ if write_mode is None:
|
|
|
+ binding.pop("write_mode")
|
|
|
+ bindings[logical_ref] = repository.persist_dataset_binding(
|
|
|
+ deployment_id=deployment_id,
|
|
|
+ logical_ref=logical_ref,
|
|
|
+ binding=binding,
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ )
|
|
|
+ ordered = list(bindings.values())
|
|
|
+ for binding in ordered[:2]:
|
|
|
+ _catalog_input_artifact(
|
|
|
+ connection,
|
|
|
+ binding=binding,
|
|
|
+ correlation_id=sample_correlation,
|
|
|
+ artifact=sample_artifact,
|
|
|
+ schema_fields=schema_fields,
|
|
|
+ )
|
|
|
+ components = (
|
|
|
+ connection.execute(
|
|
|
+ text(
|
|
|
+ "SELECT id::text AS id, component_kind, "
|
|
|
+ "rule_version_id::text AS rule_version_id "
|
|
|
+ "FROM public.dataflow_component_bindings "
|
|
|
+ "WHERE dataflow_version_id = CAST(:id AS uuid) "
|
|
|
+ "ORDER BY order_no"
|
|
|
+ ),
|
|
|
+ {"id": released["id"]},
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .all()
|
|
|
+ )
|
|
|
+ compiler = BoundSqlPlanService(
|
|
|
+ repository,
|
|
|
+ CompilerRegistry({"polars": PolarsRuleCompiler()}),
|
|
|
+ )
|
|
|
+ gate = PhysicalPlanPublicationService(
|
|
|
+ repository,
|
|
|
+ test_runner=ServerOwnedPhysicalPreflightRunner(store),
|
|
|
+ )
|
|
|
+ plans = []
|
|
|
+ for component, input_binding, output_binding in (
|
|
|
+ (components[0], ordered[0], ordered[1]),
|
|
|
+ (components[1], ordered[1], ordered[2]),
|
|
|
+ ):
|
|
|
+ plan = compiler.compile_and_persist(
|
|
|
+ component_binding_id=component["id"],
|
|
|
+ rule_version_id=component["rule_version_id"],
|
|
|
+ input_schema_snapshot_id=(
|
|
|
+ input_snapshot["id"]
|
|
|
+ if input_binding is ordered[0]
|
|
|
+ else output_snapshot["id"]
|
|
|
+ ),
|
|
|
+ output_schema_snapshot_id=output_snapshot["id"],
|
|
|
+ input_binding_id=input_binding["id"],
|
|
|
+ output_binding_id=output_binding["id"],
|
|
|
+ )
|
|
|
+ gate.validate(plan["id"], actor_uid)
|
|
|
+ gate.test(plan["id"], actor_uid)
|
|
|
+ plans.append(gate.publish(plan["id"], actor_uid))
|
|
|
+ return {
|
|
|
+ "bindings": {
|
|
|
+ "input": ordered[0],
|
|
|
+ "intermediate": ordered[1],
|
|
|
+ "output": ordered[2],
|
|
|
+ },
|
|
|
+ "components": components,
|
|
|
+ "physical_plans": plans,
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+def _schedule():
|
|
|
+ return {
|
|
|
+ "schema_version": "1.0",
|
|
|
+ "timezone": "Asia/Shanghai",
|
|
|
+ "triggers": [{"type": "manual"}],
|
|
|
+ "max_concurrency": 1,
|
|
|
+ "timeout_seconds": 300,
|
|
|
+ "retry": {"max_attempts": 1, "delay_seconds": 0},
|
|
|
+ "conflict_policy": "queue",
|
|
|
+ "backfill": {"max_days": 1, "max_runs": 1},
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+def _cleanup_chain(connection, ids):
|
|
|
+ """Delete only rows owned by this E2E identity chain, in FK order."""
|
|
|
+
|
|
|
+ params = {
|
|
|
+ **ids,
|
|
|
+ "snapshot_ids": ids["schema_snapshot_ids"],
|
|
|
+ }
|
|
|
+ statements = (
|
|
|
+ "DELETE FROM public.runner_task_executions "
|
|
|
+ "WHERE dataflow_uid = CAST(:dataflow_uid AS uuid)",
|
|
|
+ "UPDATE public.dataflow_deployments SET canary_evidence_id = NULL, "
|
|
|
+ "previous_active_deployment_id = NULL "
|
|
|
+ "WHERE dataflow_uid = CAST(:dataflow_uid AS uuid)",
|
|
|
+ "DELETE FROM public.rule_sql_staging_receipts WHERE deployment_id IN "
|
|
|
+ "(SELECT id FROM public.dataflow_deployments "
|
|
|
+ "WHERE dataflow_uid = CAST(:dataflow_uid AS uuid))",
|
|
|
+ "DELETE FROM public.rule_run_artifacts WHERE binding_id IN "
|
|
|
+ "(SELECT b.id FROM public.dataflow_dataset_bindings b "
|
|
|
+ "JOIN public.dataflow_deployments d "
|
|
|
+ "ON d.id = b.dataflow_deployment_id "
|
|
|
+ "WHERE d.dataflow_uid = CAST(:dataflow_uid AS uuid))",
|
|
|
+ "DELETE FROM public.rule_violation_samples WHERE rule_run_id IN "
|
|
|
+ "(SELECT r.id FROM public.rule_runs r "
|
|
|
+ "JOIN public.dataflow_deployments d ON d.id = r.deployment_id "
|
|
|
+ "WHERE d.dataflow_uid = CAST(:dataflow_uid AS uuid))",
|
|
|
+ "DELETE FROM public.rule_runs WHERE deployment_id IN "
|
|
|
+ "(SELECT id FROM public.dataflow_deployments "
|
|
|
+ "WHERE dataflow_uid = CAST(:dataflow_uid AS uuid))",
|
|
|
+ "DELETE FROM public.dataflow_deployment_operation_leases "
|
|
|
+ "WHERE dataflow_uid = CAST(:dataflow_uid AS uuid)",
|
|
|
+ "DELETE FROM public.dataflow_deployment_operations "
|
|
|
+ "WHERE deployment_id IN "
|
|
|
+ "(SELECT id FROM public.dataflow_deployments "
|
|
|
+ "WHERE dataflow_uid = CAST(:dataflow_uid AS uuid))",
|
|
|
+ "DELETE FROM public.dataflow_deployment_transitions "
|
|
|
+ "WHERE deployment_id IN "
|
|
|
+ "(SELECT id FROM public.dataflow_deployments "
|
|
|
+ "WHERE dataflow_uid = CAST(:dataflow_uid AS uuid))",
|
|
|
+ "DELETE FROM public.dataflow_canary_evidence WHERE deployment_id IN "
|
|
|
+ "(SELECT id FROM public.dataflow_deployments "
|
|
|
+ "WHERE dataflow_uid = CAST(:dataflow_uid AS uuid))",
|
|
|
+ "DELETE FROM public.rule_publication_audits "
|
|
|
+ "WHERE rule_execution_plan_id IN "
|
|
|
+ "(SELECT p.id FROM public.rule_execution_plans p "
|
|
|
+ "JOIN public.dataflow_component_bindings b "
|
|
|
+ "ON b.id = p.component_binding_id "
|
|
|
+ "JOIN public.dataflow_versions v ON v.id = b.dataflow_version_id "
|
|
|
+ "WHERE v.dataflow_uid = CAST(:dataflow_uid AS uuid))",
|
|
|
+ "DELETE FROM public.rule_test_evidence "
|
|
|
+ "WHERE rule_execution_plan_id IN "
|
|
|
+ "(SELECT p.id FROM public.rule_execution_plans p "
|
|
|
+ "JOIN public.dataflow_component_bindings b "
|
|
|
+ "ON b.id = p.component_binding_id "
|
|
|
+ "JOIN public.dataflow_versions v ON v.id = b.dataflow_version_id "
|
|
|
+ "WHERE v.dataflow_uid = CAST(:dataflow_uid AS uuid))",
|
|
|
+ "DELETE FROM public.rule_compile_evidence "
|
|
|
+ "WHERE rule_execution_plan_id IN "
|
|
|
+ "(SELECT p.id FROM public.rule_execution_plans p "
|
|
|
+ "JOIN public.dataflow_component_bindings b "
|
|
|
+ "ON b.id = p.component_binding_id "
|
|
|
+ "JOIN public.dataflow_versions v ON v.id = b.dataflow_version_id "
|
|
|
+ "WHERE v.dataflow_uid = CAST(:dataflow_uid AS uuid))",
|
|
|
+ "DELETE FROM public.rule_execution_plans "
|
|
|
+ "WHERE component_binding_id IN "
|
|
|
+ "(SELECT b.id FROM public.dataflow_component_bindings b "
|
|
|
+ "JOIN public.dataflow_versions v ON v.id = b.dataflow_version_id "
|
|
|
+ "WHERE v.dataflow_uid = CAST(:dataflow_uid AS uuid))",
|
|
|
+ "DELETE FROM public.dataflow_dataset_bindings "
|
|
|
+ "WHERE dataflow_deployment_id IN "
|
|
|
+ "(SELECT id FROM public.dataflow_deployments "
|
|
|
+ "WHERE dataflow_uid = CAST(:dataflow_uid AS uuid))",
|
|
|
+ "DELETE FROM public.dataflow_deployments "
|
|
|
+ "WHERE dataflow_uid = CAST(:dataflow_uid AS uuid)",
|
|
|
+ "DELETE FROM public.dataflow_component_bindings "
|
|
|
+ "WHERE dataflow_version_id IN "
|
|
|
+ "(SELECT id FROM public.dataflow_versions "
|
|
|
+ "WHERE dataflow_uid = CAST(:dataflow_uid AS uuid))",
|
|
|
+ "DELETE FROM public.dataflow_versions "
|
|
|
+ "WHERE dataflow_uid = CAST(:dataflow_uid AS uuid)",
|
|
|
+ "DELETE FROM public.standard_rule_bindings "
|
|
|
+ "WHERE standard_version_id IN "
|
|
|
+ "(SELECT id FROM public.data_standard_versions "
|
|
|
+ "WHERE standard_uid = CAST(:standard_uid AS uuid))",
|
|
|
+ "DELETE FROM public.data_standard_versions "
|
|
|
+ "WHERE standard_uid = CAST(:standard_uid AS uuid)",
|
|
|
+ "DELETE FROM public.data_standards "
|
|
|
+ "WHERE standard_uid = CAST(:standard_uid AS uuid)",
|
|
|
+ "DELETE FROM public.rule_publication_audits "
|
|
|
+ "WHERE rule_version_id IN (SELECT id FROM public.data_rule_versions "
|
|
|
+ "WHERE rule_uid = CAST(:rule_uid AS uuid))",
|
|
|
+ "DELETE FROM public.rule_validation_attempts "
|
|
|
+ "WHERE rule_version_id IN (SELECT id FROM public.data_rule_versions "
|
|
|
+ "WHERE rule_uid = CAST(:rule_uid AS uuid))",
|
|
|
+ "DELETE FROM public.rule_logical_test_evidence "
|
|
|
+ "WHERE logical_plan_id IN (SELECT lp.id "
|
|
|
+ "FROM public.rule_logical_plans lp "
|
|
|
+ "JOIN public.data_rule_versions rv ON rv.id = lp.rule_version_id "
|
|
|
+ "WHERE rv.rule_uid = CAST(:rule_uid AS uuid))",
|
|
|
+ "DELETE FROM public.rule_logical_compile_evidence "
|
|
|
+ "WHERE logical_plan_id IN (SELECT lp.id "
|
|
|
+ "FROM public.rule_logical_plans lp "
|
|
|
+ "JOIN public.data_rule_versions rv ON rv.id = lp.rule_version_id "
|
|
|
+ "WHERE rv.rule_uid = CAST(:rule_uid AS uuid))",
|
|
|
+ "DELETE FROM public.rule_logical_plans "
|
|
|
+ "WHERE rule_version_id IN (SELECT id FROM public.data_rule_versions "
|
|
|
+ "WHERE rule_uid = CAST(:rule_uid AS uuid))",
|
|
|
+ "DELETE FROM public.rule_validation_profiles "
|
|
|
+ "WHERE rule_version_id IN (SELECT id FROM public.data_rule_versions "
|
|
|
+ "WHERE rule_uid = CAST(:rule_uid AS uuid))",
|
|
|
+ "DELETE FROM public.rule_generation_attempts "
|
|
|
+ "WHERE generation_run_id IN (SELECT id "
|
|
|
+ "FROM public.rule_generation_runs "
|
|
|
+ "WHERE created_by = CAST(:actor_uid AS uuid))",
|
|
|
+ "DELETE FROM public.rule_generation_runs "
|
|
|
+ "WHERE created_by = CAST(:actor_uid AS uuid)",
|
|
|
+ "DELETE FROM public.rule_artifacts "
|
|
|
+ "WHERE rule_version_id IN (SELECT id FROM public.data_rule_versions "
|
|
|
+ "WHERE rule_uid = CAST(:rule_uid AS uuid))",
|
|
|
+ "DELETE FROM public.data_rule_versions "
|
|
|
+ "WHERE rule_uid = CAST(:rule_uid AS uuid)",
|
|
|
+ "DELETE FROM public.data_rules WHERE rule_uid = CAST(:rule_uid AS uuid)",
|
|
|
+ "DELETE FROM public.data_schema_snapshots "
|
|
|
+ "WHERE id = ANY(CAST(:snapshot_ids AS uuid[]))",
|
|
|
+ "DELETE FROM public.users WHERE id = CAST(:actor_uid AS uuid)",
|
|
|
+ )
|
|
|
+ for statement in statements:
|
|
|
+ connection.execute(text(statement), params)
|
|
|
+
|
|
|
+
|
|
|
+def _free_port() -> int:
|
|
|
+ with socket.socket() as sock:
|
|
|
+ sock.bind(("127.0.0.1", 0))
|
|
|
+ return int(sock.getsockname()[1])
|
|
|
+
|
|
|
+
|
|
|
+@contextmanager
|
|
|
+def _local_model_server():
|
|
|
+ assert MODEL_PYTHON.is_file(), f"prepared MLX runtime is missing: {MODEL_PYTHON}"
|
|
|
+ assert SERVER.is_file(), f"local model server is missing: {SERVER}"
|
|
|
+ port = _free_port()
|
|
|
+ env = {
|
|
|
+ **os.environ,
|
|
|
+ "HF_HOME": HF_HOME,
|
|
|
+ "PYTHONUNBUFFERED": "1",
|
|
|
+ }
|
|
|
+ process = subprocess.Popen(
|
|
|
+ [
|
|
|
+ str(MODEL_PYTHON),
|
|
|
+ str(SERVER),
|
|
|
+ "--host",
|
|
|
+ "127.0.0.1",
|
|
|
+ "--port",
|
|
|
+ str(port),
|
|
|
+ "--model",
|
|
|
+ MODEL_NAME,
|
|
|
+ "--max-tokens",
|
|
|
+ "2048",
|
|
|
+ ],
|
|
|
+ cwd=ROOT,
|
|
|
+ env=env,
|
|
|
+ stdout=subprocess.PIPE,
|
|
|
+ stderr=subprocess.STDOUT,
|
|
|
+ text=True,
|
|
|
+ )
|
|
|
+ health_url = f"http://127.0.0.1:{port}/health"
|
|
|
+ deadline = time.monotonic() + 180
|
|
|
+ output = []
|
|
|
+ try:
|
|
|
+ while time.monotonic() < deadline:
|
|
|
+ if process.poll() is not None:
|
|
|
+ if process.stdout is not None:
|
|
|
+ output.append(process.stdout.read())
|
|
|
+ raise AssertionError(
|
|
|
+ "local model server exited before readiness:\n"
|
|
|
+ + "".join(output)[-4000:]
|
|
|
+ )
|
|
|
+ try:
|
|
|
+ with urllib.request.urlopen(health_url, timeout=1) as response:
|
|
|
+ payload = json.load(response)
|
|
|
+ if payload == {
|
|
|
+ "status": "ready",
|
|
|
+ "model": MODEL_NAME,
|
|
|
+ "loopback_only": True,
|
|
|
+ }:
|
|
|
+ break
|
|
|
+ except Exception:
|
|
|
+ time.sleep(0.5)
|
|
|
+ else:
|
|
|
+ raise AssertionError("local model server did not become ready")
|
|
|
+ yield f"http://127.0.0.1:{port}/v1"
|
|
|
+ finally:
|
|
|
+ process.terminate()
|
|
|
+ try:
|
|
|
+ process.wait(timeout=10)
|
|
|
+ except subprocess.TimeoutExpired:
|
|
|
+ process.kill()
|
|
|
+ process.wait(timeout=5)
|
|
|
+
|
|
|
+
|
|
|
+@pytest.fixture(scope="module")
|
|
|
+def real_qwen_result():
|
|
|
+ if os.environ.get("RUN_DATA_RULE_REAL_MODEL") != "1":
|
|
|
+ pytest.skip("set RUN_DATA_RULE_REAL_MODEL=1 for the dedicated real-model gate")
|
|
|
+ started_at = time.monotonic()
|
|
|
+ with _local_model_server() as base_url:
|
|
|
+ model = OpenAICompatibleRuleModel(
|
|
|
+ client=OpenAI(
|
|
|
+ api_key="local-validation-only",
|
|
|
+ base_url=base_url,
|
|
|
+ max_retries=0,
|
|
|
+ ),
|
|
|
+ provider="local-qwen-cpu-validation",
|
|
|
+ model_name=MODEL_NAME,
|
|
|
+ max_tokens=2048,
|
|
|
+ )
|
|
|
+ result = RuleAuthoringAgent(
|
|
|
+ model=model,
|
|
|
+ timeout_seconds=240,
|
|
|
+ confidence_threshold=0.5,
|
|
|
+ max_repair_attempts=2,
|
|
|
+ ).interpret(
|
|
|
+ source_text=SOURCE_TEXT,
|
|
|
+ authoring_surface="data_flow",
|
|
|
+ context={
|
|
|
+ "rule_uid": new_governance_uid(),
|
|
|
+ "input_schema_ref": "bd:customer:raw:v1",
|
|
|
+ "output_schema_ref": "bd:customer:clean:v1",
|
|
|
+ "timezone": "Asia/Shanghai",
|
|
|
+ "fields": [
|
|
|
+ {
|
|
|
+ "name": "customer_id",
|
|
|
+ "type": "string",
|
|
|
+ "nullable": False,
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "name": "mobile",
|
|
|
+ "type": "string",
|
|
|
+ "nullable": True,
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "name": "updated_at",
|
|
|
+ "type": "timestamp",
|
|
|
+ "nullable": False,
|
|
|
+ },
|
|
|
+ ],
|
|
|
+ },
|
|
|
+ )
|
|
|
+ return result, round(time.monotonic() - started_at, 3)
|
|
|
+
|
|
|
+
|
|
|
+def test_real_qwen_authoring_produces_closed_three_step_rule_candidate(
|
|
|
+ real_qwen_result,
|
|
|
+):
|
|
|
+ result, elapsed_seconds = real_qwen_result
|
|
|
+ candidate = result["candidate"]
|
|
|
+ steps = candidate["rule_spec"]["steps"]
|
|
|
+ evidence = {
|
|
|
+ "schema_version": "1.0",
|
|
|
+ "model_provider": result["model_provider"],
|
|
|
+ "model_name": result["model_name"],
|
|
|
+ "elapsed_seconds": elapsed_seconds,
|
|
|
+ "repair_attempts": result["repair_attempts"],
|
|
|
+ "confidence": candidate["confidence"],
|
|
|
+ "ambiguities": candidate["ambiguities"],
|
|
|
+ "operations": [step["op"] for step in steps],
|
|
|
+ "model_hash": result["model_hash"],
|
|
|
+ "prompt_hash": result["prompt_hash"],
|
|
|
+ "context_hash": result["context_hash"],
|
|
|
+ "candidate_hash": result["candidate_hash"],
|
|
|
+ }
|
|
|
+ EVIDENCE.parent.mkdir(parents=True, exist_ok=True)
|
|
|
+ EVIDENCE.write_text(
|
|
|
+ json.dumps(
|
|
|
+ evidence,
|
|
|
+ ensure_ascii=False,
|
|
|
+ sort_keys=True,
|
|
|
+ indent=2,
|
|
|
+ )
|
|
|
+ + "\n",
|
|
|
+ encoding="utf-8",
|
|
|
+ )
|
|
|
+
|
|
|
+ assert candidate["ambiguities"] == []
|
|
|
+ assert candidate["confidence"] >= 0.5
|
|
|
+ assert result["status"] == "ready"
|
|
|
+ assert [step["op"] for step in steps] == [
|
|
|
+ "normalize_text",
|
|
|
+ "assert",
|
|
|
+ "deduplicate",
|
|
|
+ ]
|
|
|
+ assert steps[0]["column"] == "mobile"
|
|
|
+ assert steps[0]["trim"] is True
|
|
|
+ assert steps[1]["on_failure"] == "quarantine"
|
|
|
+ assert steps[2]["keys"] == ["customer_id"]
|
|
|
+ assert steps[2]["order_by"] == ["updated_at"]
|
|
|
+ assert steps[2]["keep"] == "last"
|
|
|
+ assert result["model_name"] == MODEL_NAME
|
|
|
+ for field in (
|
|
|
+ "model_hash",
|
|
|
+ "prompt_hash",
|
|
|
+ "context_hash",
|
|
|
+ "candidate_hash",
|
|
|
+ ):
|
|
|
+ assert len(result[field]) == 64
|
|
|
+ print(
|
|
|
+ json.dumps(
|
|
|
+ {
|
|
|
+ "model_provider": result["model_provider"],
|
|
|
+ "model_name": result["model_name"],
|
|
|
+ "elapsed_seconds": elapsed_seconds,
|
|
|
+ "repair_attempts": result["repair_attempts"],
|
|
|
+ "operations": [step["op"] for step in steps],
|
|
|
+ "candidate_hash": result["candidate_hash"],
|
|
|
+ "prompt_hash": result["prompt_hash"],
|
|
|
+ },
|
|
|
+ ensure_ascii=False,
|
|
|
+ sort_keys=True,
|
|
|
+ )
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def test_real_qwen_rule_reaches_published_production_line_and_data_factory(
|
|
|
+ real_qwen_result,
|
|
|
+):
|
|
|
+ """One correlated identity chain from real Qwen output to Data Factory."""
|
|
|
+
|
|
|
+ if os.environ.get("RUN_DATA_RULE_FULL_CHAIN") != "1":
|
|
|
+ pytest.skip(
|
|
|
+ "set RUN_DATA_RULE_FULL_CHAIN=1 for the Docker production-chain gate"
|
|
|
+ )
|
|
|
+ model_result, model_elapsed_seconds = real_qwen_result
|
|
|
+ candidate = copy.deepcopy(model_result["candidate"])
|
|
|
+ rule_spec = candidate["rule_spec"]
|
|
|
+ input_fields = [
|
|
|
+ {"name": "customer_id", "type": "string", "nullable": False},
|
|
|
+ {"name": "mobile", "type": "string", "nullable": True},
|
|
|
+ {
|
|
|
+ "name": "updated_at",
|
|
|
+ "type": "timestamp",
|
|
|
+ "nullable": False,
|
|
|
+ },
|
|
|
+ ]
|
|
|
+ raw_frame = pl.DataFrame(
|
|
|
+ {
|
|
|
+ "customer_id": ["cust-1", "cust-1", "cust-2", "cust-3"],
|
|
|
+ "mobile": [
|
|
|
+ " 13800138000 ",
|
|
|
+ "13800138000",
|
|
|
+ "not-a-mobile",
|
|
|
+ "13900139000",
|
|
|
+ ],
|
|
|
+ "updated_at": [
|
|
|
+ datetime(2026, 7, 24, 8, 0),
|
|
|
+ datetime(2026, 7, 24, 9, 0),
|
|
|
+ datetime(2026, 7, 24, 10, 0),
|
|
|
+ datetime(2026, 7, 24, 11, 0),
|
|
|
+ ],
|
|
|
+ }
|
|
|
+ )
|
|
|
+ store = _artifact_store()
|
|
|
+ platform_engine = create_engine(
|
|
|
+ PLATFORM_DATABASE_URL,
|
|
|
+ pool_pre_ping=True,
|
|
|
+ )
|
|
|
+ actor_uid = new_governance_uid()
|
|
|
+ data_source_uid = new_governance_uid()
|
|
|
+ dataflow_uid = new_governance_uid()
|
|
|
+ deployment_id = new_governance_uid()
|
|
|
+ physical_sample_correlation = new_governance_uid()
|
|
|
+ logical_sample = store.write(
|
|
|
+ raw_frame,
|
|
|
+ new_governance_uid(),
|
|
|
+ 1800,
|
|
|
+ schema_fields=input_fields,
|
|
|
+ )
|
|
|
+ physical_sample = store.write(
|
|
|
+ raw_frame,
|
|
|
+ physical_sample_correlation,
|
|
|
+ 1800,
|
|
|
+ schema_fields=input_fields,
|
|
|
+ )
|
|
|
+ generated_artifact_refs = []
|
|
|
+ kestra_flows = []
|
|
|
+ created_ids = {
|
|
|
+ "actor_uid": actor_uid,
|
|
|
+ "dataflow_uid": dataflow_uid,
|
|
|
+ "rule_uid": rule_spec["rule_uid"],
|
|
|
+ "standard_uid": None,
|
|
|
+ "schema_snapshot_ids": [],
|
|
|
+ }
|
|
|
+ with Session(platform_engine) as connection:
|
|
|
+ try:
|
|
|
+ connection.execute(
|
|
|
+ text(
|
|
|
+ "INSERT INTO public.users "
|
|
|
+ "(id, username, display_name, password_hash, status) "
|
|
|
+ "VALUES (CAST(:id AS uuid), :username, "
|
|
|
+ "'M5 Qwen E2E', 'not-a-login-secret', 'active')"
|
|
|
+ ),
|
|
|
+ {
|
|
|
+ "id": actor_uid,
|
|
|
+ "username": f"m5-qwen-{actor_uid[:8]}",
|
|
|
+ },
|
|
|
+ )
|
|
|
+ repository = DataRuleRepository(connection)
|
|
|
+ snapshots = []
|
|
|
+ owned_snapshot_ids = []
|
|
|
+ for schema_ref in (
|
|
|
+ rule_spec["input_schema_ref"],
|
|
|
+ rule_spec["output_schema_ref"],
|
|
|
+ ):
|
|
|
+ snapshot = {
|
|
|
+ "schema_ref": schema_ref,
|
|
|
+ "schema_hash": canonical_schema_hash(input_fields),
|
|
|
+ "fields": input_fields,
|
|
|
+ "source_revision": "m5-real-qwen:1",
|
|
|
+ }
|
|
|
+ existing_snapshot = repository.find_schema_snapshot(
|
|
|
+ schema_ref=schema_ref,
|
|
|
+ schema_hash=snapshot["schema_hash"],
|
|
|
+ )
|
|
|
+ persisted_snapshot = repository.persist_schema_snapshot(
|
|
|
+ snapshot=snapshot
|
|
|
+ )
|
|
|
+ snapshots.append(persisted_snapshot)
|
|
|
+ if existing_snapshot is None:
|
|
|
+ owned_snapshot_ids.append(persisted_snapshot["id"])
|
|
|
+ input_snapshot, output_snapshot = snapshots
|
|
|
+ created_ids["schema_snapshot_ids"] = owned_snapshot_ids
|
|
|
+ validation_context = {
|
|
|
+ "input_schema_snapshot_id": input_snapshot["id"],
|
|
|
+ "input_schema_hash": input_snapshot["schema_hash"],
|
|
|
+ "input_fields": input_fields,
|
|
|
+ "output_schema_snapshot_id": output_snapshot["id"],
|
|
|
+ "output_schema_hash": output_snapshot["schema_hash"],
|
|
|
+ "output_fields": input_fields,
|
|
|
+ "input_sample_artifact_ref": logical_sample["artifact_ref"],
|
|
|
+ "input_sample_artifact_digest": logical_sample["digest"],
|
|
|
+ "golden_output_artifact_ref": None,
|
|
|
+ "golden_output_artifact_digest": None,
|
|
|
+ }
|
|
|
+ generation_evidence = {
|
|
|
+ **copy.deepcopy(model_result),
|
|
|
+ "source_text": SOURCE_TEXT,
|
|
|
+ "authoring_surface": "data_flow",
|
|
|
+ }
|
|
|
+ generation = repository.record_generation_run(
|
|
|
+ evidence=generation_evidence,
|
|
|
+ created_by=actor_uid,
|
|
|
+ validation_context=validation_context,
|
|
|
+ )
|
|
|
+ signer = GenerationReceiptSigner(
|
|
|
+ "m5-local-qwen-receipt-secret-with-entropy"
|
|
|
+ )
|
|
|
+ receipt_claims = generation_receipt_claims(
|
|
|
+ generation_run_id=generation["id"],
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ source_text=SOURCE_TEXT,
|
|
|
+ candidate_hash=model_result["candidate_hash"],
|
|
|
+ rule_spec=rule_spec,
|
|
|
+ model_hash=model_result["model_hash"],
|
|
|
+ prompt_hash=model_result["prompt_hash"],
|
|
|
+ context_hash=model_result["context_hash"],
|
|
|
+ expires_at=datetime.now(UTC) + timedelta(minutes=5),
|
|
|
+ )
|
|
|
+ rule_service = RulePublicationService(
|
|
|
+ repository,
|
|
|
+ receipt_signer=signer,
|
|
|
+ compiler=LogicalRuleCompiler(),
|
|
|
+ test_runner=ServerOwnedLogicalDryRunRunner(store),
|
|
|
+ )
|
|
|
+ draft = rule_service.create_draft(
|
|
|
+ rule_spec=rule_spec,
|
|
|
+ source_text=SOURCE_TEXT,
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ generation_receipt=signer.issue(receipt_claims),
|
|
|
+ category="data_cleaning",
|
|
|
+ source_language="zh-CN",
|
|
|
+ generated_kind="rulespec",
|
|
|
+ )
|
|
|
+ logical = rule_service.validate(draft["id"], actor_uid)
|
|
|
+ logical_test = rule_service.test(
|
|
|
+ draft["id"],
|
|
|
+ actor_uid,
|
|
|
+ plan_id=logical["plan_id"],
|
|
|
+ )
|
|
|
+ published_rule = rule_service.publish(
|
|
|
+ draft["id"],
|
|
|
+ actor_uid,
|
|
|
+ )
|
|
|
+ assert logical_test["test_evidence"]["counts"]["rows_quarantined"] == 1
|
|
|
+ assert logical_test["test_evidence"]["counts"]["rows_deduplicated"] == 1
|
|
|
+ assert published_rule["status"] == "published"
|
|
|
+
|
|
|
+ standard_spec = {
|
|
|
+ "schema_version": "1.0",
|
|
|
+ "standard_uid": new_governance_uid(),
|
|
|
+ "name": "客户手机号质量标准",
|
|
|
+ "scope": {
|
|
|
+ "object_type": "business_domain",
|
|
|
+ "schema_ref": rule_spec["output_schema_ref"],
|
|
|
+ },
|
|
|
+ "clauses": [
|
|
|
+ {
|
|
|
+ "id": "mobile_format",
|
|
|
+ "description": SOURCE_TEXT,
|
|
|
+ "severity": "error",
|
|
|
+ "rule_version_id": draft["id"],
|
|
|
+ "exception_policy": "quarantine",
|
|
|
+ }
|
|
|
+ ],
|
|
|
+ }
|
|
|
+ created_ids["standard_uid"] = standard_spec["standard_uid"]
|
|
|
+ standard = repository.create_standard_version(
|
|
|
+ standard_spec=standard_spec,
|
|
|
+ source_text=SOURCE_TEXT,
|
|
|
+ created_by=actor_uid,
|
|
|
+ )
|
|
|
+ standard = repository.publish_standard_version(
|
|
|
+ version_id=standard["id"],
|
|
|
+ published_by=actor_uid,
|
|
|
+ )
|
|
|
+ dataflow_spec = {
|
|
|
+ "schema_version": "1.0",
|
|
|
+ "dataflow_uid": dataflow_uid,
|
|
|
+ "name": "客户数据产品生产线",
|
|
|
+ "input_schema_refs": [rule_spec["input_schema_ref"]],
|
|
|
+ "output_schema_ref": rule_spec["output_schema_ref"],
|
|
|
+ "components": [
|
|
|
+ {
|
|
|
+ "id": "ai_customer_cleaning",
|
|
|
+ "type": "rule.apply",
|
|
|
+ "rule_version_id": draft["id"],
|
|
|
+ "stage": "transform",
|
|
|
+ "order": 10,
|
|
|
+ "idempotency": {
|
|
|
+ "strategy": "deduplication_key",
|
|
|
+ "key": "customer_id",
|
|
|
+ },
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "id": "customer_standard_gate",
|
|
|
+ "type": "standard.enforce",
|
|
|
+ "standard_version_id": standard["id"],
|
|
|
+ "stage": "quality_gate",
|
|
|
+ "order": 20,
|
|
|
+ },
|
|
|
+ ],
|
|
|
+ "parameters": {},
|
|
|
+ }
|
|
|
+ released = ProductionLineReleaseService(
|
|
|
+ repository,
|
|
|
+ schema_resolver=_PinnedSchemaResolver(snapshots),
|
|
|
+ ).release(
|
|
|
+ dataflow_uid=dataflow_uid,
|
|
|
+ dataflow_spec=dataflow_spec,
|
|
|
+ source_text="用AI规则和数据标准组装客户数据生产线",
|
|
|
+ created_by=actor_uid,
|
|
|
+ )
|
|
|
+ assert released["status"] == "released"
|
|
|
+ assert released["package"]["standard_version_ids"] == [standard["id"]]
|
|
|
+ assert released["package"]["rule_version_ids"] == [draft["id"]]
|
|
|
+
|
|
|
+ _persist_preparatory_deployment(
|
|
|
+ connection,
|
|
|
+ deployment_id=deployment_id,
|
|
|
+ version_id=released["id"],
|
|
|
+ )
|
|
|
+ binding_specs = (
|
|
|
+ (
|
|
|
+ "raw_customer",
|
|
|
+ input_snapshot,
|
|
|
+ "read",
|
|
|
+ None,
|
|
|
+ ),
|
|
|
+ (
|
|
|
+ "clean_customer",
|
|
|
+ output_snapshot,
|
|
|
+ "read_write",
|
|
|
+ "append",
|
|
|
+ ),
|
|
|
+ (
|
|
|
+ "customer_product",
|
|
|
+ output_snapshot,
|
|
|
+ "write",
|
|
|
+ "append",
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ bindings = {}
|
|
|
+ for logical_ref, snapshot, access_mode, write_mode in binding_specs:
|
|
|
+ binding = {
|
|
|
+ "data_source_uid": data_source_uid,
|
|
|
+ "object_kind": "parquet_artifact",
|
|
|
+ "object_ref": logical_ref,
|
|
|
+ "schema_snapshot_id": snapshot["id"],
|
|
|
+ "access_mode": access_mode,
|
|
|
+ "dialect": "polars",
|
|
|
+ "write_mode": write_mode,
|
|
|
+ }
|
|
|
+ if write_mode is None:
|
|
|
+ binding.pop("write_mode")
|
|
|
+ bindings[logical_ref] = repository.persist_dataset_binding(
|
|
|
+ deployment_id=deployment_id,
|
|
|
+ logical_ref=logical_ref,
|
|
|
+ binding=binding,
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ )
|
|
|
+ for binding in (
|
|
|
+ bindings["raw_customer"],
|
|
|
+ bindings["clean_customer"],
|
|
|
+ ):
|
|
|
+ _catalog_input_artifact(
|
|
|
+ connection,
|
|
|
+ binding=binding,
|
|
|
+ correlation_id=physical_sample_correlation,
|
|
|
+ artifact=physical_sample,
|
|
|
+ schema_fields=input_fields,
|
|
|
+ )
|
|
|
+
|
|
|
+ components = (
|
|
|
+ connection.execute(
|
|
|
+ text(
|
|
|
+ "SELECT id::text AS id, component_kind, "
|
|
|
+ "rule_version_id::text AS rule_version_id "
|
|
|
+ "FROM public.dataflow_component_bindings "
|
|
|
+ "WHERE dataflow_version_id = CAST(:id AS uuid) "
|
|
|
+ "ORDER BY order_no"
|
|
|
+ ),
|
|
|
+ {"id": released["id"]},
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .all()
|
|
|
+ )
|
|
|
+ assert [row["component_kind"] for row in components] == [
|
|
|
+ "rule.apply",
|
|
|
+ "quality.check",
|
|
|
+ ]
|
|
|
+ compiler_service = BoundSqlPlanService(
|
|
|
+ repository,
|
|
|
+ CompilerRegistry({"polars": PolarsRuleCompiler()}),
|
|
|
+ )
|
|
|
+ physical_gate = PhysicalPlanPublicationService(
|
|
|
+ repository,
|
|
|
+ test_runner=ServerOwnedPhysicalPreflightRunner(store),
|
|
|
+ )
|
|
|
+ physical_plans = []
|
|
|
+ plan_bindings = (
|
|
|
+ (
|
|
|
+ components[0],
|
|
|
+ bindings["raw_customer"],
|
|
|
+ bindings["clean_customer"],
|
|
|
+ ),
|
|
|
+ (
|
|
|
+ components[1],
|
|
|
+ bindings["clean_customer"],
|
|
|
+ bindings["customer_product"],
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ for component, input_binding, output_binding in plan_bindings:
|
|
|
+ plan = compiler_service.compile_and_persist(
|
|
|
+ component_binding_id=component["id"],
|
|
|
+ rule_version_id=component["rule_version_id"],
|
|
|
+ input_schema_snapshot_id=(
|
|
|
+ input_snapshot["id"]
|
|
|
+ if input_binding is bindings["raw_customer"]
|
|
|
+ else output_snapshot["id"]
|
|
|
+ ),
|
|
|
+ output_schema_snapshot_id=output_snapshot["id"],
|
|
|
+ input_binding_id=input_binding["id"],
|
|
|
+ output_binding_id=output_binding["id"],
|
|
|
+ )
|
|
|
+ physical_gate.validate(plan["id"], actor_uid)
|
|
|
+ physical_gate.test(plan["id"], actor_uid)
|
|
|
+ published_plan = physical_gate.publish(
|
|
|
+ plan["id"],
|
|
|
+ actor_uid,
|
|
|
+ )
|
|
|
+ physical_plans.append(published_plan)
|
|
|
+ assert len(physical_plans) == 2
|
|
|
+ assert all(item["plan_status"] == "published" for item in physical_plans)
|
|
|
+
|
|
|
+ deployable = repository.load_deployable_release(
|
|
|
+ released["id"],
|
|
|
+ environment="production",
|
|
|
+ )
|
|
|
+ executable_nodes = deployable["physical_workflow_spec"]["nodes"]
|
|
|
+ assert [node["type"] for node in executable_nodes] == [
|
|
|
+ "rule.apply",
|
|
|
+ "quality.check",
|
|
|
+ ]
|
|
|
+ published_backends = (
|
|
|
+ connection.execute(
|
|
|
+ text(
|
|
|
+ "SELECT backend FROM public.rule_execution_plans "
|
|
|
+ "WHERE component_binding_id = "
|
|
|
+ "ANY(CAST(:ids AS uuid[])) ORDER BY created_at"
|
|
|
+ ),
|
|
|
+ {"ids": [row["id"] for row in components]},
|
|
|
+ )
|
|
|
+ .scalars()
|
|
|
+ .all()
|
|
|
+ )
|
|
|
+ assert published_backends == ["polars_batch", "polars_batch"]
|
|
|
+
|
|
|
+ # Exercise the exact published plans through Runner before Kestra.
|
|
|
+ # This proves quality.check selects the safe Polars read path and
|
|
|
+ # consumes only the opaque artifact handed off by rule.apply.
|
|
|
+ shared_engine = _SharedConnectionEngine(connection)
|
|
|
+ artifact_resolver = PostgresArtifactResolver(
|
|
|
+ shared_engine,
|
|
|
+ store,
|
|
|
+ )
|
|
|
+ executor = RulePlanExecutor(
|
|
|
+ PostgresRulePlanRepository(shared_engine),
|
|
|
+ adapters={
|
|
|
+ "polars_batch": PolarsRulePlanAdapter(
|
|
|
+ artifact_store=store,
|
|
|
+ artifact_resolver=artifact_resolver,
|
|
|
+ )
|
|
|
+ },
|
|
|
+ )
|
|
|
+ transform_result = executor.execute(
|
|
|
+ executable_nodes[0],
|
|
|
+ {},
|
|
|
+ write_authorized=True,
|
|
|
+ correlation_id=physical_sample_correlation,
|
|
|
+ )
|
|
|
+ quality_result = executor.execute(
|
|
|
+ executable_nodes[1],
|
|
|
+ {"input_artifact": transform_result["artifact_ref"]},
|
|
|
+ write_authorized=False,
|
|
|
+ correlation_id=physical_sample_correlation,
|
|
|
+ )
|
|
|
+ assert transform_result["rows_in"] == 4
|
|
|
+ assert transform_result["rows_out"] == 2
|
|
|
+ assert transform_result["rows_quarantined"] == 1
|
|
|
+ assert transform_result["rows_deduplicated"] == 1
|
|
|
+ assert quality_result["rows_in"] == 2
|
|
|
+ assert quality_result["rows_out"] == 2
|
|
|
+ assert quality_result["rows_quarantined"] == 0
|
|
|
+ assert quality_result["commit_outcome"] == "not_applicable"
|
|
|
+ assert "artifact_ref" not in quality_result
|
|
|
+ generated_artifact_refs.append(
|
|
|
+ transform_result["artifact_ref"]
|
|
|
+ )
|
|
|
+
|
|
|
+ # Kestra and Runner use independent connections. Commit the exact
|
|
|
+ # immutable control-plane state before exposing the deployment.
|
|
|
+ connection.commit()
|
|
|
+
|
|
|
+ task_token_issuer = TaskTokenIssuer(
|
|
|
+ os.environ.get(
|
|
|
+ "DATA_RULE_RUNNER_TASK_TOKEN_SECRET",
|
|
|
+ "dataops-local-runner-task-token-secret-change-me",
|
|
|
+ ),
|
|
|
+ ttl_seconds=300,
|
|
|
+ )
|
|
|
+ factory = DataFlowDeploymentService(
|
|
|
+ repository,
|
|
|
+ engine=KestraAdapter(
|
|
|
+ os.environ.get(
|
|
|
+ "DATA_RULE_KESTRA_URL",
|
|
|
+ "http://127.0.0.1:18080/api/v1",
|
|
|
+ ),
|
|
|
+ os.environ.get("DATA_RULE_KESTRA_USER", "admin@dataops.local"),
|
|
|
+ os.environ.get("DATA_RULE_KESTRA_PASSWORD", "DataOpsKestra1!"),
|
|
|
+ timeout=30,
|
|
|
+ ),
|
|
|
+ token_issuer=task_token_issuer,
|
|
|
+ canary_timeout_seconds=300,
|
|
|
+ poll_interval_seconds=0.5,
|
|
|
+ )
|
|
|
+ deployment = factory.create(
|
|
|
+ released["id"],
|
|
|
+ binding_snapshot={
|
|
|
+ "input": {"binding_id": bindings["raw_customer"]["id"]},
|
|
|
+ "output": {"binding_id": bindings["customer_product"]["id"]},
|
|
|
+ },
|
|
|
+ environment="production",
|
|
|
+ schedule_plan=_schedule(),
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ reason="M5 real Qwen production-line acceptance",
|
|
|
+ idempotency_key=f"m5-create-{released['id']}",
|
|
|
+ )
|
|
|
+ connection.commit()
|
|
|
+ disabled = factory.deploy_disabled(
|
|
|
+ deployment["id"],
|
|
|
+ actor_uid,
|
|
|
+ idempotency_key=f"m5-deploy-{released['id']}",
|
|
|
+ )
|
|
|
+ assert disabled["status"] == "disabled"
|
|
|
+ assert disabled["id"] == deployment_id
|
|
|
+ assert disabled["physical_plan_hashes"] == [
|
|
|
+ item["plan_hash"] for item in physical_plans
|
|
|
+ ]
|
|
|
+ kestra_flows.append(
|
|
|
+ (
|
|
|
+ disabled["engine_namespace"],
|
|
|
+ disabled["engine_definition_id"],
|
|
|
+ )
|
|
|
+ )
|
|
|
+
|
|
|
+ canary_correlation = new_governance_uid()
|
|
|
+ canary_input = store.write(
|
|
|
+ raw_frame,
|
|
|
+ canary_correlation,
|
|
|
+ 1800,
|
|
|
+ schema_fields=input_fields,
|
|
|
+ )
|
|
|
+ generated_artifact_refs.append(canary_input["artifact_ref"])
|
|
|
+ with platform_engine.begin() as artifact_catalog:
|
|
|
+ _catalog_input_artifact(
|
|
|
+ artifact_catalog,
|
|
|
+ binding=bindings["raw_customer"],
|
|
|
+ correlation_id=canary_correlation,
|
|
|
+ artifact=canary_input,
|
|
|
+ schema_fields=input_fields,
|
|
|
+ )
|
|
|
+ canary = factory.run_canary(
|
|
|
+ disabled["id"],
|
|
|
+ {},
|
|
|
+ actor_uid,
|
|
|
+ idempotency_key=f"m5-canary-{released['id']}",
|
|
|
+ correlation_id=canary_correlation,
|
|
|
+ )
|
|
|
+ assert canary["status"] == "passed"
|
|
|
+ connection.commit()
|
|
|
+ output_row = (
|
|
|
+ connection.execute(
|
|
|
+ text(
|
|
|
+ "SELECT artifact_ref, artifact_digest, row_count, "
|
|
|
+ "schema_fields FROM public.rule_run_artifacts "
|
|
|
+ "WHERE binding_id = CAST(:binding_id AS uuid) "
|
|
|
+ "AND correlation_id = CAST(:correlation_id AS uuid) "
|
|
|
+ "AND artifact_kind = 'output' "
|
|
|
+ "AND handoff_status = 'ready' "
|
|
|
+ "ORDER BY created_at DESC LIMIT 1"
|
|
|
+ ),
|
|
|
+ {
|
|
|
+ "binding_id": bindings["customer_product"]["id"],
|
|
|
+ "correlation_id": canary_correlation,
|
|
|
+ },
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one()
|
|
|
+ )
|
|
|
+ generated_artifact_refs.append(str(output_row["artifact_ref"]))
|
|
|
+ product = store.read(
|
|
|
+ str(output_row["artifact_ref"]),
|
|
|
+ str(output_row["artifact_digest"]),
|
|
|
+ expected_schema_fields=output_row["schema_fields"],
|
|
|
+ ).collect()
|
|
|
+ assert product.select("customer_id", "mobile").sort(
|
|
|
+ "customer_id"
|
|
|
+ ).to_dicts() == [
|
|
|
+ {"customer_id": "cust-1", "mobile": "13800138000"},
|
|
|
+ {"customer_id": "cust-3", "mobile": "13900139000"},
|
|
|
+ ]
|
|
|
+ run_rows = (
|
|
|
+ connection.execute(
|
|
|
+ text(
|
|
|
+ "SELECT component_binding_id::text, rows_in, rows_out, "
|
|
|
+ "rows_quarantined, status FROM public.rule_runs "
|
|
|
+ "WHERE deployment_id = CAST(:deployment_id AS uuid) "
|
|
|
+ "AND correlation_id = CAST(:correlation_id AS uuid) "
|
|
|
+ "ORDER BY created_at"
|
|
|
+ ),
|
|
|
+ {
|
|
|
+ "deployment_id": disabled["id"],
|
|
|
+ "correlation_id": canary_correlation,
|
|
|
+ },
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .all()
|
|
|
+ )
|
|
|
+ assert [row["status"] for row in run_rows] == [
|
|
|
+ "success",
|
|
|
+ "success",
|
|
|
+ ]
|
|
|
+ assert [
|
|
|
+ (
|
|
|
+ int(row["rows_in"]),
|
|
|
+ int(row["rows_out"]),
|
|
|
+ int(row["rows_quarantined"]),
|
|
|
+ )
|
|
|
+ for row in run_rows
|
|
|
+ ] == [(4, 2, 1), (2, 2, 0)]
|
|
|
+ violation_sample = (
|
|
|
+ connection.execute(
|
|
|
+ text(
|
|
|
+ "SELECT s.artifact_ref, s.sample_count, "
|
|
|
+ "s.redaction_policy, s.expires_at "
|
|
|
+ "FROM public.rule_violation_samples s "
|
|
|
+ "JOIN public.rule_runs r ON r.id = s.rule_run_id "
|
|
|
+ "WHERE r.deployment_id = "
|
|
|
+ "CAST(:deployment_id AS uuid) "
|
|
|
+ "AND r.correlation_id = "
|
|
|
+ "CAST(:correlation_id AS uuid) "
|
|
|
+ "ORDER BY s.created_at DESC LIMIT 1"
|
|
|
+ ),
|
|
|
+ {
|
|
|
+ "deployment_id": disabled["id"],
|
|
|
+ "correlation_id": canary_correlation,
|
|
|
+ },
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one()
|
|
|
+ )
|
|
|
+ violation_artifact = store.describe(
|
|
|
+ str(violation_sample["artifact_ref"])
|
|
|
+ )
|
|
|
+ generated_artifact_refs.append(
|
|
|
+ str(violation_sample["artifact_ref"])
|
|
|
+ )
|
|
|
+ assert int(violation_sample["sample_count"]) == 1
|
|
|
+ assert (
|
|
|
+ violation_sample["redaction_policy"]
|
|
|
+ == "rule-violation-default-v1"
|
|
|
+ )
|
|
|
+ assert violation_sample["expires_at"] > datetime.now(UTC)
|
|
|
+ assert len(violation_artifact["digest"]) == 64
|
|
|
+ active = factory.activate(
|
|
|
+ disabled["id"],
|
|
|
+ canary["id"],
|
|
|
+ actor_uid,
|
|
|
+ idempotency_key=f"m5-activate-{released['id']}",
|
|
|
+ )
|
|
|
+ assert active["status"] == "active"
|
|
|
+ connection.commit()
|
|
|
+
|
|
|
+ manual_correlation = new_governance_uid()
|
|
|
+ manual_input = store.write(
|
|
|
+ raw_frame,
|
|
|
+ manual_correlation,
|
|
|
+ 1800,
|
|
|
+ schema_fields=input_fields,
|
|
|
+ )
|
|
|
+ generated_artifact_refs.append(manual_input["artifact_ref"])
|
|
|
+ with platform_engine.begin() as artifact_catalog:
|
|
|
+ _catalog_input_artifact(
|
|
|
+ artifact_catalog,
|
|
|
+ binding=bindings["raw_customer"],
|
|
|
+ correlation_id=manual_correlation,
|
|
|
+ artifact=manual_input,
|
|
|
+ schema_fields=input_fields,
|
|
|
+ )
|
|
|
+ manual_response = factory.execute_active(
|
|
|
+ active["id"],
|
|
|
+ {},
|
|
|
+ actor_uid,
|
|
|
+ idempotency_key=f"m5-execute-{released['id']}",
|
|
|
+ correlation_id=manual_correlation,
|
|
|
+ )
|
|
|
+ assert manual_response["status"] == "success"
|
|
|
+ assert manual_response["correlation_id"] == manual_correlation
|
|
|
+ manual_output = (
|
|
|
+ connection.execute(
|
|
|
+ text(
|
|
|
+ "SELECT artifact_ref, artifact_digest, row_count "
|
|
|
+ "FROM public.rule_run_artifacts "
|
|
|
+ "WHERE binding_id = CAST(:binding_id AS uuid) "
|
|
|
+ "AND correlation_id = CAST(:correlation_id AS uuid) "
|
|
|
+ "AND artifact_kind = 'output' "
|
|
|
+ "AND handoff_status = 'ready' "
|
|
|
+ "ORDER BY created_at DESC LIMIT 1"
|
|
|
+ ),
|
|
|
+ {
|
|
|
+ "binding_id": bindings["customer_product"]["id"],
|
|
|
+ "correlation_id": manual_correlation,
|
|
|
+ },
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one()
|
|
|
+ )
|
|
|
+ generated_artifact_refs.append(str(manual_output["artifact_ref"]))
|
|
|
+ assert int(manual_output["row_count"]) == 2
|
|
|
+
|
|
|
+ released_v2 = ProductionLineReleaseService(
|
|
|
+ repository,
|
|
|
+ schema_resolver=_PinnedSchemaResolver(snapshots),
|
|
|
+ ).release(
|
|
|
+ dataflow_uid=dataflow_uid,
|
|
|
+ dataflow_spec=dataflow_spec,
|
|
|
+ source_text="客户数据生产线第二次不可变发布",
|
|
|
+ created_by=actor_uid,
|
|
|
+ )
|
|
|
+ assert released_v2["version_no"] == released["version_no"] + 1
|
|
|
+ deployment_v2_id = new_governance_uid()
|
|
|
+ physical_v2 = _materialize_physical_release(
|
|
|
+ connection=connection,
|
|
|
+ repository=repository,
|
|
|
+ store=store,
|
|
|
+ released=released_v2,
|
|
|
+ deployment_id=deployment_v2_id,
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ data_source_uid=data_source_uid,
|
|
|
+ input_snapshot=input_snapshot,
|
|
|
+ output_snapshot=output_snapshot,
|
|
|
+ sample_correlation=physical_sample_correlation,
|
|
|
+ sample_artifact=physical_sample,
|
|
|
+ schema_fields=input_fields,
|
|
|
+ )
|
|
|
+ connection.commit()
|
|
|
+ deployment_v2 = factory.create(
|
|
|
+ released_v2["id"],
|
|
|
+ binding_snapshot={
|
|
|
+ "input": {"binding_id": physical_v2["bindings"]["input"]["id"]},
|
|
|
+ "output": {"binding_id": physical_v2["bindings"]["output"]["id"]},
|
|
|
+ },
|
|
|
+ environment="production",
|
|
|
+ schedule_plan=_schedule(),
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ reason="M5 second immutable production-line release",
|
|
|
+ idempotency_key=f"m5-create-{released_v2['id']}",
|
|
|
+ )
|
|
|
+ connection.commit()
|
|
|
+ disabled_v2 = factory.deploy_disabled(
|
|
|
+ deployment_v2["id"],
|
|
|
+ actor_uid,
|
|
|
+ idempotency_key=f"m5-deploy-{released_v2['id']}",
|
|
|
+ )
|
|
|
+ kestra_flows.append(
|
|
|
+ (
|
|
|
+ disabled_v2["engine_namespace"],
|
|
|
+ disabled_v2["engine_definition_id"],
|
|
|
+ )
|
|
|
+ )
|
|
|
+ canary_v2_correlation = new_governance_uid()
|
|
|
+ canary_v2_input = store.write(
|
|
|
+ raw_frame,
|
|
|
+ canary_v2_correlation,
|
|
|
+ 1800,
|
|
|
+ schema_fields=input_fields,
|
|
|
+ )
|
|
|
+ generated_artifact_refs.append(canary_v2_input["artifact_ref"])
|
|
|
+ with platform_engine.begin() as artifact_catalog:
|
|
|
+ _catalog_input_artifact(
|
|
|
+ artifact_catalog,
|
|
|
+ binding=physical_v2["bindings"]["input"],
|
|
|
+ correlation_id=canary_v2_correlation,
|
|
|
+ artifact=canary_v2_input,
|
|
|
+ schema_fields=input_fields,
|
|
|
+ )
|
|
|
+ canary_v2 = factory.run_canary(
|
|
|
+ disabled_v2["id"],
|
|
|
+ {},
|
|
|
+ actor_uid,
|
|
|
+ idempotency_key=f"m5-canary-{released_v2['id']}",
|
|
|
+ correlation_id=canary_v2_correlation,
|
|
|
+ )
|
|
|
+ assert canary_v2["status"] == "passed"
|
|
|
+ connection.commit()
|
|
|
+ v2_output = (
|
|
|
+ connection.execute(
|
|
|
+ text(
|
|
|
+ "SELECT artifact_ref, artifact_digest, row_count "
|
|
|
+ "FROM public.rule_run_artifacts "
|
|
|
+ "WHERE binding_id = CAST(:binding_id AS uuid) "
|
|
|
+ "AND correlation_id = CAST(:correlation_id AS uuid) "
|
|
|
+ "AND artifact_kind = 'output' "
|
|
|
+ "AND handoff_status = 'ready' "
|
|
|
+ "ORDER BY created_at DESC LIMIT 1"
|
|
|
+ ),
|
|
|
+ {
|
|
|
+ "binding_id": physical_v2["bindings"]["output"]["id"],
|
|
|
+ "correlation_id": canary_v2_correlation,
|
|
|
+ },
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one()
|
|
|
+ )
|
|
|
+ generated_artifact_refs.append(str(v2_output["artifact_ref"]))
|
|
|
+ assert int(v2_output["row_count"]) == 2
|
|
|
+ v2_runs = (
|
|
|
+ connection.execute(
|
|
|
+ text(
|
|
|
+ "SELECT rows_in, rows_out, rows_quarantined, status "
|
|
|
+ "FROM public.rule_runs "
|
|
|
+ "WHERE deployment_id = CAST(:deployment_id AS uuid) "
|
|
|
+ "AND correlation_id = CAST(:correlation_id AS uuid) "
|
|
|
+ "ORDER BY created_at"
|
|
|
+ ),
|
|
|
+ {
|
|
|
+ "deployment_id": disabled_v2["id"],
|
|
|
+ "correlation_id": canary_v2_correlation,
|
|
|
+ },
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .all()
|
|
|
+ )
|
|
|
+ assert [
|
|
|
+ (
|
|
|
+ row["status"],
|
|
|
+ int(row["rows_in"]),
|
|
|
+ int(row["rows_out"]),
|
|
|
+ int(row["rows_quarantined"]),
|
|
|
+ )
|
|
|
+ for row in v2_runs
|
|
|
+ ] == [
|
|
|
+ ("success", 4, 2, 1),
|
|
|
+ ("success", 2, 2, 0),
|
|
|
+ ]
|
|
|
+ active_v2 = factory.activate(
|
|
|
+ disabled_v2["id"],
|
|
|
+ canary_v2["id"],
|
|
|
+ actor_uid,
|
|
|
+ idempotency_key=f"m5-activate-{released_v2['id']}",
|
|
|
+ )
|
|
|
+ assert active_v2["status"] == "active"
|
|
|
+ assert active_v2["previous_active_deployment_id"] == active["id"]
|
|
|
+ assert repository.get_deployment(active["id"])["status"] == "superseded"
|
|
|
+ connection.commit()
|
|
|
+
|
|
|
+ execute_v2_correlation = new_governance_uid()
|
|
|
+ execute_v2_input = store.write(
|
|
|
+ raw_frame,
|
|
|
+ execute_v2_correlation,
|
|
|
+ 1800,
|
|
|
+ schema_fields=input_fields,
|
|
|
+ )
|
|
|
+ generated_artifact_refs.append(execute_v2_input["artifact_ref"])
|
|
|
+ with platform_engine.begin() as artifact_catalog:
|
|
|
+ _catalog_input_artifact(
|
|
|
+ artifact_catalog,
|
|
|
+ binding=physical_v2["bindings"]["input"],
|
|
|
+ correlation_id=execute_v2_correlation,
|
|
|
+ artifact=execute_v2_input,
|
|
|
+ schema_fields=input_fields,
|
|
|
+ )
|
|
|
+ executed_v2 = factory.execute_active(
|
|
|
+ active_v2["id"],
|
|
|
+ {},
|
|
|
+ actor_uid,
|
|
|
+ idempotency_key=f"m5-execute-{released_v2['id']}",
|
|
|
+ correlation_id=execute_v2_correlation,
|
|
|
+ )
|
|
|
+ assert executed_v2["status"] == "success"
|
|
|
+ v2_plan_hashes = (
|
|
|
+ connection.execute(
|
|
|
+ text(
|
|
|
+ "SELECT DISTINCT plan_hash FROM public.rule_runs "
|
|
|
+ "WHERE deployment_id = CAST(:deployment_id AS uuid) "
|
|
|
+ "AND correlation_id = CAST(:correlation_id AS uuid) "
|
|
|
+ "ORDER BY plan_hash"
|
|
|
+ ),
|
|
|
+ {
|
|
|
+ "deployment_id": active_v2["id"],
|
|
|
+ "correlation_id": execute_v2_correlation,
|
|
|
+ },
|
|
|
+ )
|
|
|
+ .scalars()
|
|
|
+ .all()
|
|
|
+ )
|
|
|
+ assert v2_plan_hashes == sorted(
|
|
|
+ item["plan_hash"] for item in physical_v2["physical_plans"]
|
|
|
+ )
|
|
|
+
|
|
|
+ rollback = factory.rollback(
|
|
|
+ active_v2["id"],
|
|
|
+ actor_uid,
|
|
|
+ idempotency_key=f"m5-rollback-{released_v2['id']}",
|
|
|
+ )
|
|
|
+ assert rollback["rolled_back"]["status"] == "rolled_back"
|
|
|
+ assert rollback["active"]["id"] == active["id"]
|
|
|
+ assert rollback["active"]["status"] == "active"
|
|
|
+ assert (
|
|
|
+ factory.engine.get_flow(
|
|
|
+ active["engine_namespace"],
|
|
|
+ active["engine_definition_id"],
|
|
|
+ )["disabled"]
|
|
|
+ is False
|
|
|
+ )
|
|
|
+ assert (
|
|
|
+ factory.engine.get_flow(
|
|
|
+ active_v2["engine_namespace"],
|
|
|
+ active_v2["engine_definition_id"],
|
|
|
+ )["disabled"]
|
|
|
+ is True
|
|
|
+ )
|
|
|
+ connection.commit()
|
|
|
+
|
|
|
+ restored_correlation = new_governance_uid()
|
|
|
+ restored_input = store.write(
|
|
|
+ raw_frame,
|
|
|
+ restored_correlation,
|
|
|
+ 1800,
|
|
|
+ schema_fields=input_fields,
|
|
|
+ )
|
|
|
+ generated_artifact_refs.append(restored_input["artifact_ref"])
|
|
|
+ with platform_engine.begin() as artifact_catalog:
|
|
|
+ _catalog_input_artifact(
|
|
|
+ artifact_catalog,
|
|
|
+ binding=bindings["raw_customer"],
|
|
|
+ correlation_id=restored_correlation,
|
|
|
+ artifact=restored_input,
|
|
|
+ schema_fields=input_fields,
|
|
|
+ )
|
|
|
+ restored_execution = factory.execute_active(
|
|
|
+ rollback["active"]["id"],
|
|
|
+ {},
|
|
|
+ actor_uid,
|
|
|
+ idempotency_key=f"m5-execute-restored-{released['id']}",
|
|
|
+ correlation_id=restored_correlation,
|
|
|
+ )
|
|
|
+ assert restored_execution["status"] == "success"
|
|
|
+ restored_plan_hashes = (
|
|
|
+ connection.execute(
|
|
|
+ text(
|
|
|
+ "SELECT DISTINCT plan_hash FROM public.rule_runs "
|
|
|
+ "WHERE deployment_id = CAST(:deployment_id AS uuid) "
|
|
|
+ "AND correlation_id = CAST(:correlation_id AS uuid) "
|
|
|
+ "ORDER BY plan_hash"
|
|
|
+ ),
|
|
|
+ {
|
|
|
+ "deployment_id": rollback["active"]["id"],
|
|
|
+ "correlation_id": restored_correlation,
|
|
|
+ },
|
|
|
+ )
|
|
|
+ .scalars()
|
|
|
+ .all()
|
|
|
+ )
|
|
|
+ assert restored_plan_hashes == sorted(
|
|
|
+ item["plan_hash"] for item in physical_plans
|
|
|
+ )
|
|
|
+
|
|
|
+ chain_evidence = {
|
|
|
+ "schema_version": "1.0",
|
|
|
+ "model": {
|
|
|
+ "provider": model_result["model_provider"],
|
|
|
+ "name": model_result["model_name"],
|
|
|
+ "elapsed_seconds": model_elapsed_seconds,
|
|
|
+ "candidate_hash": model_result["candidate_hash"],
|
|
|
+ },
|
|
|
+ "generation_run_id": generation["id"],
|
|
|
+ "rule_version_id": draft["id"],
|
|
|
+ "logical_plan_hash": logical["plan_hash"],
|
|
|
+ "logical_test_evidence_id": logical_test["test_evidence_id"],
|
|
|
+ "standard_version_id": standard["id"],
|
|
|
+ "dataflow_uid": dataflow_uid,
|
|
|
+ "dataflow_version_id": released["id"],
|
|
|
+ "package_hash": released["package"]["package_hash"],
|
|
|
+ "physical_plan_hashes": [item["plan_hash"] for item in physical_plans],
|
|
|
+ "runner_preflight": {
|
|
|
+ "correlation_id": physical_sample_correlation,
|
|
|
+ "transform": {
|
|
|
+ "rows_in": transform_result["rows_in"],
|
|
|
+ "rows_out": transform_result["rows_out"],
|
|
|
+ "rows_quarantined": transform_result["rows_quarantined"],
|
|
|
+ "rows_deduplicated": transform_result["rows_deduplicated"],
|
|
|
+ },
|
|
|
+ "quality": {
|
|
|
+ "rows_in": quality_result["rows_in"],
|
|
|
+ "rows_out": quality_result["rows_out"],
|
|
|
+ "rows_quarantined": quality_result["rows_quarantined"],
|
|
|
+ },
|
|
|
+ },
|
|
|
+ "deployment_id": disabled["id"],
|
|
|
+ "deployment_status": active["status"],
|
|
|
+ "engine_definition_id": disabled["engine_definition_id"],
|
|
|
+ "correlation_id": disabled["correlation_id"],
|
|
|
+ "canary": {
|
|
|
+ "correlation_id": canary_correlation,
|
|
|
+ "evidence_id": canary["id"],
|
|
|
+ "execution_id": canary["execution_id"],
|
|
|
+ "status": canary["status"],
|
|
|
+ "product_artifact_digest": str(output_row["artifact_digest"]),
|
|
|
+ "product_row_count": int(output_row["row_count"]),
|
|
|
+ "run_counts": [
|
|
|
+ {
|
|
|
+ "component_binding_id": row["component_binding_id"],
|
|
|
+ "rows_in": int(row["rows_in"]),
|
|
|
+ "rows_out": int(row["rows_out"]),
|
|
|
+ "rows_quarantined": int(row["rows_quarantined"]),
|
|
|
+ }
|
|
|
+ for row in run_rows
|
|
|
+ ],
|
|
|
+ "violation_evidence": {
|
|
|
+ "artifact_digest": violation_artifact["digest"],
|
|
|
+ "sample_count": int(
|
|
|
+ violation_sample["sample_count"]
|
|
|
+ ),
|
|
|
+ "redaction_policy": violation_sample[
|
|
|
+ "redaction_policy"
|
|
|
+ ],
|
|
|
+ "expires_at": violation_sample[
|
|
|
+ "expires_at"
|
|
|
+ ].isoformat(),
|
|
|
+ },
|
|
|
+ },
|
|
|
+ "manual_execution": {
|
|
|
+ "correlation_id": manual_correlation,
|
|
|
+ "execution_id": manual_response["execution_id"],
|
|
|
+ "state": manual_response["status"],
|
|
|
+ "product_artifact_digest": str(manual_output["artifact_digest"]),
|
|
|
+ "product_row_count": int(manual_output["row_count"]),
|
|
|
+ },
|
|
|
+ "version_transition": {
|
|
|
+ "v1": {
|
|
|
+ "dataflow_version_id": released["id"],
|
|
|
+ "deployment_id": active["id"],
|
|
|
+ "status_after_rollback": rollback["active"]["status"],
|
|
|
+ },
|
|
|
+ "v2": {
|
|
|
+ "dataflow_version_id": released_v2["id"],
|
|
|
+ "deployment_id": active_v2["id"],
|
|
|
+ "canary_correlation_id": canary_v2_correlation,
|
|
|
+ "canary_evidence_id": canary_v2["id"],
|
|
|
+ "product_artifact_digest": str(v2_output["artifact_digest"]),
|
|
|
+ "product_row_count": int(v2_output["row_count"]),
|
|
|
+ "status_after_rollback": rollback["rolled_back"]["status"],
|
|
|
+ "physical_plan_hashes": [
|
|
|
+ item["plan_hash"] for item in physical_v2["physical_plans"]
|
|
|
+ ],
|
|
|
+ "production_execution": {
|
|
|
+ "correlation_id": execute_v2_correlation,
|
|
|
+ "execution_id": executed_v2["execution_id"],
|
|
|
+ "plan_hashes": v2_plan_hashes,
|
|
|
+ },
|
|
|
+ },
|
|
|
+ "restored_deployment_id": rollback["active"]["id"],
|
|
|
+ "restored_execution": {
|
|
|
+ "correlation_id": restored_correlation,
|
|
|
+ "execution_id": restored_execution["execution_id"],
|
|
|
+ "plan_hashes": restored_plan_hashes,
|
|
|
+ },
|
|
|
+ },
|
|
|
+ }
|
|
|
+ FULL_CHAIN_EVIDENCE.parent.mkdir(parents=True, exist_ok=True)
|
|
|
+ FULL_CHAIN_EVIDENCE.write_text(
|
|
|
+ json.dumps(
|
|
|
+ chain_evidence,
|
|
|
+ ensure_ascii=False,
|
|
|
+ indent=2,
|
|
|
+ sort_keys=True,
|
|
|
+ )
|
|
|
+ + "\n",
|
|
|
+ encoding="utf-8",
|
|
|
+ )
|
|
|
+ finally:
|
|
|
+ connection.rollback()
|
|
|
+ for namespace, flow_id in kestra_flows:
|
|
|
+ with suppress(requests.RequestException):
|
|
|
+ requests.delete(
|
|
|
+ os.environ.get(
|
|
|
+ "DATA_RULE_KESTRA_URL",
|
|
|
+ "http://127.0.0.1:18080/api/v1",
|
|
|
+ ).rstrip("/")
|
|
|
+ + f"/main/flows/{namespace}/{flow_id}",
|
|
|
+ auth=(
|
|
|
+ os.environ.get(
|
|
|
+ "DATA_RULE_KESTRA_USER",
|
|
|
+ "admin@dataops.local",
|
|
|
+ ),
|
|
|
+ os.environ.get(
|
|
|
+ "DATA_RULE_KESTRA_PASSWORD",
|
|
|
+ "DataOpsKestra1!",
|
|
|
+ ),
|
|
|
+ ),
|
|
|
+ timeout=30,
|
|
|
+ ).raise_for_status()
|
|
|
+ with platform_engine.connect() as catalog:
|
|
|
+ generated_artifact_refs.extend(
|
|
|
+ str(value)
|
|
|
+ for value in catalog.execute(
|
|
|
+ text(
|
|
|
+ "SELECT DISTINCT a.artifact_ref "
|
|
|
+ "FROM public.rule_run_artifacts a "
|
|
|
+ "JOIN public.dataflow_dataset_bindings b "
|
|
|
+ "ON b.id = a.binding_id "
|
|
|
+ "JOIN public.dataflow_deployments d "
|
|
|
+ "ON d.id = b.dataflow_deployment_id "
|
|
|
+ "WHERE d.dataflow_uid = "
|
|
|
+ "CAST(:dataflow_uid AS uuid)"
|
|
|
+ ),
|
|
|
+ {"dataflow_uid": dataflow_uid},
|
|
|
+ ).scalars()
|
|
|
+ )
|
|
|
+ generated_artifact_refs.extend(
|
|
|
+ str(value)
|
|
|
+ for value in catalog.execute(
|
|
|
+ text(
|
|
|
+ "SELECT DISTINCT s.artifact_ref "
|
|
|
+ "FROM public.rule_violation_samples s "
|
|
|
+ "JOIN public.rule_runs r ON r.id = s.rule_run_id "
|
|
|
+ "JOIN public.dataflow_deployments d "
|
|
|
+ "ON d.id = r.deployment_id "
|
|
|
+ "WHERE d.dataflow_uid = "
|
|
|
+ "CAST(:dataflow_uid AS uuid)"
|
|
|
+ ),
|
|
|
+ {"dataflow_uid": dataflow_uid},
|
|
|
+ ).scalars()
|
|
|
+ )
|
|
|
+ with platform_engine.begin() as cleanup:
|
|
|
+ _cleanup_chain(cleanup, created_ids)
|
|
|
+ platform_engine.dispose()
|
|
|
+ store.delete(logical_sample["artifact_ref"])
|
|
|
+ store.delete(physical_sample["artifact_ref"])
|
|
|
+ for artifact_ref in set(generated_artifact_refs):
|
|
|
+ with suppress(Exception):
|
|
|
+ store.delete(artifact_ref)
|