""" 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!"