DataOps-platform/tests/test_enterprise_connector_metadata_frontend_contract.py

from pathlib import Path
import re


ROOT = Path(__file__).resolve().parents[1]
COMPONENT = ROOT / "frontend/src/components/connectors/ConnectorOperations.vue"
DATABASE_PAGE = ROOT / "frontend/src/views/dataGovernance/development/enterpriseConnectors.vue"


def _method_block(source: str, name: str) -> str:
    match = re.search(rf"(?s)\b{name}\s*\([^)]*\)\s*\{{(.*?)(?=\n\s{{4}}\w+\s*\(|\n\s{{2}}\}}\n\}})", source)
    assert match, f"{name} method was not found"
    return match.group(0)


def test_connector_operations_has_fail_closed_allowlist_for_manifests_and_runs():
    assert COMPONENT.is_file(), "ConnectorOperations shared surface is missing"
    source = COMPONENT.read_text(encoding="utf-8")

    assert re.search(
        r"connectorIds:\s*\{\s*type:\s*Array,\s*required:\s*true,.*?validator:",
        source,
        re.S,
    )
    assert "/^[a-z][a-z0-9_-]{2,63}$/" in source
    assert re.search(
        r"mode:\s*\{\s*type:\s*String,\s*required:\s*true,.*?database.*?rest-catalog",
        source,
        re.S,
    )

    normalized = _method_block(source, "normalizedConnectorIds")
    assert "Array.isArray(this.connectorIds)" in normalized
    assert "new Set" in normalized
    assert "return []" in normalized

    load_all = _method_block(source, "loadAll")
    empty_guard = load_all.index("if (!this.normalizedConnectorIds.length)")
    first_api_call = load_all.index("getConnectorManifests()")
    assert empty_guard < first_api_call
    assert "this.clearConnectorState()" in load_all[:first_api_call]
    clear_state = _method_block(source, "clearConnectorState")
    assert "this.manifests = []" in clear_state
    assert "this.runs = []" in clear_state
    assert "this.graph = {}" in clear_state
    assert "return" in load_all[empty_guard:first_api_call]

    assert "const allowedConnectorIds = new Set(this.normalizedConnectorIds)" in load_all
    assert "this.manifests = (manifests.data.manifests || []).filter(item => allowedConnectorIds.has(item.connector_id))" in load_all
    assert "this.runs = (runs.data.runs || []).filter(item => allowedConnectorIds.has(item.connector_id))" in load_all


def test_database_wrapper_is_database_only_and_keeps_scope_warning():
    source = DATABASE_PAGE.read_text(encoding="utf-8")

    assert "rest-catalog" not in source
    assert "当前开放范围仅限数据库访问。文件目录、对象存储、API 与消息系统等来源将在后续版本扩展开发。" in source
    assert re.search(
        r"<connector-operations\s+:connector-ids=\"\['oracle', 'postgresql', 'sqlserver'\]\"\s+mode=\"database\"\s*/>",
        source,
    )