tests/test_descriptive_summarizer.py

"""描述性梳理统计器测试（intent-understanding-layer 任务 9）。

覆盖 Property 8（不外推）/ Property 10（语言无关）与需求 7.2–7.4、7.7。
"""

from __future__ import annotations

import math

from skills.descriptive_summary.summarizer import summarize


def test_groups_by_strength_batch_attribute():
    obs = [
        {"strength": "20μg", "batch": "B1", "attribute": "总杂", "value": "0.00"},
        {"strength": "40μg", "batch": "B2", "attribute": "总杂", "value": "0.00"},
    ]
    out = summarize(obs)
    assert out["n_groups"] == 2
    assert set(out["strengths"]) == {"20μg", "40μg"}


def test_mean_and_rsd_math():
    obs = [
        {"strength": "20μg", "batch": "B1", "attribute": "含量", "value": "99.19"},
        {"strength": "20μg", "batch": "B1", "attribute": "含量", "value": "100.00"},
        {"strength": "20μg", "batch": "B1", "attribute": "含量", "value": "97.68"},
    ]
    out = summarize(obs)
    g = out["groups"][0]
    assert g["n"] == 3
    assert math.isclose(g["mean"], (99.19 + 100.0 + 97.68) / 3, rel_tol=1e-9)
    assert g["rsd_pct"] is not None and g["rsd_pct"] > 0


def test_rsd_none_for_single_point():
    out = summarize([{"strength": "x", "batch": "B", "attribute": "a", "value": "5"}])
    assert out["groups"][0]["rsd_pct"] is None


def test_conformance_upper_limit():
    obs = [{"strength": "20μg", "batch": "B1", "attribute": "总杂", "value": "0.5"}]
    out = summarize(obs, spec_limits={"总杂": "总杂≤2.0%"})
    assert out["groups"][0]["within_spec"] is True
    obs2 = [{"strength": "20μg", "batch": "B1", "attribute": "总杂", "value": "3.0"}]
    out2 = summarize(obs2, spec_limits={"总杂": "总杂≤2.0%"})
    assert out2["groups"][0]["within_spec"] is False


def test_conformance_range_assay():
    obs = [{"strength": "20μg", "batch": "B1", "attribute": "含量", "value": "98.92"}]
    out = summarize(obs, spec_limits={"含量": "90.0~110.0%"})
    assert out["groups"][0]["within_spec"] is True


def test_conformance_none_without_spec():
    out = summarize([{"strength": "x", "batch": "B", "attribute": "总杂", "value": "0.5"}])
    assert out["groups"][0]["within_spec"] is None


def test_no_extrapolation_fields_property():
    out = summarize([{"strength": "x", "batch": "B", "attribute": "a", "value": "1"}])
    forbidden = {"shelf_life", "k", "r2", "model", "extrapolation", "target_timepoints", "arrhenius"}
    blob = str(out).lower()
    for f in forbidden:
        assert f not in blob


def test_multi_document_aggregation():
    obs = [
        {"strength": "20μg", "batch": "B1", "attribute": "总杂", "value": "0.1", "source": "a.xlsx"},
        {"strength": "20μg", "batch": "B1", "attribute": "总杂", "value": "0.2", "source": "b.xlsx"},
    ]
    out = summarize(obs)
    g = out["groups"][0]
    assert g["n"] == 2
    assert set(g["sources"]) == {"a.xlsx", "b.xlsx"}


def test_non_numeric_values_skipped():
    obs = [
        {"strength": "x", "batch": "B", "attribute": "a", "value": "样品名称"},
        {"strength": "x", "batch": "B", "attribute": "a", "value": "1.0"},
    ]
    out = summarize(obs)
    assert out["groups"][0]["n"] == 1


def test_conformance_less_than_operators():
    # ＜ / < 上限型
    obs = [{"strength": "20μg", "batch": "B1", "attribute": "溶化时限", "value": "60"}]
    out = summarize(obs, spec_limits={"溶化时限": "＜120s"})
    assert out["groups"][0]["within_spec"] is True
    out2 = summarize([{"strength": "20μg", "batch": "B1", "attribute": "溶化时限", "value": "130"}],
                     spec_limits={"溶化时限": "<120s"})
    assert out2["groups"][0]["within_spec"] is False


def test_conformance_greater_than_lower_limit():
    obs = [{"strength": "20μg", "batch": "B1", "attribute": "抗拉强度", "value": "16.98"}]
    out = summarize(obs, spec_limits={"抗拉强度": "≥1"})
    assert out["groups"][0]["within_spec"] is True
    out2 = summarize([{"strength": "20μg", "batch": "B1", "attribute": "抗拉强度", "value": "0.5"}],
                     spec_limits={"抗拉强度": ">1"})
    assert out2["groups"][0]["within_spec"] is False


def test_same_attribute_different_tables_not_merged():
    """不同来源表的同名属性不应被合并（修复过度合并）。"""
    obs = [
        {"strength": "20μg", "batch": "B1", "attribute": "含量mg", "value": "0.834", "table": "含量均匀度"},
        {"strength": "20μg", "batch": "B1", "attribute": "含量mg", "value": "0.826", "table": "含量均匀度"},
        {"strength": "20μg", "batch": "B1", "attribute": "含量mg", "value": "0.826", "table": "含量测定"},
    ]
    out = summarize(obs)
    groups = [g for g in out["groups"] if g["attribute"] == "含量mg"]
    # 两张表 → 两个分组，而非合并成一个 n=3
    assert len(groups) == 2
    tables = {g["table"] for g in groups}
    assert tables == {"含量均匀度", "含量测定"}


