63 lines
2.7 KiB
Python
63 lines
2.7 KiB
Python
"""Test classify_batch with MemRepository + FakeLLM."""
|
|
from services.classify import classify_batch
|
|
from services.llm_client import FakeLLMClient
|
|
from repository import MemRepository
|
|
|
|
|
|
class TestClassifyBatch:
|
|
def test_garbage_by_filename(self):
|
|
"""Счёт-фактура отфильтровывается без LLM."""
|
|
repo = MemRepository()
|
|
llm = FakeLLMClient({"default": '{"doc_type":"contract"}'})
|
|
|
|
repo.insert_document("счет-фактура №123.docx", "mime", "data", batch_id="b1")
|
|
|
|
result = classify_batch("b1", llm_client=llm, repo=repo)
|
|
assert result["ok"] is True
|
|
assert result["garbage"] == 1
|
|
assert result["done"] == 1
|
|
assert len(llm.calls) == 0 # LLM не вызывался
|
|
|
|
def test_garbage_by_header(self):
|
|
"""Акт сверки в тексте отфильтровывается."""
|
|
repo = MemRepository()
|
|
llm = FakeLLMClient({"default": '{"doc_type":"contract"}'})
|
|
|
|
doc = repo.insert_document("документ.docx", "mime", "data", batch_id="b1")
|
|
repo.set_document_parsed(doc["id"], [
|
|
{"type": "paragraph", "text": "АКТ СВЕРКИ взаимных расчётов", "style": ""}
|
|
])
|
|
|
|
result = classify_batch("b1", llm_client=llm, repo=repo)
|
|
assert result["garbage"] == 1
|
|
assert len(llm.calls) == 0
|
|
|
|
def test_llm_classify(self):
|
|
"""Договор классифицируется через LLM."""
|
|
repo = MemRepository()
|
|
llm = FakeLLMClient({
|
|
"default": '{"doc_type":"contract","own_number":"03700_1","doc_date":"2026-02-01","counterparty":"ЗАО XXX001"}'
|
|
})
|
|
|
|
doc = repo.insert_document("договор-XXX001.docx", "mime", "data", batch_id="b1")
|
|
repo.set_document_parsed(doc["id"], [
|
|
{"type": "paragraph", "text": "ДОГОВОР № 03700_1", "style": ""}
|
|
])
|
|
|
|
result = classify_batch("b1", llm_client=llm, repo=repo)
|
|
assert result["done"] == 1, f"done={result['done']}, failed={result['failed']}, doc={repo.get_document(doc['id'])}"
|
|
assert result["garbage"] == 0
|
|
assert len(llm.calls) == 1
|
|
|
|
updated = repo.get_document(doc["id"])
|
|
assert updated["doc_type"] == "contract"
|
|
assert updated["own_number"] == "03700_1"
|
|
assert updated["counterparty"] == "ЗАО XXX001"
|
|
|
|
def test_no_pending(self):
|
|
"""Пустой батч."""
|
|
repo = MemRepository()
|
|
result = classify_batch("empty", llm_client=FakeLLMClient(), repo=repo)
|
|
assert result["ok"] is False
|
|
assert "no pending" in result["error"]
|