import json
import re
import subprocess
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
COMPONENT = ROOT / "frontend/src/components/connectors/ConnectorOperations.vue"
DATABASE_PAGE = ROOT / "frontend/src/views/dataGovernance/development/enterpriseConnectors.vue"
METADATA_INDEX = ROOT / "frontend/src/views/dataGovernance/metadata/index.vue"
METADATA_MANAGEMENT = ROOT / "frontend/src/views/dataGovernance/metadata/components/metadataManagement.vue"
EXTERNAL_CATALOG_ACCESS = ROOT / "frontend/src/views/dataGovernance/metadata/components/externalCatalogAccess.vue"


def _balanced_block(source: str, start: int) -> str:
    """Return a JavaScript object/function block while ignoring strings/comments."""
    assert source[start] == "{"
    depth = 0
    quote = None
    escaped = False
    index = start
    while index < len(source):
        char = source[index]
        following = source[index + 1] if index + 1 < len(source) else ""
        if quote:
            if escaped:
                escaped = False
            elif char == "\\":
                escaped = True
            elif char == quote:
                quote = None
            index += 1
            continue
        if char in "'\"`":
            quote = char
            index += 1
            continue
        if char == "/" and following == "/":
            newline = source.find("\n", index)
            index = len(source) if newline < 0 else newline + 1
            continue
        if char == "/" and following == "*":
            end = source.find("*/", index + 2)
            assert end >= 0, "unterminated JavaScript comment"
            index = end + 2
            continue
        if char == "{":
            depth += 1
        elif char == "}":
            depth -= 1
            if depth == 0:
                return source[start : index + 1]
        index += 1
    raise AssertionError("unterminated JavaScript block")


def _named_object(source: str, name: str) -> str:
    match = re.search(rf"\b{re.escape(name)}\s*:\s*\{{", source)
    assert match, f"{name} object was not found"
    return _balanced_block(source, source.index("{", match.start()))


def _function_body(source: str, name: str) -> str:
    match = re.search(rf"\b(?:async\s+)?{re.escape(name)}\s*\([^)]*\)\s*\{{", source)
    assert match, f"{name} method was not found"
    block = _balanced_block(source, match.end() - 1)
    return block[1:-1]


def _validator_expression(prop: str) -> str:
    match = re.search(r"\bvalidator\s*:\s*(.+?)(?=\n\s*\})", prop, re.S)
    assert match, "prop validator was not found"
    return match.group(1).strip().rstrip(",")


def _run_node(script: str) -> dict:
    result = subprocess.run(
        ["node", "--input-type=module", "-e", script],
        check=False,
        capture_output=True,
        text=True,
        cwd=ROOT / "frontend",
    )
    assert result.returncode == 0, result.stderr
    return json.loads(result.stdout)


def _component_source() -> str:
    assert COMPONENT.is_file(), "ConnectorOperations shared surface is missing"
    return COMPONENT.read_text(encoding="utf-8")


def test_connector_prop_validators_execute_strict_database_and_rest_contracts():
    source = _component_source()
    connector_ids = _named_object(_named_object(source, "props"), "connectorIds")
    mode = _named_object(_named_object(source, "props"), "mode")
    pattern_match = re.search(
        r"const\s+CONNECTOR_ID_PATTERN\s*=\s*(/[^\n;]+/)", source
    )
    assert pattern_match, "connector ID regex declaration is missing"
    assert pattern_match.group(1) == "/^[a-z][a-z0-9_-]{2,63}$/"

    connector_validator = _validator_expression(connector_ids)
    mode_validator = _validator_expression(mode)
    mode_values_match = re.fullmatch(
        r"value\s*=>\s*\[\s*(.*?)\s*\]\.includes\(value\)", mode_validator
    )
    assert mode_values_match, "mode validator must be an explicit allowlist"
    assert re.findall(r"['\"]([^'\"]+)['\"]", mode_values_match.group(1)) == [
        "database",
        "rest-catalog",
    ]
    assert re.search(r"\btype\s*:\s*Array\b", connector_ids)
    assert re.search(r"\brequired\s*:\s*true\b", connector_ids)
    assert re.search(r"\btype\s*:\s*String\b", mode)
    assert re.search(r"\brequired\s*:\s*true\b", mode)
    assert "value.length > 0" in connector_validator
    assert re.search(r"value\.every\s*\(", connector_validator)
    assert "CONNECTOR_ID_PATTERN.test(id)" in connector_validator

    results = _run_node(
        f"""
        const CONNECTOR_ID_PATTERN = {pattern_match.group(1)};
        const connectorIdsValidator = ({connector_validator});
        const modeValidator = ({mode_validator});
        process.stdout.write(JSON.stringify({{
          connectorIds: {{
            valid: connectorIdsValidator(['oracle', 'postgresql', 'sqlserver']),
            empty: connectorIdsValidator([]),
            nonString: connectorIdsValidator(['oracle', 7]),
            tooShort: connectorIdsValidator(['ab']),
            uppercase: connectorIdsValidator(['Oracle']),
            dot: connectorIdsValidator(['oracle.prod']),
            unsafeCharacter: connectorIdsValidator(['oracle%prod']),
            maxLength: connectorIdsValidator(['a' + 'b'.repeat(63)]),
            tooLong: connectorIdsValidator(['a'.repeat(65)])
          }},
          modes: {{
            database: modeValidator('database'),
            restCatalog: modeValidator('rest-catalog'),
            metadata: modeValidator('metadata'),
            thirdMode: modeValidator('catalog-v2'),
            empty: modeValidator(''),
            typo: modeValidator('rest_catalog')
          }}
        }}));
        """
    )
    assert results["connectorIds"] == {
        "valid": True,
        "empty": False,
        "nonString": False,
        "tooShort": False,
        "uppercase": False,
        "dot": False,
        "unsafeCharacter": False,
        "maxLength": True,
        "tooLong": False,
    }
    assert results["modes"] == {
        "database": True,
        "restCatalog": True,
        "metadata": False,
        "thirdMode": False,
        "empty": False,
        "typo": False,
    }


