from __future__ import annotations

import json
import re
from pathlib import Path

INVENTORY_PATH = (
    Path(__file__).resolve().parents[3]
    / "deploy"
    / "data-access"
    / "datasets"
    / "gcp-digital-bi-dbt.json"
)
LINEAGE_CATALOG_PATH = (
    Path(__file__).resolve().parents[2]
    / "app"
    / "domains"
    / "data_access"
    / "lineage_catalogs"
    / "gcp-digital-bi-dbt.json"
)
INFRA_MAIN_PATH = (
    Path(__file__).resolve().parents[3] / "infra" / "gcp-dax" / "main.tf"
)
INFRA_VARIABLES_PATH = (
    Path(__file__).resolve().parents[3] / "infra" / "gcp-dax" / "variables.tf"
)

EXPECTED_BUSINESS_REQUESTABLE_LOCATORS = {
    ("gcp-digital-bi", dataset_id)
    for dataset_id in {
        "anls_business_operations",
        "anls_cloud",
        "anls_core",
        "aws_marketplace_datafeed",
        "billing",
        "bo",
        "dp",
        "dwh_business_operations",
        "dwh_cloud_platform",
        "dwh_core",
        "dwh_finance",
        "dwh_marketing",
        "dwh_plg",
        "dwh_sales",
        "dws_business_operations",
        "model_billing",
        "model_dbaas_billing",
        "model_dim",
        "model_fp_and_a",
        "model_msp_subclient_billing",
        "model_plg",
        "model_tidbcloud",
        "metadb_new",
    }
}
PROVIDER_MISSING_SOURCE_LOCATORS = {
    ("gcp-digital-bi", "usage_us_west2"),
}


def _inventory_rows() -> list[dict[str, object]]:
    payload = json.loads(INVENTORY_PATH.read_text(encoding="utf-8"))
    return payload["datasets"]


def _referenced_source_locators() -> set[tuple[str, str]]:
    catalog = json.loads(LINEAGE_CATALOG_PATH.read_text(encoding="utf-8"))
    referenced_upstream_node_ids = {
        str(edge["upstream_node_id"])
        for edge in catalog["edges"]
    }
    return {
        (str(node["project_id"]), str(node["dataset_id"]))
        for node in catalog["nodes"]
        if (
            node["node_type"] == "source"
            and str(node["node_id"]) in referenced_upstream_node_ids
        )
    }


def test_requestable_inventory_matches_production_catalog_scope() -> None:
    payload = json.loads(INVENTORY_PATH.read_text(encoding="utf-8"))
    requestable = {
        (str(row["project_id"]), str(row["dataset_id"]))
        for row in payload["datasets"]
        if row.get("requestable") is True
    }

    expected = EXPECTED_BUSINESS_REQUESTABLE_LOCATORS | (
        _referenced_source_locators() - PROVIDER_MISSING_SOURCE_LOCATORS
    )
    assert requestable == expected
    assert payload["metadata"]["requestable_count"] == len(expected)


def test_requestable_inventory_has_business_descriptions() -> None:
    rows = [
        row
        for row in _inventory_rows()
        if row.get("requestable") is True
    ]

    assert len(rows) == len(
        EXPECTED_BUSINESS_REQUESTABLE_LOCATORS
        | (_referenced_source_locators() - PROVIDER_MISSING_SOURCE_LOCATORS),
    )
    for row in rows:
        description = row.get("description")
        assert isinstance(description, str)
        assert description.strip() == description
        assert "\n" not in description
        assert 20 <= len(description) <= 500


def test_metadb_new_is_requestable_with_the_approved_owner() -> None:
    row = next(
        row for row in _inventory_rows() if row["dataset_id"] == "metadb_new"
    )

    assert row["project_id"] == "gcp-digital-bi"
    assert row["owner_email"] == "xukai@pingcap.com"
    assert row["requestable"] is True
    assert row["status"] == "active"


