DataOps-platform/tests/test_data_rule_frontend_contract.py

from pathlib import Path

from app.core.common.identifiers import new_governance_uid
from app.core.data_rules.contracts import validate_dataflow_spec

API = Path("frontend/src/api/dataRules.js")
AUTHORING = Path("frontend/src/components/DataRules/RuleAuthoringPanel.vue")
CATALOG = Path("frontend/src/components/DataRules/RuleCatalogPicker.vue")
EVIDENCE = Path("frontend/src/components/DataRules/CompilationEvidence.vue")
ASSEMBLER = Path(
    "frontend/src/components/DataRules/ProductionLineAssembler.vue"
)
STANDARD = Path(
    "frontend/src/views/dataGovernance/dataStandard/components/edit.vue"
)
DATAFLOW = Path(
    "frontend/src/views/dataGovernance/dataProcess/components/edit.vue"
)
FACTORY = Path(
    "frontend/src/views/dataFactory/workflow/ProductionLineDeployment.vue"
)
FACTORY_INDEX = Path("frontend/src/views/dataFactory/workflow/index.vue")


def test_rule_api_client_exposes_versions_publish_and_server_side_release():
    source = API.read_text(encoding="utf-8")

    assert "http.get('/rules/capabilities')" in source
    assert "http.post('/rules/interpret'" in source
    assert "http.post('/rules/validate'" in source
    assert "http.post('/rules/production-lines/resolve'" in source
    assert "http.post('/rules/rule-versions'" in source
    assert "http.post(`/rules/rule-versions/${versionId}/publish`)" in source
    assert "http.post(`/rules/rule-versions/${versionId}/validate`" in source
    assert "http.post(`/rules/rule-versions/${versionId}/test`" in source
    assert "http.get('/rules/catalog'" in source
    assert "/rules/catalog/assets/${assetType}/${versionId}/evidence" in source
    assert "http.post('/rules/standard-versions'" in source
    assert "http.post(`/rules/standard-versions/${versionId}/publish`)" in source
    assert "http.post(`/rules/production-lines/${dataflowUid}/release`" in source
    assert "/activate" not in source


def test_reusable_authoring_panel_preserves_human_review_gate():
    source = AUTHORING.read_text(encoding="utf-8")

    assert "authoringSurface" in source
    assert "interpretRule" in source
    assert "clarification_required" in source
    assert "$emit('candidate'" in source
    assert "createRuleVersion" in source
    assert "publishRuleVersion" in source
    assert "先生成并发布 RuleVersion" in source
    assert "rules:publish" in source
    assert "$emit('version'" in source
    assert "自动执行" not in source


def test_authoring_panel_exposes_governed_ai_lifecycle_without_code_editor():
    source = AUTHORING.read_text(encoding="utf-8")

    for token in [
        "schemaContextReady",
        "generation_receipt",
        "assumptions",
        "ambiguities",
        "modelProvenance",
        "validateRuleVersion",
        "testRuleVersion",
        "CompilationEvidence",
    ]:
        assert token in source
    assert "candidatePreview" not in source
    assert "<pre>{{ candidate" not in source


def test_catalog_picker_selects_fixed_published_asset_versions():
    source = CATALOG.read_text(encoding="utf-8")

    for token in [
        "searchPublishedCatalog",
        "assetType",
        "published",
        "schema_compatibility",
        "latest_evidence",
        "v-skeleton-loader",
        "aria-label",
        "keydown",
    ]:
        assert token in source
    assert "$emit('input', item.id)" in source
    assert "item.name" in source
    assert "rule_uid" in source
    assert "standard_uid" in source


def test_compilation_evidence_shows_trusted_chain_without_sensitive_payloads():
    source = EVIDENCE.read_text(encoding="utf-8")

    for token in [
        "generation",
        "logical_compile",
        "dry_run",
        "publication",
        "physical",
        "getRuleVersionEvidence",
    ]:
        assert token in source
    for forbidden in ["raw_sample", "secret", "password", "token"]:
        assert forbidden not in source.lower()
    assert "'ready'" in source


def test_standard_links_published_rules_and_keeps_legacy_code_read_only():
    standard = STANDARD.read_text(encoding="utf-8")

    assert 'authoring-surface="data_standard"' in standard
    assert '@version="handleRuleVersion"' in standard
    assert "RuleCatalogPicker" in standard
    assert "rule_version_id" in standard
    assert "legacyMigration" in standard
    assert "readonly" in standard
    assert "迁移" in standard
    assert "dataStandardCodeGenerate" not in standard
    assert "handleCodeGenerate" not in standard
    assert "rule_version_status" not in standard


