DataOps-platform/tests/test_enterprise_connector_metadata_navigation_contract.py

import re
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
ROUTES = ROOT / "frontend/src/router/routes.js"


def _route_block(source: str, *, indent: int, name: str) -> str:
    """Return one route object, scoped to its direct indentation level."""
    start_token = "\n" + " " * indent + "{\n"
    name_token = " " * (indent + 2) + f"name: '{name}'"
    assert source.count(name_token) == 1, f"route {name!r} must be unique"

    name_at = source.index(name_token)
    start = source.rfind(start_token, 0, name_at)
    if start >= 0:
        start += 1
    elif source.startswith(start_token[1:]):
        start = 0

    terminators = [
        "\n" + " " * indent + "},",
        "\n" + " " * indent + "}",
    ]
    ends = [(source.find(token, name_at), token) for token in terminators]
    valid_ends = [(end, token) for end, token in ends if end >= 0]
    assert start >= 0 and valid_ends, f"route {name!r} block boundaries not found"
    end, end_token = min(valid_ends)
    return source[start : end + len(end_token)]


def _assert_exact_line(block: str, *, indent: int, field: str, value: str) -> None:
    expected = " " * indent + f"{field}: {value}"
    assert re.search(rf"(?m)^{re.escape(expected)},?$", block), f"missing exact line: {expected}"


def _assert_inline_meta_permission(route: str, *, indent: int, permission: str) -> None:
    marker = " " * indent + "meta: { permissions: ["
    match = re.search(rf"(?m)^{re.escape(marker)}([^\]]*)\]", route)
    assert match, "route meta permissions array not found"
    assert match.group(1).strip() == f"'{permission}'", (
        f"permissions must be exactly ['{permission}']"
    )


def _direct_child_names(parent_route: str, *, child_indent: int) -> list[str]:
    return re.findall(
        rf"(?m)^{re.escape(' ' * (child_indent + 2))}name: '([^']+)'",
        parent_route,
    )


def test_enterprise_connectors_are_owned_only_by_data_factory():
    routes = ROUTES.read_text(encoding="utf-8")
    data_research = _route_block(routes, indent=4, name="data-governance")
    data_factory = _route_block(routes, indent=4, name="dataFactory")

    assert routes.count("name: 'enterpriseConnectors'") == 1
    assert "enterpriseConnectors" not in data_research
    assert "path: '/data-governance/development/enterprise-connectors'" not in data_research
    assert "component: 'dataGovernance/development/enterpriseConnectors'" not in data_research

    connector = _route_block(data_factory, indent=8, name="enterpriseConnectors")
    for field, value in [
        ("icon", "'mdi-connection'"),
        ("title", "'企业连接器'"),
        ("path", "'/data-governance/development/enterprise-connectors'"),
        ("children", "[]"),
        ("label", "'企业连接器'"),
        ("sort", "4"),
        ("component", "'dataGovernance/development/enterpriseConnectors'"),
        ("name", "'enterpriseConnectors'"),
    ]:
        _assert_exact_line(connector, indent=10, field=field, value=value)

    _assert_inline_meta_permission(connector, indent=10, permission="connectors:read")


def test_data_factory_connector_and_orchestration_navigation_order_is_stable():
    routes = ROUTES.read_text(encoding="utf-8")
    data_factory = _route_block(routes, indent=4, name="dataFactory")
    names = [
        "enterpriseConnectors",
        "productionLineDeployment",
        "productionLine",
        "workflowIndex",
        "dataObservability",
        "connectionPoolIndex",
    ]

    child_blocks = [_route_block(data_factory, indent=8, name=name) for name in names]
    child_names = _direct_child_names(data_factory, child_indent=8)
    connector_index = child_names.index("enterpriseConnectors")
    deployment_index = child_names.index("productionLineDeployment")

    assert deployment_index == connector_index + 1
    assert child_names[connector_index : child_names.index("connectionPoolIndex") + 1] == names
    assert [re.search(r"(?m)^\s+name: '([^']+)'", block).group(1) for block in child_blocks] == names