File size: 11,686 Bytes
19729e9
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
0e6887b
19729e9
 
0e6887b
 
 
 
 
 
 
 
19729e9
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
0e6887b
 
 
 
 
 
 
 
19729e9
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
0e6887b
 
 
19729e9
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
0e6887b
19729e9
0e6887b
19729e9
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
"""相容性 Skill(``skills.compatibility.skill``)三层契约的集成测试(任务 18)。

覆盖需求 8.2 与平台契约:
- extract:规范化 SMILES / 辅料、识别官能团、渲染结构图(RDKit 可选→降级无图)。
- compute:**纯 Python** 确定性风险分级(签名无 svc,需求 2.1);数据不足优雅拒绝(8.3)。
- explain:复用 professional_analyzer + 相容性 prompts,LLM 仅写文字、不改数值(2.3)。
- 结构图与风险色块经 ChartService / ReportService 呈现。

测试自包含:svc.llm 用 stub;RDKit 相关断言用 importorskip 守护;不触网。
导入路径由 tests/conftest.py 设置(platform 目录与仓库根加入 sys.path)。
"""

from __future__ import annotations

import importlib.util
import inspect

import pytest

from kernel.services import Services
from kernel.skill_base import ComputeResult, ExtractedData, PharmaSkill, RawInput
from skills.compatibility import rules
from skills.compatibility.skill import SKILL, CompatibilitySkill

_RDKIT_AVAILABLE = importlib.util.find_spec("rdkit") is not None


# ---------------------------------------------------------------------------
# Stub LLM / Services
# ---------------------------------------------------------------------------

class _StubLLMResult:
    def __init__(self, ok: bool, content: str = "") -> None:
        self.ok = ok
        self.content = content


class _StubLLM:
    """记录调用并返回固定文本的 LLM 桩。"""

    def __init__(self, ok: bool = True, content: str = "一、机理分析……\n二、[必须] 控制限度……") -> None:
        self._ok = ok
        self._content = content
        self.calls: list[tuple[str, str]] = []

    def complete(self, system: str, user: str, *, temperature: float = 0.3, **kwargs):
        self.calls.append((system, user))
        return _StubLLMResult(self._ok, self._content if self._ok else "")


def _svc(llm=None, lang: str = "zh") -> Services:
    return Services(llm=llm, lang=lang)


# ---------------------------------------------------------------------------
# 契约:compute 签名无 svc(需求 2.1)
# ---------------------------------------------------------------------------

def test_compute_signature_has_no_service_handle():
    """compute(self, data) 的签名不含 svc / llm 句柄(架构级阻断幻觉计算)。"""
    sig = inspect.signature(CompatibilitySkill.compute)
    params = list(sig.parameters)
    assert params == ["self", "data"]
    assert "svc" not in params and "llm" not in params


def test_skill_is_pharmaskill_and_registered_instance():
    assert isinstance(SKILL, CompatibilitySkill)
    assert isinstance(SKILL, PharmaSkill)
    assert SKILL.meta.id == "compatibility"
    # 输入类型声明含 SMILES 与 EXCIPIENT。
    kinds = {k.value for k in SKILL.meta.input_kinds}
    assert "smiles" in kinds and "excipient" in kinds


# ---------------------------------------------------------------------------
# compute:确定性风险分级
# ---------------------------------------------------------------------------

def test_compute_grades_risk_from_extracted_data():
    """compute 对 extract 产物做确定性分级(伯胺 + 乳糖 → 高风险)。"""
    data = ExtractedData(
        payload={
            "smiles": "NCCO",
            "api_name": "Test API",
            "excipients": ["乳糖"],
            "functional_groups": [{"id": "primary_amine"}],
            "structure_image": None,
        },
        method="manual",
    )
    result = SKILL.compute(data)
    assert isinstance(result, ComputeResult)
    assert result.can_proceed is True
    assert result.summary["overall_risk"] == rules.RISK_HIGH
    assert result.summary["n_high"] >= 1
    assert result.trace  # 计算追踪非空


