| """提取层触发字段派生的单元测试。 |
| |
| 覆盖： |
| - ``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" |
|
|
