from __future__ import annotations

from pathlib import Path
import sys


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

from orchestration.airflow.engine.contract_assets import (
    compile_contract_assets,
    compile_repo_contract_assets,
    stale_asset_names,
)


def test_contract_asset_compiler_projects_monthly_revrec_outputs() -> None:
    graph = compile_repo_contract_assets(REPO_ROOT)

    orders = graph.assets["finance.monthly_revrec_orders"]

    assert orders.airflow.name == "finance.monthly_revrec_orders"
    assert orders.airflow.uri == "x-datalake://finance/monthly_revrec_orders"
    assert orders.airflow.group == "curated"
    assert orders.contract_version == "1.0.0"
    assert orders.produced_by_kind == "python"
    assert orders.produced_by_ref == "monthly_cloud_billing_revrec"
    assert orders.asset_ref == "transform_python_dwh.monthly_revrec_orders"
    assert orders.airflow.extra["contract_path"] == (
        "transform/python/contracts/monthly_revrec_orders.curated.contract.yaml"
    )
    assert orders.airflow.extra["freshness_max_age"] == "P1D"

    cloud_revenue = graph.assets["finance.cloud_revenue"]
    assert cloud_revenue.airflow.extra["freshness_degrade_after"] == "PT720H"
    assert cloud_revenue.airflow.extra["freshness_timezone"] == "UTC"
    assert cloud_revenue.airflow.extra["freshness_schedule_ref"] == (
        "ingestion/extract/tidbcloud/sources/data_foundation_finance/source.yaml#schedule"
    )


def test_contract_asset_compiler_preserves_declared_lineage_edges() -> None:
    graph = compile_repo_contract_assets(REPO_ROOT)

    order_upstreams = {
        edge.upstream_ref: edge
        for edge in graph.edges
        if edge.downstream_ref == "finance.monthly_revrec_orders"
    }

    assert set(order_upstreams) == {
        "finance.salesforce_org_monthly_revrec",
        "finance.salesforce_order_daily_view",
    }
    for upstream_ref, edge in order_upstreams.items():
        assert edge.upstream_kind == "data_product"
        assert edge.required is True
        assert edge.upstream_asset_name == upstream_ref
        assert edge.downstream_asset_name == "finance.monthly_revrec_orders"


def test_contract_asset_compiler_models_missing_contract_backed_upstreams_as_external_assets() -> None:
    graph = compile_repo_contract_assets(REPO_ROOT)

    cloud_tenants = graph.external_assets["cloud_platform.cloud_tenants.cloud_tenants_snapshot"]

    assert cloud_tenants.airflow.name == "cloud_platform.cloud_tenants.cloud_tenants_snapshot"
    assert cloud_tenants.airflow.group == "external"
    assert cloud_tenants.airflow.uri == (
        "x-datalake://external/data_product/cloud_platform/cloud_tenants/cloud_tenants_snapshot"
    )
    assert all(
        edge.upstream_asset_name is not None
        for edge in graph.edges
        if edge.upstream_kind in {"data_product", "source"}
    )


def test_contract_asset_compiler_projects_dbt_seed_source_contracts() -> None:
    graph = compile_repo_contract_assets(REPO_ROOT)

    google_seed = graph.assets["seed.googlesheets.org_monthly_bill_confirmation_for_revrec"]
    payment_seed = graph.assets["seed.finance.payment_type_mapping"]

    assert google_seed.contract_kind == "source"
    assert google_seed.produced_by_kind == "dbt"
    assert google_seed.produced_by_ref == "org_monthly_bill_confirmation_for_revrec"
    assert google_seed.airflow.group == "source"
    assert google_seed.airflow.uri == (
        "x-datalake://finance/seed/googlesheets/org_monthly_bill_confirmation_for_revrec"
    )
    assert payment_seed.contract_kind == "source"
    assert payment_seed.produced_by_kind == "dbt"
    assert payment_seed.produced_by_ref == "payment_type_mapping"