def test_alibaba_cloud_is_owner_cataloged_but_not_requestable() -> None:
    row = next(
        row for row in _inventory_rows() if row["dataset_id"] == "alibaba_cloud"
    )

    assert row["project_id"] == "gcp-digital-bi"
    assert row["owner_email"] == "fanfei.kong@pingcap.com"
    assert row["requestable"] is False
    assert row["status"] == "active"
    assert row["labels"]["managed_by"] == "manual_catalog"
    assert row["labels"]["cloud_provider"] == "alibaba_cloud"


def test_every_referenced_lineage_source_dataset_is_in_the_helix_inventory() -> None:
    source_locators = _referenced_source_locators()
    rows_by_locator = {
        (str(row["project_id"]), str(row["dataset_id"])): row
        for row in _inventory_rows()
    }
    assembled_root_locators = {
        locator
        for locator, row in rows_by_locator.items()
        if row.get("labels", {}).get("lineage_root") == "true"
    }

    assert source_locators <= rows_by_locator.keys()
    assert assembled_root_locators <= source_locators
    for locator in source_locators:
        row = rows_by_locator[locator]
        assert row.get("owner_email")
        if locator in PROVIDER_MISSING_SOURCE_LOCATORS:
            assert row["requestable"] is False
            assert row["status"] == "suspended"
            assert row["labels"]["provider_missing"] == "true"
            continue
        assert row["requestable"] is True
        if row.get("labels", {}).get("lineage_root") == "true":
            assert row["owner_email"] == "xiaopeng.zhu@pingcap.com"
            assert row["status"] == "active"
            assert row["labels"]["cataloged_by"] == "dbt_lineage"
            assert row["labels"]["dbt_generated"] == "false"
            assert "managed_by" not in row["labels"]


def test_deployed_lineage_catalog_has_one_node_per_physical_relation() -> None:
    catalog = json.loads(LINEAGE_CATALOG_PATH.read_text(encoding="utf-8"))
    locators = [
        (
            str(node["project_id"]),
            str(node["dataset_id"]),
            str(node["table_id"]),
        )
        for node in catalog["nodes"]
    ]

    assert len(locators) == len(set(locators))


def test_terraform_controller_scope_covers_every_active_catalog_dataset() -> None:
    active_projects = {
        str(row["project_id"])
        for row in _inventory_rows()
        if row.get("status") == "active"
    }
    main = INFRA_MAIN_PATH.read_text(encoding="utf-8")
    variables = INFRA_VARIABLES_PATH.read_text(encoding="utf-8")
    managed_rows = re.search(
        r"managed_dataset_rows = \[(?P<body>.*?)\n  \]",
        main,
        re.DOTALL,
    )
    block = re.search(
        r'variable "managed_dataset_project_ids" \{(?P<body>.*?)\n\}',
        variables,
        re.DOTALL,
    )

    assert active_projects == {"gcp-digital-bi", "gcp-digital-bi-dev"}
    assert managed_rows is not None
    assert 'try(dataset.status, "active") == "active"' in managed_rows.group("body")
    assert "requestable" not in managed_rows.group("body")
    assert "managed_dataset_scope_digest" in main
    assert "dataset_inventory_file = local.dataset_inventory_file" not in main
    assert "contains(var.managed_dataset_project_ids, dataset.project_id)" in main
    assert "for_each = var.managed_dataset_project_ids" in main
    assert (
        "dax_dataset_iam_reconciler[each.value.project_id].name"
        in main
    )
    assert block is not None
    defaults = set(re.findall(r'"([a-z][a-z0-9-]+)"', block.group("body")))
    assert defaults == active_projects


def test_non_requestable_active_catalog_rows_retain_controller_scope() -> None:
    rows = _inventory_rows()
    active_locators = {
        (str(row["project_id"]), str(row["dataset_id"]))
        for row in rows
        if row.get("status") == "active"
    }
    requestable_locators = {
        (str(row["project_id"]), str(row["dataset_id"]))
        for row in rows
        if row.get("status") == "active" and row.get("requestable") is True
    }

    assert len(active_locators) == 66
    assert len(requestable_locators) == 47
    assert len(active_locators - requestable_locators) == 19
    assert (
        "gcp-digital-bi",
        "alibaba_cloud",
    ) in active_locators - requestable_locators