data-query-agent/tests/test_product_efficiency_formatter.py

import importlib.util
from pathlib import Path

import pandas as pd
import pytest


SCRIPT = (
    Path(__file__).parents[1]
    / ".agents"
    / "skills"
    / "odps-product-efficiency-report"
    / "scripts"
    / "format_report.py"
)
spec = importlib.util.spec_from_file_location("product_efficiency_formatter", SCRIPT)
assert spec and spec.loader
formatter = importlib.util.module_from_spec(spec)
spec.loader.exec_module(formatter)


CONFIG = {
    "experiment_buckets": list("012345678"),
    "control_buckets": list("9abcdef"),
}


def raw_facts(*, dau: int = 0, all_exposure_pv: int = 0) -> pd.DataFrame:
    rows = []
    for bucket in formatter.HEX:
        row = {
            "stat_date": "20260808",
            "app_type": "4",
            "version_code": "all",
            "bucket": bucket,
            **{column: 0 for column in formatter.RAW_FACT_COLUMNS},
        }
        row["dau"] = dau
        row["all_exposure_pv"] = all_exposure_pv
        rows.append(row)
    return pd.DataFrame(rows)


def test_rejects_zero_dau_with_nonzero_behavior_facts() -> None:
    with pytest.raises(ValueError, match="zero DAU.*nonzero behavior.*rootSessionId"):
        formatter.build_report(raw_facts(all_exposure_pv=1), CONFIG)


def test_legitimate_all_zero_facts_keep_blank_rates() -> None:
    report = formatter.build_report(raw_facts(), CONFIG)

    assert len(report) == 20
    assert len(report.columns) == 65
    assert pd.isna(report.loc[0, "全部曝光PV/DAU"])
    assert report.loc[0, "全部STR（分享PV/曝光PV）"] == ""


def test_positive_dau_rates_are_unchanged() -> None:
    report = formatter.build_report(raw_facts(dau=10, all_exposure_pv=20), CONFIG)

    assert report.loc[0, "全部曝光PV/DAU"] == 2.0
    assert report.loc[0, "全部STR（分享PV/曝光PV）"] == "0.0000%"