7dcffbd10d
CounterpartyIndex.resolve: порядок шагов ИНН -> точное имя -> алиас -> нормализованное имя (OQ-128, §3 п. 4). Алиас уникален (ALIAS_TAKEN), поэтому владелец алиаса берётся с match_kind = ALIAS, даже если есть ЮЛ-тёзки по нормализованному имени; NULL остаётся при нескольких совпадениях только по имени. Правило общее для маппинга (Срез 2c) и дорезолва свода на чтение (§3 п. 7). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
458 lines
21 KiB
Python
458 lines
21 KiB
Python
"""API сводной ОСВ версии (spec consolidated-osv «Страница свода и сводка»; design D-1, D-16): страница, фильтры,
|
||
сводка с пробелами, суммы-строки, права и видимость, производительность без N+1."""
|
||
|
||
import time
|
||
import uuid
|
||
from decimal import Decimal
|
||
from typing import Any
|
||
|
||
import pytest
|
||
from sqlalchemy import event, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from src.domain.aliases import normalize_alias
|
||
from src.models.access import AppUser
|
||
from src.models.consolidation import ConsolidationDocument, ConsolidationVersion
|
||
from src.models.directories import CompanyGroup, LegalEntity, LegalEntityAlias, RegionalOffice
|
||
from src.models.enums import ConsolidationStatus, CounterpartyMatchKind, PeriodQuarter
|
||
from src.models.files import Document
|
||
from src.models.osv import OsvEntry
|
||
from tests.conftest import AuthClientFactory
|
||
from tests.factories import make_company_group, make_confirmed_osv, make_consolidation, make_legal_entity
|
||
|
||
HTTP_OK = 200
|
||
HTTP_FORBIDDEN = 403
|
||
HTTP_NOT_FOUND = 404
|
||
HTTP_UNPROCESSABLE = 422
|
||
KA_RF = "ka_rf@armka.local"
|
||
CURATOR = "curator_go@armka.local"
|
||
FOREIGN = "client_manager@armka.local" # Краснодарский филиал
|
||
ROSTOV = "Ростовский региональный филиал"
|
||
YEAR = 2024
|
||
Q1, Q2, Q3 = PeriodQuarter.Q1, PeriodQuarter.Q2, PeriodQuarter.Q3
|
||
D = Decimal
|
||
ZERO = D("0.00")
|
||
ROWS_TOTAL, UNMAPPED_TOTAL, AUTO_MATCHED_TOTAL, DOCUMENTS_TOTAL = 8, 2, 2, 7
|
||
PERF_MAX_SECONDS = 5.0
|
||
PERF_MAX_SELECTS = 10
|
||
|
||
|
||
class World:
|
||
"""Версия РБ Q3 2024: периметр {А, Б}; вне периметра — ЧКФ (та же ГК), «Внешний» (имя), Ф (алиас)."""
|
||
|
||
version: ConsolidationVersion
|
||
a: LegalEntity
|
||
b: LegalEntity
|
||
chkf: LegalEntity
|
||
external: LegalEntity
|
||
aliased: LegalEntity
|
||
|
||
|
||
def _quarter_rows(world: World, quarter: PeriodQuarter) -> list[dict[str, Any]]:
|
||
"""Строки ОСВ 60 «А»: opening/turnover/closing по кварталам 100/10/110, 110/10/120, 120/10/130."""
|
||
step = {Q1: 0, Q2: 10, Q3: 20}[quarter]
|
||
amounts = {
|
||
"opening_debit": D(100 + step),
|
||
"turnover_debit": D(10),
|
||
"closing_debit": D(110 + step),
|
||
}
|
||
return [
|
||
{
|
||
"name": "ООО Б",
|
||
"inn": world.b.inn,
|
||
"counterparty_id": world.b.id,
|
||
"counterparty_match_kind": CounterpartyMatchKind.INN,
|
||
**amounts,
|
||
},
|
||
{
|
||
"name": "ООО ЧКФ",
|
||
"counterparty_id": world.chkf.id,
|
||
"counterparty_match_kind": CounterpartyMatchKind.MANUAL,
|
||
**amounts,
|
||
},
|
||
{"name": "Неизвестный контрагент", **amounts},
|
||
{"name": "ООО Внешний", **amounts},
|
||
{"name": "Пекарня Ф", **amounts},
|
||
]
|
||
|
||
|
||
async def _bind(session: AsyncSession, version: ConsolidationVersion, *documents: Any) -> None:
|
||
for document in documents:
|
||
session.add(ConsolidationDocument(consolidation_version_id=version.id, document_id=document.id))
|
||
await session.flush()
|
||
|
||
|
||
async def _world(session: AsyncSession, *, status: ConsolidationStatus = ConsolidationStatus.DATA_CORRECTION) -> World:
|
||
office = (await session.execute(select(RegionalOffice).where(RegionalOffice.name == ROSTOV))).scalar_one()
|
||
group: CompanyGroup = await make_company_group(session, name="ГК Русский бисквит", office=office)
|
||
world = World()
|
||
world.a = await make_legal_entity(session, name="ООО А", group=group)
|
||
world.b = await make_legal_entity(session, name="ООО Б", group=group)
|
||
world.chkf = await make_legal_entity(session, name="ООО ЧКФ", group=group)
|
||
world.external = await make_legal_entity(session, name="ООО Внешний")
|
||
world.aliased = await make_legal_entity(session, name="Фирма Эф")
|
||
session.add(
|
||
LegalEntityAlias(
|
||
legal_entity_id=world.aliased.id, alias="Пекарня Ф", normalized_alias=normalize_alias("Пекарня Ф")
|
||
)
|
||
)
|
||
user = (await session.execute(select(AppUser).where(AppUser.email == KA_RF))).scalar_one()
|
||
_, world.version = await make_consolidation(
|
||
session, group=group, year=YEAR, quarter=Q3, members=[world.a, world.b], initiator=user, status=status
|
||
)
|
||
documents = []
|
||
for quarter in (Q1, Q2, Q3):
|
||
documents.append(
|
||
await make_confirmed_osv(
|
||
session,
|
||
entity=world.a,
|
||
account="60",
|
||
year=YEAR,
|
||
quarter=quarter,
|
||
is_cumulative=False,
|
||
rows=_quarter_rows(world, quarter),
|
||
)
|
||
)
|
||
documents.append(
|
||
await make_confirmed_osv(
|
||
session,
|
||
entity=world.b,
|
||
account="60",
|
||
year=YEAR,
|
||
quarter=quarter,
|
||
is_cumulative=False,
|
||
rows=[
|
||
{
|
||
"name": "ООО А",
|
||
"counterparty_id": world.a.id,
|
||
"counterparty_match_kind": CounterpartyMatchKind.MANUAL,
|
||
"turnover_credit": D("5.00"),
|
||
},
|
||
{"name": "Неизвестный Б", "turnover_credit": D("1.00")},
|
||
],
|
||
)
|
||
)
|
||
documents.append(
|
||
await make_confirmed_osv(
|
||
session,
|
||
entity=world.a,
|
||
account="62",
|
||
year=YEAR,
|
||
quarter=Q3,
|
||
is_cumulative=True,
|
||
rows=[
|
||
{
|
||
"name": "ООО Б",
|
||
"inn": world.b.inn,
|
||
"counterparty_id": world.b.id,
|
||
"counterparty_match_kind": CounterpartyMatchKind.INN,
|
||
"closing_credit": D("42.50"),
|
||
}
|
||
],
|
||
)
|
||
)
|
||
await _bind(session, world.version, *documents)
|
||
return world
|
||
|
||
|
||
def _url(world: World, suffix: str = "osv") -> str:
|
||
return f"/consolidations/versions/{world.version.id}/{suffix}"
|
||
|
||
|
||
async def _page(client: Any, world: World, query: str = "") -> dict[str, Any]:
|
||
response = await client.get(f"{_url(world)}?{query}")
|
||
assert response.status_code == HTTP_OK, response.text
|
||
body: dict[str, Any] = response.json()
|
||
return body
|
||
|
||
|
||
def _row(body: dict[str, Any], entity: str, account: str, name: str) -> dict[str, Any]:
|
||
return next(
|
||
row
|
||
for row in body["items"]
|
||
if (row["legal_entity_name"], row["account_code"], row["counterparty_name"]) == (entity, account, name)
|
||
)
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_page_rows_order_total_and_money_strings(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
world = await _world(seeded)
|
||
client = await auth_client(KA_RF)
|
||
|
||
body = await _page(client, world)
|
||
|
||
assert (body["total"], body["limit"], body["offset"]) == (ROWS_TOTAL, 500, 0)
|
||
assert [(r["legal_entity_name"], r["account_code"], r["row_number"]) for r in body["items"]] == [
|
||
("ООО А", "60", 1), ("ООО А", "60", 2), ("ООО А", "60", 3), ("ООО А", "60", 4), ("ООО А", "60", 5),
|
||
("ООО А", "62", 1), ("ООО Б", "60", 1), ("ООО Б", "60", 2),
|
||
] # fmt: skip
|
||
row = _row(body, "ООО А", "60", "ООО Б")
|
||
assert row["period"] == "2024 Q3" and row["is_cumulative"] is False
|
||
assert (row["opening_debit"], row["turnover_debit"], row["closing_debit"]) == ("100.00", "30.00", "130.00")
|
||
assert (row["opening_credit"], row["turnover_credit"], row["closing_credit"]) == ("0.00", "0.00", "0.00")
|
||
assert len(row["entry_ids"]) == 3 # noqa: PLR2004
|
||
assert all(isinstance(entry_id, str) for entry_id in row["entry_ids"])
|
||
assert row["warnings"] == []
|
||
assert row["legal_entity_id"] == str(world.a.id) and row["legal_entity_inn"] == world.a.inn
|
||
assert row["account_root"] == "60" and row["counterparty_inn"] == world.b.inn
|
||
assert (row["counterparty_id"], row["counterparty_entity_name"]) == (str(world.b.id), "ООО Б")
|
||
assert (row["counterparty_type"], row["match_kind"], row["match_stored"]) == ("GSK", "INN", True)
|
||
cumulative = _row(body, "ООО А", "62", "ООО Б")
|
||
assert cumulative["is_cumulative"] is True and cumulative["closing_credit"] == "42.50"
|
||
assert len(cumulative["entry_ids"]) == 1
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_counterparty_types_and_read_time_resolution(
|
||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||
) -> None:
|
||
world = await _world(seeded)
|
||
client = await auth_client(KA_RF)
|
||
|
||
body = await _page(client, world)
|
||
|
||
chkf = _row(body, "ООО А", "60", "ООО ЧКФ")
|
||
assert (chkf["counterparty_type"], chkf["match_kind"], chkf["match_stored"]) == ("GK", "MANUAL", True)
|
||
unknown = _row(body, "ООО А", "60", "Неизвестный контрагент")
|
||
assert (unknown["counterparty_id"], unknown["counterparty_type"], unknown["match_kind"]) == (None, "EXTERNAL", None)
|
||
assert unknown["match_stored"] is False and unknown["counterparty_entity_name"] is None
|
||
by_name = _row(body, "ООО А", "60", "ООО Внешний")
|
||
assert (by_name["counterparty_id"], by_name["match_kind"], by_name["match_stored"]) == (
|
||
str(world.external.id), "NAME", False) # fmt: skip
|
||
assert by_name["counterparty_type"] == "EXTERNAL"
|
||
by_alias = _row(body, "ООО А", "60", "Пекарня Ф")
|
||
assert (by_alias["counterparty_id"], by_alias["match_kind"], by_alias["match_stored"]) == (
|
||
str(world.aliased.id), "ALIAS", False) # fmt: skip
|
||
assert by_alias["counterparty_entity_name"] == "Фирма Эф"
|
||
assert _row(body, "ООО Б", "60", "ООО А")["counterparty_type"] == "GSK"
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_read_time_resolution_alias_wins_over_namesakes(
|
||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||
) -> None:
|
||
"""OQ-128 (§3 п. 4, п. 7): после ручного сопоставления алиас приоритетнее ЮЛ-тёзок по нормализованному имени —
|
||
дорезолв на чтение отдаёт владельца алиаса (`ALIAS`, `match_stored = false`), а не «несопоставлено»."""
|
||
world = await _world(seeded)
|
||
await make_legal_entity(seeded, name="ООО Мю Трейд", inn="6601000001")
|
||
await make_legal_entity(seeded, name="АО Мю Трейд", inn="6601000002")
|
||
owner = await make_legal_entity(seeded, name="ООО Совсем Другое Имя", inn="6601000003")
|
||
seeded.add(
|
||
LegalEntityAlias(legal_entity_id=owner.id, alias="Мю Трейд", normalized_alias=normalize_alias("Мю Трейд"))
|
||
)
|
||
document = await make_confirmed_osv(
|
||
seeded,
|
||
entity=world.a,
|
||
account="70",
|
||
year=YEAR,
|
||
quarter=Q3,
|
||
is_cumulative=True,
|
||
rows=[{"name": "Мю Трейд", "closing_debit": D(1)}, {"name": "Ню Трейд", "closing_debit": D(1)}],
|
||
)
|
||
await _bind(seeded, world.version, document)
|
||
await make_legal_entity(seeded, name="ООО Ню Трейд", inn="6601000004")
|
||
await make_legal_entity(seeded, name="АО Ню Трейд", inn="6601000005")
|
||
client = await auth_client(KA_RF)
|
||
|
||
body = await _page(client, world)
|
||
|
||
resolved = _row(body, "ООО А", "70", "Мю Трейд")
|
||
assert (resolved["counterparty_id"], resolved["match_kind"], resolved["match_stored"]) == (
|
||
str(owner.id), "ALIAS", False) # fmt: skip
|
||
# тёзки без алиаса по-прежнему неоднозначны
|
||
ambiguous = _row(body, "ООО А", "70", "Ню Трейд")
|
||
assert (ambiguous["counterparty_id"], ambiguous["match_kind"]) == (None, None)
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_summary_counts_and_page_total_agree(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
world = await _world(seeded)
|
||
client = await auth_client(KA_RF)
|
||
|
||
summary = (await client.get(_url(world, "osv/summary"))).json()
|
||
|
||
assert summary == {
|
||
"rows_total": ROWS_TOTAL, "unmapped_total": UNMAPPED_TOTAL, "auto_matched_total": AUTO_MATCHED_TOTAL,
|
||
"documents_total": DOCUMENTS_TOTAL, "gaps": [],
|
||
} # fmt: skip
|
||
unmapped = await _page(client, world, "unmapped=true")
|
||
assert unmapped["total"] == summary["unmapped_total"]
|
||
assert len(unmapped["items"]) == UNMAPPED_TOTAL
|
||
assert all(row["counterparty_id"] is None for row in unmapped["items"])
|
||
mapped = await _page(client, world, "unmapped=false")
|
||
assert mapped["total"] == ROWS_TOTAL - UNMAPPED_TOTAL
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_filter_by_match_kind_repeated_param(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
world = await _world(seeded)
|
||
client = await auth_client(KA_RF)
|
||
|
||
body = await _page(client, world, "match_kind=NAME&match_kind=ALIAS")
|
||
summary = (await client.get(_url(world, "osv/summary"))).json()
|
||
|
||
assert body["total"] == summary["auto_matched_total"] == AUTO_MATCHED_TOTAL
|
||
assert {row["match_kind"] for row in body["items"]} == {"NAME", "ALIAS"}
|
||
assert (await _page(client, world, "match_kind=MANUAL"))["total"] == 2 # noqa: PLR2004 - ЧКФ и «ООО А» у Б
|
||
assert (await _page(client, world, "match_kind=INN"))["total"] == 2 # noqa: PLR2004 - 60 и 62
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_filters_q_entity_account_root_type(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
world = await _world(seeded)
|
||
client = await auth_client(KA_RF)
|
||
|
||
by_name = await _page(client, world, "q=%D0%B2%D0%BD%D0%B5%D1%88") # «внеш» — регистр не важен
|
||
assert [row["counterparty_name"] for row in by_name["items"]] == ["ООО Внешний"]
|
||
by_inn = await _page(client, world, f"q={world.b.inn[:6]}") # префикс ИНН контрагента
|
||
assert {row["counterparty_name"] for row in by_inn["items"]} == {"ООО Б"} and by_inn["total"] == 2 # noqa: PLR2004
|
||
assert (await _page(client, world, f"legal_entity_id={world.b.id}"))["total"] == 2 # noqa: PLR2004
|
||
assert (await _page(client, world, "account_root=62"))["total"] == 1
|
||
assert (await _page(client, world, "counterparty_type=GK"))["total"] == 1
|
||
assert (await _page(client, world, "counterparty_type=GSK"))["total"] == 3 # noqa: PLR2004
|
||
both = await _page(client, world, f"legal_entity_id={world.a.id}&account_root=60&counterparty_type=EXTERNAL")
|
||
assert both["total"] == 3 # noqa: PLR2004 - неизвестный, «Внешний», «Пекарня Ф»
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_pagination_window_and_limit_bounds(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
world = await _world(seeded)
|
||
client = await auth_client(KA_RF)
|
||
|
||
body = await _page(client, world, "limit=2&offset=4")
|
||
|
||
assert (body["total"], body["limit"], body["offset"], len(body["items"])) == (ROWS_TOTAL, 2, 4, 2)
|
||
assert [row["row_number"] for row in body["items"]] == [5, 1] # пятая строка 60 «А», затем 62 «А»
|
||
assert (await _page(client, world, "limit=2000"))["limit"] == 2000 # noqa: PLR2004
|
||
assert (await _page(client, world, f"offset={ROWS_TOTAL}"))["items"] == []
|
||
for bad in ("limit=2001", "limit=0", "offset=-1", "counterparty_type=OTHER", "match_kind=BOGUS"):
|
||
assert (await client.get(f"{_url(world)}?{bad}")).status_code == HTTP_UNPROCESSABLE, bad
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_gap_is_reported_and_has_no_rows(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
world = await _world(seeded)
|
||
# у «Б» отвязываем ОСВ 60 за Q2: остаются Q1 и Q3 → пробел
|
||
link = (
|
||
await seeded.execute(
|
||
select(ConsolidationDocument)
|
||
.join(Document, Document.id == ConsolidationDocument.document_id)
|
||
.where(
|
||
ConsolidationDocument.consolidation_version_id == world.version.id,
|
||
Document.legal_entity_id == world.b.id,
|
||
Document.period_quarter == Q2,
|
||
)
|
||
)
|
||
).scalar_one()
|
||
await seeded.delete(link)
|
||
await seeded.flush()
|
||
client = await auth_client(KA_RF)
|
||
|
||
summary = (await client.get(_url(world, "osv/summary"))).json()
|
||
body = await _page(client, world, f"legal_entity_id={world.b.id}")
|
||
|
||
assert summary["gaps"] == [
|
||
{"legal_entity_id": str(world.b.id), "legal_entity_name": "ООО Б", "account_code": "60",
|
||
"missing_quarters": ["Q2"]}
|
||
] # fmt: skip
|
||
assert summary["documents_total"] == DOCUMENTS_TOTAL - 1
|
||
assert body["total"] == 0
|
||
assert summary["rows_total"] == ROWS_TOTAL - 2
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_cancelled_version_is_readable(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
world = await _world(seeded, status=ConsolidationStatus.CANCELLED)
|
||
client = await auth_client(KA_RF)
|
||
|
||
assert (await _page(client, world))["total"] == ROWS_TOTAL
|
||
assert (await client.get(_url(world, "osv/summary"))).status_code == HTTP_OK
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_rights_and_visibility(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
world = await _world(seeded)
|
||
curator = await auth_client(CURATOR)
|
||
foreign = await auth_client(FOREIGN)
|
||
admin = await auth_client("admin@armka.local") # права CONSOLIDATION_VIEW у роли нет
|
||
|
||
assert (await curator.get(_url(world))).status_code == HTTP_OK
|
||
assert (await curator.get(_url(world, "osv/summary"))).status_code == HTTP_OK
|
||
for suffix in ("osv", "osv/summary"):
|
||
hidden = await foreign.get(_url(world, suffix))
|
||
unknown = await foreign.get(f"/consolidations/versions/{uuid.uuid4()}/{suffix}")
|
||
assert hidden.status_code == unknown.status_code == HTTP_NOT_FOUND
|
||
assert hidden.json() == unknown.json()
|
||
assert (await admin.get(_url(world, suffix))).status_code == HTTP_FORBIDDEN
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_read_is_side_effect_free(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
"""Дорезолв не пишется: `osv_entry.counterparty_id` строки «ООО Внешний» остаётся пустым."""
|
||
world = await _world(seeded)
|
||
client = await auth_client(KA_RF)
|
||
|
||
await _page(client, world)
|
||
|
||
resolved = (
|
||
await seeded.execute(
|
||
select(OsvEntry.counterparty_id).where(OsvEntry.counterparty_name.in_(["ООО Внешний", "Пекарня Ф"]))
|
||
)
|
||
).scalars()
|
||
assert set(resolved) == {None}
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_large_version_is_fast_without_n_plus_one(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
"""Свод из 5 400 исходных строк (2 ЮЛ × 3 счёта × 3 квартала × 300): ответ быстро, SELECT-ов — константа."""
|
||
office = (await seeded.execute(select(RegionalOffice).where(RegionalOffice.name == ROSTOV))).scalar_one()
|
||
group = await make_company_group(seeded, name="ГК Большая", office=office)
|
||
members = [await make_legal_entity(seeded, name=f"ООО Большая {n}", group=group) for n in range(2)]
|
||
counterparties = [await make_legal_entity(seeded, name=f"Контрагент {n:03d}", group=group) for n in range(50)]
|
||
user = (await seeded.execute(select(AppUser).where(AppUser.email == KA_RF))).scalar_one()
|
||
_, version = await make_consolidation(
|
||
seeded, group=group, year=YEAR, quarter=Q3, members=members, initiator=user,
|
||
status=ConsolidationStatus.DATA_CORRECTION,
|
||
) # fmt: skip
|
||
documents = []
|
||
for entity in members:
|
||
for account in ("60", "62", "76"):
|
||
for quarter in (Q1, Q2, Q3):
|
||
rows = [
|
||
{
|
||
"name": f"Контрагент {n:03d}" if n % 2 else f"Неизвестный {n}",
|
||
"counterparty_id": counterparties[n % 50].id if n % 3 == 0 else None,
|
||
"counterparty_match_kind": CounterpartyMatchKind.MANUAL if n % 3 == 0 else None,
|
||
"opening_debit": D("100.00"),
|
||
"turnover_debit": D("10.00"),
|
||
"closing_debit": D("110.00"),
|
||
}
|
||
for n in range(300)
|
||
]
|
||
documents.append(
|
||
await make_confirmed_osv(
|
||
seeded, entity=entity, account=account, year=YEAR, quarter=quarter, is_cumulative=False,
|
||
rows=rows,
|
||
)
|
||
) # fmt: skip
|
||
await _bind(seeded, version, *documents)
|
||
client = await auth_client(KA_RF)
|
||
statements: list[str] = []
|
||
connection = (await seeded.connection()).sync_connection
|
||
listener = lambda _c, _cur, statement, *_rest: statements.append(statement) # noqa: E731
|
||
event.listen(connection, "before_cursor_execute", listener)
|
||
try:
|
||
started = time.perf_counter()
|
||
response = await client.get(f"/consolidations/versions/{version.id}/osv?limit=2000")
|
||
elapsed = time.perf_counter() - started
|
||
finally:
|
||
event.remove(connection, "before_cursor_execute", listener)
|
||
|
||
assert response.status_code == HTTP_OK, response.text
|
||
body = response.json()
|
||
assert body["total"] == 2 * 3 * 300 and len(body["items"]) == 1800 # noqa: PLR2004 - 5 400 исходных строк → 1 800
|
||
selects = [s for s in statements if s.lstrip().upper().startswith("SELECT")]
|
||
print(f"\nPERF osv: {elapsed:.3f}s, SELECT={len(selects)}, statements={len(statements)}") # noqa: T201
|
||
assert elapsed < PERF_MAX_SECONDS
|
||
assert len(selects) <= PERF_MAX_SELECTS, selects
|