DFiP_Budget_planing/api/tests/integration/test_svod2_paging_api.py

"""Тесты-оракулы нового режима 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,
    "ооо лютик", "дс-2", "дс-1", "70",
]

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 _strip_nulls(value):
    if isinstance(value, dict):
        return {k: _strip_nulls(v) for k, v in value.items() if v is not None}
    if isinstance(value, list):
        return [_strip_nulls(v) for v in value]
    return value


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
    # Режим страниц отдаёт data целиком (с null), ответ без параметров - без null-полей (AURORA-1848).
    assert _strip_nulls(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)


# Решение пользователя 08.11 (R-30): серверный поиск совпадает с тем, что видно на экране. Пустое (null)
# поле contract_detail не затирает такое же поле contract_summary: экран получает data без null-полей,
# и в {...header, ...contract_summary, ...contract_detail} побеждает последнее НЕпустое значение.
# Строка 910004 имеет contract_summary (ООО Лютик, ДС-2, 70), но НЕ имеет contract_detail.
LINE_NO_DETAIL = 910004
# Строка 910001: contract_summary (ООО Ромашка, ДС-1), в contract_detail counterparty = NULL, reference = Д-1.
LINE_EMPTY_DETAIL_FIELD = 910001


@pytest.mark.parametrize("sheet", SHEETS)
@pytest.mark.parametrize("search", ["ооо лютик", "дс-2", "70"])
def test_search_summary_value_found_when_detail_row_missing(
    client, svod2_data, headers, sheet, search
):
    full = legacy_full(client, sheet, headers)
    expected = screen_rows(full, search)
    rows, _ = _all_pages(client, sheet, headers, 3, view="compact", search=search)
    assert _line_ids(rows) == _line_ids(expected)
    if sheet == "GO":
        assert LINE_NO_DETAIL in _line_ids(rows)


@pytest.mark.parametrize("search,found", [("ооо ромашка", True), ("дс-1", False), ("д-1", True)])
def test_search_summary_value_found_when_detail_field_is_empty(
    client, svod2_data, headers, search, found
):
    """detail.counterparty = NULL, summary.counterparty = «ООО Ромашка»: строка находится по значению summary.

    reference у detail непустой (Д-1) и перекрывает summary (ДС-1): по «дс-1» строка не находится.
    """
    full = legacy_full(client, "GO", headers)
    assert (LINE_EMPTY_DETAIL_FIELD in _line_ids(screen_rows(full, search))) is found
    rows, _ = _all_pages(client, "GO", headers, 3, view="compact", search=search)
    assert (LINE_EMPTY_DETAIL_FIELD in _line_ids(rows)) is found


def test_search_detail_value_overrides_summary_value(client, svod2_data, svod2_sql, headers):
    """Непустое поле contract_detail по-прежнему перекрывает contract_summary: старое значение не находится."""
    svod2_sql(
        "INSERT INTO v3.contract_summary (line_id, counterparty, reference) "
        "VALUES (910002, 'Старый контрагент', 'Старая ссылка')"
    )
    for search, found in (("старый контрагент", False), ("ромашка плюс", True), ("старая ссылка", False)):
        full = legacy_full(client, "GO", headers)
        assert (910002 in _line_ids(screen_rows(full, search))) is found
        rows, _ = _all_pages(client, "GO", headers, 3, view="compact", search=search)
        assert (910002 in _line_ids(rows)) is found, search


@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