tests/test_descriptive_skill.py

"""质量属性梳理 Skill 测试（intent-understanding-layer 任务 10）。

覆盖需求 7.1、7.5、7.6：注册发现、extract→compute→explain 全链路、无外推。
"""

from __future__ import annotations

from kernel.registry import SkillRegistry
from kernel.skill_base import RawInput
from skills.descriptive_summary.skill import DescriptiveSummarySkill


def _confirmed_sheet_raw():
    """构造一个含已确认任务单的 RawInput（SL-0010 风格）。"""
    task_sheet = {
        "extracted_items": [
            {"field": "spec_limit", "value": "总杂≤2.0%", "source_ref": "限度 总杂≤2.0%",
             "group": {"attribute": "总杂"}},
            {"field": "value", "value": "0.00", "source_ref": "总杂% 0",
             "group": {"strength": "20μg", "batch": "B1", "attribute": "总杂"}},
            {"field": "value", "value": "0.00", "source_ref": "总杂% 0",
             "group": {"strength": "40μg", "batch": "B2", "attribute": "总杂"}},
            {"field": "value", "value": "99.19", "source_ref": "含量 99.19",
             "group": {"strength": "20μg", "batch": "B1", "attribute": "含量"}},
        ]
    }
    return RawInput(goal="梳理各规格质量属性", extra={"task_sheet": task_sheet})


class _Svc:
    llm = None
    lang = "zh"


def test_skill_discovered_by_registry():
    reg = SkillRegistry()
    reg.discover("skills")
    assert reg.get("descriptive_summary") is not None


def test_full_extract_compute_explain():
    skill = DescriptiveSummarySkill()
    svc = _Svc()
    data = skill.extract(_confirmed_sheet_raw(), svc)
    assert data.method == "task_sheet"
    result = skill.compute(data)
    assert result.can_proceed is True
    assert result.summary["n_groups"] == 3
    sections = skill.explain(result, svc)
    assert "overview" in sections.sections
    assert "data_table" in sections.sections
    assert "<table" in sections.sections["data_table"]


def test_compute_refuses_without_observations():
    skill = DescriptiveSummarySkill()
    data = skill.extract(RawInput(goal="x", extra={"task_sheet": {"extracted_items": []}}), _Svc())
    result = skill.compute(data)
    assert result.can_proceed is False
    assert result.refusal is not None


def test_no_extrapolation_in_result():
    skill = DescriptiveSummarySkill()
    result = skill.compute(skill.extract(_confirmed_sheet_raw(), _Svc()))
    blob = str(result.summary).lower()
    for forbidden in ("shelf_life", "arrhenius", "k_value", "target_timepoints"):
        assert forbidden not in blob


def test_conformance_evaluated_when_spec_present():
    skill = DescriptiveSummarySkill()
    result = skill.compute(skill.extract(_confirmed_sheet_raw(), _Svc()))
    total_imp_groups = [g for g in result.summary["groups"] if g["attribute"] == "总杂"]
    assert total_imp_groups
    assert all(g["within_spec"] is True for g in total_imp_groups)


def test_headless_render_inputs_returns_rawinput():
    skill = DescriptiveSummarySkill()
    out = skill.render_inputs(object())  # 无任何控件属性
    assert isinstance(out, RawInput)


def test_tolerates_llm_variable_field_names():
    """LLM 把 field 命名为属性名（非严格 "value"）时仍能识别为观测（回归 bug）。"""
    skill = DescriptiveSummarySkill()
    task_sheet = {
        "extracted_items": [
            # LLM 风格：field 是属性名，没有严格的 "value"/"spec_limit"
            {"field": "膜厚", "value": "0.05", "source_ref": "膜厚 0.05",
             "group": {"strength": "20μg", "attribute": "膜厚"}},
            {"field": "总杂%", "value": "0.00", "source_ref": "总杂 0",
             "group": {"strength": "40μg", "attribute": "总杂"}},
            # 规格限度按内容识别（含 ≤ 符号、含"限度"词）
            {"field": "限度", "value": "总杂≤2.0%", "source_ref": "限度 总杂≤2.0%",
             "group": {"attribute": "总杂"}},
        ]
    }
    raw = RawInput(goal="梳理", extra={"task_sheet": task_sheet})
    data = skill.extract(raw, _Svc())
    obs = data.payload["observations"]
    assert len(obs) == 2  # 两个观测项被识别
    assert data.payload["spec_limits"].get("总杂") == "总杂≤2.0%"
    result = skill.compute(data)
    assert result.can_proceed is True


def test_strength_recovered_from_field_name():
    """group 缺 strength 时，从 field/source 中回退提取规格 token。"""
    skill = DescriptiveSummarySkill()
    task_sheet = {
        "extracted_items": [
            {"field": "物理特性-20μg 膜厚", "value": "0.05", "source_ref": "20μg 膜厚 0.05",
             "group": {"attribute": "膜厚"}},
        ]
    }
    data = skill.extract(RawInput(goal="x", extra={"task_sheet": task_sheet}), _Svc())
    assert data.payload["observations"][0]["strength"] == "20μg"


