78 lines
3.1 KiB
Python
78 lines
3.1 KiB
Python
|
"""Unit tests for LlmPipeline — mock LLM client and embedding provider."""
|
||
|
|
from __future__ import annotations
|
||
|
|
from unittest.mock import MagicMock, patch
|
||
|
|
import json
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
|
||
|
|
def _make_pipeline():
|
||
|
|
with patch("app.infrastructure.perception.llm_pipeline.get_llm_client") as mock_llm_fn, \
|
||
|
|
patch("app.infrastructure.perception.llm_pipeline.OpenAICompatibleEmbeddingProvider") as mock_emb_cls:
|
||
|
|
|
||
|
|
mock_client = MagicMock()
|
||
|
|
mock_client.chat.return_value = MagicMock(content='{"obligations":[{"text":"test obligation","deontic":"must","subject":"OEM","object":"system","condition":""}],"deadlines":[{"date":"2026-07-01","description":"实施截止"}],"scope":"适用于M1类车辆","penalties":"罚款","impact_level":"high"}')
|
||
|
|
mock_llm_fn.return_value = mock_client
|
||
|
|
|
||
|
|
mock_emb = MagicMock()
|
||
|
|
mock_emb.embed_texts.return_value = [[0.1] * 1024, [0.9] * 1024]
|
||
|
|
mock_emb_cls.return_value = mock_emb
|
||
|
|
|
||
|
|
from app.infrastructure.perception.llm_pipeline import LlmPipeline
|
||
|
|
return LlmPipeline(), mock_client, mock_emb
|
||
|
|
|
||
|
|
|
||
|
|
def test_extract_structure_returns_dict():
|
||
|
|
pipeline, mock_client, _ = _make_pipeline()
|
||
|
|
event = {
|
||
|
|
"id": "evt-001",
|
||
|
|
"standard_code": "GB 18384-2025",
|
||
|
|
"title": "电动汽车安全要求",
|
||
|
|
"summary": "新增 IP67 级别防护",
|
||
|
|
"source_label": "CATARC",
|
||
|
|
"tags": ["电池安全"],
|
||
|
|
}
|
||
|
|
result = pipeline.extract_structure(event)
|
||
|
|
assert isinstance(result, dict)
|
||
|
|
assert "obligations" in result
|
||
|
|
assert "impact_level" in result
|
||
|
|
|
||
|
|
|
||
|
|
def test_assess_impact_returns_list():
|
||
|
|
pipeline, mock_client, _ = _make_pipeline()
|
||
|
|
mock_client.chat.return_value = MagicMock(content='[{"doc_id":"d1","doc_name":"Safety Manual","score":0.85,"key_clauses":"§4.2","recommendation":"更新第4章"}]')
|
||
|
|
mock_retrieval = MagicMock()
|
||
|
|
chunk = MagicMock()
|
||
|
|
chunk.doc_id = "d1"
|
||
|
|
chunk.doc_title = "Safety Manual"
|
||
|
|
chunk.score = 0.85
|
||
|
|
chunk.text = "relevant text"
|
||
|
|
chunk.section_title = "§4.2"
|
||
|
|
mock_retrieval.retrieve.return_value = [chunk]
|
||
|
|
event = {
|
||
|
|
"standard_code": "GB 18384-2025",
|
||
|
|
"title": "电动汽车安全要求",
|
||
|
|
"obligations": [{"text": "OEM shall comply"}],
|
||
|
|
}
|
||
|
|
result = pipeline.assess_impact(event, mock_retrieval)
|
||
|
|
assert isinstance(result, list)
|
||
|
|
|
||
|
|
|
||
|
|
def test_compute_diff_no_change():
|
||
|
|
pipeline, _, mock_emb = _make_pipeline()
|
||
|
|
mock_emb.embed_texts.return_value = [[0.5] * 1024, [0.5] * 1024]
|
||
|
|
result = pipeline.compute_diff("paragraph one", "paragraph one")
|
||
|
|
assert isinstance(result, dict)
|
||
|
|
assert "changed_sections" in result
|
||
|
|
assert "change_summary" in result
|
||
|
|
|
||
|
|
|
||
|
|
def test_compute_diff_detects_change():
|
||
|
|
pipeline, mock_client, mock_emb = _make_pipeline()
|
||
|
|
mock_emb.embed_texts.return_value = [
|
||
|
|
[1.0] + [0.0] * 1023,
|
||
|
|
[0.0] + [1.0] + [0.0] * 1022,
|
||
|
|
]
|
||
|
|
mock_client.chat.return_value = MagicMock(content='{"change_type":"tightened","summary":"Requirement tightened"}')
|
||
|
|
result = pipeline.compute_diff("old paragraph text", "new tighter requirement text")
|
||
|
|
assert isinstance(result["changed_sections"], list)
|