"""Tests for the dbt-derived, read-only DAX table-lineage projection."""

from __future__ import annotations

import json

import pytest

from app.domains.data_access import lineage
from app.domains.data_access.lineage_catalog import build_dbt_lineage_catalog
from app.domains.data_access.lineage_inventory import sync_lineage_root_datasets
from scripts import seed_data_access_resources as seed_script


@pytest.fixture(autouse=True)
def _clear_catalog_cache():
    lineage.clear_lineage_catalog_cache()
    yield
    lineage.clear_lineage_catalog_cache()


def _manifest() -> dict:
    return {
        "metadata": {"generated_at": "2026-08-20T00:00:00Z"},
        "nodes": {
            "model.bi_dev.stg_orders": {
                "resource_type": "model",
                "database": "gcp-digital-bi",
                "schema": "stg_orders",
                "alias": "stg_orders",
                "config": {"materialized": "view"},
                "description": "Normalized orders",
            },
            "model.bi_dev.orders_dashboard": {
                "resource_type": "model",
                "database": "gcp-digital-bi",
                "schema": "dws_orders",
                "alias": "orders_dashboard",
                "config": {"materialized": "table"},
                "description": "Dashboard relation",
            },
            "test.bi_dev.orders_dashboard_not_null": {
                "resource_type": "test",
                "database": "gcp-digital-bi",
                "schema": "dbt_test__audit",
                "alias": "orders_dashboard_not_null",
            },
        },
        "sources": {
            "source.bi_dev.erp.orders": {
                "resource_type": "source",
                "database": "gcp-digital-bi",
                "schema": "raw_erp",
                "identifier": "orders",
                "name": "orders",
                "description": "ERP extract",
            },
        },
        "parent_map": {
            "model.bi_dev.stg_orders": ["source.bi_dev.erp.orders"],
            "model.bi_dev.orders_dashboard": ["model.bi_dev.stg_orders"],
            "test.bi_dev.orders_dashboard_not_null": [
                "model.bi_dev.orders_dashboard",
            ],
        },
    }


def _install_catalog(monkeypatch, tmp_path) -> None:
    catalog = build_dbt_lineage_catalog(
        _manifest(),
        catalog_id="gcp-digital-bi-dbt",
        source_version="dbt-main-fixture",
    )
    path = tmp_path / "gcp-digital-bi-dbt.json"
    path.write_text(json.dumps(catalog), encoding="utf-8")
    monkeypatch.setattr(lineage, "LINEAGE_CATALOG_DIR", tmp_path)


def test_builder_projects_only_relations_and_direct_dbt_dependencies() -> None:
    catalog = build_dbt_lineage_catalog(
        _manifest(),
        catalog_id="gcp-digital-bi-dbt",
        source_version="dbt-main-fixture",
    )

    assert [node["node_id"] for node in catalog["nodes"]] == [
        "model.bi_dev.orders_dashboard",
        "model.bi_dev.stg_orders",
        "source.bi_dev.erp.orders",
    ]
    assert catalog["edges"] == [
        {
            "upstream_node_id": "model.bi_dev.stg_orders",
            "downstream_node_id": "model.bi_dev.orders_dashboard",
            "relationship": "ref",
        },
        {
            "upstream_node_id": "source.bi_dev.erp.orders",
            "downstream_node_id": "model.bi_dev.stg_orders",
            "relationship": "source",
        },
    ]


def test_table_lineage_returns_a_root_to_serving_path(monkeypatch, tmp_path) -> None:
    _install_catalog(monkeypatch, tmp_path)

    graph = lineage.get_table_lineage(
        project_id="gcp-digital-bi",
        dataset_id="dws_orders",
        table_id="orders_dashboard",
    )

    assert graph.status == "available"
    assert [(node.table_id, node.layer, node.level) for node in graph.nodes] == [
        ("orders", "source", 0),
        ("stg_orders", "staging", 1),
        ("orders_dashboard", "serving", 2),
    ]
    assert [node.table_id for node in graph.nodes if node.is_root] == ["orders"]
    assert [node.table_id for node in graph.nodes if node.is_target] == [
        "orders_dashboard",
    ]
    assert graph.truncated is False


