From 174ec73732a8ee1650cefdfdfe22501168893495 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Wed, 28 Oct 2026 21:11:17 +0400 Subject: [PATCH] =?UTF-8?q?test(api):=20=D1=81=D0=B2=D0=BE=D0=B4=202=20?= =?UTF-8?q?=E2=80=94=20Python-=D0=BE=D1=80=D0=B0=D0=BA=D1=83=D0=BB=20?= =?UTF-8?q?=D1=8D=D0=BA=D1=80=D0=B0=D0=BD=D0=B0,=20=D1=81=D0=B2=D0=B5?= =?UTF-8?q?=D1=80=D0=B5=D0=BD=D0=BD=D1=8B=D0=B9=20=D1=81=20=D0=B2=D0=B5?= =?UTF-8?q?=D0=BA=D1=82=D0=BE=D1=80=D0=B0=D0=BC=D0=B8=20JS?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- api/tests/svod2_oracle.py | 177 ++++++++++++++++++++++++++++ api/tests/unit/test_svod2_oracle.py | 51 ++++++++ 2 files changed, 228 insertions(+) create mode 100644 api/tests/svod2_oracle.py create mode 100644 api/tests/unit/test_svod2_oracle.py diff --git a/api/tests/svod2_oracle.py b/api/tests/svod2_oracle.py new file mode 100644 index 0000000..176cfc8 --- /dev/null +++ b/api/tests/svod2_oracle.py @@ -0,0 +1,177 @@ +"""Python-порт оракула экрана «до» свода 2 (perf/analysis/svod2_search_oracle.mjs). + +Поведение сверяется с векторами, снятыми в Node.js (tests/fixtures/*.json). +""" + +from __future__ import annotations + +import math +from decimal import Decimal + +FILTER_COLUMNS = ("smeta_type", "smeta_direction", "org_name") + +# ECMAScript WhiteSpace + LineTerminator (design § 3). +_JS_WHITESPACE = ( + "\u0009\u000a\u000b\u000c\u000d   " + "           " + "

  　" +) + + +def js_trim(s: str) -> str: + return s.strip(_JS_WHITESPACE) + + +def js_number_text(x: int | float) -> str: + """ECMAScript Number::toString (radix 10).""" + f = float(x) + if math.isnan(f): + return "NaN" + if f == 0: + return "0" + if math.isinf(f): + return "Infinity" if f > 0 else "-Infinity" + sign = "-" if f < 0 else "" + # Кратчайшая запись цифр: repr(float) -> Decimal -> (digits, exponent). + t = Decimal(repr(abs(f))).as_tuple() + digits = "".join(map(str, t.digits)).rstrip("0") + # значение = 0.d1d2...dk * 10^n + n = len(t.digits) + t.exponent + k = len(digits) + if not digits: + return "0" + # убрать ведущие нули (Decimal('0.001') даёт digits (1,), без лишних; страховка) + stripped = digits.lstrip("0") + n -= len(digits) - len(stripped) + digits, k = stripped, len(stripped) + if k <= n <= 21: + body = digits + "0" * (n - k) + elif 0 < n <= 21: + body = digits[:n] + "." + digits[n:] + elif -6 < n <= 0: + body = "0." + "0" * (-n) + digits + else: + e = n - 1 + es = ("+" if e >= 0 else "-") + str(abs(e)) + body = (digits[0] + ("." + digits[1:] if k > 1 else "")) + "e" + es + return sign + body + + +def js_string(value: object) -> str: + """String(value ?? '').""" + if value is None: + return "" + if isinstance(value, bool): + return "true" if value else "false" + if isinstance(value, (int, float)): + return js_number_text(value) + return str(value) + + +def matches_search(row: dict, query: str) -> bool: + data = row.get("data") or {} + merged = { + **(data.get("header") or {}), + **(data.get("contract_summary") or {}), + **(data.get("contract_detail") or {}), + } + return any(query in js_string(v).lower() for v in merged.values()) + + +def _by_search(rows: list[dict], search: str) -> list[dict]: + q = js_trim(search or "").lower() + return [r for r in rows if matches_search(r, q)] if q else rows + + +def _cell(row: dict, col: str): + return ((row.get("data") or {}).get("header") or {}).get(col) + + +def _equals(cell, filter_value: str) -> bool: + if cell is None: + return False + return js_trim(js_string(cell).lower()) == js_trim(filter_value.lower()) + + +def _by_filters(rows: list[dict], filters: dict[str, str], skip: str | None) -> list[dict]: + active = [c for c in FILTER_COLUMNS if c != skip and filters.get(c)] + return [r for r in rows if all(_equals(_cell(r, c), filters[c]) for c in active)] + + +def screen_rows(rows, search: str = "", filters=None) -> list[dict]: + return _by_filters(_by_search(rows, search), filters or {}, None) + + +def screen_found(rows, search: str) -> tuple[int, int]: + return len(_by_search(rows, search)), len(rows) + + +def screen_facets(rows, search: str = "", filters=None) -> dict: + filters = filters or {} + searched = _by_search(rows, search) + result: dict = {"total": len(rows), "search_total": len(searched)} + for col in FILTER_COLUMNS: + counts: dict = {} + for r in _by_filters(searched, filters, col): + v = _cell(r, col) + if v is not None: + counts[v] = counts.get(v, 0) + 1 + result[col] = dict(sorted(counts.items())) + return result + + +_Q = ("adj_current", "adj_ssp", "adj_rf", "adj_reserve", "payment_date", "payment_amount", + "payment_amount_ho", "payment_amount_rf", "payment_act", "booking_amount", + "actual_m1", "actual_m2", "actual_m3") + + +def _build_paths() -> tuple[str, ...]: + p = ["line_id"] + p += [f"header.{f}" for f in ("source_form", "source_id", "smeta_type", "smeta_direction", + "org_name", "name", "item_id", "num_group", "justification")] + p += [f"plan.plan_q{i}" for i in range(1, 5)] + p += ["allocation.internal_order", "allocation.property_object"] + p += [f"sequestration.DFIP.adj_q{i}" for i in range(1, 5)] + p += [f"reserve.amount_q{i}" for i in range(1, 5)] + p += [f"approved.approved_q{i}" for i in range(1, 5)] + p += [f"contract_detail.{f}" for f in ("counterparty", "reference", "subject", "currency", + "ceiling_amount")] + p += [f"contract_detail.expenses_q{i}" for i in range(1, 5)] + p += ["contract_detail.vat_rate", "contract_detail.deadline"] + for q in range(1, 5): + p += [f"q{q}.{f}" for f in _Q] + if q == 4: + p.append("q4.actual_spod") + if q <= 3: + p += [f"q{q}.transfer_to_q{k}" for k in range(q + 1, 5)] + p.append(f"q{q}.transfer_to_economy") + if q >= 2: + p += [f"q{q}.{f}" for f in ("plan_revision_eco_change", "plan_revision_item_adj", + "plan_revision_increase", "plan_revision_sequester")] + return tuple(p) + + +COMPACT_PATHS: tuple[str, ...] = _build_paths() + + +def compact_projection(data: dict) -> dict: + """jsonb_strip_nulls(jsonb_build_object(...)) по COMPACT_PATHS. + + Группы и подгруппы присутствуют всегда; листья None/отсутствующие опускаются. + """ + out: dict = {} + for path in COMPACT_PATHS: + parts = path.split(".") + src = data + for part in parts[:-1]: + src = src.get(part) if isinstance(src, dict) else None + dst = out + for part in parts[:-1]: + dst = dst.setdefault(part, {}) + leaf = src.get(parts[-1]) if isinstance(src, dict) else None + if len(parts) == 1: + if leaf is not None: + out[parts[0]] = leaf + elif leaf is not None: + dst[parts[-1]] = leaf + return out diff --git a/api/tests/unit/test_svod2_oracle.py b/api/tests/unit/test_svod2_oracle.py new file mode 100644 index 0000000..19422e5 --- /dev/null +++ b/api/tests/unit/test_svod2_oracle.py @@ -0,0 +1,51 @@ +import json +from pathlib import Path + +import pytest + +from tests.svod2_oracle import ( + COMPACT_PATHS, + FILTER_COLUMNS, + compact_projection, + js_number_text, + screen_facets, + screen_found, + screen_rows, +) + +FIX = Path(__file__).resolve().parents[1] / "fixtures" +NUM = json.loads((FIX / "js_number_text_vectors.json").read_text()) +SCREEN = json.loads((FIX / "svod2_screen_vectors.json").read_text()) + + +def test_js_number_text_matches_node_vectors(): + for v in NUM: + assert js_number_text(json.loads(v["input"])) == v["js"], v + + +@pytest.mark.parametrize("case", SCREEN["cases"], ids=lambda c: f"{c['search']!r}-{c['filters']}") +def test_screen_port_matches_node(case): + rows = SCREEN["rows"] + assert [r["data"]["line_id"] for r in screen_rows(rows, case["search"], case["filters"])] == case["line_ids"] + assert screen_found(rows, case["search"]) == (case["found"]["found"], case["found"]["total"]) + f = screen_facets(rows, case["search"], case["filters"]) + assert (f["total"], f["search_total"]) == (case["facets"]["total"], case["facets"]["search_total"]) + for col in FILTER_COLUMNS: + assert f[col] == dict(case["facets"][col]) + + +def test_compact_projection_keeps_zero_drops_null_and_unknown(): + out = compact_projection({"line_id": 1, "header": {"source_form": "FORM_1", "name": None, "vsp_id": 5}, + "plan": {"plan_q1": 0, "plan_q2": None}}) + assert out["line_id"] == 1 + assert out["header"] == {"source_form": "FORM_1"} + assert out["plan"] == {"plan_q1": 0} + assert out["sequestration"] == {"DFIP": {}} + assert "transfer_to_q1" not in out["q1"] and set(out) >= {"q1", "q2", "q3", "q4", "contract_detail"} + + +def test_compact_paths_shape(): + assert len(COMPACT_PATHS) == len(set(COMPACT_PATHS)) + assert "q4.actual_spod" in COMPACT_PATHS and "q1.actual_spod" not in COMPACT_PATHS + assert "q1.transfer_to_q2" in COMPACT_PATHS and "q3.transfer_to_q4" in COMPACT_PATHS + assert "q4.transfer_to_economy" not in COMPACT_PATHS and "q1.plan_revision_increase" not in COMPACT_PATHS