95 lines
4.0 KiB
Python
95 lines
4.0 KiB
Python
"""Самопроверка compare_responses (без стенда)."""
|
|
import gzip
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "analysis"))
|
|
|
|
import compare_responses as cr # noqa: E402
|
|
|
|
|
|
def test_strip_nulls_pg_semantics():
|
|
v = {"a": None, "b": {"c": None}, "d": [None, {"e": None, "f": 1}], "g": {}}
|
|
assert cr.strip_nulls(v) == {"b": {}, "d": [None, {"f": 1}], "g": {}}
|
|
|
|
|
|
def test_strip_response_keeps_wrapper_and_sort_order():
|
|
body = {"success": True, "message": None, "count": 1,
|
|
"result": [{"row_type": "INPUT", "depth": 3, "sort_order": None, "data": {"x": None, "y": 1}}]}
|
|
out = cr.strip_response(body)
|
|
assert out["message"] is None
|
|
assert out["result"][0]["sort_order"] is None
|
|
assert out["result"][0]["data"] == {"y": 1}
|
|
assert body["result"][0]["data"] == {"x": None, "y": 1} # исходное не изменено
|
|
|
|
|
|
def test_first_diff_path():
|
|
a = {"result": [{"data": {"q1": {"m1": 1}}}]}
|
|
b = {"result": [{"data": {"q1": {"m1": 2}}}]}
|
|
assert cr.first_diff(a, b).startswith("$.result[0].data.q1.m1")
|
|
assert cr.first_diff(a, a) is None
|
|
assert cr.first_diff({"a": [1]}, {"a": [1, 2]}).startswith("$.a")
|
|
assert cr.first_diff({"a": 1}, {"a": 1, "b": 2}).startswith("$")
|
|
|
|
|
|
def _write(d: Path, name: str, body: dict, status: int = 200):
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
(d / f"{name}.json.gz").write_bytes(gzip.compress(json.dumps(body).encode()))
|
|
idx = d / "index.json"
|
|
cur = json.loads(idx.read_text()) if idx.exists() else {}
|
|
cur[name] = status
|
|
idx.write_text(json.dumps(cur))
|
|
|
|
|
|
def _body(data):
|
|
return {"success": True, "message": None, "count": 1,
|
|
"result": [{"row_type": "INPUT", "depth": 1, "sort_order": 1, "data": data}]}
|
|
|
|
|
|
def test_compare_exact_and_strip_modes(tmp_path):
|
|
b, a = tmp_path / "before", tmp_path / "after"
|
|
for name in ("sheet-FORM_2-AHR-none-1", "report-5-2026-LIMIT"):
|
|
_write(b, name, _body({"x": None, "y": 1}))
|
|
_write(a, name, _body({"y": 1}))
|
|
n, bad = cr.compare(b, a, "exact")
|
|
assert n == 0 and len(bad) == 2
|
|
n, bad = cr.compare(b, a, "strip-nulls")
|
|
assert n == 1 and len(bad) == 1 and "report-5-2026-LIMIT" in bad[0]
|
|
|
|
|
|
def test_compare_status_mismatch(tmp_path):
|
|
b, a = tmp_path / "before", tmp_path / "after"
|
|
_write(b, "svod1-MAIN", _body({"y": 1}), 200)
|
|
_write(a, "svod1-MAIN", _body({"y": 1}), 500)
|
|
n, bad = cr.compare(b, a, "exact")
|
|
assert n == 0 and len(bad) == 1
|
|
|
|
|
|
def test_compare_missing_after(tmp_path):
|
|
b, a = tmp_path / "before", tmp_path / "after"
|
|
_write(b, "svod1-MAIN", _body({}))
|
|
a.mkdir()
|
|
n, bad = cr.compare(b, a, "exact")
|
|
assert n == 0 and len(bad) == 1
|
|
|
|
|
|
def test_build_requests_minimal_manifest():
|
|
man = {"sheets": [
|
|
{"form_id": 1, "form_type": "FORM_2", "sheet": "AHR", "direction": None, "year": 2026, "url": "", "rows": 1, "owner": "rf"},
|
|
{"form_id": 2, "form_type": "FORM_1", "sheet": "AHR", "direction": "Support", "year": 2026, "url": "", "rows": 1, "owner": "ssp"},
|
|
], "project_reports": [{"project_id": 5, "year": 2026, "report_type": "LIMIT", "url": "", "rows": 1}]}
|
|
reqs = cr.build_requests(man)
|
|
by = {n: (r, p) for r, n, p in reqs}
|
|
assert by["sheet-FORM_2-AHR-none-1"] == ("rf", "/api/v1/form/1/sheet/AHR")
|
|
assert by["sheet-FORM_2-AHR-none-1-sections"] == ("rf", "/api/v1/form/1/sheet/AHR?sections=plan,q1")
|
|
assert by["sheet-FORM_1-AHR-Support-2"] == ("dfip", "/api/v1/form/2/sheet/AHR?direction=Support")
|
|
assert sum(n.startswith("svod1-") for n in by) == 5
|
|
assert sum(n.startswith("svod2-") for n in by) == 2
|
|
assert by["svod1-MAIN"] == ("dfip", "/api/v1/svod/1/MAIN/2026")
|
|
assert by["report-5-2026-LIMIT"][0] == "rf"
|
|
assert by["project-summary-5"] == ("rf", "/api/v1/projects/5/summary")
|
|
assert {"projects-plan-summary", "projects-fact-summary", "projects-detail-summary"} <= set(by)
|
|
assert not any("rollup" in n for n in by)
|
|
assert cr.is_stripped("sheet-x") and cr.is_stripped("svod2-GO") and not cr.is_stripped("report-1")
|