53 lines
2.0 KiB
Python
53 lines
2.0 KiB
Python
"""Нагрузочный тест: массовые apply_ops (ADD → UPDATE → DELETE).
|
||
|
||
Прогоняет N операций каждого типа одним вызовом apply_ops, замеряет время,
|
||
проверяет целостность spec_current/spec_events (ничего не потеряно, не продублировано).
|
||
"""
|
||
import os
|
||
import time
|
||
|
||
import pytest
|
||
|
||
from db import spec_events, spec_current
|
||
from db.connection import query
|
||
|
||
pytestmark = pytest.mark.load
|
||
|
||
N = int(os.getenv("LOAD_OPS", "5000"))
|
||
|
||
CID = "load-cid-111"
|
||
SID = "load-sid-111"
|
||
DID = "load-did-111"
|
||
|
||
|
||
def test_mass_apply_ops(db):
|
||
# ADD N строк
|
||
ops = [{"action": "ADD", "new_row": {"name": f"услуга {i}", "price": i, "qty": 1, "sum": i}} for i in range(N)]
|
||
t0 = time.time()
|
||
s = spec_events.apply_ops(CID, SID, DID, ops, "pid", {})
|
||
t_add = time.time() - t0
|
||
assert s["added"] == N
|
||
assert len(spec_current.list_by_contract(CID)) == N
|
||
|
||
# UPDATE всех N строк
|
||
hashes = [spec_events._hash(f"услуга {i}") for i in range(N)]
|
||
ups = [{"action": "UPDATE", "target_hash": h, "new_values": {"price": i + 1}} for i, h in enumerate(hashes)]
|
||
t0 = time.time()
|
||
s = spec_events.apply_ops(CID, SID, DID, ups, "pid", {})
|
||
t_upd = time.time() - t0
|
||
assert s["updated"] == N
|
||
assert spec_current.list_by_contract(CID)[0]["price"] == 1 # первая строка обновлена
|
||
|
||
# DELETE всех N строк
|
||
dels = [{"action": "DELETE", "target_hash": h} for h in hashes]
|
||
t0 = time.time()
|
||
s = spec_events.apply_ops(CID, SID, DID, dels, "pid", {})
|
||
t_del = time.time() - t0
|
||
assert s["deleted"] == N
|
||
assert spec_current.list_by_contract(CID) == []
|
||
|
||
total = query("SELECT count(*) AS c FROM spec_events WHERE contract_id = %s", (CID,))
|
||
assert total[0]["c"] == N * 3 # ADD + UPDATE + DELETE, ничего не потеряно
|
||
|
||
print(f"\n[load] ops={N} add={t_add:.1f}с upd={t_upd:.1f}с del={t_del:.1f}с")
|