261 lines
9.3 KiB
Python
261 lines
9.3 KiB
Python
"""Тесты-оракулы нового режима API свода 2 (постраничная выдача, поиск, фильтры, варианты, compact).
|
||
|
||
Оракул — функции экрана «до» из tests.svod2_oracle, применённые к прежнему полному ответу.
|
||
"""
|
||
import pytest
|
||
|
||
from tests.integration.conftest import SVOD2_YEAR, legacy_full
|
||
from tests.svod2_oracle import (
|
||
compact_projection,
|
||
screen_facets,
|
||
screen_found,
|
||
screen_rows,
|
||
)
|
||
|
||
SMETA_TYPE = "КВ"
|
||
SMETA_TYPE_VARIANT = SMETA_TYPE.lower() + " "
|
||
|
||
SEARCHES = [
|
||
"аРЕНДА", "ё", "Ё", "ромаш", "лютик", "ромашка плюс", "1234.5", "1234.50", " аренда ",
|
||
" ", "1", ",", "777.25", "<cd_id>", "я" * 5000,
|
||
]
|
||
|
||
FILTERS = {
|
||
"type-case-space": {"smeta_type": SMETA_TYPE_VARIANT},
|
||
"org-twin": {"org_name": "Филиал Ёлки"},
|
||
"org-case-variant": {"org_name": "филиал ёлки "},
|
||
"unknown": {"smeta_type": "нет такого"},
|
||
"dir+org": {"smeta_direction": "Развитие", "org_name": "Филиал Ёлки"},
|
||
"search+type": {"search": "аренда", "smeta_type": SMETA_TYPE_VARIANT},
|
||
"search+type-hit": {"search": "о", "smeta_type": SMETA_TYPE_VARIANT},
|
||
"long": {"org_name": "я" * 5000},
|
||
}
|
||
|
||
SHEETS = ["GO", "RF"]
|
||
|
||
|
||
def _url(sheet: str, suffix: str = "") -> str:
|
||
return f"/api/v1/svod/2/{sheet}/{SVOD2_YEAR}{suffix}"
|
||
|
||
|
||
def _line_ids(rows: list[dict]) -> list[int]:
|
||
return [r["data"]["line_id"] for r in rows]
|
||
|
||
|
||
def _resolve_search(search: str, data: dict) -> str:
|
||
return str(data["cd_id"]) if search == "<cd_id>" else search
|
||
|
||
|
||
def _split(params: dict) -> tuple[str, dict]:
|
||
filters = dict(params)
|
||
search = filters.pop("search", "")
|
||
return search, filters
|
||
|
||
|
||
def _all_pages(client, sheet, headers, limit, **params) -> tuple[list[dict], list[int]]:
|
||
"""Все страницы нового режима: (строки, total каждой страницы)."""
|
||
rows: list[dict] = []
|
||
totals: list[int] = []
|
||
offset = 0
|
||
while True:
|
||
r = client.get(
|
||
_url(sheet),
|
||
params={**params, "offset": offset, "limit": limit},
|
||
headers=headers,
|
||
)
|
||
assert r.status_code == 200, r.text
|
||
body = r.json()
|
||
totals.append(body["total"])
|
||
rows.extend(body["result"])
|
||
if len(body["result"]) < limit:
|
||
return rows, totals
|
||
offset += limit
|
||
|
||
|
||
@pytest.fixture
|
||
def headers(admin_tokens, auth_headers):
|
||
return auth_headers(admin_tokens)
|
||
|
||
|
||
def test_legacy_mode_has_no_total(client, svod2_data, headers):
|
||
r = client.get(_url("GO", "?offset=0&limit=2"), headers=headers)
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
assert body["count"] <= 2
|
||
assert len(body["result"]) <= 2
|
||
assert "total" not in body
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"q", ["offset=-1", "limit=0", "sort_by=bad", "sort_direction=bad", "view=bad"]
|
||
)
|
||
def test_invalid_params_422(client, svod2_data, headers, q):
|
||
r = client.get(_url("GO", f"?{q}"), headers=headers)
|
||
assert r.status_code == 422
|
||
|
||
|
||
@pytest.mark.parametrize("sheet", SHEETS)
|
||
def test_pages_stitch_to_full_list(client, svod2_data, headers, sheet):
|
||
full = legacy_full(client, sheet, headers)
|
||
assert full
|
||
rows, totals = _all_pages(client, sheet, headers, 2, view="full")
|
||
assert _line_ids(rows) == _line_ids(full)
|
||
assert totals and all(t == len(full) for t in totals)
|
||
|
||
|
||
@pytest.mark.parametrize("sheet", SHEETS)
|
||
def test_page_past_end_has_total(client, svod2_data, headers, sheet):
|
||
full = legacy_full(client, sheet, headers)
|
||
r = client.get(
|
||
_url(sheet),
|
||
params={"view": "full", "offset": len(full), "limit": 2},
|
||
headers=headers,
|
||
)
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
assert body["result"] == []
|
||
assert body["count"] == 0
|
||
assert body["total"] == len(full)
|
||
|
||
|
||
def test_view_full_rows_equal_legacy_rows(client, svod2_data, headers):
|
||
full = legacy_full(client, "RF", headers)
|
||
r = client.get(
|
||
_url("RF"), params={"view": "full", "offset": 0, "limit": 50}, headers=headers
|
||
)
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
assert "total" in body
|
||
assert body["result"] == full[:50]
|
||
|
||
|
||
@pytest.mark.parametrize("sheet", SHEETS)
|
||
@pytest.mark.parametrize("search", SEARCHES, ids=[str(i) for i in range(len(SEARCHES))])
|
||
def test_search_matches_screen_oracle(client, svod2_data, headers, sheet, search):
|
||
search = _resolve_search(search, svod2_data)
|
||
full = legacy_full(client, sheet, headers)
|
||
expected = screen_rows(full, search)
|
||
rows, totals = _all_pages(client, sheet, headers, 3, view="compact", search=search)
|
||
assert _line_ids(rows) == _line_ids(expected)
|
||
assert all(t == len(expected) for t in totals)
|
||
|
||
|
||
@pytest.mark.parametrize("sheet", SHEETS)
|
||
@pytest.mark.parametrize("name", list(FILTERS))
|
||
def test_filters_match_screen_oracle(client, svod2_data, headers, sheet, name):
|
||
params = FILTERS[name]
|
||
search, filters = _split(params)
|
||
full = legacy_full(client, sheet, headers)
|
||
expected = screen_rows(full, search, filters)
|
||
rows, totals = _all_pages(client, sheet, headers, 3, view="compact", **params)
|
||
assert _line_ids(rows) == _line_ids(expected)
|
||
assert all(t == len(expected) for t in totals)
|
||
|
||
|
||
def _facets_to_dict(result: dict) -> dict:
|
||
out = {"total": result["total"], "search_total": result["search_total"]}
|
||
for col in ("smeta_type", "smeta_direction", "org_name"):
|
||
out[col] = {item["value"]: item["count"] for item in result[col]}
|
||
return out
|
||
|
||
|
||
@pytest.mark.parametrize("sheet", SHEETS)
|
||
@pytest.mark.parametrize("name", list(FILTERS))
|
||
@pytest.mark.parametrize("search", SEARCHES[:6], ids=[str(i) for i in range(6)])
|
||
def test_facets_match_screen_oracle(client, svod2_data, headers, sheet, name, search):
|
||
_, filters = _split(FILTERS[name])
|
||
full = legacy_full(client, sheet, headers)
|
||
expected = screen_facets(full, search, filters)
|
||
r = client.get(
|
||
_url(sheet, "/facets"), params={"search": search, **filters}, headers=headers
|
||
)
|
||
assert r.status_code == 200, r.text
|
||
assert _facets_to_dict(r.json()["result"]) == expected
|
||
|
||
|
||
@pytest.mark.parametrize("sheet", SHEETS)
|
||
@pytest.mark.parametrize("search", SEARCHES, ids=[str(i) for i in range(len(SEARCHES))])
|
||
def test_found_matches_screen_oracle(client, svod2_data, headers, sheet, search):
|
||
search = _resolve_search(search, svod2_data)
|
||
full = legacy_full(client, sheet, headers)
|
||
r = client.get(_url(sheet, "/facets"), params={"search": search}, headers=headers)
|
||
assert r.status_code == 200, r.text
|
||
result = r.json()["result"]
|
||
assert (result["search_total"], result["total"]) == screen_found(full, search)
|
||
|
||
|
||
@pytest.mark.parametrize("sheet", SHEETS)
|
||
def test_compact_is_projection_of_full(client, svod2_data, headers, sheet):
|
||
params = {"offset": 0, "limit": 5}
|
||
full = client.get(_url(sheet), params={**params, "view": "full"}, headers=headers)
|
||
compact = client.get(_url(sheet), params={**params, "view": "compact"}, headers=headers)
|
||
assert full.status_code == 200 and compact.status_code == 200
|
||
full_rows = full.json()["result"]
|
||
compact_rows = compact.json()["result"]
|
||
assert full_rows and len(full_rows) == len(compact_rows)
|
||
for f, c in zip(full_rows, compact_rows):
|
||
assert (c["row_type"], c["depth"], c["sort_order"]) == (
|
||
f["row_type"], f["depth"], f["sort_order"],
|
||
)
|
||
assert c["data"] == compact_projection(f["data"])
|
||
|
||
|
||
def test_rf_ckk_only_value_is_searchable(client, svod2_data, headers):
|
||
r = client.get(
|
||
_url("RF"),
|
||
params={"view": "compact", "search": svod2_data["ckk_only_value"]},
|
||
headers=headers,
|
||
)
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
assert body["total"] >= 1
|
||
assert body["result"], "строка с значением только из ККК не найдена поиском"
|
||
|
||
|
||
@pytest.mark.parametrize("sheet", SHEETS)
|
||
def test_isp_foreign_org_is_empty(client, svod2_data, isp_tokens, auth_headers, sheet):
|
||
isp = auth_headers(isp_tokens)
|
||
r = client.get(
|
||
_url(sheet),
|
||
params={"view": "full", "org_ids": svod2_data["org_b"], "offset": 0, "limit": 50},
|
||
headers=isp,
|
||
)
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
assert body["result"] == []
|
||
assert body["total"] == 0
|
||
|
||
f = client.get(
|
||
_url(sheet, "/facets"), params={"org_ids": svod2_data["org_b"]}, headers=isp
|
||
)
|
||
assert f.status_code == 200
|
||
result = f.json()["result"]
|
||
assert result["total"] == 0 and result["search_total"] == 0
|
||
assert result["smeta_type"] == []
|
||
assert result["smeta_direction"] == []
|
||
assert result["org_name"] == []
|
||
|
||
|
||
@pytest.mark.parametrize("sheet", SHEETS)
|
||
def test_isp_sees_only_own_org(
|
||
client, svod2_data, headers, isp_tokens, auth_headers, sheet
|
||
):
|
||
admin_rows = legacy_full(client, sheet, headers)
|
||
own = [
|
||
r for r in admin_rows
|
||
if r["data"]["header"]["source_org_unit_id"] == svod2_data["org_a"]
|
||
]
|
||
assert own
|
||
r = client.get(
|
||
_url(sheet),
|
||
params={"view": "full", "offset": 0, "limit": 50},
|
||
headers=auth_headers(isp_tokens),
|
||
)
|
||
assert r.status_code == 200
|
||
assert r.json()["total"] == len(own)
|
||
|
||
|
||
def test_facets_unknown_sheet_404(client, svod2_data, headers):
|
||
r = client.get(_url("XX", "/facets"), headers=headers)
|
||
assert r.status_code == 404
|