feat: testgen pipeline tests — generate → parse → textify → prompt (20 tests)

deploy/tests/pipeline/test_testgen_pipeline.py

@@ -0,0 +1,365 @@
+"""
+test_testgen_pipeline.py — Сквозной тест: testgen → parse → textify → LLM-промпт.
+
+Пайплайн (без БД, без HTTP):
+1. testgen генерирует DOCX (договор, спека, допник)
+2. parse.py парсит байты → elements_json
+3. process.py::_elements_to_text() → plain text для LLM
+4. llm_prompt.py строит промпт → проверяем структуру
+
+Точка входа: elements_json — именно туда приходят данные после парсинга.
+"""
+import io
+import json
+import sys
+import os
+import pytest
+
+# Добавить пути для импортов
+_deploy = os.path.join(os.path.dirname(__file__), "..", "..")
+_testgen = os.path.join(_deploy, "..", "..", "testgen")
+sys.path.insert(0, _deploy)
+sys.path.insert(0, _testgen)
+
+from compare.parse import parse_file
+from compare.process import _elements_to_text
+
+
+# ═══════════════════════════════════════════════════════════════
+# Helpers
+# ═══════════════════════════════════════════════════════════════
+
+def _build_and_parse(build_fn, ctx):
+    """Построить docx через testgen, отдать байты парсеру. Возвращает parsed dict."""
+    from docx import Document
+    doc = build_fn(ctx)
+    buf = io.BytesIO()
+    doc.save(buf)
+    return parse_file("test.docx", buf.getvalue())
+
+
+def _elements_to_dicts(elements):
+    """Привести elements к dict для сравнения (убрать лишнее)."""
+    result = []
+    for el in elements:
+        d = {"type": el["type"]}
+        if el["type"] == "paragraph":
+            d["text"] = el.get("text", "")
+        elif el["type"] == "table":
+            d["row_count"] = len(el.get("rows", []))
+            d["col_count"] = len(el["rows"][0]) if el.get("rows") else 0
+            d["first_cell"] = el["rows"][0][0] if el.get("rows") and el["rows"][0] else ""
+        result.append(d)
+    return result
+
+
+# ═══════════════════════════════════════════════════════════════
+# Tests
+# ═══════════════════════════════════════════════════════════════
+
+class TestGenerateParseTextify:
+    """Полный пайплайн: testgen → parse → textify."""
+
+    # ── Данные ────────────────────────────────────────────────
+
+    @pytest.fixture
+    def pools(self):
+        import pools
+        return pools
+
+    @pytest.fixture
+    def templates(self):
+        import templates
+        return templates
+
+    @pytest.fixture
+    def contract_ctx(self, pools):
+        ci = 0
+        c = pools.COMPANIES[ci]
+        num = pools.contract_number(ci)
+        date = pools.random_date()
+        return {
+            "company": c, "number": num, "date": date, "company_idx": ci,
+            "comment": "Тестовый контракт для пайплайна.",
+        }
+
+    @pytest.fixture
+    def spec_ctx(self, pools):
+        ci = 0
+        c = pools.COMPANIES[ci]
+        num = pools.contract_number(ci)
+        date = pools.random_date()
+        svcs, total, total_str = pools.pick_services(5)
+        for s in svcs:
+            s["date_start"] = pools.random_date(2026, 3, 3)
+        return {
+            "company": c, "number": num, "date": date, "contract_date": date,
+            "version": 1, "services": svcs, "total": total, "total_str": total_str,
+            "label": f"Абонентские услуги с {date}",
+            "comment": "Тестовая спецификация для пайплайна.",
+        }
+
+    @pytest.fixture
+    def addendum_ctx(self, pools):
+        ci = 0
+        c = pools.COMPANIES[ci]
+        num = pools.contract_number(ci)
+        date = pools.random_date()
+        inst, itot, itot_s = pools.pick_services(2)
+        mon, mtot, mtot_s = pools.pick_services(3)
+        return {
+            "company": c, "number": num, "date": date, "contract_date": pools.random_date(),
+            "addendum_num": 1,
+            "services_install": inst, "install_total": itot, "install_total_str": itot_s,
+            "services_monthly": mon, "monthly_total": mtot, "monthly_total_str": mtot_s,
+            "comment": "Тестовый допник для пайплайна.",
+        }
+
+    # ── Этап 1: Генерация → парсинг ───────────────────────────
+
+    def test_generate_contract_parse_ok(self, templates, contract_ctx):
+        """Договор: генерируется и парсится без ошибок."""
+        result = _build_and_parse(templates.build_contract, contract_ctx)
+        assert result["status"] == "parsed", f"Parse failed: {result.get('error')}"
+        assert result["element_count"] > 0
+
+    def test_generate_contract_has_paragraphs(self, templates, contract_ctx):
+        """Договор: в elements есть paragraphs."""
+        result = _build_and_parse(templates.build_contract, contract_ctx)
+        paragraphs = [e for e in result["elements"] if e["type"] == "paragraph"]
+        assert len(paragraphs) >= 3, f"Expected >=3 paragraphs, got {len(paragraphs)}"
+
+    def test_generate_contract_has_tables(self, templates, contract_ctx):
+        """Договор: в elements есть таблица реквизитов."""
+        result = _build_and_parse(templates.build_contract, contract_ctx)
+        tables = [e for e in result["elements"] if e["type"] == "table"]
+        assert len(tables) >= 1, f"Expected >=1 table, got {len(tables)}"
+
+    def test_generate_spec_parse_ok(self, templates, spec_ctx):
+        """Спецификация: генерируется и парсится без ошибок."""
+        result = _build_and_parse(templates.build_spec, spec_ctx)
+        assert result["status"] == "parsed"
+        assert result["element_count"] > 0
+
+    def test_generate_spec_has_services_table(self, templates, spec_ctx):
+        """Спецификация: таблица услуг найдена."""
+        result = _build_and_parse(templates.build_spec, spec_ctx)
+        tables = [e for e in result["elements"] if e["type"] == "table"]
+        assert len(tables) >= 1
+        # Первая таблица — услуги, должна быть НЕ пустой
+        svc_table = tables[0]
+        assert len(svc_table["rows"]) >= 2  # заголовок + минимум 1 строка
+
+    def test_generate_addendum_parse_ok(self, templates, addendum_ctx):
+        """Допсоглашение: генерируется и парсится без ошибок."""
+        result = _build_and_parse(templates.build_addendum, addendum_ctx)
+        assert result["status"] == "parsed"
+        assert result["element_count"] > 0
+
+    def test_generate_addendum_has_tables(self, templates, addendum_ctx):
+        """Допсоглашение: таблицы разовых и абонентских услуг."""
+        result = _build_and_parse(templates.build_addendum, addendum_ctx)
+        tables = [e for e in result["elements"] if e["type"] == "table"]
+        # Должно быть минимум 3 таблицы: реквизиты + разовые + абонентские
+        assert len(tables) >= 3, f"Expected >=3 tables, got {len(tables)}"
+
+    # ── Этап 2: elements_json → textify ────────────────────────
+
+    def test_textify_contract(self, templates, contract_ctx):
+        """Договор after textify содержит ключевые слова."""
+        result = _build_and_parse(templates.build_contract, contract_ctx)
+        text = _elements_to_text(result["elements"])
+        assert "Договор" in text or "ДОГОВОР" in text
+        assert "НУБЕС" in text
+        assert len(text) > 200, f"Text too short: {len(text)} chars"
+
+    def test_textify_spec_contains_table_markers(self, templates, spec_ctx):
+        """Спецификация after textify: пайп-таблицы на месте."""
+        result = _build_and_parse(templates.build_spec, spec_ctx)
+        text = _elements_to_text(result["elements"])
+        # Должны быть маркеры таблиц (--- Таблица ... ---)
+        assert "Таблица" in text
+        # Должны быть пайпы (формат ячеек)
+        assert "|" in text
+        # Должны быть названия услуг
+        for s in spec_ctx["services"]:
+            # Проверяем первые 30 символов названия
+            short_name = s["name"][:30]
+            assert short_name in text, f"Service '{short_name}' not found in textified output"
+
+    def test_textify_spec_contains_total(self, templates, spec_ctx):
+        """Спецификация: итоговая сумма в тексте."""
+        result = _build_and_parse(templates.build_spec, spec_ctx)
+        text = _elements_to_text(result["elements"])
+        assert "Итого" in text
+
+    def test_textify_addendum_structure(self, templates, addendum_ctx):
+        """Допсоглашение after textify: структура сохраняется."""
+        result = _build_and_parse(templates.build_addendum, addendum_ctx)
+        text = _elements_to_text(result["elements"])
+        assert "Соглашение" in text or "СОГЛАШЕНИЕ" in text
+        assert "Инсталляц" in text or "Разовые" in text
+        assert "Абонентские" in text
+
+    # ── Этап 3: elements_json → промпт LLM ─────────────────────
+
+    def test_build_extract_prompt_from_elements(self, templates, spec_ctx):
+        """Из elements_json спецификации строится extract-промпт.
+        Пустая current_spec → extract mode (FALLBACK_EXTRACT)."""
+        from llm_prompt import build_prompt
+        result = _build_and_parse(templates.build_spec, spec_ctx)
+        doc_text = _elements_to_text(result["elements"])
+
+        # Пустая current_spec = первый документ = extract-промпт
+        prompt_text, prompt_id = build_prompt([], doc_text)
+        assert len(prompt_text) > 100
+        assert "{doc_text}" not in prompt_text  # переменная подставлена
+        assert doc_text in prompt_text  # текст документа внутри промпта
+        assert "JSON" in prompt_text or "json" in prompt_text
+
+    def test_build_diff_prompt_from_elements(self, templates, addendum_ctx):
+        """Из elements_json допсоглашения строится diff-промпт.
+        Непустая current_spec → diff mode (FALLBACK_DIFF)."""
+        from llm_prompt import build_prompt
+        result = _build_and_parse(templates.build_addendum, addendum_ctx)
+        doc_text = _elements_to_text(result["elements"])
+
+        # Непустая current_spec = не первый документ = diff-промпт
+        current_spec = [
+            {"hash": "h1", "name": "Аренда стойко-места", "price": 50000, "qty": 1, "sum": 50000, "date_start": "2026-01-01"},
+        ]
+        prompt_text, prompt_id = build_prompt(current_spec, doc_text)
+        assert len(prompt_text) > 100
+        assert doc_text in prompt_text
+        assert "UPDATE" in prompt_text or "DELETE" in prompt_text or "ADD" in prompt_text
+
+    # ── Этап 4: edge cases ─────────────────────────────────────
+
+    def test_parse_empty_docx(self):
+        """Пустой docx — парсится без ошибок, но без элементов."""
+        from docx import Document
+        doc = Document()
+        buf = io.BytesIO()
+        doc.save(buf)
+        result = parse_file("empty.docx", buf.getvalue())
+        assert result["status"] == "parsed"
+        assert result["element_count"] == 0
+
+    def test_parse_docx_text_only(self):
+        """DOCX только с текстом — корректно парсится."""
+        from docx import Document
+        doc = Document()
+        doc.add_paragraph("Привет мир")
+        doc.add_paragraph("Вторая строка")
+        buf = io.BytesIO()
+        doc.save(buf)
+        result = parse_file("text.docx", buf.getvalue())
+        assert result["status"] == "parsed"
+        assert result["element_count"] == 2
+        assert all(e["type"] == "paragraph" for e in result["elements"])
+
+    def test_parse_docx_table_only(self):
+        """DOCX только с таблицей — корректно парсится."""
+        from docx import Document
+        doc = Document()
+        table = doc.add_table(rows=2, cols=3)
+        table.rows[0].cells[0].text = "A"
+        table.rows[0].cells[1].text = "B"
+        table.rows[0].cells[2].text = "C"
+        table.rows[1].cells[0].text = "1"
+        table.rows[1].cells[1].text = "2"
+        table.rows[1].cells[2].text = "3"
+        buf = io.BytesIO()
+        doc.save(buf)
+        result = parse_file("table.docx", buf.getvalue())
+        assert result["status"] == "parsed"
+        tables = [e for e in result["elements"] if e["type"] == "table"]
+        assert len(tables) == 1
+        assert tables[0]["rows"] == [["A", "B", "C"], ["1", "2", "3"]]
+
+    def test_textify_preserves_table_data(self):
+        """Textify сохраняет все данные таблицы (без потерь)."""
+        elements = [
+            {"type": "table", "rows": [
+                ["Услуга", "Цена", "Кол-во"],
+                ["Аренда стойки", "50000", "2"],
+                ["IP-адрес", "300", "8"],
+            ]}
+        ]
+        text = _elements_to_text(elements)
+        assert "Услуга" in text
+        assert "Аренда стойки" in text
+        assert "50000" in text
+        assert "2" in text
+        assert "IP-адрес" in text
+        assert "300" in text
+        assert "8" in text
+
+    # ── Этап 5: валидация manifest-подобной структуры ──────────
+
+    def test_contract_elements_structure(self, templates, contract_ctx):
+        """Элементы договора имеют правильную структуру."""
+        result = _build_and_parse(templates.build_contract, contract_ctx)
+        for el in result["elements"]:
+            assert "type" in el
+            assert el["type"] in ("paragraph", "table")
+            if el["type"] == "paragraph":
+                assert "text" in el
+                assert isinstance(el["text"], str)
+            elif el["type"] == "table":
+                assert "rows" in el
+                assert isinstance(el["rows"], list)
+                assert len(el["rows"]) > 0
+                assert all(isinstance(row, list) for row in el["rows"])
+
+    def test_multiple_contracts_different_content(self, templates, pools):
+        """Разные контракты дают разный elements_json."""
+        import hashlib
+
+        results = []
+        for ci in range(3):
+            c = pools.COMPANIES[ci]
+            ctx = {
+                "company": c, "number": pools.contract_number(ci),
+                "date": pools.random_date(), "company_idx": ci,
+                "comment": f"Контракт {ci}",
+            }
+            result = _build_and_parse(templates.build_contract, ctx)
+            # Сериализуем для сравнения
+            txt = _elements_to_text(result["elements"])
+            results.append(txt)
+
+        # Все три должны быть разными
+        hashes = [hashlib.md5(t.encode()).hexdigest() for t in results]
+        assert len(set(hashes)) == 3, f"All 3 contracts produced same text!"
+
+    def test_spec_v1_v2_produce_different_tables(self, templates, pools):
+        """Спецификации v1 и v2 дают разные таблицы (разные цены/объёмы)."""
+        c = pools.COMPANIES[0]
+        base_ctx = {
+            "company": c, "number": pools.contract_number(0),
+            "date": pools.random_date(), "contract_date": pools.random_date(),
+            "version": 1,
+            "label": "Абонентские услуги",
+        }
+
+        # v1
+        svcs_v1, tot_v1, tot_str_v1 = pools.pick_services(5)
+        ctx_v1 = {**base_ctx, "services": svcs_v1, "total": tot_v1, "total_str": tot_str_v1}
+        r1 = _build_and_parse(templates.build_spec, ctx_v1)
+        t1 = _elements_to_text(r1["elements"])
+
+        # v2 — меняем цены
+        import copy
+        svcs_v2 = copy.deepcopy(svcs_v1)
+        if svcs_v2:
+            svcs_v2[0]["price"] = round(svcs_v2[0]["price"] * 1.25, 2)
+            svcs_v2[0]["price_str"] = pools.format_price(svcs_v2[0]["price"])
+            svcs_v2[0]["sum"] = round(svcs_v2[0]["price"] * svcs_v2[0]["qty"], 2)
+            svcs_v2[0]["sum_str"] = pools.format_price(svcs_v2[0]["sum"])
+        tot_v2 = round(sum(s["sum"] for s in svcs_v2), 2)
+        ctx_v2 = {**base_ctx, "version": 2, "services": svcs_v2, "total": tot_v2, "total_str": pools.format_price(tot_v2)}
+        r2 = _build_and_parse(templates.build_spec, ctx_v2)
+        t2 = _elements_to_text(r2["elements"])
+
+        assert t1 != t2, "v1 and v2 specs produced identical text!"