DataOps-platform/tests/test_data_rule_api.py

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",
    }