feat(Ф4): pytest — 22 юнит-теста (contracts + garbage + upload + llm + repo)

2026-06-28 09:24:34 +04:00
parent 4d2d5a7dc0
commit 0776369d13
7 changed files with 214 additions and 2 deletions
@@ -0,0 +1,110 @@
+"""Unit tests — no DB, no network, just pure logic."""
+import pytest
+from contracts import ParseResult, ClassifyResult, GroupingResult, CompareOp
+from services.upload import parse_multipart
+from services.classify import _is_garbage_by_filename, _is_garbage_by_header
+
+
+class TestContracts:
+    """Dataclass creation and validation."""
+
+    def test_parse_result(self):
+        r = ParseResult(status="parsed", element_count=10, elements=[])
+        assert r.status == "parsed"
+        assert r.element_count == 10
+
+    def test_classify_from_llm(self):
+        r = ClassifyResult.from_llm({
+            "doc_type": "contract",
+            "own_number": "03700_1",
+            "doc_date": "2026-02-01",
+            "counterparty": "ЗАО XXX001",
+        })
+        assert r.doc_type == "contract"
+        assert r.own_number == "03700_1"
+        assert r.counterparty == "ЗАО XXX001"
+
+    def test_grouping_result_empty(self):
+        r = GroupingResult(contract_number="03700_1")
+        assert r.contract_number == "03700_1"
+        assert r.documents == []
+
+    def test_compare_op_add(self):
+        op = CompareOp(action="ADD", new_row={"name": "Тест", "price": 100})
+        assert op.action == "ADD"
+        assert op.new_row["price"] == 100
+
+    def test_compare_op_from_llm(self):
+        op = CompareOp.from_llm({
+            "action": "UPDATE",
+            "target_id": "r1",
+            "new_values": {"price": 200},
+            "comment": "change",
+        })
+        assert op.action == "UPDATE"
+        assert op.new_values["price"] == 200
+        assert op.comment == "change"
+
+
+class TestGarbageFilter:
+    """Stage 1-2 garbage detection."""
+
+    def test_filename_garbage_invoice(self):
+        assert _is_garbage_by_filename("счет-фактура №123.docx") is True
+
+    def test_filename_garbage_act(self):
+        assert _is_garbage_by_filename("Акт сверки за май.pdf") is True
+
+    def test_filename_not_garbage(self):
+        assert _is_garbage_by_filename("договор-XXX001.docx") is False
+        assert _is_garbage_by_filename("спецификация услуг.pdf") is False
+
+    def test_header_garbage(self):
+        text = "СЧЕТ-ФАКТУРА № 123 от 01.01.2026\nПоставщик: ООО Ромашка"
+        assert _is_garbage_by_header(text) is True
+
+    def test_header_not_garbage(self):
+        text = "ДОГОВОР № 03700_1\nг. Москва\n\nИсполнитель обязуется..."
+        assert _is_garbage_by_header(text) is False
+
+
+class TestParseMultipart:
+    """Pure multipart parsing without HTTP."""
+
+    def test_simple_file(self, sample_docx_bytes):
+        boundary = b"----testboundary"
+        body = (
+            b"------testboundary\r\n"
+            b'Content-Disposition: form-data; name="files"; filename="test.docx"\r\n'
+            b"Content-Type: application/octet-stream\r\n\r\n"
+            + sample_docx_bytes +
+            b"\r\n------testboundary--\r\n"
+        )
+        result = parse_multipart(body, f"multipart/form-data; boundary=----testboundary")
+        assert result["filename"] == "test.docx"
+        assert result["file_data"] == sample_docx_bytes
+
+    def test_with_contract_id(self, sample_docx_bytes):
+        boundary = b"----testboundary"
+        body = (
+            b"------testboundary\r\n"
+            b'Content-Disposition: form-data; name="files"; filename="test.docx"\r\n\r\n'
+            + sample_docx_bytes +
+            b"\r\n------testboundary\r\n"
+            b'Content-Disposition: form-data; name="contract_id"\r\n\r\n'
+            b"550e8400-e29b-41d4-a716-446655440000"
+            b"\r\n------testboundary--\r\n"
+        )
+        result = parse_multipart(body, f"multipart/form-data; boundary=----testboundary")
+        assert result["contract_id"] == "550e8400-e29b-41d4-a716-446655440000"
+
+    def test_path_traversal_rejected(self):
+        boundary = b"----testboundary"
+        body = (
+            b"------testboundary\r\n"
+            b'Content-Disposition: form-data; name="files"; filename="../etc/passwd"\r\n\r\n'
+            b"evil"
+            b"\r\n------testboundary--\r\n"
+        )
+        with pytest.raises(ValueError, match="invalid filename"):
+            parse_multipart(body, f"multipart/form-data; boundary=----testboundary")
@@ -0,0 +1,59 @@
+"""Unit tests for LLM client and Repository — with fakes."""
+from services.llm_client import FakeLLMClient
+from repository import MemRepository
+
+
+class TestFakeLLM:
+    def test_exact_match(self):
+        client = FakeLLMClient({"hello": "world"})
+        assert client.complete("hello") == "world"
+
+    def test_partial_match(self):
+        client = FakeLLMClient({"classify": '{"doc_type":"contract"}'})
+        assert "contract" in client.complete("classify this document")
+
+    def test_default_fallback(self):
+        client = FakeLLMClient({"default": '{"ok":true}'})
+        assert client.complete("unknown prompt") == '{"ok":true}'
+
+    def test_call_history(self):
+        client = FakeLLMClient({"a": "1", "b": "2"})
+        client.complete("a")
+        client.complete("b")
+        assert len(client.calls) == 2
+        assert client.calls[0] == "a"
+
+
+class TestMemRepository:
+    def test_insert_and_retrieve(self, mem_repo):
+        doc = mem_repo.insert_document("test.docx", "application/vnd...", "base64data", batch_id="batch-1")
+        assert doc["filename"] == "test.docx"
+        assert doc["classify_status"] == "pending"
+
+    def test_list_pending(self, mem_repo):
+        mem_repo.insert_document("a.docx", "mime", "data", batch_id="b1")
+        mem_repo.insert_document("b.docx", "mime", "data", batch_id="b1")
+        mem_repo.insert_document("c.docx", "mime", "data", batch_id="b2")
+        pending = mem_repo.list_pending("b1")
+        assert len(pending) == 2
+
+    def test_set_classification(self, mem_repo):
+        doc = mem_repo.insert_document("test.docx", "mime", "data", batch_id="b1")
+        mem_repo.set_classification(doc["id"], "contract", "03700", None, "2026-02-01", "XXX")
+        updated = mem_repo.documents[doc["id"]]
+        assert updated["doc_type"] == "contract"
+        assert updated["classify_status"] == "classified"
+
+    def test_set_garbage(self, mem_repo):
+        doc = mem_repo.insert_document("счет.docx", "mime", "data", batch_id="b1")
+        mem_repo.set_classify_garbage(doc["id"], "filename_regex")
+        assert mem_repo.documents[doc["id"]]["doc_type"] == "garbage"
+
+    def test_contract_lifecycle(self, mem_repo):
+        cid = mem_repo.insert_contract("03700_1", "ЗАО XXX")
+        assert cid
+        doc = mem_repo.insert_document("test.docx", "mime", "data")
+        mem_repo.insert_supplement(cid, doc["id"], "initial")
+        supps = mem_repo.list_supplements(cid)
+        assert len(supps) == 1
+        assert supps[0]["type"] == "initial"