def test_source_table_lineage_returns_the_downstream_reference_path(
    monkeypatch,
    tmp_path,
) -> None:
    _install_catalog(monkeypatch, tmp_path)

    graph = lineage.get_table_lineage(
        project_id="gcp-digital-bi",
        dataset_id="raw_erp",
        table_id="orders",
    )

    assert graph.status == "available"
    assert graph.direction == "downstream"
    assert [(node.table_id, node.level) for node in graph.nodes] == [
        ("orders", 0),
        ("stg_orders", 1),
        ("orders_dashboard", 2),
    ]
    assert [node.table_id for node in graph.nodes if node.is_target] == ["orders"]
    assert [(edge.upstream_node_id, edge.downstream_node_id) for edge in graph.edges] == [
        ("model.bi_dev.stg_orders", "model.bi_dev.orders_dashboard"),
        ("source.bi_dev.erp.orders", "model.bi_dev.stg_orders"),
    ]


def test_table_lineage_does_not_invent_a_path_for_untracked_tables(monkeypatch, tmp_path) -> None:
    _install_catalog(monkeypatch, tmp_path)

    graph = lineage.get_table_lineage(
        project_id="gcp-digital-bi",
        dataset_id="dws_orders",
        table_id="manual_table",
    )

    assert graph.status == "not_tracked"
    assert graph.nodes == ()
    assert graph.edges == ()


def test_lineage_catalog_lists_referenced_source_tables_for_request_authoring(
    monkeypatch,
    tmp_path,
) -> None:
    _install_catalog(monkeypatch, tmp_path)

    tables = lineage.list_lineage_catalog_tables(
        project_id="gcp-digital-bi",
        dataset_id="raw_erp",
    )

    assert [(table.table_id, table.table_type) for table in tables] == [
        ("orders", "TABLE"),
    ]


def test_lineage_catalog_search_matches_table_name_and_description(
    monkeypatch,
    tmp_path,
) -> None:
    _install_catalog(monkeypatch, tmp_path)

    name_matches = lineage.search_lineage_catalog_tables("ORDERS_DASH")
    description_matches = lineage.search_lineage_catalog_tables("erp extract")

    assert [
        (match.project_id, match.dataset_id, match.table.table_id)
        for match in name_matches
    ] == [("gcp-digital-bi", "dws_orders", "orders_dashboard")]
    assert [
        (
            match.project_id,
            match.dataset_id,
            match.table.table_id,
            match.table.description,
        )
        for match in description_matches
    ] == [("gcp-digital-bi", "raw_erp", "orders", "ERP extract")]


def test_lineage_catalog_lists_referenced_source_tables_from_external_project(
    monkeypatch,
    tmp_path,
) -> None:
    manifest = _manifest()
    manifest["sources"]["source.bi_dev.erp.orders"]["database"] = "gcp-digital-bi-dev"
    catalog = build_dbt_lineage_catalog(
        manifest,
        catalog_id="gcp-digital-bi-dbt",
        source_version="dbt-main-fixture",
    )
    path = tmp_path / "gcp-digital-bi-dbt.json"
    path.write_text(json.dumps(catalog), encoding="utf-8")
    monkeypatch.setattr(lineage, "LINEAGE_CATALOG_DIR", tmp_path)

    tables = lineage.list_lineage_catalog_tables(
        project_id="gcp-digital-bi-dev",
        dataset_id="raw_erp",
    )

    assert [(table.table_id, table.table_type) for table in tables] == [
        ("orders", "TABLE"),
    ]