def test_normalized_connector_ids_deduplicates_valid_values_and_fails_closed():
    source = _component_source()
    pattern_match = re.search(
        r"const\s+CONNECTOR_ID_PATTERN\s*=\s*(/[^\n;]+/)", source
    )
    assert pattern_match, "connector ID regex declaration is missing"
    normalized = _function_body(source, "normalizedConnectorIds")
    assert "Array.isArray(this.connectorIds)" in normalized
    assert "CONNECTOR_ID_PATTERN.test(id)" in normalized
    assert "return []" in normalized
    assert "new Set(this.connectorIds)" in normalized

    results = _run_node(
        f"""
        const CONNECTOR_ID_PATTERN = {pattern_match.group(1)};
        const normalize = new Function(
          'CONNECTOR_ID_PATTERN', `return function normalizedConnectorIds () {{{normalized}}}`
        )(CONNECTOR_ID_PATTERN);
        const evaluate = connectorIds => normalize.call({{ connectorIds }});
        process.stdout.write(JSON.stringify({{
          deduplicated: evaluate(['oracle', 'oracle', 'sqlserver']),
          malformed: evaluate(['oracle', 'Oracle']),
          nonArray: evaluate(null)
        }}));
        """
    )
    assert results == {
        "deduplicated": ["oracle", "sqlserver"],
        "malformed": [],
        "nonArray": [],
    }


def test_load_all_executes_fail_closed_filtering_and_clears_error_state():
    source = _component_source()
    load_all = _function_body(source, "loadAll")
    clear_state = _function_body(source, "clearConnectorState")
    guard = load_all.index("if (!this.normalizedConnectorIds.length)")
    api_calls = (
        "getConnectorManifests()",
        "getConnectorRuns()",
        "getDatasourceGraph()",
    )

    assert "this.clearConnectorState()" in load_all[:guard]
    assert "return" in load_all[guard : min(load_all.index(call) for call in api_calls)]
    assert all(guard < load_all.index(call) for call in api_calls)
    assert "this.manifests = []" in clear_state
    assert "this.runs = []" in clear_state
    assert "this.graph = {}" in clear_state
    assert load_all.count("new Set(this.normalizedConnectorIds)") == 1
    assert re.search(
        r"this\.manifests\s*=.*?\.filter\(item\s*=>\s*allowedConnectorIds\.has\(item\.connector_id\)\)",
        load_all,
    )
    assert re.search(
        r"this\.runs\s*=.*?\.filter\(item\s*=>\s*allowedConnectorIds\.has\(item\.connector_id\)\)",
        load_all,
    )
    assert "await Promise.all([getConnectorManifests(), getConnectorRuns(), getDatasourceGraph()])" in load_all
    catch_block = re.search(r"catch\s*\([^)]*\)\s*\{(.*?)\}", load_all, re.S)
    assert catch_block and "this.clearConnectorState()" in catch_block.group(1)

    results = _run_node(
        f"""
        const buildLoadAll = new Function(
          'getConnectorManifests', 'getConnectorRuns', 'getDatasourceGraph',
          `return async function loadAll () {{{load_all}}}`
        );
        async function invoke(ids, failRuns) {{
          const calls = [];
          const errors = [];
          const context = {{
            normalizedConnectorIds: ids,
            loading: false,
            manifests: [{{ connector_id: 'stale' }}],
            runs: [{{ connector_id: 'stale' }}],
            graph: {{ node_count: 99 }},
            clearConnectorState () {{ this.manifests = []; this.runs = []; this.graph = {{}}; }},
            $snackbar: {{ error: error => errors.push(String(error)) }}
          }};
          const response = data => Promise.resolve({{ data }});
          const loadAll = buildLoadAll(
            () => {{ calls.push('manifests'); return response({{ manifests: [{{ connector_id: 'oracle' }}, {{ connector_id: 'rest-catalog' }}] }}); }},
            () => {{ calls.push('runs'); return failRuns ? Promise.reject(new Error('runs unavailable')) : response({{ runs: [{{ connector_id: 'oracle' }}, {{ connector_id: 'rest-catalog' }}] }}); }},
            () => {{ calls.push('graph'); return response({{ summary: {{ node_count: 1, edge_count: 2 }} }}); }}
          );
          await loadAll.call(context);
          return {{ calls, errors, manifests: context.manifests, runs: context.runs, graph: context.graph, loading: context.loading }};
        }}
        const empty = await invoke([], false);
        const filtered = await invoke(['oracle'], false);
        const failed = await invoke(['oracle'], true);
        process.stdout.write(JSON.stringify({{ empty, filtered, failed }}));
        """
    )
    assert results["empty"] == {
        "calls": [],
        "errors": [],
        "manifests": [],
        "runs": [],
        "graph": {},
        "loading": False,
    }
    assert results["filtered"]["calls"] == ["manifests", "runs", "graph"]
    assert results["filtered"]["manifests"] == [{"connector_id": "oracle"}]
    assert results["filtered"]["runs"] == [{"connector_id": "oracle"}]
    assert results["filtered"]["graph"] == {"node_count": 1, "edge_count": 2}
    assert results["failed"]["manifests"] == []
    assert results["failed"]["runs"] == []
    assert results["failed"]["graph"] == {}
    assert len(results["failed"]["errors"]) == 1