def test_compute_refuses_when_no_excipients():
    """无辅料时 compute 优雅拒绝(need 8.3)。"""
    data = ExtractedData(payload={"smiles": "CCO", "excipients": [], "functional_groups": []})
    result = SKILL.compute(data)
    assert result.can_proceed is False
    assert result.refusal is not None
    assert "辅料" in result.refusal["reason"]


def test_compute_injects_structure_figure_when_available():
    """extract 渲染出结构图时,compute 把其放入 figures 供报告注入。"""
    data = ExtractedData(
        payload={
            "smiles": "CCO",
            "excipients": ["乳糖"],
            "functional_groups": [{"id": "primary_amine"}],
            "structure_image": "data:image/png;base64,AAAA",
        }
    )
    result = SKILL.compute(data)
    assert "structure" in result.figures
    assert result.figures["structure"]["image_base64"].startswith("data:image/png")


def test_compute_is_deterministic():
    data = ExtractedData(
        payload={
            "smiles": "NCCO",
            "excipients": ["乳糖", "硬脂酸镁"],
            "functional_groups": [{"id": "primary_amine"}, {"id": "ester"}],
        }
    )
    first = SKILL.compute(data).summary
    again = SKILL.compute(data).summary
    assert first == again


# ---------------------------------------------------------------------------
# explain:LLM 仅写文字;风险矩阵 / 结构渲染收归 report_service(需求 3.1)
# ---------------------------------------------------------------------------

def _assemble_html(result, sections, lang: str = "zh") -> str:
    """用 ReportService 组装相容性报告 HTML(矩阵/结构/KPI 由 report_service 渲染)。"""
    from services.report_service import ReportService

    return ReportService().assemble(SKILL.meta, result, sections, lang=lang)


def test_explain_returns_only_mechanism_and_matrix_rendered_by_report_service():
    data = ExtractedData(
        payload={
            "smiles": "NCCO",
            "api_name": "Test API",
            "excipients": ["乳糖"],
            "functional_groups": [{"id": "primary_amine"}],
            "structure_image": None,
        }
    )
    result = SKILL.compute(data)
    llm = _StubLLM(ok=True)
    sections = SKILL.explain(result, _svc(llm))

    # LLM 被调用一次(机理 + 处方)。
    assert len(llm.calls) == 1
    # 机理段来自 LLM 文本。
    assert "机理分析" in sections.sections["mechanism"]
    # 风险矩阵/结构不再由 explain 产出(已收归 report_service)。
    assert "risk_matrix" not in sections.sections
    assert "structure" not in sections.sections
    # compute 产出结构化矩阵数据;report_service 据此渲染高风险徽章与矩阵。
    assert result.summary["risk_matrix_2d"]["rows"]
    html = _assemble_html(result, sections, "zh")
    assert "高风险" in html
    assert "compat-matrix" in html


def test_explain_does_not_alter_compute_numbers():
    """explain 不得修改 compute 的风险分级(需求 2.3)。"""
    data = ExtractedData(
        payload={
            "smiles": "NCCO",
            "excipients": ["乳糖"],
            "functional_groups": [{"id": "primary_amine"}],
        }
    )
    result = SKILL.compute(data)
    overall_before = result.summary["overall_risk"]
    SKILL.explain(result, _svc(_StubLLM(ok=True)))
    assert result.summary["overall_risk"] == overall_before


def test_explain_degrades_gracefully_without_llm():
    """svc.llm 为 None 时,explain 用确定性摘要兜底,不崩溃。"""
    data = ExtractedData(
        payload={
            "smiles": "NCCO",
            "excipients": ["乳糖"],
            "functional_groups": [{"id": "primary_amine"}],
        }
    )
    result = SKILL.compute(data)
    sections = SKILL.explain(result, _svc(llm=None))
    assert "mechanism" in sections.sections
    # 兜底摘要应包含确定性依据文本。
    assert sections.sections["mechanism"]


