"""提取层触发字段派生的单元测试。 覆盖: - ``StabilityDataExtractor`` 列式表格启发式解析(含分组列); - skill 层确定性触发派生(spec_type / role / design_type / grouping / 规格限度)。 导入路径由 tests/conftest.py 设置。 """ from utils.stability_data_extractor import StabilityDataExtractor from skills.stability.skill import ( _spec_from_goal, _design_type_from_goal, _infer_spec_type_from_name, _infer_role_from_name, _enrich_analysis_triggers, ) class _FailingInvoker: """强制走启发式提取:模拟 LLM 不可用 / 失败,绝不触网。""" def invoke(self, *a, **k): return type("_R", (), {"success": False, "content": ""})() def _heuristic_extractor() -> StabilityDataExtractor: return StabilityDataExtractor(model_invoker=_FailingInvoker()) # --------------------------------------------------------------------------- # 列式表格启发式解析 # --------------------------------------------------------------------------- def test_extractor_parses_simple_column_table(): text = ( "Batch,Condition,Time(Months),Total_Impurities(%)\n" "B1,25C_60RH,0,0.10\n" "B1,25C_60RH,3,0.18\n" "B1,25C_60RH,6,0.25\n" ) res = _heuristic_extractor().extract_from_text(text, "预测货架期") assert res["extraction_method"] == "heuristic" assert len(res["batches"]) == 1 cond = res["batches"][0]["conditions"][0] assert cond["timepoints"] == [0.0, 3.0, 6.0] assert cond["cqa_data"][0]["values"] == [0.10, 0.18, 0.25] def test_extractor_detects_grouping_dimension(): text = ( "Strength,Time(Months),Degradation(%)\n" "50mg,0,0.10\n" "50mg,6,0.52\n" "100mg,0,0.10\n" "100mg,6,0.35\n" ) res = _heuristic_extractor().extract_from_text(text, "跨组评估") assert res.get("grouping_dimension") == "strength" gids = {b.get("group_id") for b in res["batches"]} assert {"50mg", "100mg"} <= gids for b in res["batches"]: if b.get("group_id"): assert b["factor_levels"].get("strength") == b["group_id"] def test_extractor_multi_cqa_columns(): text = ( "Time(Months),Assay(%),Total_Impurities(%)\n" "0,99.8,0.10\n" "6,98.9,0.92\n" "12,97.5,2.28\n" ) res = _heuristic_extractor().extract_from_text(text, "质量平衡") cqas = {c["cqa_name"] for c in res["batches"][0]["conditions"][0]["cqa_data"]} assert cqas == {"含量", "总杂质"} # --------------------------------------------------------------------------- # 目标文本解析 # --------------------------------------------------------------------------- def test_spec_from_goal_lower(): val, st = _spec_from_goal("规格标准设定为下限 90.0%") assert val == 90.0 and st == "lower" def test_spec_from_goal_upper(): val, st = _spec_from_goal("设定降解产物接受上限为 8.0%") assert val == 8.0 and st == "upper" def test_spec_from_goal_acceptance_limit(): val, st = _spec_from_goal("在接受限度为 1.2% 时推断货架期") assert val == 1.2 and st == "upper" # 小数值默认上限型 def test_design_type_matrixing(): dtype, factors = _design_type_from_goal("采用 ICH Q1D 的 2/3 时间点矩阵法(Matrixing)设计") assert dtype == "matrixing" and factors == ["time"] def test_design_type_bracketing(): dtype, factors = _design_type_from_goal("采用括号法 Bracketing,仅测试最高/最低规格") assert dtype == "bracketing" and "strength" in factors def test_infer_spec_type_and_role(): assert _infer_spec_type_from_name("含量") == "lower" assert _infer_spec_type_from_name("Assay") == "lower" assert _infer_spec_type_from_name("总杂质") == "upper" assert _infer_role_from_name("含量", "lower") == "assay" assert _infer_role_from_name("总杂质", "upper") == "degradant" # --------------------------------------------------------------------------- # 触发派生只补缺、不覆盖(非回归) # --------------------------------------------------------------------------- def test_enrich_does_not_override_existing_spec_type(): data = { "primary_cqa": "总杂质", "specification_limit": 0.5, "batches": [{ "batch_id": "B1", "conditions": [{ "condition_id": "25C_60RH", "cqa_data": [{"cqa_name": "总杂质", "spec_type": "upper", "role": "degradant", "values": [0.1, 0.2, 0.3]}], }], }], } _enrich_analysis_triggers("预测 36 个月货架期", data) cqa = data["batches"][0]["conditions"][0]["cqa_data"][0] assert cqa["spec_type"] == "upper" # 未被覆盖 assert cqa["role"] == "degradant"