def test_operations_await_refresh_and_rest_configuration_is_mode_gated():
    source = _component_source()
    markup = source.split("<script>", 1)[0]
    dry_run = _function_body(source, "dryRun")
    for field in ("base_url", "allowed_host"):
        fields = re.findall(
            rf'<v-text-field\b[^>]*v-model\.trim="form\.{field}"[^>]*/>', markup
        )
        assert len(fields) == 1
        assert "v-if=\"mode === 'rest-catalog'\"" in fields[0]
    assert re.search(
        r"if\s*\(this\.mode\s*===\s*'rest-catalog'\)\s*Object\.assign\(config,\s*\{\s*base_url:\s*this\.form\.base_url,\s*allowed_host:\s*this\.form\.allowed_host\s*\}\s*\)",
        dry_run,
    )
    assert '@click="loadAll"' in markup
    for name, connector_call in (
        ("dryRun", "executeConnectorRun"),
        ("cancel", "cancelConnectorRun"),
        ("resume", "resumeConnectorRun"),
    ):
        body = _function_body(source, name)
        assert body.index(f"await {connector_call}") < body.index("await this.loadAll()")

    results = _run_node(
        f"""
        const buildDryRun = new Function(
          'executeConnectorRun', `return async function dryRun () {{{dry_run}}}`
        );
        async function invoke(mode) {{
          const payloads = [];
          const events = [];
          const context = {{
            mode,
            selected: {{ connector_id: 'oracle', version: '1.0.0' }},
            form: {{ source_uid: 'source-1', credential_ref: 'env:TEST', base_url: 'https://catalog.example', allowed_host: 'catalog.example' }},
            dialog: true,
            loadAll: async () => events.push('loadAll'),
            $snackbar: {{ error: error => {{ throw error; }} }}
          }};
          const dryRun = buildDryRun(async payload => {{ events.push('execute'); payloads.push(payload); }});
          await dryRun.call(context);
          return {{ payload: payloads[0], events, dialog: context.dialog }};
        }}
        const database = await invoke('database');
        const restCatalog = await invoke('rest-catalog');
        process.stdout.write(JSON.stringify({{ database, restCatalog }}));
        """
    )
    assert results["database"]["payload"]["config"] == {"credential_ref": "env:TEST"}
    assert results["restCatalog"]["payload"]["config"] == {
        "credential_ref": "env:TEST",
        "base_url": "https://catalog.example",
        "allowed_host": "catalog.example",
    }
    assert results["database"]["events"] == ["execute", "loadAll"]
    assert results["restCatalog"]["events"] == ["execute", "loadAll"]
    assert results["database"]["dialog"] is False


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

    assert re.search(r"<h1\b[^>]*>\s*企业连接器\s*</h1>", source)
    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,
    )