def test_duplicate_dbt_source_names_resolve_as_one_physical_relation(
    monkeypatch,
    tmp_path,
) -> None:
    manifest = _manifest()
    manifest["sources"]["source.bi_dev.legacy_erp.orders"] = {
        "resource_type": "source",
        "database": "gcp-digital-bi",
        "schema": "raw_erp",
        "identifier": "orders",
        "name": "legacy_orders",
        "description": "The same ERP extract under a legacy dbt source name",
    }
    catalog = build_dbt_lineage_catalog(
        manifest,
        catalog_id="gcp-digital-bi-dbt",
        source_version="dbt-main-fixture",
    )
    path = tmp_path / "gcp-digital-bi-dbt.json"
    path.write_text(json.dumps(catalog), encoding="utf-8")
    monkeypatch.setattr(lineage, "LINEAGE_CATALOG_DIR", tmp_path)

    graph = lineage.get_table_lineage(
        project_id="gcp-digital-bi",
        dataset_id="raw_erp",
        table_id="orders",
    )

    assert graph.status == "available"
    assert graph.direction == "downstream"
    assert (
        sum(node.dataset_id == "raw_erp" and node.table_id == "orders" for node in graph.nodes)
        == 1
    )


def test_deployed_catalog_resolves_a_known_model_lineage() -> None:
    graph = lineage.get_table_lineage(
        project_id="gcp-digital-bi",
        dataset_id="model_tidbcloud",
        table_id="dws_cloud_clusters_provider_distribution_v2",
    )

    assert graph.status == "available"
    assert graph.source_version == "dbt-main-5139557ce1276adb5b07d740d5a71ed8cfe636b2"
    assert any(node.is_root and node.node_type == "source" for node in graph.nodes)
    assert graph.nodes[-1].is_target is True


def test_deployed_metadb_catalog_exposes_source_reference_counts() -> None:
    tables = lineage.list_lineage_catalog_tables(
        project_id="gcp-digital-bi",
        dataset_id="metadb",
    )

    references = {table.table_id: table.downstream_reference_count for table in tables}
    assert references["cluster_qps"] == 1
    assert references["dbaas_backups"] == 1
    assert references["billing_packages"] == 0


def test_deployed_metadb_catalog_marks_source_tables_as_downstream_lineage() -> None:
    tables = lineage.list_lineage_catalog_tables(
        project_id="gcp-digital-bi",
        dataset_id="metadb",
    )

    directions = {table.table_id: table.lineage_direction for table in tables}
    assert directions["cluster_qps"] == "downstream"
    assert directions["billing_packages"] == "downstream"


def test_missing_source_dataset_is_added_as_requestable_owner_metadata() -> None:
    catalog = build_dbt_lineage_catalog(
        _manifest(),
        catalog_id="gcp-digital-bi-dbt",
        source_version="dbt-main-fixture",
    )
    catalog["nodes"].append(
        {
            "node_id": "source.bi_dev.unused.audit_log",
            "project_id": "gcp-digital-bi",
            "dataset_id": "unused_audit",
            "table_id": "audit_log",
            "node_type": "source",
            "materialization": "source",
            "description": "Declared but not referenced by a dbt dependency",
        }
    )
    inventory = {
        "metadata": {"inventory": "fixture"},
        "datasets": [
            {
                "project_id": "gcp-digital-bi",
                "dataset_id": "dws_orders",
                "owner_email": "explicit.owner@pingcap.com",
                "requestable": True,
                "status": "active",
            }
        ],
    }

    projected, additions = sync_lineage_root_datasets(inventory, catalog)

    assert [(item.project_id, item.dataset_id, item.owner_email) for item in additions] == [
        ("gcp-digital-bi", "raw_erp", "xiaopeng.zhu@pingcap.com"),
    ]
    assert not any(row["dataset_id"] == "unused_audit" for row in projected["datasets"])
    assert inventory["datasets"][0]["owner_email"] == "explicit.owner@pingcap.com"
    added = next(row for row in projected["datasets"] if row["dataset_id"] == "raw_erp")
    assert added["requestable"] is True
    assert added["status"] == "active"
    assert added["labels"]["lineage_root"] == "true"
    assert added["labels"]["cataloged_by"] == "dbt_lineage"
    assert added["labels"]["dbt_generated"] == "false"
    assert "managed_by" not in added["labels"]
    assert projected["metadata"]["lineage_root_dataset_count"] == 1
    assert projected["metadata"]["requestable_count"] == 2


