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 "listDataFlowDeployments" in source
    assert "/deploy-disabled" in source
    assert "/canary" in source
    assert "/activate" in source
    assert "/rollback" 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_governed_trial_activation_and_rollback():
    page = FACTORY.read_text(encoding="utf-8")
    index = FACTORY_INDEX.read_text(encoding="utf-8")

    assert "数据生产线投产" in page
    assert "getRuleCapabilities" in page
    assert "data_factory_readiness" in page
    assert "listDataFlowDeployments" in page
    assert "listDataFlowDeploymentCandidates" in page
    assert "createDataFlowDeployment" in page
    assert "deployDataFlowDisabled" in page
    assert "runDataFlowCanary" in page
    assert "activateDataFlowDeployment" in page
    assert "rollbackDataFlowDeployment" in page
    assert "reconcileDataFlowDeploy" in page
    assert "reconcileDataFlowActivate" in page
    assert "reconcileDataFlowRollback" in page
    assert "dataflows:deploy" in page
    assert "dataflows:canary" in page
    assert "rules:execute" in page
    assert "dataflows:activate" in page
    assert "dataflows:rollback" in page
    assert "生产线包 Hash" in page
    assert "试生产证据" in page
    assert "状态履历" in page
    assert "生产线入厂" in page
    assert "window.localStorage.getItem(storageKey)" in page
    assert "window.localStorage.setItem(storageKey, value)" in page
    assert "error?.error?.code === 'terminal_conflict'" in page
    assert "Number(error?.code) === 409" not in page
    assert "unknown_operations" in page
    assert "v-dialog" in page
    assert "ProductionLineDeployment" in index