83 lines
2.4 KiB
Python
83 lines
2.4 KiB
Python
"""
|
|
Tests pour le repository SQLAlchemy.
|
|
"""
|
|
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from pricewatch.app.core.schema import DebugInfo, DebugStatus, FetchMethod, ProductSnapshot
|
|
from pricewatch.app.db.models import Base, Product, ScrapingLog
|
|
from pricewatch.app.db.repository import ProductRepository
|
|
|
|
|
|
@pytest.fixture
|
|
def session() -> Session:
|
|
"""Session SQLite in-memory pour tests repository."""
|
|
engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
SessionLocal = sessionmaker(bind=engine)
|
|
session = SessionLocal()
|
|
try:
|
|
yield session
|
|
finally:
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
|
|
def _make_snapshot(reference: str | None) -> ProductSnapshot:
|
|
return ProductSnapshot(
|
|
source="amazon",
|
|
url="https://example.com/product",
|
|
fetched_at=datetime(2026, 1, 14, 12, 0, 0),
|
|
title="Produit test",
|
|
price=199.99,
|
|
currency="EUR",
|
|
shipping_cost=0.0,
|
|
reference=reference,
|
|
images=["https://example.com/img1.jpg"],
|
|
specs={"Couleur": "Noir"},
|
|
debug=DebugInfo(
|
|
method=FetchMethod.HTTP,
|
|
status=DebugStatus.SUCCESS,
|
|
errors=["Avertissement"],
|
|
notes=["OK"],
|
|
),
|
|
)
|
|
|
|
|
|
def test_save_snapshot_creates_product(session: Session):
|
|
"""Le repository persiste produit + log."""
|
|
repo = ProductRepository(session)
|
|
snapshot = _make_snapshot(reference="B08N5WRWNW")
|
|
|
|
product_id = repo.save_snapshot(snapshot)
|
|
session.commit()
|
|
|
|
product = session.query(Product).one()
|
|
assert product.id == product_id
|
|
assert product.reference == "B08N5WRWNW"
|
|
assert len(product.images) == 1
|
|
assert len(product.specs) == 1
|
|
assert len(product.price_history) == 1
|
|
|
|
log = session.query(ScrapingLog).one()
|
|
assert log.product_id == product_id
|
|
assert log.errors == ["Avertissement"]
|
|
assert log.notes == ["OK"]
|
|
|
|
|
|
def test_save_snapshot_without_reference(session: Session):
|
|
"""Sans reference, le produit n'est pas cree mais le log existe."""
|
|
repo = ProductRepository(session)
|
|
snapshot = _make_snapshot(reference=None)
|
|
|
|
product_id = repo.save_snapshot(snapshot)
|
|
session.commit()
|
|
|
|
assert product_id is None
|
|
assert session.query(Product).count() == 0
|
|
assert session.query(ScrapingLog).count() == 1
|