from __future__ import annotations

import os
from pathlib import Path
import subprocess
import sys


REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT))

from orchestration.airflow.engine.contract_assets import AirflowAssetSpec
from orchestration.airflow.engine.dbt_assets import build_airflow_asset, load_dbt_asset_config


DBT_REVREC_CONFIG_PATH = REPO_ROOT / "orchestration" / "airflow" / "sources" / "dbt-revrec.yaml"


def test_dbt_revrec_config_binds_dbt_nodes_to_contract_assets() -> None:
    config = load_dbt_asset_config(DBT_REVREC_CONFIG_PATH, repo_root=REPO_ROOT)

    assert config.select == ("+salesforce_org_monthly_revrec", "+salesforce_order_daily_view")

    order_daily = config.node_bindings["salesforce_order_daily_view"]
    assert order_daily.contract_id == "finance.salesforce_order_daily_view"
    assert order_daily.outlet.name == "finance.salesforce_order_daily_view"
    assert order_daily.outlet.uri == "x-datalake://finance/salesforce_order_daily_view"
    assert {asset.name for asset in order_daily.inlets} == {
        "core.date_spine",
        "crm.nexuscrm.stg_salesforce_order_item",
        "crm.nexuscrm.stg_salesforce_product",
        "crm.salesforce_orders",
        "finance.salesforce_revrec",
    }

    payment_seed = config.node_bindings["payment_type_mapping"]
    assert payment_seed.resource_type == "seed"
    assert payment_seed.contract_id == "seed.finance.payment_type_mapping"
    assert payment_seed.outlet.uri == "x-datalake://finance/seed/finance/payment_type_mapping"

    org_revrec = config.node_bindings["salesforce_org_monthly_revrec"]
    assert "seed.finance.payment_type_mapping" in {asset.name for asset in org_revrec.inlets}


def test_dbt_revrec_schedule_uses_only_external_roots_of_the_selected_slice() -> None:
    config = load_dbt_asset_config(DBT_REVREC_CONFIG_PATH, repo_root=REPO_ROOT)
    schedule_names = {asset.name for asset in config.schedule_assets}

    assert "finance.salesforce_order_daily_view" not in schedule_names
    assert "finance.salesforce_revrec" not in schedule_names
    assert "seed.googlesheets.org_monthly_bill_confirmation_for_revrec" not in schedule_names

    assert {
        "billing.metadb.billing2_invoices_snapshot",
        "cloud_platform.cloud_tenants.cloud_tenants_snapshot",
        "crm.nexuscrm.account_snapshot",
        "crm.nexuscrm.order_product__c_snapshot",
        "crm.nexuscrm.order_snapshot",
        "crm.nexuscrm.opportunity_snapshot",
        "crm.nexuscrm.org__c_snapshot",
        "crm.nexuscrm.product2_snapshot",
        "finance.data_foundation_finance.salesforce_account_snapshot",
        "finance.salesforce.revenue_c_snapshot",
    }.issubset(schedule_names)


def test_dbt_revrec_dag_registers_cosmos_with_manifest_and_contract_converters() -> None:
    dag_source = (REPO_ROOT / "orchestration" / "airflow" / "dags" / "dbt_revrec_assets.py").read_text()
    shim_source = (REPO_ROOT / "orchestration" / "airflow" / "bin" / "dbt-databend-shim").read_text()

    assert "DbtTaskGroup(" in dag_source
    assert "load_method=LoadMode.DBT_MANIFEST" in dag_source
    assert "build_contract_asset_converters(DBT_REVREC.node_bindings)" in dag_source
    assert "emit_datasets=False" in dag_source
    assert "dbt_executable_path=DBT_REVREC.dbt_executable_path" in dag_source
    assert "transform/dbt/engine/run_dbt.py" in shim_source


