366 lines
18 KiB
Python
366 lines
18 KiB
Python
"""
|
|
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!"
|