def test_explain_falls_back_when_llm_fails():
    data = ExtractedData(
        payload={"smiles": "NCCO", "excipients": ["乳糖"], "functional_groups": [{"id": "primary_amine"}]}
    )
    result = SKILL.compute(data)
    sections = SKILL.explain(result, _svc(_StubLLM(ok=False)))
    assert sections.sections["mechanism"]  # 非空兜底


def test_explain_english_language():
    data = ExtractedData(
        payload={"smiles": "NCCO", "excipients": ["乳糖"], "functional_groups": [{"id": "primary_amine"}]}
    )
    result = SKILL.compute(data)
    sections = SKILL.explain(result, _svc(_StubLLM(ok=True), lang="en"))
    # 英文报告:风险矩阵由 report_service 渲染为英文标签。
    html = _assemble_html(result, sections, "en")
    assert "High risk" in html


# ---------------------------------------------------------------------------
# extract:规范化 + 官能团识别 + 结构图(RDKit 可选)
# ---------------------------------------------------------------------------

def test_extract_parses_excipients_heuristically():
    """无 RDKit 依赖路径:辅料按行解析,payload schema 完整。"""
    raw = RawInput(smiles="", excipient="乳糖\n硬脂酸镁")
    data = SKILL.extract(raw, _svc())
    assert isinstance(data, ExtractedData)
    assert "乳糖" in data.payload["excipients"]
    assert "硬脂酸镁" in data.payload["excipients"]
    # 无 SMILES 时附提示。
    assert "未提供 SMILES" in data.notes


@pytest.mark.skipif(not _RDKIT_AVAILABLE, reason="RDKit 未安装,跳过结构识别断言")
def test_extract_identifies_functional_groups_with_rdkit():
    """RDKit 可用时,乙醇胺(NCCO)应识别出伯胺并渲染结构图。"""
    raw = RawInput(smiles="NCCO", excipient="乳糖")
    data = SKILL.extract(raw, _svc())
    group_ids = {
        (g.get("id") if isinstance(g, dict) else g)
        for g in data.payload["functional_groups"]
    }
    assert "primary_amine" in group_ids
    # 结构图为 data URI。
    assert data.payload["structure_image"] is None or data.payload["structure_image"].startswith("data:image")


def test_extract_degrades_when_rdkit_unavailable(monkeypatch):
    """RDKit 不可用时,extract 跳过结构图与官能团识别但不崩溃。"""
    # 注入一个声称不可用的渲染器。
    class _NoRDKit:
        is_available = False

    skill = CompatibilitySkill(molecule_renderer=_NoRDKit())
    raw = RawInput(smiles="NCCO", excipient="乳糖")
    data = skill.extract(raw, _svc())
    assert data.payload["functional_groups"] == []
    assert data.payload["structure_image"] is None
    assert "RDKit 不可用" in data.notes


# ---------------------------------------------------------------------------
# 端到端:extract → compute → explain(RDKit 可选)
# ---------------------------------------------------------------------------

def test_end_to_end_pipeline_smiles_plus_excipient():
    raw = RawInput(smiles="NCCO", excipient="乳糖")
    llm = _StubLLM(ok=True)
    svc = _svc(llm)

    data = SKILL.extract(raw, svc)
    result = SKILL.compute(data)          # 注意:compute 不接收 svc
    sections = SKILL.explain(result, svc)

    assert result.can_proceed is True
    assert result.summary["overall_risk"] in (
        rules.RISK_NONE, rules.RISK_LOW, rules.RISK_MEDIUM, rules.RISK_HIGH
    )
    # explain 仅产出机理叙述;矩阵/结构由 report_service 渲染。
    assert "mechanism" in sections.sections
    assert "risk_matrix" not in sections.sections


def test_can_handle_scoring():
    assert SKILL.can_handle(RawInput(smiles="CCO", excipient="乳糖")) == pytest.approx(0.9)
    assert SKILL.can_handle(RawInput(excipient="乳糖")) == pytest.approx(0.6)
    assert SKILL.can_handle(RawInput(smiles="CCO")) == pytest.approx(0.4)
    assert SKILL.can_handle(RawInput()) == pytest.approx(0.0)