453 lines
24 KiB
Python
453 lines
24 KiB
Python
"""Сценарий регрессии на стенде с нуля (regress.sh, этап stand): только stdlib, запуск `python3 -I`.
|
||
|
||
Для каждого сценария expected.json: консолидация ГК/ГСК за период → блокировка → загрузка файлов фикстур в версию →
|
||
ожидание распознавания (tabular-parser, scan-gateway + эмулятор ИС) → подтверждение ОСВ-документов → проверки свода
|
||
(summary, пробелы, контрольные строки), экспорта xlsx, шахматки, форм ББ/ОФР; плюс негатив (повтор создания — 409,
|
||
блокировка без права EDIT — 403). Итог — report.json (проверки) и snapshots/<сценарий>.*.json (весь свод, шахматка,
|
||
формы); с --baseline снимки сравниваются с прошлым прогоном (расхождения — в report.json, раздел diffs).
|
||
|
||
python3 -I stand_scenario.py --api http://127.0.0.1:18000 --fixtures ../fixtures --expected expected.json --out DIR
|
||
[--baseline DIR_ПРОШЛОГО_stand] [--timeout 600]
|
||
|
||
Код возврата: 0 — все проверки зелёные, 1 — есть красные (расхождения со снимком прошлого прогона красными не считаются,
|
||
их надо просмотреть).
|
||
"""
|
||
|
||
import argparse
|
||
import io
|
||
import json
|
||
import re
|
||
import sys
|
||
import time
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
import uuid
|
||
import zipfile
|
||
from decimal import Decimal
|
||
from pathlib import Path
|
||
|
||
MONEY = ("opening_debit", "opening_credit", "turnover_debit", "turnover_credit", "closing_debit", "closing_credit")
|
||
PENDING_FILE = {"UPLOADED", "PROCESSING"}
|
||
# locator документов из OCR содержит id задания (file:<uuid>.json;table:0) — в снимке заменяется
|
||
UUID_RE = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}")
|
||
TO_CONFIRM = {"RECOGNIZED", "NEEDS_CHECK"}
|
||
|
||
|
||
class Api:
|
||
def __init__(self, base: str) -> None:
|
||
self.base = base.rstrip("/") + "/api/v1"
|
||
self.token = ""
|
||
|
||
def call(self, method: str, path: str, body=None, *, raw=False, ctype=None, params=None):
|
||
url = self.base + path + ("?" + urllib.parse.urlencode(params, doseq=True) if params else "")
|
||
data, headers = None, {}
|
||
if self.token:
|
||
headers["Authorization"] = f"Bearer {self.token}"
|
||
if body is not None:
|
||
if ctype:
|
||
data, headers["Content-Type"] = body, ctype
|
||
else:
|
||
data, headers["Content-Type"] = json.dumps(body).encode(), "application/json"
|
||
req = urllib.request.Request(url, data=data, method=method, headers=headers)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=120) as r:
|
||
payload = r.read()
|
||
return r.status, (payload if raw else (json.loads(payload) if payload else None)), dict(r.headers)
|
||
except urllib.error.HTTPError as e:
|
||
payload = e.read()
|
||
try:
|
||
return e.code, json.loads(payload), dict(e.headers)
|
||
except ValueError:
|
||
return e.code, payload.decode(errors="replace"), dict(e.headers)
|
||
|
||
def ok(self, method: str, path: str, body=None, **kw):
|
||
status, data, _ = self.call(method, path, body, **kw)
|
||
if status >= 300:
|
||
raise RuntimeError(f"{method} {path}: {status} {str(data)[:300]}")
|
||
return data
|
||
|
||
def login(self, email: str) -> str:
|
||
self.token = ""
|
||
self.token = self.ok("POST", "/dev/token", {"email": email})["access_token"]
|
||
return self.token
|
||
|
||
def upload(self, path: str, file: Path):
|
||
boundary = uuid.uuid4().hex
|
||
name = file.name.replace('"', "%22")
|
||
body = (
|
||
f'--{boundary}\r\nContent-Disposition: form-data; name="files"; filename="{name}"\r\n'
|
||
"Content-Type: application/octet-stream\r\n\r\n"
|
||
).encode() + file.read_bytes() + f"\r\n--{boundary}--\r\n".encode()
|
||
return self.ok("POST", path, body, ctype=f"multipart/form-data; boundary={boundary}")
|
||
|
||
def pages(self, path: str, params=None, limit=200):
|
||
offset, items = 0, []
|
||
while True:
|
||
page = self.ok("GET", path, params={**(params or {}), "limit": limit, "offset": offset})
|
||
items += page["items"]
|
||
offset += limit
|
||
if offset >= page["total"]:
|
||
return items
|
||
|
||
|
||
class Report:
|
||
def __init__(self) -> None:
|
||
self.checks: list[dict] = []
|
||
self.diffs: list[dict] = []
|
||
|
||
def check(self, name: str, ok: bool, detail: str = "") -> bool:
|
||
self.checks.append({"name": name, "ok": bool(ok), "detail": detail})
|
||
print(("ok " if ok else "FAIL ") + name + (f": {detail}" if detail else ""), flush=True)
|
||
return ok
|
||
|
||
|
||
def dec(v) -> Decimal:
|
||
return Decimal(str(v)) if v not in (None, "") else Decimal(0)
|
||
|
||
|
||
def scrub(obj, inn_by_le: dict):
|
||
"""Снимок без идентификаторов стенда: id ЮЛ → ИНН, прочие id и списки id убираются (меняются на каждом стенде)."""
|
||
if isinstance(obj, list):
|
||
return [scrub(x, inn_by_le) for x in obj]
|
||
if not isinstance(obj, dict):
|
||
return obj
|
||
out = {}
|
||
for k, v in obj.items():
|
||
if isinstance(v, str) and v in inn_by_le:
|
||
out[k] = inn_by_le[v]
|
||
elif k == "id" or k.endswith(("_id", "_ids")):
|
||
continue
|
||
else:
|
||
out[k] = scrub(v, inn_by_le)
|
||
return out
|
||
|
||
|
||
def wait_health(base: str, timeout: float) -> dict:
|
||
deadline, last = time.time() + timeout, None
|
||
while time.time() < deadline:
|
||
try:
|
||
with urllib.request.urlopen(base.rstrip("/") + "/healthcheck", timeout=5) as r:
|
||
last = json.loads(r.read())
|
||
if last.get("nats") == "ok":
|
||
return last
|
||
except (OSError, ValueError) as e:
|
||
last = {"error": str(e)}
|
||
time.sleep(2)
|
||
raise RuntimeError(f"ядро не готово за {timeout} с: {last}")
|
||
|
||
|
||
def xlsx_rows(content: bytes) -> dict[str, int]:
|
||
"""Число строк <row> на листе по имени листа (без openpyxl: только stdlib)."""
|
||
z = zipfile.ZipFile(io.BytesIO(content))
|
||
wb = z.read("xl/workbook.xml").decode()
|
||
rels = z.read("xl/_rels/workbook.xml.rels").decode()
|
||
targets = dict(re.findall(r'<Relationship[^>]*Id="([^"]+)"[^>]*Target="([^"]+)"', rels))
|
||
targets |= {k: v for v, k in re.findall(r'<Relationship[^>]*Target="([^"]+)"[^>]*Id="([^"]+)"', rels)}
|
||
out = {}
|
||
for name, rid in re.findall(r'<sheet[^>]*name="([^"]+)"[^>]*r:id="([^"]+)"', wb):
|
||
target = targets[rid].lstrip("/")
|
||
target = target if target.startswith("xl/") else "xl/" + target
|
||
out[name] = len(re.findall(r"<row[ >]", z.read(target).decode()))
|
||
return out
|
||
|
||
|
||
def run_scenario(api: Api, sc: dict, ctx: dict, fixtures: Path, rep: Report, out: Path, timeout: float) -> None:
|
||
sid = sc["id"]
|
||
by_inn = ctx["by_inn"]
|
||
status, card, _ = api.call(
|
||
"POST",
|
||
"/consolidations",
|
||
{"company_group_id": ctx["group_id"], "related_group_id": ctx["related_id"], "year": sc["year"], "quarter": sc["quarter"]},
|
||
)
|
||
if not rep.check(f"{sid}: создание консолидации", status == 201, f"{status}"):
|
||
return
|
||
vid, cid = card["id"], card["consolidation_id"]
|
||
rep.check(f"{sid}: карточка версии", card["status"] == "INITIATED" and card["version_no"] == 1, f"status={card['status']}")
|
||
st2, body2, _ = api.call(
|
||
"POST",
|
||
"/consolidations",
|
||
{"company_group_id": ctx["group_id"], "related_group_id": ctx["related_id"], "year": sc["year"], "quarter": sc["quarter"]},
|
||
)
|
||
rep.check(f"{sid}: повтор создания — 409", st2 == 409, f"{st2} {str(body2)[:120]}")
|
||
api.ok("POST", f"/consolidations/versions/{vid}/lock")
|
||
|
||
# загрузка
|
||
sent = 0
|
||
for up in sc["uploads"]:
|
||
le = by_inn[up["inn"]]
|
||
files = sorted(p for pat in up["files"] for p in fixtures.glob(pat) if p.is_file())
|
||
for f in files:
|
||
res = api.upload(f"/consolidations/{cid}/legal-entities/{le}/files", f)
|
||
if res["rejected"]:
|
||
rep.check(f"{sid}: загрузка {f.name}", False, str(res["rejected"])[:200])
|
||
sent += len(res["accepted"])
|
||
expected_files = sum(len([p for pat in up["files"] for p in fixtures.glob(pat) if p.is_file()]) for up in sc["uploads"])
|
||
rep.check(f"{sid}: файлы приняты", sent == expected_files, f"{sent} из {expected_files}")
|
||
|
||
# распознавание
|
||
t0, files = time.time(), []
|
||
while time.time() - t0 < timeout:
|
||
files = [f for up in sc["uploads"] for f in api.pages(f"/consolidations/{cid}/legal-entities/{by_inn[up['inn']]}/files")]
|
||
if files and not any(f["status"] in PENDING_FILE for f in files):
|
||
break
|
||
time.sleep(3)
|
||
by_status: dict[str, int] = {}
|
||
for f in files:
|
||
by_status[f["status"]] = by_status.get(f["status"], 0) + 1
|
||
rep.check(
|
||
f"{sid}: распознавание завершено за {int(time.time() - t0)} с",
|
||
by_status.get("PROCESSED", 0) == expected_files,
|
||
json.dumps(by_status, ensure_ascii=False),
|
||
)
|
||
for f in files:
|
||
if f["status"] != "PROCESSED":
|
||
job = f.get("job") or {}
|
||
rep.check(f"{sid}: файл {f['original_name']}", False, f"{f['status']} {job.get('error_code')} {job.get('error_message')}")
|
||
|
||
# подтверждение ОСВ-документов (чистые уже CONFIRMED — автоподтверждение OQ-157)
|
||
confirmed = 0
|
||
for f in files:
|
||
for d in f["documents"]:
|
||
if d["document_type"] == "OSV" and d["status"] in TO_CONFIRM:
|
||
st, body, _ = api.call("POST", f"/documents/{d['id']}/confirm")
|
||
ok = st in (200, 201) or (st == 409 and "ALREADY_CONFIRMED" in str(body))
|
||
rep.check(f"{sid}: подтверждение {f['original_name']} / {d['locator']}", ok, f"{st}") if not ok else None
|
||
confirmed += 1
|
||
docs = [(f["original_name"], d) for f in files for d in f["documents"]]
|
||
(out / "snapshots" / f"{sid}.documents.json").write_text(
|
||
json.dumps(
|
||
sorted(
|
||
[
|
||
{k: d.get(k) for k in ("document_type", "osv_account_code", "period_year", "period_quarter", "is_cumulative", "period_check", "counterparty_check", "locator", "issue_counts")}
|
||
| {"file": n, "status_after_recognition": d["status"], "locator": UUID_RE.sub("<uuid>", str(d["locator"]))}
|
||
for n, d in docs
|
||
],
|
||
key=lambda x: (x["file"], str(x["locator"])),
|
||
),
|
||
ensure_ascii=False,
|
||
indent=1,
|
||
)
|
||
)
|
||
print(f"{sid}: подтверждено вручную {confirmed}", flush=True)
|
||
for kind in ("OSV", "BALANCE", "INCOME"):
|
||
doc = next((d for _, d in docs if d["document_type"] == kind), None)
|
||
if doc:
|
||
ctx["documents"][f"{sid}-{kind.lower()}"] = doc["id"]
|
||
|
||
# свод
|
||
summary = api.ok("GET", f"/consolidations/versions/{vid}/osv/summary")
|
||
for k, v in sc["summary"].items():
|
||
rep.check(f"{sid}: summary.{k}", summary[k] == v, f"{summary[k]} (ожидалось {v})")
|
||
inn_by_le = {v: k for k, v in by_inn.items()}
|
||
gaps = sorted([inn_by_le.get(g["legal_entity_id"], g["legal_entity_name"]), g["account_code"], sorted(g["missing_quarters"])] for g in summary["gaps"])
|
||
rep.check(f"{sid}: пробелы", gaps == sorted(sc["gaps"]), f"{len(gaps)}: {gaps}" if gaps != sorted(sc["gaps"]) else f"{len(gaps)}")
|
||
card = api.ok("GET", f"/consolidations/versions/{vid}")
|
||
rep.check(f"{sid}: статус версии после загрузки", card["status"] == "DATA_CORRECTION", card["status"])
|
||
|
||
for row in sc["rows"]:
|
||
items = [
|
||
r
|
||
for r in api.pages(f"/consolidations/versions/{vid}/osv", {"legal_entity_id": by_inn[row["inn"]], "q": row["q"]})
|
||
if r["account_code"] == row["account_code"]
|
||
]
|
||
if not rep.check(f"{sid}: строка «{row['name']}» найдена", len(items) == 1, f"строк: {len(items)}"):
|
||
continue
|
||
r, bad = items[0], []
|
||
for k, v in row["expect"].items():
|
||
if k in MONEY:
|
||
if dec(r[k]) != dec(v):
|
||
bad.append(f"{k}={r[k]} (ожидалось {v})")
|
||
elif k == "entry_ids":
|
||
if len(r["entry_ids"]) != v:
|
||
bad.append(f"entry_ids={len(r['entry_ids'])} (ожидалось {v})")
|
||
elif r.get(k) != v:
|
||
bad.append(f"{k}={r.get(k)} (ожидалось {v})")
|
||
rep.check(f"{sid}: строка «{row['name']}» суммы", not bad, "; ".join(bad))
|
||
|
||
rows = api.pages(f"/consolidations/versions/{vid}/osv", limit=2000)
|
||
rep.check(f"{sid}: свод постранично = rows_total", len(rows) == summary["rows_total"], f"{len(rows)}")
|
||
snap_rows = sorted(
|
||
(
|
||
{
|
||
"le": inn_by_le.get(r["legal_entity_id"], r["legal_entity_name"]),
|
||
"account": r["account_code"],
|
||
"row": r["row_number"],
|
||
"cp": r["counterparty_name"],
|
||
"cp_inn": r["counterparty_inn"],
|
||
"type": r["counterparty_type"],
|
||
"match": r["match_kind"],
|
||
}
|
||
| {k: str(dec(r[k])) for k in MONEY}
|
||
for r in rows
|
||
),
|
||
key=lambda x: (x["le"], str(x["account"]), str(x["cp"]), x["row"] or 0),
|
||
)
|
||
(out / "snapshots" / f"{sid}.osv.json").write_text(json.dumps(snap_rows, ensure_ascii=False, indent=0))
|
||
types: dict[str, int] = {}
|
||
for r in rows:
|
||
types[r["counterparty_type"]] = types.get(r["counterparty_type"], 0) + 1
|
||
print(f"{sid}: типы контрагентов {types}", flush=True)
|
||
|
||
# экспорт
|
||
st, content, headers = api.call("GET", f"/consolidations/versions/{vid}/osv/export", raw=True)
|
||
if rep.check(f"{sid}: экспорт свода 200", st == 200, f"{st}"):
|
||
sheets = xlsx_rows(content)
|
||
(out / f"{sid}.svod.xlsx").write_bytes(content)
|
||
main_rows = sheets.get("Сводная ОСВ", -1) - 1
|
||
gaps_rows = sheets.get("Пробелы", -1) - 1
|
||
rep.check(f"{sid}: экспорт — строк = rows_total", main_rows == summary["rows_total"], f"{sheets}")
|
||
rep.check(f"{sid}: экспорт — пробелов = summary.gaps", gaps_rows == len(summary["gaps"]), f"{gaps_rows}")
|
||
cd = headers.get("Content-Disposition") or headers.get("content-disposition") or ""
|
||
rep.check(f"{sid}: экспорт — Content-Disposition", "attachment" in cd and ".xlsx" in cd, cd[:120])
|
||
|
||
# чек-лист
|
||
st, checklist, _ = api.call("GET", f"/consolidations/versions/{vid}/checklist")
|
||
rep.check(f"{sid}: чек-лист 200", st == 200, f"{st}")
|
||
|
||
# шахматка: все предустановленные правила + результирующая таблица
|
||
rules = api.ok("GET", f"/consolidations/versions/{vid}/elimination-rules")["rules"]
|
||
matrix = {}
|
||
for rule in rules:
|
||
m = api.ok("GET", f"/consolidations/versions/{vid}/intercompany-matrix/preset/{rule['id']}")
|
||
names = {c["id"]: inn_by_le.get(c["id"], c["name"]) for c in m["companies"]}
|
||
matrix[f"{rule['rule_no']} {rule['statement_type']} {rule['side_1']} {rule['measure']}"] = sorted(
|
||
[names.get(c["company_a_id"]), names.get(c["company_b_id"]), str(dec(c["amount_a"])), str(dec(c["amount_b"])), str(dec(c["variance_amount"])), c["one_sided"]]
|
||
for c in m["cells"]
|
||
)
|
||
st, result_table, _ = api.call("GET", f"/consolidations/versions/{vid}/intercompany-matrix/result-table")
|
||
rep.check(f"{sid}: результирующая таблица 200", st == 200, f"{st}")
|
||
st, totals, _ = api.call("GET", f"/consolidations/versions/{vid}/elimination-totals")
|
||
rep.check(f"{sid}: итоги элиминации 200", st == 200, f"{st}")
|
||
rep.check(f"{sid}: правил элиминации", len(rules) > 0, f"{len(rules)}")
|
||
# ручная шахматка «Дт <счёт> / Кт <счёт>» — как в test_intercompany_matrix_golden ядра; предустановленные правила
|
||
# собирают стороны шире одного счёта (например, правило 6 — ещё 76.06 и 76.АВ), их суммы — только в снимке
|
||
accounts = {acc["code"]: acc["id"] for acc in ctx["accounts"]}
|
||
for m in sc["matrix_manual"]:
|
||
body = {"debit_account_ids": [accounts[m["debit"]]], "credit_account_ids": [accounts[m["credit"]]]}
|
||
st, res, _ = api.call("POST", f"/consolidations/versions/{vid}/intercompany-matrix/manual", body)
|
||
if not rep.check(f"{sid}: шахматка Дт {m['debit']} / Кт {m['credit']} — 200", st == 200, f"{st} {str(res)[:200]}"):
|
||
continue
|
||
got = {
|
||
(inn_by_le.get(c["company_a_id"]), inn_by_le.get(c["company_b_id"])): [str(dec(c["amount_a"])), str(dec(c["amount_b"]))]
|
||
for c in res["cells"]
|
||
}
|
||
matrix[f"manual Дт {m['debit']} / Кт {m['credit']}"] = sorted([*k, *v] for k, v in got.items())
|
||
for a_inn, b_inn, amount_a, amount_b in m["cells"]:
|
||
cell = got.get((a_inn, b_inn))
|
||
rep.check(
|
||
f"{sid}: шахматка Дт {m['debit']} / Кт {m['credit']}, {a_inn} → {b_inn}",
|
||
cell is not None and dec(cell[0]) == dec(amount_a) and dec(cell[1]) == dec(amount_b),
|
||
f"{cell} (ожидалось {[amount_a, amount_b]})",
|
||
)
|
||
|
||
(out / "snapshots" / f"{sid}.matrix.json").write_text(
|
||
json.dumps(
|
||
{"rules": len(rules), "matrix": matrix, "result_table": scrub(result_table, inn_by_le), "totals": scrub(totals, inn_by_le)},
|
||
ensure_ascii=False, indent=1, default=str,
|
||
)
|
||
)
|
||
|
||
# формы ББ/ОФР из PDF (эмулятор ИС): типы документов, строки, проверки
|
||
stmts = {}
|
||
for inn, want in sc["statements"].items():
|
||
got = sorted(d["document_type"] for n, d in docs if d["document_type"] in ("BALANCE", "INCOME") and by_inn[inn] in {f["legal_entity_id"] for f in files if f["original_name"] == n})
|
||
rep.check(f"{sid}: формы {inn}", got == sorted(want), f"{got}")
|
||
for n, d in docs:
|
||
if d["document_type"] in ("BALANCE", "INCOME"):
|
||
s = api.ok("GET", f"/documents/{d['id']}/statement")
|
||
stmts[f"{n} / {d['document_type']}"] = {
|
||
"status": s["document"]["status"],
|
||
"form_variant": s["document"]["form_variant"],
|
||
"lines": len(s["lines"]),
|
||
"codes": sorted(str(ln["line_code"]) for ln in s["lines"] if ln["line_code"]),
|
||
"failed": sorted(c.get("code", str(c)) if isinstance(c, dict) else str(c) for c in s["checks"]["failed"]),
|
||
"issues": sorted(i["code"] for i in s["issues"]),
|
||
}
|
||
rep.check(f"{sid}: форма {n} / {d['document_type']} — строки есть", len(s["lines"]) > 0, f"строк {len(s['lines'])}, статус {s['document']['status']}")
|
||
(out / "snapshots" / f"{sid}.statements.json").write_text(json.dumps(stmts, ensure_ascii=False, indent=1))
|
||
ctx["versions"][sid] = vid
|
||
|
||
|
||
def negative(api: Api, exp: dict, ctx: dict, rep: Report) -> None:
|
||
vid = next(iter(ctx["versions"].values()), None)
|
||
if not vid:
|
||
return
|
||
api.login(exp["readonly_user"])
|
||
st, _, _ = api.call("GET", f"/consolidations/versions/{vid}/osv/summary")
|
||
rep.check("права: эксперт читает свод", st == 200, f"{st}")
|
||
st, body, _ = api.call("POST", f"/consolidations/versions/{vid}/lock")
|
||
rep.check("права: эксперт не берёт блокировку — 403", st == 403, f"{st} {str(body)[:120]}")
|
||
api.login(exp["user"])
|
||
st, _, _ = api.call("GET", "/consolidations/versions/" + str(uuid.uuid4()))
|
||
rep.check("несуществующая версия — 404", st == 404, f"{st}")
|
||
|
||
|
||
def compare(out: Path, baseline: Path, rep: Report) -> None:
|
||
for cur in sorted((out / "snapshots").glob("*.json")):
|
||
old = baseline / "snapshots" / cur.name
|
||
if not old.exists():
|
||
rep.diffs.append({"file": cur.name, "detail": "нет в прошлом прогоне"})
|
||
continue
|
||
a, b = json.loads(old.read_text()), json.loads(cur.read_text())
|
||
if a == b:
|
||
continue
|
||
if isinstance(a, list) and isinstance(b, list):
|
||
sa, sb = {json.dumps(x, sort_keys=True, ensure_ascii=False) for x in a}, {json.dumps(x, sort_keys=True, ensure_ascii=False) for x in b}
|
||
detail = f"было {len(a)}, стало {len(b)}; ушло {len(sa - sb)}, появилось {len(sb - sa)}"
|
||
examples = sorted(sb - sa)[:5]
|
||
else:
|
||
ka = set(a) if isinstance(a, dict) else set()
|
||
changed = sorted(k for k in set(a) | set(b) if a.get(k) != b.get(k)) if isinstance(a, dict) else []
|
||
detail = f"изменились ключи: {changed[:10]}" + (f" (+{len(changed) - 10})" if len(changed) > 10 else "")
|
||
examples = [] if ka else [str(b)[:200]]
|
||
rep.diffs.append({"file": cur.name, "detail": detail, "examples": examples})
|
||
print(f"сравнение с {baseline}: расхождений {len(rep.diffs)}", flush=True)
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--api", required=True)
|
||
ap.add_argument("--fixtures", required=True, type=Path)
|
||
ap.add_argument("--expected", required=True, type=Path)
|
||
ap.add_argument("--out", required=True, type=Path)
|
||
ap.add_argument("--baseline", type=Path)
|
||
ap.add_argument("--timeout", type=float, default=600)
|
||
a = ap.parse_args()
|
||
exp = json.loads(a.expected.read_text())
|
||
(a.out / "snapshots").mkdir(parents=True, exist_ok=True)
|
||
rep = Report()
|
||
api = Api(a.api)
|
||
ctx: dict = {}
|
||
try:
|
||
health = wait_health(a.api, 180)
|
||
rep.check("ядро готово", True, json.dumps(health, ensure_ascii=False)[:200])
|
||
api.login(exp["user"])
|
||
groups = api.ok("GET", "/directories/company-groups", params={"q": exp["company_group"]})["items"]
|
||
group = next(g for g in groups if g["name"] == exp["company_group"])
|
||
related = next(r for r in api.ok("GET", f"/directories/company-groups/{group['id']}/related-groups") if r["name"] == exp["related_group"])
|
||
les = api.pages("/directories/legal-entities", {"company_group_id": group["id"]}, limit=200)
|
||
ctx = {"group_id": group["id"], "related_id": related["id"], "by_inn": {le["inn"]: le["id"] for le in les}, "versions": {}, "documents": {}}
|
||
ctx["accounts"] = api.ok("GET", "/directories/accounts")
|
||
rep.check("dev-seed: ГК, ГСК, ЮЛ", len(les) > 0 and all(i in ctx["by_inn"] for sc in exp["scenarios"] for i in (u["inn"] for u in sc["uploads"])), f"ЮЛ {len(les)}")
|
||
for sc in exp["scenarios"]:
|
||
try:
|
||
run_scenario(api, sc, ctx, a.fixtures, rep, a.out, a.timeout)
|
||
except Exception as e: # noqa: BLE001 — сценарий падает целиком, следующий идёт
|
||
rep.check(f"{sc['id']}: сценарий без исключений", False, f"{type(e).__name__}: {e}")
|
||
negative(api, exp, ctx, rep)
|
||
except Exception as e: # noqa: BLE001
|
||
rep.check("сценарий без исключений", False, f"{type(e).__name__}: {e}")
|
||
if a.baseline and (a.baseline / "snapshots").is_dir():
|
||
compare(a.out, a.baseline, rep)
|
||
ids = {k: ctx[k] for k in ("group_id", "versions", "documents")} if ctx else {}
|
||
report = {"checks": rep.checks, "diffs": rep.diffs, "baseline": str(a.baseline or ""), "ids": ids}
|
||
(a.out / "report.json").write_text(json.dumps(report, ensure_ascii=False, indent=1))
|
||
red = [c for c in rep.checks if not c["ok"]]
|
||
print(f"проверок {len(rep.checks)}, красных {len(red)}, расхождений со снимком {len(rep.diffs)}")
|
||
return 1 if red else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|