41 lines
1.3 KiB
Python
41 lines
1.3 KiB
Python
"""
|
|
Tests de charge legere pour la persistence (100 snapshots).
|
|
"""
|
|
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from pricewatch.app.core.schema import DebugInfo, DebugStatus, FetchMethod, ProductSnapshot
|
|
from pricewatch.app.db.models import Base, Product
|
|
from pricewatch.app.db.repository import ProductRepository
|
|
|
|
|
|
def test_bulk_save_100_snapshots():
|
|
"""Le repository persiste 100 snapshots sans erreur."""
|
|
engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
session = sessionmaker(bind=engine)()
|
|
|
|
try:
|
|
repo = ProductRepository(session)
|
|
for idx in range(100):
|
|
snapshot = ProductSnapshot(
|
|
source="amazon",
|
|
url=f"https://example.com/product/{idx}",
|
|
fetched_at=datetime(2026, 1, 14, 14, 0, 0),
|
|
title=f"Produit {idx}",
|
|
price=10.0 + idx,
|
|
currency="EUR",
|
|
reference=f"REF-{idx}",
|
|
debug=DebugInfo(method=FetchMethod.HTTP, status=DebugStatus.SUCCESS),
|
|
)
|
|
repo.save_snapshot(snapshot)
|
|
session.commit()
|
|
|
|
assert session.query(Product).count() == 100
|
|
finally:
|
|
session.close()
|
|
engine.dispose()
|