def test_spec_limit_normalized_match_attribute_with_unit_suffix():
    """限度键与观测属性名存在单位后缀差异时，归一化回退仍能挂载限度。"""
    # 观测属性 "总杂%"，限度键 "总杂" —— 精确匹配失败，归一化匹配成功。
    obs = [{"strength": "20μg", "batch": "B1", "attribute": "总杂%", "value": "0.5"}]
    out = summarize(obs, spec_limits={"总杂": "总杂≤2.0%"})
    g = out["groups"][0]
    assert g["spec_limit"] == "总杂≤2.0%"
    assert g["within_spec"] is True


def test_spec_limit_normalized_match_assay_mg_per_g():
    """'含量mg/g' 观测可匹配 '含量' 限度键。"""
    obs = [{"strength": "20μg", "batch": "B1", "attribute": "含量mg/g", "value": "98.0"}]
    out = summarize(obs, spec_limits={"含量": "90.0~110.0%"})
    assert out["groups"][0]["within_spec"] is True


def test_exact_match_still_preferred_over_normalized():
    """精确键存在时优先精确匹配，不被归一化覆盖。"""
    obs = [{"strength": "x", "batch": "B", "attribute": "总杂", "value": "0.5"}]
    out = summarize(obs, spec_limits={"总杂": "总杂≤2.0%"})
    assert out["groups"][0]["spec_limit"] == "总杂≤2.0%"


def test_display_fields_consistent_format():
    """mean_display / rsd_display 为统一定点串，供表格与叙述共用。"""
    obs = [
        {"strength": "20μg", "batch": "B1", "attribute": "含量", "value": "99.19"},
        {"strength": "20μg", "batch": "B1", "attribute": "含量", "value": "100.00"},
        {"strength": "20μg", "batch": "B1", "attribute": "含量", "value": "97.68"},
    ]
    g = summarize(obs)["groups"][0]
    assert g["mean_display"] == f"{g['mean']:.4g}"
    assert g["rsd_display"] == f"{g['rsd_pct']:.2f}"


def test_single_point_flag():
    """n=1 分组带 single_point 标注；n≥2 不带。"""
    out = summarize([
        {"strength": "x", "batch": "B", "attribute": "a", "value": "5"},
        {"strength": "y", "batch": "B", "attribute": "a", "value": "5"},
        {"strength": "y", "batch": "B", "attribute": "a", "value": "6"},
    ])
    by_strength = {g["strength"]: g for g in out["groups"]}
    assert by_strength["x"]["single_point"] is True
    assert by_strength["y"]["single_point"] is False
    assert by_strength["y"]["rsd_display"] is not None


def test_censored_value_excluded_from_mean_but_counted():
    """删失值（＞100）不参与均值/RSD，但计入观测总数并单独保留。"""
    out = summarize([
        {"strength": "20μg", "batch": "B1", "attribute": "耐折度", "value": "＞100"},
    ])
    g = out["groups"][0]
    assert g["mean"] is None          # 无真实点测定 → 无均值
    assert g["rsd_pct"] is None
    assert g["n"] == 1                 # 仍计为一个观测
    assert g["n_numeric"] == 0
    assert g["censored_values"] == ["＞100"]


def test_censored_gt_satisfies_lower_limit():
    """＞100 对下限 ≥50 → 必然符合。"""
    out = summarize(
        [{"strength": "20μg", "batch": "B1", "attribute": "耐折度", "value": "＞100"}],
        spec_limits={"耐折度": "≥50"},
    )
    assert out["groups"][0]["within_spec"] is True


def test_censored_lt_satisfies_upper_limit():
    """<0.05 对上限 ≤1.0 → 必然符合。"""
    out = summarize(
        [{"strength": "20μg", "batch": "B1", "attribute": "单杂", "value": "<0.05"}],
        spec_limits={"单杂": "≤1.0"},
    )
    assert out["groups"][0]["within_spec"] is True


def test_censored_gt_violates_upper_limit():
    """＞100 对上限 ≤80 → 必然违反。"""
    out = summarize(
        [{"strength": "20μg", "batch": "B1", "attribute": "x", "value": "＞100"}],
        spec_limits={"x": "≤80"},
    )
    assert out["groups"][0]["within_spec"] is False


def test_censored_indeterminate_returns_none():
    """＞100 对上限 ≤120 → 真值可能超限，无法确定 → None（不臆断）。"""
    out = summarize(
        [{"strength": "20μg", "batch": "B1", "attribute": "x", "value": "＞100"}],
        spec_limits={"x": "≤120"},
    )
    assert out["groups"][0]["within_spec"] is None


def test_mixed_numeric_and_censored_conformance():
    """数值点 + 删失值混合：全部确定满足 → True。"""
    out = summarize(
        [
            {"strength": "20μg", "batch": "B1", "attribute": "耐折度", "value": "120"},
            {"strength": "20μg", "batch": "B1", "attribute": "耐折度", "value": "＞100"},
        ],
        spec_limits={"耐折度": "≥50"},
    )
    g = out["groups"][0]
    assert g["n"] == 2 and g["n_numeric"] == 1
    assert g["mean"] == 120.0          # 仅数值点参与均值
    assert g["within_spec"] is True