def test_source_dataset_does_not_expand_or_promote_without_dbt_target() -> None:
    catalog = build_dbt_lineage_catalog(
        _manifest(),
        catalog_id="gcp-digital-bi-dbt",
        source_version="dbt-main-fixture",
    )
    inventory = {
        "metadata": {"inventory": "fixture", "requestable_count": 0},
        "datasets": [
            {
                "project_id": "gcp-digital-bi",
                "dataset_id": "raw_erp",
                "owner_email": "erp.owner@pingcap.com",
                "requestable": False,
                "status": "active",
            }
        ],
    }

    projected, additions = sync_lineage_root_datasets(inventory, catalog)

    assert additions == ()
    assert projected["datasets"][0]["owner_email"] == "erp.owner@pingcap.com"
    assert projected["datasets"][0]["requestable"] is False
    assert projected["metadata"]["requestable_count"] == 0
    assert inventory["datasets"][0]["requestable"] is False


def test_existing_root_is_promoted_when_reachable_from_requestable_dbt_target() -> None:
    catalog = build_dbt_lineage_catalog(
        _manifest(),
        catalog_id="gcp-digital-bi-dbt",
        source_version="dbt-main-fixture",
    )
    inventory = {
        "metadata": {"inventory": "fixture", "requestable_count": 1},
        "datasets": [
            {
                "project_id": "gcp-digital-bi",
                "dataset_id": "dws_orders",
                "owner_email": "serving.owner@pingcap.com",
                "requestable": True,
                "status": "active",
            },
            {
                "project_id": "gcp-digital-bi",
                "dataset_id": "raw_erp",
                "owner_email": "erp.owner@pingcap.com",
                "requestable": False,
                "status": "active",
            },
        ],
    }

    projected, additions = sync_lineage_root_datasets(inventory, catalog)

    assert additions == ()
    by_dataset = {row["dataset_id"]: row for row in projected["datasets"]}
    assert by_dataset["raw_erp"]["owner_email"] == "erp.owner@pingcap.com"
    assert by_dataset["raw_erp"]["requestable"] is True
    assert projected["metadata"]["requestable_count"] == 2


def test_unreachable_source_is_not_projected_and_stale_derived_root_is_removed() -> None:
    manifest = _manifest()
    manifest["nodes"]["model.bi_dev.orphan_stage"] = {
        "resource_type": "model",
        "database": "gcp-digital-bi",
        "schema": "stg_orphan",
        "alias": "orphan_stage",
        "config": {"materialized": "view"},
    }
    manifest["sources"]["source.bi_dev.orphan.events"] = {
        "resource_type": "source",
        "database": "gcp-digital-bi",
        "schema": "raw_orphan",
        "identifier": "events",
        "name": "events",
    }
    manifest["parent_map"]["model.bi_dev.orphan_stage"] = [
        "source.bi_dev.orphan.events",
    ]
    catalog = build_dbt_lineage_catalog(
        manifest,
        catalog_id="gcp-digital-bi-dbt",
        source_version="dbt-main-fixture",
    )
    inventory = {
        "metadata": {"inventory": "fixture"},
        "datasets": [
            {
                "project_id": "gcp-digital-bi",
                "dataset_id": "dws_orders",
                "owner_email": "serving.owner@pingcap.com",
                "requestable": True,
                "status": "active",
            },
            {
                "project_id": "gcp-digital-bi",
                "dataset_id": "raw_stale",
                "owner_email": "xiaopeng.zhu@pingcap.com",
                "requestable": True,
                "status": "active",
                "labels": {
                    "cataloged_by": "dbt_lineage",
                    "lineage_root": "true",
                    "dbt_generated": "false",
                },
            },
        ],
    }

    projected, additions = sync_lineage_root_datasets(inventory, catalog)

    dataset_ids = {row["dataset_id"] for row in projected["datasets"]}
    assert dataset_ids == {"dws_orders", "raw_erp"}
    assert "raw_orphan" not in dataset_ids
    assert "raw_stale" not in dataset_ids
    assert [(item.dataset_id, item.owner_email) for item in additions] == [
        ("raw_erp", "xiaopeng.zhu@pingcap.com"),
    ]


