qananasikq/encar
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add tests

Browse Source
This commit is contained in:
qananasikq
2026-04-16 18:11:14 +03:00
parent 3ce7866643
commit 1d3c69120e
6 changed files with 736 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

414
tests/test_load.py Normal file
View File

@@ -0,0 +1,414 @@
"""
Нагрузочные тесты: 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.client = TestClient(app)
@classmethod
def tearDownClass(cls):
import os
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()