418 lines
17 KiB
Python
418 lines
17 KiB
Python
"""
|
||
Нагрузочные тесты: API endpoints, batch upsert, конкурентный доступ к БД.
|
||
|
||
Запуск:
|
||
pytest tests/test_load.py -v -s
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import concurrent.futures
|
||
import time
|
||
import random
|
||
import string
|
||
import unittest
|
||
from datetime import datetime, timezone
|
||
|
||
from fastapi.testclient import TestClient
|
||
|
||
from encar_scraper.api.app import create_app
|
||
from encar_scraper.core.config import DatabaseConfig, RedisConfig, Settings
|
||
from encar_scraper.storage.db import PersistenceService
|
||
from encar_scraper.storage.schemas import CarRecord, ImageRecord
|
||
|
||
|
||
# ── helpers ──────────────────────────────────────────────────────────────
|
||
|
||
def _sqlite_settings() -> Settings:
|
||
return Settings(
|
||
database=DatabaseConfig(url="sqlite:///test_load.db", echo=False, auto_create_tables=True),
|
||
redis=RedisConfig(url="redis://localhost:6379/15"),
|
||
)
|
||
|
||
|
||
def _make_car(idx: int, *, images: int = 3) -> CarRecord:
|
||
uid = f"load-{idx}-{''.join(random.choices(string.ascii_lowercase, k=4))}"
|
||
return CarRecord(
|
||
parser_id=uid,
|
||
brand=random.choice(["Hyundai", "Kia", "BMW", "Mercedes", "Genesis"]),
|
||
model=f"Model-{idx % 50}",
|
||
year=random.randint(2010, 2025),
|
||
price=random.randint(5_000_000, 80_000_000),
|
||
currency="KRW",
|
||
mileage=random.randint(0, 300_000),
|
||
country="KR",
|
||
origin="ENCAR",
|
||
origin_url=f"https://encar.com/cars/{uid}",
|
||
origin_id=uid,
|
||
slug=uid,
|
||
selling_type="NA",
|
||
body_type="SEDAN",
|
||
last_seen_at=datetime.now(timezone.utc),
|
||
images=[
|
||
ImageRecord(
|
||
fullres_image=f"https://ci.encar.com/photo/{uid}/{j}.jpg",
|
||
preview_image=f"https://ci.encar.com/photo/{uid}/{j}_thumb.jpg",
|
||
order_index=j,
|
||
)
|
||
for j in range(images)
|
||
],
|
||
)
|
||
|
||
|
||
# ── DB load tests ───────────────────────────────────────────────────────
|
||
|
||
class TestDatabaseLoad(unittest.TestCase):
|
||
"""Тесты производительности PersistenceService на SQLite."""
|
||
|
||
@classmethod
|
||
def setUpClass(cls):
|
||
import os
|
||
if os.path.exists("test_load.db"):
|
||
os.remove("test_load.db")
|
||
cls.settings = _sqlite_settings()
|
||
cls.persistence = PersistenceService(cls.settings)
|
||
cls.persistence.create_tables()
|
||
|
||
@classmethod
|
||
def tearDownClass(cls):
|
||
cls.persistence.engine.dispose()
|
||
import os
|
||
if os.path.exists("test_load.db"):
|
||
os.remove("test_load.db")
|
||
|
||
def test_batch_upsert_500_cars(self):
|
||
"""Пакетная вставка 500 авто с изображениями — должна пройти за <10 сек."""
|
||
records = [_make_car(i, images=5) for i in range(500)]
|
||
t0 = time.perf_counter()
|
||
result = self.persistence.upsert_cars_batch(records)
|
||
elapsed = time.perf_counter() - t0
|
||
|
||
print(f"\n[batch_upsert_500] inserted={result['inserted']}, "
|
||
f"updated={result['updated']}, images={result['images_upserted']}, "
|
||
f"time={elapsed:.2f}s")
|
||
|
||
self.assertEqual(result["inserted"], 500)
|
||
self.assertEqual(result["images_upserted"], 2500)
|
||
self.assertLess(elapsed, 10, "Batch upsert 500 cars should complete in <10s")
|
||
|
||
def test_batch_upsert_update_existing(self):
|
||
"""Повторный upsert 200 существующих записей — обновление без дублей."""
|
||
records = [_make_car(i, images=3) for i in range(10000, 10200)]
|
||
self.persistence.upsert_cars_batch(records)
|
||
|
||
# Обновляем цену и перезаписываем
|
||
for r in records:
|
||
r.price = 99_999_999
|
||
t0 = time.perf_counter()
|
||
result = self.persistence.upsert_cars_batch(records)
|
||
elapsed = time.perf_counter() - t0
|
||
|
||
print(f"\n[batch_update_200] updated={result['updated']}, time={elapsed:.2f}s")
|
||
self.assertEqual(result["updated"], 200)
|
||
self.assertLess(elapsed, 10)
|
||
|
||
def test_concurrent_upsert_batches(self):
|
||
"""3 параллельных потока по 100 upsert-ов — нет deadlock / corruption."""
|
||
def worker(thread_id: int):
|
||
batch = [_make_car(thread_id * 10000 + i) for i in range(100)]
|
||
return self.persistence.upsert_cars_batch(batch)
|
||
|
||
t0 = time.perf_counter()
|
||
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:
|
||
futures = [pool.submit(worker, tid) for tid in range(3)]
|
||
results = [f.result() for f in concurrent.futures.as_completed(futures)]
|
||
elapsed = time.perf_counter() - t0
|
||
|
||
total_inserted = sum(r["inserted"] for r in results)
|
||
print(f"\n[concurrent_upsert] total_inserted={total_inserted}, time={elapsed:.2f}s")
|
||
self.assertEqual(total_inserted, 300)
|
||
self.assertLess(elapsed, 30)
|
||
|
||
def test_single_upsert_throughput(self):
|
||
"""50 последовательных upsert_car — замер throughput."""
|
||
records = [_make_car(50000 + i, images=2) for i in range(50)]
|
||
t0 = time.perf_counter()
|
||
for r in records:
|
||
self.persistence.upsert_car(r)
|
||
elapsed = time.perf_counter() - t0
|
||
rps = 50 / elapsed if elapsed > 0 else 0
|
||
|
||
print(f"\n[single_upsert_50] time={elapsed:.2f}s, rps={rps:.1f}")
|
||
self.assertLess(elapsed, 15, "50 individual upserts should complete in <15s")
|
||
|
||
def test_mark_sold_large_set(self):
|
||
"""mark_sold на 300 origin_id — должен быть быстрым."""
|
||
records = [_make_car(60000 + i, images=1) for i in range(300)]
|
||
self.persistence.upsert_cars_batch(records)
|
||
|
||
active_ids = {r.origin_id for r in records[:150]}
|
||
t0 = time.perf_counter()
|
||
sold = self.persistence.mark_sold_not_in_listing(active_origin_ids=active_ids, lane="encar")
|
||
elapsed = time.perf_counter() - t0
|
||
|
||
print(f"\n[mark_sold_300] marked_sold={sold}, time={elapsed:.2f}s")
|
||
self.assertLess(elapsed, 5)
|
||
|
||
def test_sync_run_lifecycle(self):
|
||
"""start → finish sync_run 50 раз — измеряем overhead."""
|
||
t0 = time.perf_counter()
|
||
for _ in range(50):
|
||
run_id = self.persistence.start_sync_run("encar")
|
||
self.persistence.finish_sync_run(
|
||
run_id,
|
||
status="completed",
|
||
ids_fetched=1000,
|
||
cars_upserted=950,
|
||
cars_failed=50,
|
||
images_upserted=15000,
|
||
)
|
||
elapsed = time.perf_counter() - t0
|
||
|
||
print(f"\n[sync_run_lifecycle_50] time={elapsed:.2f}s")
|
||
self.assertLess(elapsed, 5)
|
||
|
||
|
||
# ── API load tests ──────────────────────────────────────────────────────
|
||
|
||
class TestAPILoad(unittest.TestCase):
|
||
"""Нагрузочные тесты на FastAPI endpoints через TestClient."""
|
||
|
||
@classmethod
|
||
def setUpClass(cls):
|
||
import os
|
||
if os.path.exists("test_api_load.db"):
|
||
os.remove("test_api_load.db")
|
||
|
||
settings = Settings(
|
||
database=DatabaseConfig(url="sqlite:///test_api_load.db", echo=False, auto_create_tables=True),
|
||
redis=RedisConfig(url="redis://localhost:6379/15"),
|
||
)
|
||
app = create_app(settings)
|
||
app.state.persistence.create_tables()
|
||
|
||
# Предзаполняем БД 200 авто
|
||
persistence: PersistenceService = app.state.persistence
|
||
persistence.upsert_cars_batch([_make_car(i, images=4) for i in range(200)])
|
||
|
||
cls.app = app
|
||
cls.client = TestClient(app)
|
||
|
||
@classmethod
|
||
def tearDownClass(cls):
|
||
import os
|
||
cls.client.close()
|
||
cls.app.state.persistence.engine.dispose()
|
||
if os.path.exists("test_api_load.db"):
|
||
os.remove("test_api_load.db")
|
||
|
||
def test_health_check_rapid(self):
|
||
"""100 GET /health подряд — все 200, <5 сек."""
|
||
t0 = time.perf_counter()
|
||
for _ in range(100):
|
||
r = self.client.get("/health")
|
||
self.assertEqual(r.status_code, 200)
|
||
elapsed = time.perf_counter() - t0
|
||
|
||
print(f"\n[health_100] time={elapsed:.2f}s, rps={100 / elapsed:.0f}")
|
||
self.assertLess(elapsed, 5)
|
||
|
||
def test_list_cars_paginated_rapid(self):
|
||
"""100 GET /api/v1/cars с разными страницами — все 200, <10 сек."""
|
||
t0 = time.perf_counter()
|
||
for page in range(1, 101):
|
||
r = self.client.get("/api/v1/cars", params={"page": (page % 10) + 1, "per_page": 20})
|
||
self.assertEqual(r.status_code, 200)
|
||
data = r.json()
|
||
self.assertIn("items", data)
|
||
elapsed = time.perf_counter() - t0
|
||
|
||
print(f"\n[list_cars_100] time={elapsed:.2f}s, rps={100 / elapsed:.0f}")
|
||
self.assertLess(elapsed, 10)
|
||
|
||
def test_list_cars_with_filters(self):
|
||
"""50 GET /api/v1/cars с brand/year фильтрами."""
|
||
t0 = time.perf_counter()
|
||
for i in range(50):
|
||
brand = random.choice(["Hyundai", "Kia", "BMW", "Mercedes", "Genesis"])
|
||
r = self.client.get(
|
||
"/api/v1/cars",
|
||
params={"brand": brand, "year_min": 2015, "year_max": 2023, "per_page": 50},
|
||
)
|
||
self.assertEqual(r.status_code, 200)
|
||
elapsed = time.perf_counter() - t0
|
||
|
||
print(f"\n[filtered_cars_50] time={elapsed:.2f}s, rps={50 / elapsed:.0f}")
|
||
self.assertLess(elapsed, 10)
|
||
|
||
def test_get_car_by_id_rapid(self):
|
||
"""50 GET /api/v1/cars/{id} — все 200."""
|
||
t0 = time.perf_counter()
|
||
for car_id in range(1, 51):
|
||
r = self.client.get(f"/api/v1/cars/{car_id}")
|
||
self.assertIn(r.status_code, (200, 404))
|
||
elapsed = time.perf_counter() - t0
|
||
|
||
print(f"\n[get_car_50] time={elapsed:.2f}s, rps={50 / elapsed:.0f}")
|
||
self.assertLess(elapsed, 10)
|
||
|
||
def test_stats_endpoint_rapid(self):
|
||
"""50 GET /api/v1/stats — агрегация по всей таблице."""
|
||
t0 = time.perf_counter()
|
||
for _ in range(50):
|
||
r = self.client.get("/api/v1/stats")
|
||
self.assertEqual(r.status_code, 200)
|
||
data = r.json()
|
||
self.assertGreater(data["total_cars"], 0)
|
||
elapsed = time.perf_counter() - t0
|
||
|
||
print(f"\n[stats_50] time={elapsed:.2f}s, rps={50 / elapsed:.0f}")
|
||
self.assertLess(elapsed, 10)
|
||
|
||
def test_concurrent_api_requests(self):
|
||
"""20 конкурентных запросов к разным endpoints — нет 500-х."""
|
||
def make_request(idx: int):
|
||
if idx % 3 == 0:
|
||
return self.client.get("/health")
|
||
elif idx % 3 == 1:
|
||
return self.client.get("/api/v1/cars", params={"page": 1, "per_page": 10})
|
||
else:
|
||
return self.client.get("/api/v1/stats")
|
||
|
||
t0 = time.perf_counter()
|
||
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as pool:
|
||
futures = [pool.submit(make_request, i) for i in range(20)]
|
||
responses = [f.result() for f in concurrent.futures.as_completed(futures)]
|
||
elapsed = time.perf_counter() - t0
|
||
|
||
for r in responses:
|
||
self.assertNotEqual(r.status_code, 500, f"Got 500: {r.text}")
|
||
|
||
print(f"\n[concurrent_api_20] time={elapsed:.2f}s, all_ok={all(r.status_code == 200 for r in responses)}")
|
||
self.assertLess(elapsed, 10)
|
||
|
||
def test_large_page_size(self):
|
||
"""GET /api/v1/cars с per_page=100 — проверяем сериализацию больших ответов."""
|
||
t0 = time.perf_counter()
|
||
for _ in range(20):
|
||
r = self.client.get("/api/v1/cars", params={"page": 1, "per_page": 100})
|
||
self.assertEqual(r.status_code, 200)
|
||
data = r.json()
|
||
# Каждый car содержит images — проверяем что сериализация прошла
|
||
for item in data["items"]:
|
||
self.assertIn("images", item)
|
||
elapsed = time.perf_counter() - t0
|
||
|
||
print(f"\n[large_page_20] time={elapsed:.2f}s, rps={20 / elapsed:.0f}")
|
||
self.assertLess(elapsed, 10)
|
||
|
||
|
||
# ── Data integrity under load ───────────────────────────────────────────
|
||
|
||
class TestDataIntegrity(unittest.TestCase):
|
||
"""Проверка целостности данных при массовых операциях."""
|
||
|
||
@classmethod
|
||
def setUpClass(cls):
|
||
import os
|
||
if os.path.exists("test_integrity.db"):
|
||
os.remove("test_integrity.db")
|
||
cls.settings = Settings(
|
||
database=DatabaseConfig(url="sqlite:///test_integrity.db", echo=False, auto_create_tables=True),
|
||
redis=RedisConfig(url="redis://localhost:6379/15"),
|
||
)
|
||
cls.persistence = PersistenceService(cls.settings)
|
||
cls.persistence.create_tables()
|
||
|
||
@classmethod
|
||
def tearDownClass(cls):
|
||
cls.persistence.engine.dispose()
|
||
import os
|
||
if os.path.exists("test_integrity.db"):
|
||
os.remove("test_integrity.db")
|
||
|
||
def test_no_duplicates_after_repeated_upserts(self):
|
||
"""3 раза upsert одного батча — ровно N уникальных записей в БД."""
|
||
records = [_make_car(90000 + i) for i in range(100)]
|
||
|
||
for _ in range(3):
|
||
self.persistence.upsert_cars_batch(records)
|
||
|
||
from sqlalchemy import select, func
|
||
from encar_scraper.storage.models import Car
|
||
|
||
with self.persistence.session_scope() as session:
|
||
count = session.execute(
|
||
select(func.count(Car.id)).where(Car.origin_id.like("load-90%"))
|
||
).scalar()
|
||
|
||
self.assertEqual(count, 100, f"Expected 100 unique cars, got {count}")
|
||
|
||
def test_images_replaced_not_accumulated(self):
|
||
"""При upsert с новыми images — старые удаляются, а не копятся."""
|
||
rec = _make_car(80001, images=5)
|
||
self.persistence.upsert_car(rec)
|
||
|
||
# Обновляем с 2 изображениями
|
||
rec.images = [
|
||
ImageRecord(fullres_image="https://new/1.jpg", preview_image="https://new/1t.jpg", order_index=0),
|
||
ImageRecord(fullres_image="https://new/2.jpg", preview_image="https://new/2t.jpg", order_index=1),
|
||
]
|
||
self.persistence.upsert_car(rec)
|
||
|
||
from sqlalchemy import select, func
|
||
from encar_scraper.storage.models import Car, Image
|
||
|
||
with self.persistence.session_scope() as session:
|
||
car = session.execute(select(Car).where(Car.origin_id == rec.origin_id)).scalar_one()
|
||
img_count = session.execute(
|
||
select(func.count(Image.id)).where(Image.car_id == car.id)
|
||
).scalar()
|
||
|
||
self.assertEqual(img_count, 2, f"Expected 2 images after replace, got {img_count}")
|
||
|
||
def test_dedup_within_batch(self):
|
||
"""Батч с дубликатами origin_id — дедупликация работает."""
|
||
base = _make_car(70001)
|
||
dupe = _make_car(70001)
|
||
dupe.parser_id = base.parser_id
|
||
dupe.origin_id = base.origin_id
|
||
dupe.origin_url = base.origin_url
|
||
dupe.price = 12345
|
||
|
||
records = [base, dupe]
|
||
result = self.persistence.upsert_cars_batch(records)
|
||
|
||
self.assertEqual(result["inserted"] + result["updated"], 1,
|
||
"Dedup should reduce 2 records to 1")
|
||
|
||
def test_batch_1000_cars_integrity(self):
|
||
"""Вставка 1000 авто — проверяем что ВСЕ записи в БД с верными images."""
|
||
records = [_make_car(200000 + i, images=3) for i in range(1000)]
|
||
result = self.persistence.upsert_cars_batch(records)
|
||
|
||
self.assertEqual(result["inserted"], 1000)
|
||
self.assertEqual(result["images_upserted"], 3000)
|
||
|
||
from sqlalchemy import select, func
|
||
from encar_scraper.storage.models import Car, Image
|
||
|
||
with self.persistence.session_scope() as session:
|
||
car_count = session.execute(
|
||
select(func.count(Car.id)).where(Car.origin_id.like("load-20____-%"))
|
||
).scalar()
|
||
img_count = session.execute(
|
||
select(func.count(Image.id)).where(
|
||
Image.car_id.in_(
|
||
select(Car.id).where(Car.origin_id.like("load-20____-%"))
|
||
)
|
||
)
|
||
).scalar()
|
||
|
||
self.assertEqual(car_count, 1000)
|
||
self.assertEqual(img_count, 3000)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|