38 lines
1.4 KiB
Python
38 lines
1.4 KiB
Python
"""Нагрузочный тест: конкурентная запись в SQLite (WAL + busy_timeout).
|
|
|
|
N_THREADS потоков пишут по PER ADD-операций каждый. Проверяет, что конкурентная
|
|
запись не теряет данные (WAL сериализует writers, busy_timeout ждёт).
|
|
"""
|
|
import os
|
|
|
|
import pytest
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
from db import spec_events, spec_current
|
|
|
|
pytestmark = pytest.mark.load
|
|
|
|
N_THREADS = int(os.getenv("LOAD_THREADS", "4"))
|
|
PER = int(os.getenv("LOAD_PER", "200"))
|
|
|
|
|
|
def test_concurrent_writes(db):
|
|
def writer(tid):
|
|
cid = f"c{tid}"
|
|
for i in range(PER):
|
|
spec_events.apply_ops(
|
|
cid, f"s{tid}", f"d{tid}",
|
|
[{"action": "ADD", "new_row": {"name": f"svc{i}", "price": i}}],
|
|
"pid", {},
|
|
)
|
|
|
|
with ThreadPoolExecutor(max_workers=N_THREADS) as ex:
|
|
list(ex.map(writer, range(N_THREADS)))
|
|
|
|
# Никаких потерь: у каждого потока ровно PER строк
|
|
for tid in range(N_THREADS):
|
|
n = len(spec_current.list_by_contract(f"c{tid}"))
|
|
assert n == PER, f"c{tid}: {n} != {PER} (потеря данных при конкурентной записи)"
|
|
|
|
print(f"\n[load] потоков={N_THREADS} строк/поток={PER} всего={N_THREADS * PER}")
|