def test_contract_asset_compiler_projects_source_contracts_without_layers() -> None:
    graph = compile_repo_contract_assets(REPO_ROOT)

    source_asset = graph.assets["source.salesforce.revenue_c_external"]

    assert source_asset.contract_kind == "source"
    assert source_asset.layer is None
    assert source_asset.airflow.group == "source"
    assert source_asset.airflow.uri == "x-datalake://finance/source/salesforce/revenue_c_external"
    assert "layer" not in source_asset.airflow.extra


def test_contract_asset_manifest_is_deterministic_and_cleanup_can_find_stale_assets() -> None:
    graph = compile_repo_contract_assets(REPO_ROOT)
    manifest = graph.to_manifest()

    asset_names = [asset["airflow"]["name"] for asset in manifest["assets"]]
    edge_pairs = [
        (edge["downstream_ref"], edge["upstream_kind"], edge["upstream_ref"])
        for edge in manifest["edges"]
    ]

    assert asset_names == sorted(asset_names)
    assert edge_pairs == sorted(edge_pairs)
    assert stale_asset_names(
        [
            "finance.monthly_revrec_orders",
            "seed.googlesheets.org_monthly_bill_confirmation_for_revrec",
            "finance.retired_product",
        ],
        graph,
    ) == {"finance.retired_product"}


def test_contract_asset_compiler_rejects_contracts_outside_v0_profile(tmp_path: Path) -> None:
    contract_path = tmp_path / "multi_asset.contract.yaml"
    contract_path.write_text(
        """
schema_version: 0.1.0
kind: product
id: finance.multi_asset_product
name: multi_asset_product
version: 0.1.0
layer: curated
owner:
  type: team
  name: data-platform
domain:
  name: finance
purpose: test contract outside the v0 orchestration profile
usage:
  intended:
    - test
  prohibited:
    - test
assets:
  - id: first
    kind: table
    ref: first
    produced_by:
      kind: python
      ref: job
  - id: second
    kind: table
    ref: second
    produced_by:
      kind: python
      ref: job
schema:
  kind: relational
  columns:
    - name: id
      type: VARCHAR
freshness:
  max_age: P1D
tests:
  - id: smoke
    type: sql
    severity: informational
    description: smoke
    sql: select 1
governance:
  classifications:
    - financial
agent_permissions:
  default: deny
compatibility:
  change_policy: semver
semantics:
  terms:
    - name: smoke
      description: smoke
""".strip()
    )

    try:
        compile_contract_assets([contract_path], repo_root=tmp_path)
    except ValueError as error:
        assert "v0 orchestration compilation profile" in str(error)
    else:
        raise AssertionError("compiler should reject multi-asset contracts in the v0 profile")


def test_contract_asset_compiler_rejects_missing_producer_in_v0_profile(tmp_path: Path) -> None:
    contract_path = tmp_path / "missing_producer.contract.yaml"
    contract_path.write_text(
        """
schema_version: 0.1.0
kind: product
id: finance.missing_producer_product
name: missing_producer_product
version: 0.1.0
layer: curated
owner:
  type: team
  name: data-platform
domain:
  name: finance
purpose: test contract outside the v0 orchestration profile
usage:
  intended:
    - test
  prohibited:
    - test
assets:
  - id: missing_producer
    kind: table
    ref: missing_producer
schema:
  kind: relational
  columns:
    - name: id
      type: VARCHAR
freshness:
  max_age: P1D
tests:
  - id: smoke
    type: sql
    severity: informational
    description: smoke
    sql: select 1
governance:
  classifications:
    - financial
agent_permissions:
  default: deny
compatibility:
  change_policy: semver
semantics:
  terms:
    - name: smoke
      description: smoke
""".strip()
    )

    try:
        compile_contract_assets([contract_path], repo_root=tmp_path)
    except ValueError as error:
        assert "produced_by" in str(error)
        assert "v0 orchestration compilation profile" in str(error)
    else:
        raise AssertionError("compiler should reject missing produced_by in the v0 profile")