from __future__ import annotations import pathlib import sys from datetime import date from decimal import Decimal from types import SimpleNamespace import pytest pd = pytest.importorskip("pandas") sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) import transform.python.parity.monthly_cloud_billing_revrec as parity from transform.python.parity.monthly_cloud_billing_revrec import ( CaseSpec, build_summary, case_summary, case_has_failures, coerce_inputs, compare_frames, common_values, discover_common_cases, fetch_bigquery_query, filter_inputs, format_summary_table, has_failures, parse_bool, resolve_selector, run_case, value_gaps, ) def test_parse_bool_handles_bigquery_json_strings() -> None: assert parse_bool("true") is True assert parse_bool("false") is False assert parse_bool("1") is True assert parse_bool("0") is False assert parse_bool(None) is None def test_coerce_inputs_parses_order_and_billing_booleans() -> None: orders = pd.DataFrame( [ { "date_day": "2026-05-21", "order_service_start_date": "2026-04-01", "order_is_terminated": "false", "is_commitment_plan": "true", } ] ) billing = pd.DataFrame( [ { "billing_start_date": "2026-04-01", "billing_end_date": "2026-04-30", "is_salesforce_customer": "false", "total_amount": "10.00", } ] ) coerced_orders, coerced_billing = coerce_inputs(orders, billing) assert not bool(coerced_orders.loc[0, "order_is_terminated"]) assert bool(coerced_orders.loc[0, "is_commitment_plan"]) assert not bool(coerced_billing.loc[0, "is_salesforce_customer"]) def test_output_numeric_columns_follow_revenue_categories() -> None: expected = { column for category in parity.job.REVENUE_CATEGORIES for column in (f"{category}_revenue_amount", f"adjusted_{category}_amount") } assert expected <= parity.OUTPUT_NUMERIC assert parity.values_match("lake_revenue_amount", "1.004", "1.00", Decimal("0.01")) def test_monthly_revrec_parity_fails_on_empty_outputs() -> None: summary = { "collections": compare_frames( pd.DataFrame(columns=["bill_month", "adjust_tenant_id"]), pd.DataFrame(columns=["bill_month", "adjust_tenant_id"]), key_columns=("bill_month", "adjust_tenant_id"), tolerance=Decimal("0.01"), sample_limit=1, ), "orders": compare_frames( pd.DataFrame(columns=["tenant_id", "bill_month", "order_number"]), pd.DataFrame(columns=["tenant_id", "bill_month", "order_number"]), key_columns=("tenant_id", "bill_month", "order_number"), tolerance=Decimal("0.01"), sample_limit=1, ), } assert summary["collections"]["row_counts"] == {"bigquery": 0, "databend": 0, "common": 0} assert summary["orders"]["row_counts"] == {"bigquery": 0, "databend": 0, "common": 0} assert has_failures(summary) assert case_has_failures(summary, allow_empty_matches=True) def _empty_output_case( *, revenue: str = "recognised", bigquery_billing_rows: int = 0, databend_billing_rows: int = 0, ) -> dict: return { "billing_month": "2025-01", "order_date": "2026-06-01", "revenue": revenue, "input_rows": { "bigquery": {"billing": bigquery_billing_rows, "orders": 10}, "databend": {"billing": databend_billing_rows, "orders": 10}, }, "collections": compare_frames( pd.DataFrame(columns=["bill_month", "adjust_tenant_id"]), pd.DataFrame(columns=["bill_month", "adjust_tenant_id"]), key_columns=("bill_month", "adjust_tenant_id"), tolerance=Decimal("0.01"), sample_limit=1, ), "orders": compare_frames( pd.DataFrame(columns=["tenant_id", "bill_month", "order_number"]), pd.DataFrame(columns=["tenant_id", "bill_month", "order_number"]), key_columns=("tenant_id", "bill_month", "order_number"), tolerance=Decimal("0.01"), sample_limit=1, ), } def test_empty_match_allowance_only_applies_to_recognised_zero_billing_inputs() -> None: recognised_empty = _empty_output_case() predicted_empty = _empty_output_case(revenue="predicted") nonempty_input_empty = _empty_output_case(bigquery_billing_rows=1, databend_billing_rows=1) assert not case_has_failures(recognised_empty, allow_empty_matches=True) assert case_has_failures(predicted_empty, allow_empty_matches=True) assert case_has_failures(nonempty_input_empty, allow_empty_matches=True) summary = case_summary(recognised_empty, allow_empty_matches=True) table = format_summary_table([summary]) assert summary["empty_match"] is True assert summary["failed"] is False assert "PASS_EMPTY" in table def test_common_case_selectors_use_intersection_and_latest() -> None: available = common_values( ["2026-02", "2026-04", "2026-01"], ["2026-03", "2026-04", "2026-01"], ) assert available == ["2026-01", "2026-04"] assert resolve_selector("latest", available, "billing_month") == ["2026-04"] assert resolve_selector("all", available, "billing_month") == ["2026-01", "2026-04"] assert resolve_selector("2026-01", available, "billing_month") == ["2026-01"] assert value_gaps(["2026-01", "2026-03"], ["2026-01", "2026-02"]) == ["2026-03"] def test_selector_rejects_non_common_values() -> None: try: resolve_selector("2026-05", ["2026-03", "2026-04"], "billing_month") except RuntimeError as exc: assert "not present in both BigQuery and Databend" in str(exc) else: raise AssertionError("expected unavailable selector to fail") def test_bigquery_fetch_counts_exact_query_and_rejects_truncation(monkeypatch: pytest.MonkeyPatch) -> None: calls = [] def fake_bq_json(query: str, project: str, *, max_rows: int) -> list[dict]: calls.append({"query": query, "project": project, "max_rows": max_rows}) if "COUNT(*)" in query: return [{"row_count": "2"}] return [{"value": "one"}] monkeypatch.setattr(parity, "bq_json", fake_bq_json) with pytest.raises(RuntimeError, match="parity fetch was truncated"): fetch_bigquery_query( "project", "SELECT * FROM dataset.table ORDER BY id", count_query="SELECT * FROM dataset.table", label="test input", ) assert calls == [ { "query": "SELECT COUNT(*) AS row_count FROM (SELECT * FROM dataset.table) AS parity_source", "project": "project", "max_rows": 1, }, { "query": "SELECT * FROM dataset.table ORDER BY id", "project": "project", "max_rows": 3, }, ] def test_discover_common_cases_uses_common_billing_months_and_order_dates( monkeypatch: pytest.MonkeyPatch, ) -> None: args = SimpleNamespace( bigquery_project="project", bigquery_dataset="dataset", databend_input_db="db", billing_month="all", order_date="latest", ) def fake_bq_values(project: str, dataset: str, table: str, column: str) -> list[str]: assert project == "project" assert dataset == "dataset" if table == "salesforce_org_monthly_revrec": return ["2026-01", "2026-02", "2026-04"] assert table == "salesforce_order_daily_view" assert column == "date_day" return ["2026-05-30", "2026-06-01"] def fake_db_values(input_db: str, table: str, column: str) -> list[str]: assert input_db == "db" if table == "salesforce_org_monthly_revrec": return ["2026-02", "2026-03", "2026-04"] assert table == "salesforce_order_daily_view" assert column == "date_day" return ["2026-05-31", "2026-06-01"] monkeypatch.setattr(parity, "fetch_bigquery_values", fake_bq_values) monkeypatch.setattr(parity, "fetch_databend_values", fake_db_values) cases, discovery = discover_common_cases(args) assert cases == [ CaseSpec(billing_month="2026-02", order_date="2026-06-01"), CaseSpec(billing_month="2026-04", order_date="2026-06-01"), ] assert discovery["common_billing_months"] == ["2026-02", "2026-04"] assert discovery["common_order_dates"] == ["2026-06-01"] assert discovery["bigquery_only_billing_months"] == ["2026-01"] assert discovery["databend_only_billing_months"] == ["2026-03"] assert discovery["bigquery_only_order_dates"] == ["2026-05-30"] assert discovery["databend_only_order_dates"] == ["2026-05-31"] def test_filter_inputs_slices_batched_months_dates_and_revenue_modes() -> None: orders = pd.DataFrame( [ { "order_id": "selected", "date_day": date(2026, 6, 1), "order_service_start_date": date(2026, 1, 1), }, { "order_id": "wrong-date", "date_day": date(2026, 5, 31), "order_service_start_date": date(2026, 1, 1), }, { "order_id": "future-service", "date_day": date(2026, 6, 1), "order_service_start_date": date(2026, 7, 1), }, ] ) billing = pd.DataFrame( [ { "_row": "recognised-selected", "billing_month": "2026-04", "actual_revenue_recognition_date": date(2026, 5, 1), }, { "_row": "predicted-selected", "billing_month": "2026-04", "actual_revenue_recognition_date": None, }, { "_row": "wrong-month", "billing_month": "2026-03", "actual_revenue_recognition_date": None, }, ] ) recognised_orders, recognised_billing = filter_inputs( orders, billing, billing_month="2026-04", order_date="2026-06-01", revenue="recognised", ) predicted_orders, predicted_billing = filter_inputs( orders, billing, billing_month="2026-04", order_date="2026-06-01", revenue="predicted", ) assert recognised_orders["order_id"].tolist() == ["selected"] assert recognised_billing["_row"].tolist() == ["recognised-selected"] assert predicted_orders["order_id"].tolist() == ["selected"] assert predicted_billing["_row"].tolist() == ["predicted-selected"] def test_run_case_slices_batched_inputs_before_processing(monkeypatch: pytest.MonkeyPatch) -> None: args = SimpleNamespace(numeric_tolerance="0.01", sample_limit=1) orders = pd.DataFrame( [ { "date_day": date(2026, 6, 1), "order_service_start_date": date(2026, 1, 1), "order_number": "selected-order", }, { "date_day": date(2026, 5, 31), "order_service_start_date": date(2026, 1, 1), "order_number": "wrong-date", }, ] ) billing = pd.DataFrame( [ { "billing_month": "2026-04", "actual_revenue_recognition_date": None, "tenant_id": "selected-tenant", }, { "billing_month": "2026-04", "actual_revenue_recognition_date": date(2026, 5, 1), "tenant_id": "recognised-tenant", }, { "billing_month": "2026-03", "actual_revenue_recognition_date": None, "tenant_id": "wrong-month", }, ] ) calls = [] def fake_process(*, df_orders: pd.DataFrame, df_billing: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]: calls.append((df_orders.copy(), df_billing.copy())) bill_month = df_billing.iloc[0]["billing_month"] tenant_id = df_billing.iloc[0]["tenant_id"] order_number = df_orders.iloc[0]["order_number"] return ( pd.DataFrame( [ { "bill_month": bill_month, "adjust_tenant_id": tenant_id, "total_amount": Decimal("1.00"), } ] ), pd.DataFrame( [ { "tenant_id": tenant_id, "bill_month": bill_month, "order_number": order_number, "revenue_amount": Decimal("1.00"), } ] ), ) monkeypatch.setattr(parity.job, "process", fake_process) result = run_case( args, CaseSpec(billing_month="2026-04", order_date="2026-06-01"), "predicted", bq_orders=orders, bq_billing=billing, db_orders=orders.copy(), db_billing=billing.copy(), ) assert result["input_rows"] == { "bigquery": {"billing": 1, "orders": 1}, "databend": {"billing": 1, "orders": 1}, } assert result["collections"]["row_counts"] == {"bigquery": 1, "databend": 1, "common": 1} assert result["orders"]["row_counts"] == {"bigquery": 1, "databend": 1, "common": 1} assert [call[0]["order_number"].tolist() for call in calls] == [["selected-order"], ["selected-order"]] assert [call[1]["tenant_id"].tolist() for call in calls] == [["selected-tenant"], ["selected-tenant"]] def test_case_summary_counts_missing_keys_and_field_mismatches() -> None: case = { "billing_month": "2026-04", "order_date": "2026-06-01", "revenue": "recognised", "collections": { "row_counts": {"bigquery": 1, "databend": 1, "common": 1}, "missing_keys": { "bigquery_only": {"count": 0, "examples": []}, "databend_only": {"count": 0, "examples": []}, }, "fields": {"total_amount": {"matches": 0, "mismatches": 1, "examples": []}}, }, "orders": { "row_counts": {"bigquery": 1, "databend": 0, "common": 0}, "missing_keys": { "bigquery_only": {"count": 1, "examples": [["tenant", "2026-04", "order"]]}, "databend_only": {"count": 0, "examples": []}, }, "fields": {}, }, } summary = case_summary(case) table = format_summary_table([summary]) assert summary["collections_missing_keys"] == 0 assert summary["collections_field_mismatches"] == 1 assert summary["orders_missing_keys"] == 1 assert summary["orders_field_mismatches"] == 0 assert summary["failed"] is True assert "FAIL" in table def test_summary_only_keeps_full_failing_case_details() -> None: case = { "billing_month": "2026-04", "order_date": "2026-06-01", "revenue": "predicted", "collections": { "row_counts": {"bigquery": 1, "databend": 1, "common": 1}, "missing_keys": { "bigquery_only": {"count": 0, "examples": []}, "databend_only": {"count": 0, "examples": []}, }, "fields": { "total_amount": { "matches": 0, "mismatches": 1, "examples": [{"key": ["2026-04", "tenant"], "bigquery": "1", "databend": "2"}], } }, }, "orders": { "row_counts": {"bigquery": 1, "databend": 1, "common": 1}, "missing_keys": { "bigquery_only": {"count": 0, "examples": []}, "databend_only": {"count": 0, "examples": []}, }, "fields": {}, }, } summary = build_summary( {"common_billing_months": ["2026-04"], "common_order_dates": ["2026-06-01"]}, [case], allow_empty_matches=False, summary_only=True, ) assert "cases" not in summary assert summary["case_summaries"][0]["failed"] is True assert summary["failing_cases"][0]["collections"]["fields"]["total_amount"]["examples"] if __name__ == "__main__": raise SystemExit(pytest.main([__file__]))