test: нагрузочные/стресс-тесты (маркер load): массовый pipeline, apply_ops, конкуренция, HTTP-стресс

tests/load/test_load_concurrency.py

@@ -0,0 +1,37 @@
+"""Нагрузочный тест: конкурентная запись в 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", "16"))
+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}")