test: набор pytest-тестов под site/ (unit + интеграционные + безопасность)

tests/test_process_pipeline.py

@@ -0,0 +1,96 @@
+"""Интеграционные тесты 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("просто текст") == "просто текст"