test(projects): гонка technical_number и переполнение счётчика (AURORA-1831)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev 2026-10-28 18:14:33 +04:00
parent 95b74bc5e5
commit 663f45ee04

View File

@ -0,0 +1,131 @@
"""AURORA-1831: гонка technical_number в v3.add_project и переполнение счётчика.
Тест гонки коммитит в общую тестовую БД (два независимых соединения, фикстура
client с общей сессией не подходит), поэтому создаваемые проекты удаляются в finally.
Тест переполнения ничего не коммитит: всё откатывается.
"""
import asyncio
import re
import pytest
from sqlalchemy import text
from sqlalchemy.exc import DBAPIError
from sqlalchemy.ext.asyncio import create_async_engine
from src.core.config import settings
ORG_UNIT_ID = 2 # в фикстуре у этой оргединицы нет проектов
ADMIN_ID = 1
CALL_ADD_PROJECT = text(
"SELECT project_id FROM v3.add_project(:name, 2026, :org, 'project', NULL, "
"'Открытие ВСП', 'Типовой', 'own', 'Тестовая улица 1')"
)
async def _begin_as_admin(conn):
await conn.execute(
text("SELECT set_config('app.user_id', :uid, true)"), {"uid": str(ADMIN_ID)}
)
async def _wait_for_lock_waiter(monitor_conn, timeout: float = 10.0):
"""Ждёт, пока какой-либо бэкенд встанет в ожидание блокировки (pg_stat_activity)."""
deadline = asyncio.get_running_loop().time() + timeout
while asyncio.get_running_loop().time() < deadline:
row = await monitor_conn.execute(
text(
"SELECT count(*) FROM pg_stat_activity "
"WHERE datname = current_database() AND wait_event_type = 'Lock' "
"AND pid <> pg_backend_pid()"
)
)
if row.scalar_one() > 0:
return
await asyncio.sleep(0.02)
raise AssertionError("вторая транзакция не встала в ожидание блокировки")
async def _technical_number(conn, project_id: int) -> str:
row = await conn.execute(
text("SELECT technical_number FROM v3.project WHERE id = :id"), {"id": project_id}
)
return row.scalar_one()
@pytest.mark.asyncio
async def test_parallel_add_project_gets_distinct_sequential_numbers():
engine = create_async_engine(settings.DATABASE_URL, echo=False)
created: list[int] = []
conn_a = await engine.connect()
conn_b = await engine.connect()
monitor = await engine.connect()
try:
trans_a = await conn_a.begin()
await _begin_as_admin(conn_a)
pid_a = (
await conn_a.execute(CALL_ADD_PROJECT, {"name": "PT_conc_a", "org": ORG_UNIT_ID})
).scalar_one()
created.append(pid_a)
async def run_b():
async with conn_b.begin():
await _begin_as_admin(conn_b)
pid = (
await conn_b.execute(
CALL_ADD_PROJECT, {"name": "PT_conc_b", "org": ORG_UNIT_ID}
)
).scalar_one()
created.append(pid)
return pid
task_b = asyncio.create_task(run_b())
# До правки B упирается в unique index, после правки - в advisory lock.
# В обоих случаях он встаёт в ожидание блокировки до COMMIT транзакции A.
await _wait_for_lock_waiter(monitor)
await trans_a.commit()
pid_b = await asyncio.wait_for(task_b, timeout=10)
async with engine.connect() as check:
tn_a = await _technical_number(check, pid_a)
tn_b = await _technical_number(check, pid_b)
assert tn_a != tn_b
assert re.fullmatch(r"\d{4}_0002\d\d", tn_a)
assert int(tn_b[-2:]) == int(tn_a[-2:]) + 1
finally:
await conn_a.close()
await conn_b.close()
await monitor.close()
async with engine.begin() as cleanup:
await _begin_as_admin(cleanup)
for pid in created:
await cleanup.execute(text("SELECT v3.del_project(:id)"), {"id": pid})
await engine.dispose()
@pytest.mark.asyncio
async def test_add_project_counter_overflow_raises_clear_error():
engine = create_async_engine(settings.DATABASE_URL, echo=False)
try:
async with engine.connect() as conn:
trans = await conn.begin()
try:
await _begin_as_admin(conn)
await conn.execute(
text(
"INSERT INTO v3.project (name, level, org_unit_id, technical_number, "
"project_type, vsp_format, placement_type, object_address) "
"VALUES ('PT_overflow', 'project', :org, '2601_000299', "
"'Открытие ВСП', 'Типовой', 'own', 'Тестовая улица 1')"
),
{"org": ORG_UNIT_ID},
)
with pytest.raises(DBAPIError) as exc_info:
await conn.execute(
CALL_ADD_PROJECT, {"name": "PT_overflow_next", "org": ORG_UNIT_ID}
)
assert "счётчик" in str(exc_info.value.orig)
finally:
await trans.rollback()
finally:
await engine.dispose()