def test_metadata_wrapper_exposes_two_tabs_and_gates_external_catalog_by_permission():
    source = METADATA_INDEX.read_text(encoding="utf-8")
    markup = source.split("<script>", 1)[0]

    assert len(re.findall(r"<v-tab(?![-\w])", markup)) == 2
    assert len(re.findall(r"<v-tab-item(?![-\w])", markup)) == 2
    assert "元数据管理" in markup
    assert "外部目录接入" in markup
    assert re.search(r"<v-tab\b[^>]*v-if=\"canReadConnectors\"[^>]*>\s*外部目录接入\s*</v-tab>", markup)
    assert re.search(r"<v-tab-item\b[^>]*v-if=\"canReadConnectors\"[^>]*>", markup)
    assert re.search(r"<external-catalog-access\b", markup)
    assert "<connector-operations" not in source

    computed = _named_object(source, "computed")
    connector_permissions = _function_body(computed, "connectorPermissions")
    can_read = _function_body(computed, "canReadConnectors")
    result = _run_node(
        f"""
        const connectorPermissions = new Function(`return function connectorPermissions () {{{connector_permissions}}}`)();
        const canReadConnectors = new Function(`return function canReadConnectors () {{{can_read}}}`)();
        const evaluate = state => {{
          const context = {{ $store: {{ state }} }};
          context.connectorPermissions = connectorPermissions.call(context);
          return canReadConnectors.call(context);
        }};
        process.stdout.write(JSON.stringify({{
          authorized: evaluate({{ user: {{ userInfo: {{ permissions: ['connectors:read'] }} }} }}),
          unrelated: evaluate({{ user: {{ userInfo: {{ permissions: ['metadata:read'] }} }} }}),
          missingUser: evaluate({{}}),
          malformedPermissions: evaluate({{ user: {{ userInfo: {{ permissions: 'connectors:read' }} }} }})
        }}));
        """
    )
    assert result == {
        "authorized": True,
        "unrelated": False,
        "missingUser": False,
        "malformedPermissions": False,
    }


def test_metadata_wrapper_resets_external_tab_when_connector_read_is_revoked():
    source = METADATA_INDEX.read_text(encoding="utf-8")
    watcher = _named_object(source, "watch")
    reset_active_tab = _function_body(watcher, "canReadConnectors")

    result = _run_node(
        f"""
        const resetActiveTab = new Function(`return function canReadConnectors (canReadConnectors) {{{reset_active_tab}}}`)();
        const revoked = {{ activeTab: 1 }};
        const authorized = {{ activeTab: 1 }};
        const management = {{ activeTab: 0 }};
        resetActiveTab.call(revoked, false);
        resetActiveTab.call(authorized, true);
        resetActiveTab.call(management, false);
        process.stdout.write(JSON.stringify({{ revoked, authorized, management }}));
        """
    )
    assert result == {
        "revoked": {"activeTab": 0},
        "authorized": {"activeTab": 1},
        "management": {"activeTab": 0},
    }


def test_metadata_management_preserves_crud_audit_and_research_link_after_move():
    source = METADATA_MANAGEMENT.read_text(encoding="utf-8")

    for marker in (
        "新增元数据",
        "handleSubmit",
        "handleDelete",
        "getMetaDataList",
        "handleAddAudit",
        "handleAuditSubmit",
        "数据研发中心",
        "/data-governance/development",
        "智能助手",
    ):
        assert marker in source
    assert "import FilterList from '../../components/Filter'" in source
    assert "import Edit from './edit'" in source
    assert "import AddAudit from './addAudit'" in source


def test_external_catalog_access_is_rest_only_and_cannot_mount_when_unauthorized():
    index_source = METADATA_INDEX.read_text(encoding="utf-8")
    access_source = EXTERNAL_CATALOG_ACCESS.read_text(encoding="utf-8")
    markup = access_source.split("<script>", 1)[0]

    assert "受控 HTTPS 元数据目录" in access_source
    assert "不是任意业务 API 采集" in access_source
    assert "企业 URL、允许主机、凭据与 UAT 由外部负责" in access_source
    assert re.search(
        r"<connector-operations\s+:connector-ids=\"\['rest-catalog'\]\"\s+mode=\"rest-catalog\"\s*/>",
        markup,
    )
    assert "ConnectorOperations" in access_source
    assert not any(name in access_source for name in ("oracle", "postgresql", "sqlserver"))

    index_markup = index_source.split("<script>", 1)[0]
    external_tab_item = re.search(
        r"<v-tab-item\b(?P<attrs>[^>]*)>\s*<external-catalog-access\b",
        index_markup,
        re.S,
    )
    assert external_tab_item, "external catalog component must be inside a tab item"
    assert 'v-if="canReadConnectors"' in external_tab_item.group("attrs")
    assert index_markup.count("<external-catalog-access") == 1