145 lines
5.1 KiB
Python
145 lines
5.1 KiB
Python
"""
|
|
Unit tests for the StockIngestionPipeline.
|
|
"""
|
|
from decimal import Decimal
|
|
import uuid
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from oem_crawler.db.models import Base, Vehicle, VehicleOption, VehiclePriceHistory, CrawlRun
|
|
from oem_crawler.db.pipeline import StockIngestionPipeline
|
|
from oem_crawler.db.schemas import (
|
|
OptionSchema,
|
|
VehicleIngestSchema,
|
|
VehicleStatus,
|
|
compute_content_hash,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def in_memory_pipeline():
|
|
engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
test_session_factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
|
|
pipeline = StockIngestionPipeline(session_factory=test_session_factory)
|
|
return pipeline, test_session_factory
|
|
|
|
|
|
def test_pipeline_insert_update_and_idempotency(in_memory_pipeline):
|
|
pipeline, session_factory = in_memory_pipeline
|
|
|
|
test_vehicle = VehicleIngestSchema(
|
|
oem_vehicle_id="TEST-MB-EQE-001",
|
|
brand="MERCEDES-BENZ",
|
|
model_name="EQE 350 4MATIC",
|
|
series="EQE",
|
|
stock_type="used",
|
|
status=VehicleStatus.IN_STOCK,
|
|
price_gross=Decimal("65900.00"),
|
|
price_net=Decimal("55378.15"),
|
|
currency="EUR",
|
|
mileage_km=15200,
|
|
power_kw=215,
|
|
power_hp=292,
|
|
battery_capacity=Decimal("90.56"),
|
|
battery_capacity_unit="kwh",
|
|
charging_power_ac_kw=Decimal("11.00"),
|
|
charging_power_dc_kw=Decimal("170.00"),
|
|
electric_range_wltp_km=620,
|
|
dealer_name="Mercedes-Benz Niederlassung Berlin",
|
|
location_city="Berlin",
|
|
location_postal_code="10115",
|
|
latitude=Decimal("52.5200"),
|
|
longitude=Decimal("13.4050"),
|
|
options=[
|
|
OptionSchema(option_code="P20", name="Fahrassistenz-Paket Plus", category="Packages", is_standard=False),
|
|
OptionSchema(option_code="413", name="Panorama-Schiebedach", category="Roof", is_standard=False),
|
|
],
|
|
raw_payload={"id": "TEST-MB-EQE-001"},
|
|
)
|
|
|
|
# 1. First Run -> Insert
|
|
def stream_once():
|
|
yield test_vehicle
|
|
|
|
run1 = pipeline.run_crawler_pipeline(
|
|
brand="MERCEDES-BENZ",
|
|
crawler_name="test-crawler",
|
|
vehicle_stream=stream_once(),
|
|
market_country="DE",
|
|
)
|
|
|
|
assert run1.items_discovered == 1
|
|
assert run1.items_inserted == 1
|
|
assert run1.items_updated == 0
|
|
assert run1.items_unmodified == 0
|
|
assert run1.items_failed == 0
|
|
|
|
with session_factory() as s:
|
|
veh = s.query(Vehicle).filter_by(oem_vehicle_id="TEST-MB-EQE-001").first()
|
|
assert veh is not None
|
|
assert veh.price_gross == Decimal("65900.00")
|
|
assert veh.battery_capacity == Decimal("90.56")
|
|
assert veh.battery_capacity_unit == "kwh"
|
|
assert veh.charging_power_ac_kw == Decimal("11.00")
|
|
assert veh.charging_power_dc_kw == Decimal("170.00")
|
|
assert veh.electric_range_wltp_km == 620
|
|
assert veh.location_city == "Berlin"
|
|
assert veh.latitude == Decimal("52.5200")
|
|
assert veh.longitude == Decimal("13.4050")
|
|
|
|
# Check options
|
|
options = s.query(VehicleOption).filter_by(vehicle_id=veh.id).all()
|
|
assert len(options) == 2
|
|
assert {opt.option_code for opt in options} == {"P20", "413"}
|
|
|
|
# Check price history
|
|
hist = s.query(VehiclePriceHistory).filter_by(vehicle_id=veh.id).all()
|
|
assert len(hist) == 1
|
|
assert hist[0].price_gross == Decimal("65900.00")
|
|
|
|
# 2. Second Run -> Unmodified
|
|
run2 = pipeline.run_crawler_pipeline(
|
|
brand="MERCEDES-BENZ",
|
|
crawler_name="test-crawler",
|
|
vehicle_stream=stream_once(),
|
|
market_country="DE",
|
|
)
|
|
assert run2.items_discovered == 1
|
|
assert run2.items_inserted == 0
|
|
assert run2.items_updated == 0
|
|
assert run2.items_unmodified == 1
|
|
assert run2.items_failed == 0
|
|
|
|
with session_factory() as s:
|
|
veh = s.query(Vehicle).filter_by(oem_vehicle_id="TEST-MB-EQE-001").first()
|
|
hist = s.query(VehiclePriceHistory).filter_by(vehicle_id=veh.id).all()
|
|
assert len(hist) == 1 # No duplicate price history
|
|
|
|
# 3. Third Run -> Price Drop Update
|
|
test_vehicle_updated = test_vehicle.model_copy()
|
|
test_vehicle_updated.price_gross = Decimal("63900.00")
|
|
|
|
def stream_updated():
|
|
yield test_vehicle_updated
|
|
|
|
run3 = pipeline.run_crawler_pipeline(
|
|
brand="MERCEDES-BENZ",
|
|
crawler_name="test-crawler",
|
|
vehicle_stream=stream_updated(),
|
|
market_country="DE",
|
|
)
|
|
assert run3.items_discovered == 1
|
|
assert run3.items_inserted == 0
|
|
assert run3.items_updated == 1
|
|
assert run3.items_unmodified == 0
|
|
assert run3.items_failed == 0
|
|
|
|
with session_factory() as s:
|
|
veh = s.query(Vehicle).filter_by(oem_vehicle_id="TEST-MB-EQE-001").first()
|
|
assert veh.price_gross == Decimal("63900.00")
|
|
hist = s.query(VehiclePriceHistory).filter_by(vehicle_id=veh.id).order_by(VehiclePriceHistory.recorded_at).all()
|
|
assert len(hist) == 2
|
|
assert hist[0].price_gross == Decimal("65900.00")
|
|
assert hist[1].price_gross == Decimal("63900.00")
|