135 lines
6.2 KiB
Python
135 lines
6.2 KiB
Python
"""SQL-нормализаторы свода 2 (миграция 0053): регистр, обрезка, запись числа как в JS, компактная проекция."""
|
||
import json
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from tests.integration.conftest import legacy_full
|
||
from tests.svod2_oracle import compact_projection
|
||
|
||
NUM = json.loads(
|
||
(Path(__file__).resolve().parent.parent / "fixtures" / "js_number_text_vectors.json").read_text(
|
||
encoding="utf-8"
|
||
)
|
||
)
|
||
|
||
|
||
def test_js_number_text_matches_node_vectors(svod2_sql):
|
||
for v in NUM:
|
||
got = svod2_sql(
|
||
"SELECT v3.svod2_js_number_text((CAST(:x AS jsonb) #>> '{}')::float8) AS t",
|
||
{"x": v["input"]},
|
||
)[0][0]
|
||
assert got == v["js"], v
|
||
|
||
|
||
def test_js_number_text_ignores_session_extra_float_digits(svod2_sql):
|
||
svod2_sql("SET LOCAL extra_float_digits = 0", None)
|
||
got = svod2_sql("SELECT v3.svod2_js_number_text(0.1::float8 + 0.2::float8) AS t", None)[0][0]
|
||
assert got == "0.30000000000000004"
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"src",
|
||
["АБВГДЕЁЖЗИЙКЛМНОПРСТУФХЦЧШЩЪЫЬЭЮЯ", "ЄІЇЎЂЏҐ", "ABCxyz", "ÀÉÖÞ", "×", "аРЕНДА 1"],
|
||
)
|
||
def test_casefold_matches_js_lower(svod2_sql, src):
|
||
assert svod2_sql("SELECT v3.svod2_casefold(:s) AS t", {"s": src})[0][0] == src.lower()
|
||
|
||
|
||
def test_casefold_independent_of_collation(svod2_sql):
|
||
got = svod2_sql("""SELECT v3.svod2_casefold('ЁЛКА' COLLATE "C") AS t""", None)[0][0]
|
||
assert got == "ёлка"
|
||
|
||
|
||
def test_js_trim_ecmascript_whitespace(svod2_sql):
|
||
assert svod2_sql("SELECT v3.svod2_js_trim(:s) AS t", {"s": " x \t"})[0][0] == "x"
|
||
assert svod2_sql("SELECT v3.svod2_js_trim(:s) AS t", {"s": " x"})[0][0] == " x"
|
||
|
||
|
||
def test_compact_matches_python_projection(svod2_sql, svod2_data, client, admin_tokens, auth_headers):
|
||
full = legacy_full(client, "RF", auth_headers(admin_tokens))[0]["data"]
|
||
got = svod2_sql("SELECT v3.svod2_compact(CAST(:d AS jsonb)) AS c", {"d": json.dumps(full)})[0][0]
|
||
assert got == compact_projection(full)
|
||
|
||
|
||
def test_single_signature_without_overloads(svod2_sql):
|
||
rows = svod2_sql(
|
||
"SELECT proname, pronargs FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace "
|
||
"WHERE nspname = 'v3' AND proname IN ('v_svod2_go', 'v_svod2_rf', 'v_svod2_go_export')",
|
||
None,
|
||
)
|
||
assert sorted((r.proname, r.pronargs) for r in rows) == [
|
||
("v_svod2_go", 11), ("v_svod2_go_export", 4), ("v_svod2_rf", 11)]
|
||
|
||
|
||
@pytest.mark.parametrize("fn", ["v_svod2_go", "v_svod2_rf"])
|
||
def test_page_total_count_and_order(svod2_sql, svod2_data, admin_tokens, fn):
|
||
# admin_tokens: запрос к API нужен, чтобы откат сессии клиента сработал (loop_ref задаётся при обращении к get_db)
|
||
full = svod2_sql(f"SELECT source_line_id FROM v3.{fn}(2031)", None)
|
||
page = svod2_sql(
|
||
f"SELECT source_line_id, total_count FROM v3.{fn}(2031, NULL, 0, 2, 'org_name', 'asc', "
|
||
"NULL, NULL, NULL, NULL, false)", None)
|
||
assert [r.source_line_id for r in page] == [r.source_line_id for r in full][:2]
|
||
assert {r.total_count for r in page} == {len(full)}
|
||
|
||
|
||
def test_facets_totals(svod2_sql, svod2_data, admin_tokens):
|
||
rows = {(r.kind, r.value): r.n for r in svod2_sql(
|
||
"SELECT * FROM v3.v_svod2_facets('GO', 2031, NULL, NULL, NULL, NULL, NULL)", None)}
|
||
assert rows[("total", None)] == rows[("search_total", None)] == len(
|
||
svod2_sql("SELECT 1 FROM v3.v_svod2_go(2031)", None))
|
||
|
||
|
||
SEARCHES = ["", "аренда", "АРЕНДА", "ё", "1234.5", "1234.50", " кв ", "1", "true", "777.25", "нет такого"]
|
||
|
||
|
||
@pytest.mark.parametrize("sheet,fn", [("GO", "v_svod2_go"), ("RF", "v_svod2_rf")])
|
||
@pytest.mark.parametrize("search", SEARCHES)
|
||
def test_search_matches_screen_oracle(svod2_sql, svod2_data, client, admin_tokens, auth_headers,
|
||
sheet, fn, search):
|
||
from tests.svod2_oracle import screen_rows
|
||
|
||
rows = legacy_full(client, sheet, auth_headers(admin_tokens))
|
||
expected = [r["data"]["line_id"] for r in screen_rows(rows, search)]
|
||
got = svod2_sql(
|
||
f"SELECT source_line_id, total_count FROM v3.{fn}(2031, NULL, NULL, NULL, 'org_name', 'asc', :s)",
|
||
{"s": search},
|
||
)
|
||
assert sorted(r.source_line_id for r in got) == sorted(expected)
|
||
if got:
|
||
assert {r.total_count for r in got} == {len(expected)}
|
||
|
||
|
||
@pytest.mark.parametrize("sheet", ["GO", "RF"])
|
||
@pytest.mark.parametrize("search,smeta_type", [("", None), ("1", "кв "), ("аренда", "АХР"), ("", "Операц")])
|
||
def test_facets_match_screen_oracle(svod2_sql, svod2_data, client, admin_tokens, auth_headers,
|
||
sheet, search, smeta_type):
|
||
from tests.svod2_oracle import screen_facets
|
||
|
||
rows = legacy_full(client, sheet, auth_headers(admin_tokens))
|
||
filters = {"smeta_type": smeta_type} if smeta_type else {}
|
||
expected = screen_facets(rows, search, filters)
|
||
got = svod2_sql(
|
||
"SELECT kind, value, n FROM v3.v_svod2_facets(:sh, 2031, NULL, :s, :t, NULL, NULL)",
|
||
{"sh": sheet, "s": search or None, "t": smeta_type},
|
||
)
|
||
assert next(r.n for r in got if r.kind == "total") == expected["total"]
|
||
assert next(r.n for r in got if r.kind == "search_total") == expected["search_total"]
|
||
for col in ("smeta_type", "smeta_direction", "org_name"):
|
||
assert {r.value: r.n for r in got if r.kind == col} == expected[col]
|
||
|
||
|
||
def test_facets_search_subplan_is_not_correlated(svod2_sql):
|
||
import re
|
||
|
||
plan = "\n".join(r[0] for r in svod2_sql(
|
||
"EXPLAIN (VERBOSE, COSTS OFF) "
|
||
"SELECT * FROM v3.v_svod2_facets('GO', 2031, NULL, 'аренда', NULL, NULL, NULL)", None))
|
||
# IN по svod2_lines должен быть hashed SubPlan (считается один раз); коррелированный
|
||
# считался бы на каждую строку. Не-hashed SubPlan в Filter внутри svod2_lines (EXISTS поиска)
|
||
# законен, поэтому смотрим только на выражения в Output.
|
||
outputs = [l for l in plan.splitlines() if "Output:" in l and "SubPlan" in l]
|
||
assert any("hashed SubPlan" in l for l in outputs), plan
|
||
assert not any(re.search(r"(?<!hashed )SubPlan \d+", l) for l in outputs), plan
|