def test_dbt_revrec_dag_renders_contract_assets_and_dbt_tests() -> None:
    import pytest

    airflow_models = pytest.importorskip("airflow.models")

    bag = airflow_models.DagBag(
        dag_folder=str(REPO_ROOT / "orchestration" / "airflow" / "dags" / "dbt_revrec_assets.py"),
        include_examples=False,
        safe_mode=False,
    )
    assert bag.import_errors == {}

    dag = bag.dags["orchestration_dbt_revrec_assets"]
    order_daily_run = dag.get_task("dbt_revrec.salesforce_order_daily_view.run")
    order_daily_test = dag.get_task("dbt_revrec.salesforce_order_daily_view.test")
    org_revrec_run = dag.get_task("dbt_revrec.salesforce_org_monthly_revrec.run")
    org_revrec_test = dag.get_task("dbt_revrec.salesforce_org_monthly_revrec.test")
    payment_seed = dag.get_task("dbt_revrec.payment_type_mapping_seed")
    google_seed = dag.get_task("dbt_revrec.org_monthly_bill_confirmation_for_revrec_seed")

    assert {asset.name for asset in order_daily_run.outlets} == {"finance.salesforce_order_daily_view"}
    assert {asset.name for asset in org_revrec_run.outlets} == {"finance.salesforce_org_monthly_revrec"}
    assert order_daily_test.upstream_task_ids == {"dbt_revrec.salesforce_order_daily_view.run"}
    assert org_revrec_test.upstream_task_ids == {"dbt_revrec.salesforce_org_monthly_revrec.run"}
    assert payment_seed.__class__.__name__ == "DbtSeedLocalOperator"
    assert getattr(payment_seed, "select", None) == "fqn:pingcap_lakehouse.payment_type_mapping"
    assert {asset.name for asset in payment_seed.outlets} == {"seed.finance.payment_type_mapping"}
    assert google_seed.__class__.__name__ == "DbtSeedLocalOperator"
    assert getattr(google_seed, "select", None) == (
        "fqn:pingcap_lakehouse.org_monthly_bill_confirmation_for_revrec"
    )
    assert {asset.name for asset in google_seed.outlets} == {
        "seed.googlesheets.org_monthly_bill_confirmation_for_revrec"
    }


def test_dbt_databend_shim_does_not_require_root_env_file(tmp_path: Path) -> None:
    shim = REPO_ROOT / "orchestration" / "airflow" / "bin" / "dbt-databend-shim"
    fake_bin = tmp_path / "bin"
    fake_bin.mkdir()
    fake_uv = fake_bin / "uv"
    fake_uv.write_text(
        "#!/usr/bin/env bash\n"
        "printf 'UV_PROJECT_ENVIRONMENT=%s\\n' \"$UV_PROJECT_ENVIRONMENT\"\n"
        "printf 'PYTHONUNBUFFERED=%s\\n' \"$PYTHONUNBUFFERED\"\n"
        "printf '%s\\n' \"$@\"\n"
    )
    fake_uv.chmod(0o755)

    runtime_root = tmp_path / "runtime-root"
    runtime_root.mkdir()
    env = {
        **os.environ,
        "DATALAKE_AIRFLOW_REPO_ROOT": str(runtime_root),
        "PATH": f"{fake_bin}{os.pathsep}{os.environ['PATH']}",
    }

    without_env = subprocess.run(
        [str(shim), "--version"],
        check=True,
        capture_output=True,
        env=env,
        text=True,
    ).stdout.splitlines()
    assert "UV_PROJECT_ENVIRONMENT=/opt/airflow/.venvs/pingcap-datalake-transform" in without_env
    assert "PYTHONUNBUFFERED=1" in without_env
    assert "--env-file" not in without_env

    (runtime_root / ".env").write_text("DATABEND_HOST=example.invalid\n")
    with_env = subprocess.run(
        [str(shim), "--version"],
        check=True,
        capture_output=True,
        env=env,
        text=True,
    ).stdout.splitlines()
    assert "--env-file" in with_env
    assert str(runtime_root / ".env") in with_env


def test_build_airflow_asset_preserves_contract_group() -> None:
    import pytest

    pytest.importorskip("airflow.sdk")

    asset = build_airflow_asset(
        AirflowAssetSpec(
            name="finance.salesforce_order_daily_view",
            uri="x-datalake://finance/salesforce_order_daily_view",
            group="curated",
            extra={"contract_id": "finance.salesforce_order_daily_view"},
        )
    )

    assert asset.group == "curated"