97 lines
4.1 KiB
Python
97 lines
4.1 KiB
Python
"""Интеграционные тесты services/process.py — run_pipeline с Fake LLM."""
|
||
import json
|
||
|
||
from db import contracts, documents, supplements, spec_current
|
||
from db.connection import query
|
||
from services import process
|
||
|
||
|
||
def _seed_contract(n_supps=2):
|
||
"""Создать contract + n документов (parsed) + n supplements."""
|
||
c = contracts.insert("03700_1")
|
||
for i in range(n_supps):
|
||
d = documents.insert(f"d{i}.docx", "application/octet-stream", "raw", batch_id="b1")
|
||
documents.set_parsed(d["id"], [{"type": "paragraph", "text": f"строка {i}"}])
|
||
supplements.insert(c["id"], d["id"], "initial" if i == 0 else "additional")
|
||
return c
|
||
|
||
|
||
class TestFullReplace:
|
||
def test_no_duplicates(self, db, monkeypatch):
|
||
c = _seed_contract(2)
|
||
|
||
def fake_call(current_spec, doc_text, build_prompt_fn, llm_client=None):
|
||
ops = [
|
||
{"action": "ADD", "new_row": {"name": "Аренда стойко-места", "price": 50000, "qty": 1, "sum": 50000}},
|
||
{"action": "ADD", "new_row": {"name": "IP-адрес IPv4", "price": 300, "qty": 8, "sum": 2400}},
|
||
{"action": "ADD", "new_row": {"name": "Канал 1 Гбит/с", "price": 20000, "qty": 1, "sum": 20000}},
|
||
]
|
||
return ({"mode": "full_replace", "ops": ops}, "pid")
|
||
|
||
monkeypatch.setattr("services.llm.call_llm", fake_call)
|
||
|
||
events = list(process.run_pipeline(c["id"], "", lambda cur, txt: ("p", "pid")))
|
||
|
||
rows = spec_current.list_by_contract(c["id"])
|
||
# 2 full_replace, но без дублей — всегда 3 строки, не 6
|
||
assert len(rows) == 3
|
||
assert {r["name"] for r in rows} == {"Аренда стойко-места", "IP-адрес IPv4", "Канал 1 Гбит/с"}
|
||
|
||
# история сохранена: 2 full_replace × 3 ADD = 6 событий
|
||
ev = query("SELECT count(*) AS c FROM spec_events WHERE contract_id = %s", (c["id"],))
|
||
assert ev[0]["c"] == 6
|
||
|
||
# не было ошибок извлечения
|
||
assert not any(e.get("type") == "extract_error" for e in events)
|
||
|
||
|
||
class TestTargetIdTranslation:
|
||
def test_update_applies(self, db, monkeypatch):
|
||
c = _seed_contract(2)
|
||
|
||
def fake_call(current_spec, doc_text, build_prompt_fn, llm_client=None):
|
||
if not current_spec:
|
||
return ({"mode": "partial", "ops": [
|
||
{"action": "ADD", "new_row": {"name": "Аренда", "price": 50000, "qty": 1, "sum": 50000}},
|
||
]}, "pid")
|
||
return ({"mode": "partial", "ops": [
|
||
{"action": "UPDATE", "target_id": "r1", "new_values": {"price": 55000, "sum": 55000}},
|
||
]}, "pid")
|
||
|
||
monkeypatch.setattr("services.llm.call_llm", fake_call)
|
||
|
||
list(process.run_pipeline(c["id"], "", lambda cur, txt: ("p", "pid")))
|
||
|
||
rows = spec_current.list_by_contract(c["id"])
|
||
assert len(rows) == 1
|
||
assert rows[0]["price"] == 55000 # UPDATE применился, а не ушёл в UNRESOLVED
|
||
|
||
upd = query("SELECT status FROM spec_events WHERE contract_id = %s AND action = 'UPDATE'", (c["id"],))
|
||
assert upd[0]["status"] == "applied"
|
||
|
||
|
||
class TestNoSupplements:
|
||
def test_error_event(self, db):
|
||
evs = list(process.run_pipeline("nonexistent", "", lambda cur, txt: ("p", "pid")))
|
||
assert any(e.get("type") == "error" for e in evs)
|
||
|
||
|
||
class TestElementsToText:
|
||
def test_list(self):
|
||
ej = [
|
||
{"type": "paragraph", "text": "Привет"},
|
||
{"type": "table", "rows": [["a", "b"], ["c", "d"]]},
|
||
]
|
||
assert process._elements_to_text(ej) == "Привет\na | b\nc | d"
|
||
|
||
def test_dict_value(self):
|
||
ej = {"Value": [{"type": "paragraph", "text": "X"}]}
|
||
assert process._elements_to_text(ej) == "X"
|
||
|
||
def test_string_json(self):
|
||
ej = json.dumps([{"type": "paragraph", "text": "Y"}])
|
||
assert process._elements_to_text(ej) == "Y"
|
||
|
||
def test_plain_string(self):
|
||
assert process._elements_to_text("просто текст") == "просто текст"
|