def _cu_sheet_raw():
    """构造含含量均匀度逐单位含量（%）的任务单（SL-0010 20μg 五单位）。"""
    vals = ["99.19", "100.0", "97.68", "96.0", "98.81"]
    items = [
        {"field": "含量%", "value": v, "source_ref": f"含量% {v}",
         "group": {"strength": "20μg", "batch": "SL-0010-25052601",
                   "attribute": "含量%", "table": "含量均匀度"}}
        for v in vals
    ]
    return RawInput(goal="梳理含量均匀度", extra={"task_sheet": {"extracted_items": items}})


def test_content_uniformity_acceptance_value_in_report():
    """含量均匀度百分含量列应计算 AV 并进入报告专段；判定来自确定性药典计算。"""
    skill = DescriptiveSummarySkill()
    svc = _Svc()
    result = skill.compute(skill.extract(_cu_sheet_raw(), svc))
    cu_groups = [g for g in result.summary["groups"] if g.get("acceptance_value")]
    assert cu_groups, "应识别出含量均匀度百分含量分组并计算 AV"
    av = cu_groups[0]["acceptance_value"]
    assert av["standard"] == "chp" and abs(av["acceptance_value"] - 5.074) < 0.01
    assert cu_groups[0]["within_spec"] is True  # AV≤15 → 合格
    sections = skill.explain(result, svc)
    assert "content_uniformity" in sections.sections
    assert "AV" in sections.sections["content_uniformity"]


def test_content_uniformity_skips_mg_per_g_column():
    """mg/g 绝对含量列不应被当作 AV 判定对象。"""
    skill = DescriptiveSummarySkill()
    items = [
        {"field": "含量mg/g", "value": v, "source_ref": f"含量mg/g {v}",
         "group": {"strength": "20μg", "batch": "B", "attribute": "含量mg/g", "table": "含量均匀度"}}
        for v in ("0.834", "0.83", "0.824")
    ]
    raw = RawInput(goal="x", extra={"task_sheet": {"extracted_items": items}})
    result = skill.compute(skill.extract(raw, _Svc()))
    assert all(not g.get("acceptance_value") for g in result.summary["groups"])


def test_censored_value_through_task_sheet_path():
    """任务单/LLM 路径下的删失观测（耐折度＞100、限度≥50）应被识别并正确判定。"""
    skill = DescriptiveSummarySkill()
    task_sheet = {
        "extracted_items": [
            {"field": "耐折度", "value": "＞100", "source_ref": "耐折度 ＞100",
             "group": {"strength": "20μg", "attribute": "耐折度"}},
            {"field": "spec_limit", "value": "≥50", "source_ref": "限度 ≥50",
             "group": {"attribute": "耐折度"}},
        ]
    }
    raw = RawInput(goal="梳理", extra={"task_sheet": task_sheet})
    result = skill.compute(skill.extract(raw, _Svc()))
    g = next(x for x in result.summary["groups"] if x["attribute"] == "耐折度")
    assert g["censored_values"] == ["＞100"]
    assert g["mean"] is None
    assert g["within_spec"] is True  # ＞100 ⟹ ≥50


def test_assay_percent_range_conformance_end_to_end():
    """assay 百分含量% 挂上 90~110% 区间限度后应判定符合。"""
    skill = DescriptiveSummarySkill()
    task_sheet = {
        "extracted_items": [
            {"field": "百分含量%", "value": "98.92", "source_ref": "百分含量% 98.92",
             "group": {"strength": "20μg", "attribute": "百分含量%", "table": "含量"}},
            {"field": "spec_limit", "value": "90.0~110.0%", "source_ref": "限度",
             "group": {"attribute": "百分含量%"}},
        ]
    }
    raw = RawInput(goal="梳理", extra={"task_sheet": task_sheet})
    result = skill.compute(skill.extract(raw, _Svc()))
    g = next(x for x in result.summary["groups"] if x["attribute"] == "百分含量%")
    assert g["within_spec"] is True


def test_humanized_group_has_no_internal_field_names():
    """喂给 LLM 的记录必须是本地化自然语言键值，绝不含内部字段名/布尔（防泄漏）。"""
    skill = DescriptiveSummarySkill()
    result = skill.compute(skill.extract(_cu_sheet_raw(), _Svc()))
    g = next(x for x in result.summary["groups"] if x.get("acceptance_value"))
    rec = skill._humanize_group(g, is_en=False)
    blob = str(rec)
    for forbidden in ("within_spec", "single_point", "n_numeric", "n_censored",
                      "mean_display", "rsd_display", "true", "false", "True", "False"):
        assert forbidden not in blob
    assert rec["符合性"] in ("符合", "不符合", "未判定")
    assert "接受值" in rec  # AV 以自然语言串呈现