DM-Agent / tests /test_analyzer.py
unknown
refactor: consolidate repo structure, remove upload_to_github duplication
3291500
Raw History Blame Contribute Delete
2.66 kB
"""Tests for src.analyzer — core aggregation and analysis logic."""
import sys
from pathlib import Path
import pandas as pd
import numpy as np
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from src.analyzer import _agg_group, _safe_pct_change, compute_wow
def _make_df(rows: list[dict]) -> pd.DataFrame:
return pd.DataFrame(rows)
class TestSafePctChange:
def test_normal(self):
assert _safe_pct_change(150, 100) == 50.0
def test_decrease(self):
assert _safe_pct_change(80, 100) == -20.0
def test_zero_prev(self):
assert _safe_pct_change(100, 0) is None
def test_nan_prev(self):
assert _safe_pct_change(100, float("nan")) is None
class TestAggGroup:
def test_single_row(self):
df = _make_df([{
"Cost": 100, "Ads.Sales": 500, "Impressions": 10000,
"CTR": 0.02, "GA4.Sessions": 50, "CVR": 0.05,
}])
result = _agg_group(df, ["Cost", "Ads.Sales", "ROAS", "CTR", "CVR", "CPC"])
assert result["Cost"] == 100
assert result["Ads.Sales"] == 500
assert result["ROAS"] == 5.0
assert abs(result["CTR"] - 0.02) < 1e-6
assert abs(result["CVR"] - 0.05) < 1e-6
def test_multi_row_weighted_ctr(self):
df = _make_df([
{"Cost": 100, "Ads.Sales": 300, "Impressions": 10000, "CTR": 0.01, "GA4.Sessions": 20, "CVR": 0.1},
{"Cost": 200, "Ads.Sales": 400, "Impressions": 20000, "CTR": 0.03, "GA4.Sessions": 80, "CVR": 0.05},
])
result = _agg_group(df, ["Cost", "Ads.Sales", "ROAS", "CTR", "CVR", "CPC"])
assert result["Cost"] == 300
assert result["Ads.Sales"] == 700
expected_clicks = 10000 * 0.01 + 20000 * 0.03 # 100 + 600 = 700
expected_ctr = expected_clicks / 30000
assert abs(result["CTR"] - expected_ctr) < 1e-6
assert abs(result["CPC"] - 300 / expected_clicks) < 1e-6
def test_empty_df(self):
df = _make_df([])
df = pd.DataFrame(columns=["Cost", "Ads.Sales", "Impressions", "CTR"])
result = _agg_group(df, ["Cost", "Ads.Sales"])
assert result.get("Cost", 0) == 0
assert result.get("Ads.Sales", 0) == 0
class TestComputeWow:
def test_normal_change(self):
curr = {"Cost": 200, "ROAS": 3.0}
prev = {"Cost": 100, "ROAS": 2.0}
result = compute_wow(curr, prev, ["Cost", "ROAS"])
assert result["Cost"] == 100.0
assert result["ROAS"] == 50.0
def test_zero_prev(self):
curr = {"Cost": 100}
prev = {"Cost": 0}
result = compute_wow(curr, prev, ["Cost"])
assert result["Cost"] is None