def test_catalog_seed_automatically_assembles_referenced_source_roots(tmp_path) -> None:
    catalog = build_dbt_lineage_catalog(
        _manifest(),
        catalog_id="gcp-digital-bi-dbt",
        source_version="dbt-main-fixture",
    )
    inventory_path = tmp_path / "inventory.json"
    catalog_path = tmp_path / "lineage.json"
    inventory_path.write_text(
        json.dumps(
            {
                "metadata": {"inventory": "fixture"},
                "datasets": [
                    {
                        "project_id": "gcp-digital-bi",
                        "dataset_id": "dws_orders",
                        "owner_email": "explicit.owner@pingcap.com",
                        "requestable": True,
                        "status": "active",
                    }
                ],
            }
        ),
        encoding="utf-8",
    )
    catalog_path.write_text(json.dumps(catalog), encoding="utf-8")

    rows = seed_script._load_inventory(
        inventory_path,
        lineage_catalog_path=catalog_path,
    )

    by_dataset = {str(row["dataset_id"]): row for row in rows}
    assert by_dataset["raw_erp"]["owner_email"] == "xiaopeng.zhu@pingcap.com"
    assert by_dataset["raw_erp"]["requestable"] is True
    assert by_dataset["raw_erp"]["labels"] == {
        "inventory": "gcp-digital-bi-dbt",
        "cataloged_by": "dbt_lineage",
        "dbt_generated": "false",
        "layer": "source",
        "lineage_root": "true",
    }
    assert by_dataset["dws_orders"]["owner_email"] == ("explicit.owner@pingcap.com")


@pytest.mark.asyncio
async def test_catalog_seed_can_fallback_unretained_staging_owner(
    monkeypatch,
) -> None:
    fallback_person_id = "11111111-1111-1111-1111-111111111111"

    async def _owner_person_id(owner_email: str) -> str | None:
        if owner_email == "xiaopeng.zhu@pingcap.com":
            return fallback_person_id
        return None

    monkeypatch.setattr(seed_script, "_owner_person_id", _owner_person_id)

    entries = await seed_script._catalog_entries(
        [
            {
                "project_id": "gcp-digital-bi",
                "dataset_id": "metadb_new",
                "owner_email": "xukai@pingcap.com",
            }
        ],
        source_version="staging-release-fixture",
        allow_unresolved=False,
        unresolved_owner_fallback_email="xiaopeng.zhu@pingcap.com",
    )

    assert entries[0]["owner_person_id"] == fallback_person_id
    assert entries[0]["owner_resolution_status"] == "resolved"


@pytest.mark.asyncio
async def test_catalog_seed_rejects_unknown_fallback_owner(monkeypatch) -> None:
    async def _owner_person_id(_owner_email: str) -> str | None:
        return None

    monkeypatch.setattr(seed_script, "_owner_person_id", _owner_person_id)

    with pytest.raises(SystemExit) as exc_info:
        await seed_script._catalog_entries(
            [
                {
                    "project_id": "gcp-digital-bi",
                    "dataset_id": "metadb_new",
                    "owner_email": "xukai@pingcap.com",
                }
            ],
            source_version="staging-release-fixture",
            allow_unresolved=True,
            unresolved_owner_fallback_email="missing@pingcap.com",
        )

    assert exc_info.value.code == 4