DataOps-platform/tests/test_data_rule_api.py

from __future__ import annotations

from datetime import datetime, timedelta, timezone

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 = []

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


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 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(timezone.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
    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
    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_ref": "bd:customer:v7"},
        },
        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 agent.calls[0]["authoring_surface"] == "data_standard"
    assert repository.calls[0][0] == "record_generation_run"


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
    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",
        },
        headers=_headers(app, "editor"),
    )
    assert created.status_code == 201
    assert created.get_json()["data"]["status"] == "validated"
    rule_version_id = created.get_json()["data"]["id"]

    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",
        "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
    app.extensions["data_rule_repository"] = FakeRuleRepository()
    client = app.test_client()

    response = client.post(
        "/api/rules/rule-versions",
        json={
            "source_text": "手机号必须为11位数字",
            "rule_spec": valid_rule_spec(),
            "status": "published",
        },
        headers=_headers(app, "editor"),
    )

    assert response.status_code == 400


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