from __future__ import annotations from datetime import UTC, datetime, timedelta import pytest from app.core.common.identifiers import new_governance_uid from app.core.data_rules.contracts import rule_spec_hash from app.core.data_rules.deployment import ( OperationInProgress, OperationUnknown, TerminalConflict, ) from app.core.system.tokens import decode_access_token, issue_access_token from tests.core.data_rules.test_contracts import ( valid_dataflow_spec, valid_rule_spec, valid_standard_spec, ) from tests.core.data_rules.test_production_line import ( assertion_only_rule, published_rule, published_standard, ) class FakeAuthoringAgent: def __init__(self): self.calls = [] def interpret(self, **kwargs): self.calls.append(kwargs) return { "status": "ready", "source_text": kwargs["source_text"], "candidate_hash": "a" * 64, "context_hash": "b" * 64, "candidate": { "candidate_type": "rule", "rule_spec": valid_rule_spec(), }, } class FakeRuleRepository: def __init__(self): self.calls = [] self.catalog_items = [ { "asset_type": "rule", "version_id": new_governance_uid(), "asset_uid": new_governance_uid(), "name": "手机号规范化", "version": 3, "owner": new_governance_uid(), "status": "published", "schema_compatibility": { "input": "a" * 64, "output": "b" * 64, }, "impact_count": 2, "backend": "polars_batch", "latest_evidence": { "compile": {"status": "success"}, "test": {"status": "success"}, }, } ] def create_rule_version(self, **kwargs): self.calls.append(("create_rule_version", kwargs)) return { "id": new_governance_uid(), "rule_uid": kwargs["rule_spec"]["rule_uid"], "version_no": 1, "status": "draft", "spec_hash": rule_spec_hash(kwargs["rule_spec"]), "created": True, } def record_generation_run(self, **kwargs): self.calls.append(("record_generation_run", kwargs)) return { "id": new_governance_uid(), "correlation_id": new_governance_uid(), "decision": kwargs["evidence"]["status"], } def resolve_validation_context(self, context): self.calls.append(("resolve_validation_context", {"context": context})) return { "input_schema_snapshot_id": new_governance_uid(), "input_schema_hash": "c" * 64, "input_fields": [{"name": "mobile", "type": "string"}], "output_schema_snapshot_id": new_governance_uid(), "output_schema_hash": "d" * 64, "output_fields": [{"name": "mobile", "type": "string"}], "input_sample_artifact_ref": "minio://trusted/input.parquet", "input_sample_artifact_digest": "e" * 64, "golden_output_artifact_ref": None, "golden_output_artifact_digest": None, } def publish_rule_version(self, **kwargs): self.calls.append(("publish_rule_version", kwargs)) return { "id": kwargs["version_id"], "rule_uid": new_governance_uid(), "version_no": 1, "status": "published", "spec_hash": "a" * 64, } def create_standard_version(self, **kwargs): self.calls.append(("create_standard_version", kwargs)) return { "id": new_governance_uid(), "standard_uid": kwargs["standard_spec"]["standard_uid"], "version_no": 1, "status": "validated", "spec_hash": "b" * 64, "created": True, } def publish_standard_version(self, **kwargs): self.calls.append(("publish_standard_version", kwargs)) return { "id": kwargs["version_id"], "standard_uid": new_governance_uid(), "version_no": 1, "status": "published", "spec_hash": "b" * 64, } def search_published_assets(self, **kwargs): self.calls.append(("search_published_assets", kwargs)) return { "items": self.catalog_items, "total": 1, "limit": kwargs["limit"], "offset": kwargs["offset"], } def get_asset_evidence(self, **kwargs): self.calls.append(("get_asset_evidence", kwargs)) return { "asset_type": kwargs["asset_type"], "version_id": kwargs["version_id"], "stages": { "generation": {"status": "ready"}, "logical_compile": {"status": "success"}, "dry_run": {"status": "success"}, "publication": {"status": "published"}, "physical": [ { "backend": "polars_batch", "compile": {"status": "success"}, "test": {"status": "success"}, } ], }, } def get_published_asset(self, **kwargs): self.calls.append(("get_published_asset", kwargs)) return self.catalog_items[0] class FakeReleaseService: def __init__(self): self.calls = [] def release(self, **kwargs): self.calls.append(kwargs) return { "id": new_governance_uid(), "version_no": 1, "status": "released", "package_hash": "c" * 64, "package": { "package_hash": "c" * 64, "standard_version_ids": [], "rule_version_ids": [], }, } class FakeDeploymentService: def __init__(self): self.calls = [] self.deployment_id = new_governance_uid() self.evidence_id = new_governance_uid() def create(self, version_id, **kwargs): self.calls.append(("create", version_id, kwargs)) return {"id": self.deployment_id, "status": "draft"} def deploy_disabled(self, deployment_id, actor_uid, **kwargs): self.calls.append(("deploy_disabled", deployment_id, actor_uid, kwargs)) return {"id": deployment_id, "status": "disabled"} def run_canary(self, deployment_id, inputs, actor_uid, **kwargs): self.calls.append(("run_canary", deployment_id, inputs, actor_uid, kwargs)) return { "id": self.evidence_id, "deployment_id": deployment_id, "status": "passed", } def activate(self, deployment_id, evidence_id, actor_uid, **kwargs): self.calls.append(("activate", deployment_id, evidence_id, actor_uid, kwargs)) return {"id": deployment_id, "status": "active"} def execute_active(self, deployment_id, inputs, actor_uid, **kwargs): self.calls.append(("execute_active", deployment_id, inputs, actor_uid, kwargs)) return { "deployment_id": deployment_id, "execution_id": "execution-1", "correlation_id": kwargs["correlation_id"], "status": "success", } def rollback(self, deployment_id, actor_uid, **kwargs): self.calls.append(("rollback", deployment_id, actor_uid, kwargs)) return { "rolled_back": {"id": deployment_id, "status": "rolled_back"}, "active": {"id": new_governance_uid(), "status": "active"}, } def reconcile(self, deployment_id, action, actor_uid, **kwargs): self.calls.append(("reconcile", deployment_id, action, actor_uid, kwargs)) return { "deployment_id": deployment_id, "execution_id": "execution-1", "status": "success", } def list(self, environment=None): self.calls.append(("list", environment)) return [{"id": self.deployment_id, "status": "draft"}] class FakePublicationService: def __init__(self, repository): self.repository = repository self.plan_id = new_governance_uid() self.version_id = None def create_draft(self, **kwargs): self.repository.calls.append(("create_rule_version", kwargs)) self.version_id = new_governance_uid() return { "id": self.version_id, "rule_uid": kwargs["rule_spec"]["rule_uid"], "version_no": 1, "status": "draft", "spec_hash": rule_spec_hash(kwargs["rule_spec"]), "generation_run_id": new_governance_uid(), "created": True, } def validate(self, version_id, actor_uid): self.repository.calls.append( ( "validate_rule_version", {"version_id": version_id, "actor_uid": actor_uid}, ) ) return { "version_id": version_id, "version_status": "draft", "plan_id": self.plan_id, "plan_status": "compiled", "plan_hash": "d" * 64, } def test(self, version_id, actor_uid, *, plan_id): self.repository.calls.append( ( "test_rule_version", { "version_id": version_id, "actor_uid": actor_uid, "plan_id": plan_id, }, ) ) return { "version_id": version_id, "version_status": "validated", "plan_id": plan_id, "plan_status": "tested", "plan_hash": "d" * 64, } def publish(self, version_id, actor_uid): self.repository.calls.append( ( "publish_rule_version", {"version_id": version_id, "actor_uid": actor_uid}, ) ) return { "id": version_id, "status": "published", "plan_id": self.plan_id, "plan_status": "published", } def evidence(self, version_id): return {"version_id": version_id, "version_status": "validated"} def catalog(self, *, query, limit): return [] class FakeGraphSession: def __init__(self): self.calls = [] def run(self, query, parameters): self.calls.append((query, parameters)) return [ { "domain_id": 9, "domain_key": "customer_raw", "revision": "v2", "field_name": "customer_id", "data_type": "string", "nullable": False, "precision": None, "scale": None, "timezone": None, } ] def __enter__(self): return self def __exit__(self, *_args): return False class FakeGraphDriver: def __init__(self): self.session = FakeGraphSession() def get_session(self): return self.session class SnapshotOnlyRepository: def __init__(self): self.snapshots = {} def find_schema_snapshot(self, *, schema_ref, schema_hash): return self.snapshots.get((schema_ref, schema_hash)) def persist_schema_snapshot(self, *, snapshot): value = {"id": new_governance_uid(), **snapshot} self.snapshots[(snapshot["schema_ref"], snapshot["schema_hash"])] = value return value def _headers(app, role): token = issue_access_token( user_id=new_governance_uid(), roles=[role], secret=app.config["SECRET_KEY"], now=datetime.now(UTC), lifetime=timedelta(minutes=10), ) return {"Authorization": f"Bearer {token}"} def _use_token_identity(monkeypatch): def load(token, *, secret): claims = decode_access_token(token, secret=secret) return { "id": claims["sub"], "username": "contract-test", "display_name": "Contract Test", "roles": claims["roles"], } monkeypatch.setattr( "app.core.system.auth.load_identity_from_token", load, ) def test_rule_capabilities_and_validation_are_registered_and_governed( monkeypatch, ): from app import create_app app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True app.config["RULE_GENERATION_RECEIPT_SECRET"] = ( "dedicated-test-receipt-secret-with-entropy" ) client = app.test_client() response = client.get("/api/rules/capabilities", headers=_headers(app, "viewer")) assert response.status_code == 200 capabilities = response.get_json()["data"] assert capabilities["natural_language_authoring"] is True assert capabilities["immutable_asset_versions"] is True assert capabilities["server_side_publishing"] is True assert capabilities["production_line_release"] is True assert capabilities["data_factory_activation"] is False spec = valid_rule_spec() response = client.post( "/api/rules/validate", json={"asset_type": "rule", "spec": spec}, headers=_headers(app, "editor"), ) assert response.status_code == 200 result = response.get_json()["data"] assert result["spec_hash"] == rule_spec_hash(spec) assert result["normalized"]["rule_uid"] == spec["rule_uid"] forbidden = client.post( "/api/rules/validate", json={"asset_type": "rule", "spec": spec}, headers=_headers(app, "viewer"), ) assert forbidden.status_code == 403 def test_rule_interpret_uses_configured_agent_and_preserves_surface( monkeypatch, ): from app import create_app app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True app.config["RULE_GENERATION_RECEIPT_SECRET"] = ( "dedicated-test-receipt-secret-with-entropy" ) agent = FakeAuthoringAgent() app.extensions["data_rule_authoring_agent"] = agent repository = FakeRuleRepository() app.extensions["data_rule_repository"] = repository client = app.test_client() response = client.post( "/api/rules/interpret", json={ "source_text": "手机号去空格后必须为11位数字", "authoring_surface": "data_standard", "context": { "input_schema_snapshot_id": new_governance_uid(), "output_schema_snapshot_id": new_governance_uid(), "input_sample_artifact_ref": ("minio://trusted/input.parquet"), "golden_output_artifact_ref": None, }, }, headers=_headers(app, "editor"), ) assert response.status_code == 200 assert response.get_json()["data"]["status"] == "ready" assert response.get_json()["data"]["generation_run_id"] assert response.get_json()["data"]["generation_receipt"] assert agent.calls[0]["authoring_surface"] == "data_standard" assert repository.calls[0][0] == "resolve_validation_context" assert repository.calls[1][0] == "record_generation_run" def test_rule_interpret_preflights_receipt_signer_before_model_call( monkeypatch, ): from app import create_app app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True app.config["RULE_GENERATION_RECEIPT_SECRET"] = None agent = FakeAuthoringAgent() repository = FakeRuleRepository() app.extensions["data_rule_authoring_agent"] = agent app.extensions["data_rule_repository"] = repository client = app.test_client() response = client.post( "/api/rules/interpret", json={ "source_text": "手机号去空格后必须为11位数字", "authoring_surface": "data_standard", "context": { "input_schema_snapshot_id": new_governance_uid(), "output_schema_snapshot_id": new_governance_uid(), "input_sample_artifact_ref": ("minio://trusted/input.parquet"), "golden_output_artifact_ref": None, }, }, headers=_headers(app, "editor"), ) assert response.status_code == 503 assert agent.calls == [] assert repository.calls == [] def test_data_standard_interpret_rejects_non_rule_candidate_before_audit( monkeypatch, ): from app import create_app class StandardAgent: def interpret(self, **kwargs): return { "status": "ready", "source_text": kwargs["source_text"], "candidate_hash": "a" * 64, "context_hash": "b" * 64, "candidate": { "candidate_type": "standard", "standard_spec": valid_standard_spec(), }, } app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True app.config["RULE_GENERATION_RECEIPT_SECRET"] = "x" * 40 repository = FakeRuleRepository() app.extensions["data_rule_repository"] = repository app.extensions["data_rule_authoring_agent"] = StandardAgent() client = app.test_client() response = client.post( "/api/rules/interpret", json={ "source_text": "手机号必须为11位", "authoring_surface": "data_standard", "context": { "input_schema_snapshot_id": new_governance_uid(), "output_schema_snapshot_id": new_governance_uid(), "input_sample_artifact_ref": "minio://trusted/input.parquet", "golden_output_artifact_ref": None, }, }, headers=_headers(app, "editor"), ) assert response.status_code == 400 assert not any(call[0] == "record_generation_run" for call in repository.calls) def test_rule_interpret_and_validate_reject_unknown_fields(monkeypatch): from app import create_app app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True app.config["RULE_GENERATION_RECEIPT_SECRET"] = ( "dedicated-test-receipt-secret-with-entropy" ) repository = FakeRuleRepository() app.extensions["data_rule_repository"] = repository app.extensions["data_rule_authoring_agent"] = FakeAuthoringAgent() client = app.test_client() headers = _headers(app, "editor") interpreted = client.post( "/api/rules/interpret", json={ "source_text": "手机号必须为11位数字", "authoring_surface": "data_standard", "context": {}, "status": "published", }, headers=headers, ) validated = client.post( "/api/rules/validate", json={ "asset_type": "rule", "spec": valid_rule_spec(), "evidence": {"status": "success"}, }, headers=headers, ) assert interpreted.status_code == 400 assert validated.status_code == 400 assert repository.calls == [] def test_published_rule_catalog_uses_canonical_closed_contract(monkeypatch): from app import create_app app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True repository = FakeRuleRepository() app.extensions["data_rule_repository"] = repository client = app.test_client() headers = _headers(app, "viewer") response = client.get( "/api/rules/catalog?query=mobile&asset_type=rule&limit=10&offset=20", headers=headers, ) rejected = client.get( "/api/rules/catalog?query=mobile&status=published", headers=headers, ) assert response.status_code == 200 assert response.get_json()["data"] == { "items": repository.catalog_items, "total": 1, "limit": 10, "offset": 20, } assert repository.calls[-1] == ( "search_published_assets", { "query": "mobile", "asset_type": "rule", "limit": 10, "offset": 20, }, ) assert rejected.status_code == 400 def test_unified_catalog_defaults_to_all_assets_and_preserves_rule_alias( monkeypatch, ): from app import create_app app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True repository = FakeRuleRepository() app.extensions["data_rule_repository"] = repository client = app.test_client() headers = _headers(app, "viewer") canonical = client.get("/api/rules/catalog", headers=headers) alias = client.get("/api/rules/catalog/rule-versions", headers=headers) assert canonical.status_code == 200 assert repository.calls[0] == ( "search_published_assets", {"query": "", "asset_type": None, "limit": 50, "offset": 0}, ) assert alias.status_code == 200 assert repository.calls[1][1]["asset_type"] == "rule" def test_catalog_passes_flow_context_and_hydrates_exact_off_page_version( monkeypatch, ): import json from app import create_app app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True repository = FakeRuleRepository() app.extensions["data_rule_repository"] = repository client = app.test_client() headers = _headers(app, "viewer") inputs = ["bd:customer:v7"] output = "bd:customer_clean:v3" query = f"input_schema_refs={json.dumps(inputs)}&output_schema_ref={output}" listed = client.get(f"/api/rules/catalog?{query}", headers=headers) version_id = repository.catalog_items[0]["version_id"] exact = client.get( f"/api/rules/catalog/assets/rule/{version_id}?{query}", headers=headers, ) assert listed.status_code == 200 assert exact.status_code == 200 assert repository.calls[0][1]["input_schema_refs"] == inputs assert repository.calls[0][1]["output_schema_ref"] == output assert repository.calls[1][0] == "get_published_asset" assert repository.calls[1][1]["asset_type"] == "rule" assert repository.calls[1][1]["version_id"] == version_id assert repository.calls[1][1]["input_schema_refs"] == inputs assert repository.calls[1][1]["output_schema_ref"] == output assert repository.calls[1][1]["schema_resolver"] is not None def test_catalog_and_evidence_queries_are_closed_bounded_and_rules_read_only( monkeypatch, ): from app import create_app app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True repository = FakeRuleRepository() app.extensions["data_rule_repository"] = repository client = app.test_client() viewer = _headers(app, "viewer") version_id = new_governance_uid() evidence = client.get( f"/api/rules/catalog/assets/rule/{version_id}/evidence", headers=viewer, ) assert evidence.status_code == 200 value = evidence.get_json()["data"] assert set(value["stages"]) == { "generation", "logical_compile", "dry_run", "publication", "physical", } assert "source_text" not in str(value) assert "sample" not in str(value) assert repository.calls[-1] == ( "get_asset_evidence", {"asset_type": "rule", "version_id": version_id}, ) assert ( client.get("/api/rules/catalog?asset_type=dataflow", headers=viewer).status_code == 400 ) assert client.get("/api/rules/catalog?limit=101", headers=viewer).status_code == 400 assert client.get("/api/rules/catalog?offset=-1", headers=viewer).status_code == 400 assert ( client.get(f"/api/rules/catalog/assets/rule/{version_id}/evidence").status_code == 401 ) def test_legacy_rule_evidence_path_uses_same_safe_repository_contract( monkeypatch, ): from app import create_app app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True repository = FakeRuleRepository() app.extensions["data_rule_repository"] = repository client = app.test_client() version_id = new_governance_uid() response = client.get( f"/api/rules/rule-versions/{version_id}/evidence", headers=_headers(app, "viewer"), ) assert response.status_code == 200 assert repository.calls[-1] == ( "get_asset_evidence", {"asset_type": "rule", "version_id": version_id}, ) def test_production_line_resolve_preview_expands_standard_without_writing( monkeypatch, ): from app import create_app app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True client = app.test_client() standard_id = new_governance_uid() standard_rule_id = new_governance_uid() direct_rule_id = new_governance_uid() standard_rule = published_rule(standard_rule_id, assertion_only_rule()) direct_rule = published_rule(direct_rule_id) response = client.post( "/api/rules/production-lines/resolve", json={ "dataflow_spec": valid_dataflow_spec(standard_id, direct_rule_id), "standard_versions": { standard_id: published_standard(standard_id, standard_rule_id) }, "rule_versions": { standard_rule_id: standard_rule, direct_rule_id: direct_rule, }, "component_binding_ids": { "normalize_customer": new_governance_uid(), "customer_standard:mobile_format": new_governance_uid(), }, }, headers=_headers(app, "editor"), ) assert response.status_code == 200 result = response.get_json()["data"] assert result["preview"] is True assert result["release_ready"] is False assert result["package"]["package_hash"] assert result["package"]["standard_version_ids"] == [standard_id] def test_rule_api_rejects_invalid_or_unauthenticated_requests(monkeypatch): from app import create_app app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True client = app.test_client() assert client.get("/api/rules/capabilities").status_code == 401 response = client.post( "/api/rules/validate", json={"asset_type": "rule", "spec": {"schema_version": "1.0"}}, headers=_headers(app, "editor"), ) assert response.status_code == 400 assert "missing" not in str(response.get_json()).lower() def test_rule_and_standard_versions_are_created_then_published_by_separate_roles( monkeypatch, ): from app import create_app from tests.core.data_rules.test_contracts import valid_standard_spec app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True repository = FakeRuleRepository() app.extensions["data_rule_repository"] = repository publication = FakePublicationService(repository) app.extensions["rule_publication_service"] = publication client = app.test_client() rule_spec = valid_rule_spec() created = client.post( "/api/rules/rule-versions", json={ "source_text": "手机号必须为11位数字", "rule_spec": rule_spec, "category": "standard_clause", "generation_receipt": "signed-test-receipt", }, headers=_headers(app, "editor"), ) assert created.status_code == 201 assert created.get_json()["data"]["status"] == "draft" rule_version_id = created.get_json()["data"]["id"] compiled = client.post( f"/api/rules/rule-versions/{rule_version_id}/validate", headers=_headers(app, "editor"), ) assert compiled.status_code == 200 plan_id = compiled.get_json()["data"]["plan_id"] tested = client.post( f"/api/rules/rule-versions/{rule_version_id}/test", json={"plan_id": plan_id}, headers=_headers(app, "editor"), ) assert tested.status_code == 200 assert tested.get_json()["data"]["version_status"] == "validated" forbidden = client.post( f"/api/rules/rule-versions/{rule_version_id}/publish", headers=_headers(app, "editor"), ) assert forbidden.status_code == 403 published = client.post( f"/api/rules/rule-versions/{rule_version_id}/publish", headers=_headers(app, "admin"), ) assert published.status_code == 200 assert published.get_json()["data"]["status"] == "published" standard_spec = valid_standard_spec(rule_version_id) standard = client.post( "/api/rules/standard-versions", json={ "source_text": "客户手机号遵循统一格式", "standard_spec": standard_spec, }, headers=_headers(app, "editor"), ) assert standard.status_code == 201 standard_version_id = standard.get_json()["data"]["id"] standard_published = client.post( f"/api/rules/standard-versions/{standard_version_id}/publish", headers=_headers(app, "admin"), ) assert standard_published.status_code == 200 assert standard_published.get_json()["data"]["status"] == "published" methods = [method for method, _kwargs in repository.calls] assert methods == [ "create_rule_version", "validate_rule_version", "test_rule_version", "publish_rule_version", "create_standard_version", "publish_standard_version", ] def test_create_version_rejects_client_selected_lifecycle_status(monkeypatch): from app import create_app app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True repository = FakeRuleRepository() app.extensions["data_rule_repository"] = repository app.extensions["rule_publication_service"] = FakePublicationService(repository) client = app.test_client() response = client.post( "/api/rules/rule-versions", json={ "source_text": "手机号必须为11位数字", "rule_spec": valid_rule_spec(), "generation_receipt": "signed-test-receipt", "status": "published", }, headers=_headers(app, "editor"), ) assert response.status_code == 400 def test_generation_receipt_signer_requires_dedicated_secret(monkeypatch): from app import create_app from app.api.data_rules.routes import _receipt_signer monkeypatch.delenv("RULE_GENERATION_RECEIPT_SECRET", raising=False) app = create_app() app.config["RULE_GENERATION_RECEIPT_SECRET"] = None with ( app.app_context(), pytest.raises(RuntimeError, match="receipt secret"), ): _receipt_signer() app.config["RULE_GENERATION_RECEIPT_SECRET"] = ( "dedicated-test-receipt-secret-with-entropy" ) with app.app_context(): signer = _receipt_signer() assert signer is not None def test_rule_gates_reject_caller_supplied_compile_or_test_evidence( monkeypatch, ): from app import create_app app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True repository = FakeRuleRepository() app.extensions["rule_publication_service"] = FakePublicationService(repository) client = app.test_client() version_id = new_governance_uid() forged_compile = client.post( f"/api/rules/rule-versions/{version_id}/validate", json={"status": "success", "plan_hash": "a" * 64}, headers=_headers(app, "editor"), ) forged_test = client.post( f"/api/rules/rule-versions/{version_id}/test", json={ "plan_id": new_governance_uid(), "evidence": {"status": "success"}, }, headers=_headers(app, "editor"), ) assert forged_compile.status_code == 409 assert forged_test.status_code == 409 assert repository.calls == [] def test_data_factory_lifecycle_routes_are_closed_and_independently_governed( monkeypatch, ): from app import create_app app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True app.config["DATA_FACTORY_ACTIVATION_ENABLED"] = True service = FakeDeploymentService() app.extensions["dataflow_deployment_service"] = service client = app.test_client() admin = _headers(app, "admin") version_id = new_governance_uid() binding_id = new_governance_uid() created = client.post( "/api/rules/deployments", json={ "dataflow_version_id": version_id, "environment": "production", "binding_snapshot": { "input": {"binding_id": binding_id}, "output": {"binding_id": new_governance_uid()}, }, "schedule_plan": {"schema_version": "1.0"}, "reason": "production rollout", "idempotency_key": "create-1", }, headers=admin, ) assert created.status_code == 201 deployment_id = created.get_json()["data"]["id"] disabled = client.post( f"/api/rules/deployments/{deployment_id}/deploy-disabled", json={"reason": "disabled first", "idempotency_key": "deploy-1"}, headers=admin, ) assert disabled.status_code == 200 canary = client.post( f"/api/rules/deployments/{deployment_id}/canary", json={ "inputs": {"biz_date": "2026-07-24"}, "reason": "trial production", "idempotency_key": "canary-1", }, headers=admin, ) assert canary.status_code == 200 evidence_id = canary.get_json()["data"]["id"] active = client.post( f"/api/rules/deployments/{deployment_id}/activate", json={ "evidence_id": evidence_id, "reason": "mass production", "idempotency_key": "activate-1", }, headers=admin, ) assert active.status_code == 200 execution_correlation = new_governance_uid() executed = client.post( f"/api/rules/deployments/{deployment_id}/execute", json={ "inputs": {"biz_date": "2026-07-24"}, "reason": "manual production", "idempotency_key": "execute-1", "correlation_id": execution_correlation, }, headers=admin, ) assert executed.status_code == 200 assert executed.get_json()["data"]["correlation_id"] == (execution_correlation) reconciled = client.post( f"/api/rules/deployments/{deployment_id}/reconcile-execute", json={"idempotency_key": "execute-1"}, headers=admin, ) assert reconciled.status_code == 200 assert reconciled.get_json()["data"]["execution_id"] == "execution-1" rolled_back = client.post( f"/api/rules/deployments/{deployment_id}/rollback", json={"reason": "restore stable", "idempotency_key": "rollback-1"}, headers=admin, ) assert rolled_back.status_code == 200 forbidden = client.post( f"/api/rules/deployments/{deployment_id}/canary", json={"inputs": {}, "idempotency_key": "viewer"}, headers=_headers(app, "viewer"), ) assert forbidden.status_code == 403 inline_code = client.post( f"/api/rules/deployments/{deployment_id}/canary", json={ "inputs": {}, "idempotency_key": "unsafe", "sql": "delete from customer", }, headers=admin, ) assert inline_code.status_code == 409 assert [call[0] for call in service.calls] == [ "create", "deploy_disabled", "run_canary", "activate", "execute_active", "reconcile", "rollback", ] def test_data_factory_api_preserves_read_write_output_binding(monkeypatch): from app import create_app app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True service = FakeDeploymentService() app.extensions["dataflow_deployment_service"] = service output = { "data_source_uid": new_governance_uid(), "object_kind": "table", "object_ref": "curated.clean", "schema_snapshot_id": new_governance_uid(), "schema_hash": "a" * 64, "binding_hash": "b" * 64, "access_mode": "read_write", "write_mode": "merge", "dialect": "postgresql", } response = app.test_client().post( "/api/rules/deployments", json={ "dataflow_version_id": new_governance_uid(), "environment": "test", "binding_snapshot": { "input": {"binding_id": new_governance_uid()}, "output": output, }, "schedule_plan": {"schema_version": "1.0"}, "reason": "read-write pipeline", "idempotency_key": "read-write-output", }, headers=_headers(app, "admin"), ) assert response.status_code == 201 assert service.calls[0][2]["binding_snapshot"]["output"] == output @pytest.mark.parametrize( ("exception", "code", "disposition", "reconcile_required"), [ ( OperationInProgress("busy"), "operation_in_progress", "retain", False, ), (OperationUnknown("unknown"), "operation_unknown", "retain", True), ( TerminalConflict("terminal"), "terminal_conflict", "rotate", False, ), ], ) def test_data_factory_returns_stable_idempotency_error_contract( monkeypatch, exception, code, disposition, reconcile_required ): from app import create_app app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True service = FakeDeploymentService() def fail(*args, **kwargs): del args, kwargs raise exception service.deploy_disabled = fail app.extensions["dataflow_deployment_service"] = service response = app.test_client().post( f"/api/rules/deployments/{service.deployment_id}/deploy-disabled", json={"idempotency_key": "stable-operation-key"}, headers=_headers(app, "admin"), ) assert response.status_code == 409 error = response.get_json()["error"] assert error == { "code": code, "idempotency_key_disposition": disposition, "reconcile_required": reconcile_required, } def test_data_factory_unknown_conflict_identifies_original_reconcile_target( monkeypatch, ): from app import create_app app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True service = FakeDeploymentService() blocker = { "id": new_governance_uid(), "deployment_id": service.deployment_id, "action": "activate", "idempotency_key": "original-unknown-key", "scope_unknown_count": 2, } def fail(*args, **kwargs): del args, kwargs raise OperationUnknown("reconcile original", operation=blocker) service.deploy_disabled = fail app.extensions["dataflow_deployment_service"] = service response = app.test_client().post( f"/api/rules/deployments/{service.deployment_id}/deploy-disabled", json={"idempotency_key": "different-key"}, headers=_headers(app, "admin"), ) assert response.status_code == 409 assert response.get_json()["error"]["blocking_operation"] == blocker def test_create_rule_version_rejects_legacy_v1_payload_before_repository( monkeypatch, ): from app import create_app app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True repository = FakeRuleRepository() app.extensions["data_rule_repository"] = repository client = app.test_client() legacy = valid_rule_spec() legacy["schema_version"] = "1.0" response = client.post( "/api/rules/rule-versions", json={"source_text": "旧版规则不能再创建", "rule_spec": legacy}, headers=_headers(app, "editor"), ) assert response.status_code == 400 assert repository.calls == [] def test_dataflow_release_uses_server_assets_and_release_permission( monkeypatch, ): from app import create_app app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True service = FakeReleaseService() app.extensions["production_line_release_service"] = service client = app.test_client() flow = valid_dataflow_spec() payload = { "source_text": "客户数据生产线", "dataflow_spec": flow, } forbidden = client.post( f"/api/rules/production-lines/{flow['dataflow_uid']}/release", json=payload, headers=_headers(app, "editor"), ) assert forbidden.status_code == 403 response = client.post( f"/api/rules/production-lines/{flow['dataflow_uid']}/release", json=payload, headers=_headers(app, "admin"), ) assert response.status_code == 201 assert response.get_json()["data"]["status"] == "released" assert service.calls[0]["dataflow_uid"] == flow["dataflow_uid"] assert "standard_versions" not in service.calls[0] assert "rule_versions" not in service.calls[0] assert "component_binding_ids" not in service.calls[0] def test_dataflow_release_rejects_client_authored_schema_hashes(monkeypatch): from app import create_app app = create_app() _use_token_identity(monkeypatch) app.config["TESTING"] = True service = FakeReleaseService() app.extensions["production_line_release_service"] = service client = app.test_client() flow = valid_dataflow_spec() response = client.post( f"/api/rules/production-lines/{flow['dataflow_uid']}/release", json={ "source_text": "客户数据生产线", "dataflow_spec": flow, "input_schema_hashes": {"bd:customer_raw:v2": "a" * 64}, "output_schema_hash": "b" * 64, }, headers=_headers(app, "admin"), ) assert response.status_code == 409 assert service.calls == [] def test_default_release_service_uses_lazy_neo4j_schema_catalog(monkeypatch): from app import create_app from app.api.data_rules.routes import _release_service app = create_app() repository = SnapshotOnlyRepository() driver = FakeGraphDriver() monkeypatch.setattr("app.core.data_rules.schema_resolver.neo4j_driver", driver) app.extensions["data_rule_repository"] = repository with app.app_context(): service = _release_service() snapshot = service.schema_resolver.resolve("bd:customer_raw:v2") assert snapshot["source_revision"] == "neo4j:9:v2" assert driver.session.calls[0][1] == { "domain_key": "customer_raw", "revision": "v2", }