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.system.tokens import decode_access_token, issue_access_token
from tests.core.data_rules.test_contracts import (
    valid_dataflow_spec,
    valid_rule_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"},
                    }
                ],
            },
        }


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