from __future__ import annotations

import pathlib
import sys
from decimal import Decimal

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

from transform.dbt.parity.salesforce_order_daily_view import compare, has_failures


def _row(order_id: str, amount: str, status: str = "Activated") -> dict:
    return {
        "date_day": "2026-05-31",
        "order_id": order_id,
        "opportunity_id": "opp",
        "account_id": "acct",
        "accumulate_revenue_to_org_tenant_id": "tenant",
        "order_name": "order",
        "order_number": "ON-1",
        "order_status": status,
        "reason": None,
        "order_product_family": "TiDB Cloud",
        "order_start_date": "2026-05-01",
        "order_end_date": "2026-05-31",
        "write_off_date": None,
        "order_is_terminated": 0,
        "order_termination_date": None,
        "real_order_termination_date": None,
        "total_fees": "10.00",
        "billing_country_code": "US",
        "product_family": "TiDB Cloud",
        "tax_rate": "0",
        "opportunity_currency": "USD",
        "opportunity_exchange_rate": "1",
        "is_commitment_plan": False,
        "order_service_start_date": "2026-05-01",
        "order_service_end_date": "2026-05-31",
        "accumulated_revenue": amount,
        "accumulated_credit_amount": "0",
        "accumulated_revenue_before_adjust": amount,
        "order_net_fees": "10.00",
    }


def test_order_daily_view_parity_allows_numeric_tolerance() -> None:
    bq = _row("o1", "100.00")
    databend = _row("o1", "100.004")
    bq["order_is_terminated"] = "false"
    databend["order_is_terminated"] = 0

    summary = compare(
        [bq],
        [databend],
        numeric_tolerance=Decimal("0.01"),
        sample_limit=3,
    )

    assert summary["row_counts"] == {"bigquery": 1, "databend": 1, "common": 1}
    assert "billing_country_code" in summary["excluded_columns"]
    assert summary["fields"]["accumulated_revenue"]["daily_mismatches"] == 0
    assert summary["fields"]["order_is_terminated"]["daily_mismatches"] == 0
    assert not has_failures(summary)


def test_order_daily_view_parity_reports_daily_and_order_mismatch_counts() -> None:
    summary = compare(
        [_row("o1", "100.00"), _row("o2", "50.00")],
        [_row("o1", "100.25"), _row("o2", "50.00", status="Draft")],
        numeric_tolerance=Decimal("0.01"),
        sample_limit=1,
    )

    assert summary["fields"]["accumulated_revenue"]["daily_mismatches"] == 1
    assert summary["fields"]["accumulated_revenue"]["order_mismatches"] == 1
    assert summary["fields"]["order_status"]["daily_mismatches"] == 1
    assert summary["fields"]["order_status"]["examples"] == [
        {"date_day": "2026-05-31", "order_id": "o2", "bq": "Activated", "databend": "Draft"}
    ]
    assert has_failures(summary)


def test_order_daily_view_parity_fails_on_empty_outputs() -> None:
    summary = compare([], [], numeric_tolerance=Decimal("0.01"), sample_limit=1)

    assert summary["row_counts"] == {"bigquery": 0, "databend": 0, "common": 0}
    assert has_failures(summary)


if __name__ == "__main__":
    cases = [value for name, value in sorted(globals().items()) if name.startswith("test_") and callable(value)]
    for case in cases:
        case()
        print(f"ok  {case.__name__}")
    print(f"\n{len(cases)} passed")