def test_dataflow_assembler_uses_published_catalog_ids_without_inline_rules():
    source = ASSEMBLER.read_text(encoding="utf-8")
    dataflow = DATAFLOW.read_text(encoding="utf-8")

    assert "RuleCatalogPicker" in source
    assert "standard_version_id" in source
    assert "rule_version_id" in source
    assert "schema_compatibility" in source
    assert "releaseReadiness" in source
    assert "component_kind" in source
    assert "rule_spec" not in source
    assert "operation_code" not in source

    assert "ProductionLineAssembler" in dataflow
    assert "dataflow_spec" in dataflow
    assert "migration_metadata" in dataflow
    assert "rule_spec" not in dataflow
    assert "v-model=\"changeObj.rule\"" not in dataflow
    assert "handleRuleCandidate" not in dataflow


def test_assembler_emits_a_real_closed_dataflow_spec_for_every_station_kind():
    standard_version_id = new_governance_uid()
    apply_rule_version_id = new_governance_uid()
    quality_rule_version_id = new_governance_uid()
    normalized = validate_dataflow_spec(
        {
            "schema_version": "1.0",
            "dataflow_uid": new_governance_uid(),
            "name": "客户数据生产线",
            "description": "固定受治理资产版本",
            "input_schema_refs": ["bd:customer_raw:v1"],
            "output_schema_ref": "bd:customer_clean:v1",
            "components": [
                {
                    "id": "station_1",
                    "type": "standard.enforce",
                    "standard_version_id": standard_version_id,
                    "stage": "quality_gate",
                    "order": 1,
                },
                {
                    "id": "station_2",
                    "type": "rule.apply",
                    "rule_version_id": apply_rule_version_id,
                    "stage": "transform",
                    "order": 2,
                    "idempotency": {
                        "strategy": "partition_replace",
                        "key": "run_partition",
                    },
                },
                {
                    "id": "station_3",
                    "type": "quality.check",
                    "rule_version_id": quality_rule_version_id,
                    "stage": "quality_gate",
                    "order": 3,
                },
            ],
            "parameters": {},
        }
    )

    assert "idempotency" not in normalized["components"][0]
    assert "idempotency" in normalized["components"][1]
    assert "idempotency" not in normalized["components"][2]

    source = ASSEMBLER.read_text(encoding="utf-8")
    assert "station.component_kind === 'rule.apply'" in source
    assert "component.idempotency" in source
    assert "idempotency: {" not in source


def test_catalog_never_guesses_schema_compatibility_from_arbitrary_objects():
    source = CATALOG.read_text(encoding="utf-8")
    assembler = ASSEMBLER.read_text(encoding="utf-8")

    assert "normalizeCompatibility" in source
    assert "Object.keys(item.schema_compatibility).length" not in source
    assert "unknown" in source
    assert "schema_compatibility !== 'compatible'" in assembler
    assert "服务端发布预检" in assembler
    assert "requestSequence" in source
    assert "requestId !== this.requestSequence" in source
    assert "getPublishedCatalogAsset" in source
    assert "inputSchemaRefs" in source
    assert "outputSchemaRef" in source


def test_production_line_editor_renews_expiring_receipts_and_retries_only_safe_errors():
    source = DATAFLOW.read_text(encoding="utf-8")
    request = Path("frontend/src/utils/request.js").read_text(encoding="utf-8")

    assert "governedDraftExpiresAt" in source
    assert "data.expires_at" in source
    assert "expiresAt - Date.now() <= 60000" in source
    assert "ensureFreshDraftReservation" in source
    assert "draft_reservation_refresh_required" in source
    assert "governedDraftReplayRequired" in source
    assert "if (this.governedDraftReplayRequired) return" in source
    assert "await this.ensureDataflowUid(true)" in source
    assert source.count("await api.addDataFlow(payload)") == 2
    assert "error.response?.data" in request


def test_new_dataflow_gets_server_governed_uid_before_assembly():
    api = API.read_text(encoding="utf-8")
    dataflow = DATAFLOW.read_text(encoding="utf-8")

    assert "createProductionLineDraftIdentity" in api
    assert "/rules/production-lines/draft-identity" in api
    assert "ensureDataflowUid" in dataflow
    assert "if (!this.dataflowUid) await this.ensureDataflowUid()" in dataflow
    assert "persistedUid" in dataflow
    assert "dataflow_uid: null" not in dataflow


def test_standard_authoring_does_not_forge_validated_state_client_side():
    source = AUTHORING.read_text(encoding="utf-8")

    assert "candidate.candidate_type === 'standard'" not in source or (
        "status: 'validated'" not in source
    )
    assert "先发布规则，再在标准条款中绑定" in source


def test_data_factory_exposes_release_ready_but_no_fake_activation():
    page = FACTORY.read_text(encoding="utf-8")
    index = FACTORY_INDEX.read_text(encoding="utf-8")

    assert "数据生产线投产" in page
    assert "getRuleCapabilities" in page
    assert "production_line_release" in page
    assert "data_factory_activation" in page
    assert "发布版本" in page
    assert "执行计划" in page
    assert "发布与执行计划已经可用" in page
    assert "投产激活仍保持关闭" in page
    assert "ProductionLineDeployment" in index