110 lines
4.2 KiB
Python
110 lines
4.2 KiB
Python
"""Приёмка ББ и ОФР парсером: CLI tabular-parser из зеркального worktree (`TP_WT`), маркер `parser`.
|
|
|
|
Запуск: `$UVRUN python -m pytest synthgen/tests -q -m parser`. Без окружения тесты пропускаются.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from synthgen import forms, forms_xlsx
|
|
|
|
pytestmark = pytest.mark.parser
|
|
|
|
TP_WT = Path(os.environ.get("TP_WT") or "/Users/andrey.epifancev/projects/awp-ca/.wt/synthetic-group/tabular-parser")
|
|
PY = TP_WT / "api" / ".venv" / "bin" / "python"
|
|
|
|
|
|
def _run(path: Path) -> dict:
|
|
env = {**os.environ, "ENV_FILE": str(TP_WT / "api" / ".env.test"), "PYTHONPATH": str(TP_WT / "api")}
|
|
proc = subprocess.run(
|
|
[str(PY), "-m", "src.parser.cli", str(path)], cwd=TP_WT / "api", env=env, capture_output=True, text=True, check=False
|
|
)
|
|
assert proc.stdout.strip(), proc.stderr
|
|
return json.loads(proc.stdout)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def parsed(forms_world, tmp_path_factory):
|
|
if not PY.exists():
|
|
pytest.skip(f"нет окружения парсера: {PY}")
|
|
cfg, group, periods, books = forms_world
|
|
root = tmp_path_factory.mktemp("statements")
|
|
items = []
|
|
for m in group.members:
|
|
for p in periods:
|
|
for f in (forms.balance(books[m.code], m, p, periods, cfg.seed), forms.income(books[m.code], m, p, periods, cfg.seed)):
|
|
path = root / f"{m.code}_{p.key}_{forms_xlsx.file_name(f)}"
|
|
forms_xlsx.write(f, path)
|
|
items.append((f, path))
|
|
with ThreadPoolExecutor(max_workers=8) as pool:
|
|
results = list(pool.map(lambda fp: _run(fp[1]), items))
|
|
return [(f, p, r) for (f, p), r in zip(items, results, strict=True)]
|
|
|
|
|
|
def _doc(res):
|
|
assert len(res["documents"]) == 1, [d["document_type"] for d in res["documents"]]
|
|
return res["documents"][0]
|
|
|
|
|
|
def _issues(res):
|
|
return [(d["status"], d["checks"], [w["code"] for w in d["warnings"]]) for d in res.get("documents", [])]
|
|
|
|
|
|
def test_parser_status_and_type(parsed):
|
|
bad = []
|
|
for f, path, res in parsed:
|
|
docs = res.get("documents", [])
|
|
if res.get("status") != "done" or len(docs) != 1 or docs[0]["document_type"] != f.kind:
|
|
bad.append((path.name, res.get("status"), res.get("error"), _issues(res)))
|
|
assert not bad, bad[:5]
|
|
|
|
|
|
def test_parser_columns_and_units(parsed):
|
|
bad = []
|
|
for f, path, res in parsed:
|
|
doc = _doc(res)
|
|
got = [c["period_date"] for c in sorted(doc["columns"], key=lambda c: c["column_index"])]
|
|
want = [d.isoformat() for d in f.dates]
|
|
if got != want:
|
|
bad.append((path.name, got, want))
|
|
if "тыс" not in (doc["header"].get("units") or "").lower():
|
|
bad.append((path.name, "units", doc["header"].get("units")))
|
|
assert not bad, bad[:5]
|
|
|
|
|
|
def test_parser_checks_clean(parsed):
|
|
bad = [(path.name, _doc(res)["checks"]) for f, path, res in parsed if _doc(res)["checks"]["relations_failed"]]
|
|
assert not bad, bad[:5]
|
|
|
|
|
|
def test_parser_values_equal_form_values_and_balance(parsed):
|
|
bad = []
|
|
for f, path, res in parsed:
|
|
doc = _doc(res)
|
|
order = [c["column_index"] for c in sorted(doc["columns"], key=lambda c: c["column_index"])]
|
|
got: dict[str, list[Decimal]] = {}
|
|
for line in doc["lines"]:
|
|
if line["line_status"] != "ok":
|
|
continue
|
|
row = got.setdefault(line["line_code"], [Decimal(0)] * len(order))
|
|
for v in line["values"]:
|
|
if v["amount"] is not None:
|
|
row[order.index(v["column_index"])] += Decimal(str(v["amount"]))
|
|
codes = forms.BALANCE_CODES if f.kind == "BALANCE" else forms.INCOME_CODES
|
|
for code in codes:
|
|
for col, want in enumerate(f.values[code]):
|
|
have = got.get(code, [Decimal(0)] * len(order))[col]
|
|
if have != want:
|
|
bad.append((path.name, code, col, str(have), str(want)))
|
|
if f.kind == "BALANCE":
|
|
for col in range(3):
|
|
if got["1600"][col] != got["1700"][col]:
|
|
bad.append((path.name, "1600!=1700", col))
|
|
assert not bad, bad[:10]
|