Leopoldadmin
74b5e6afb8
- Vehicle model: 25+ fields, 5 vehicle types, soft-delete - Vehicle CRUD: 7 API endpoints with JWT auth, filter/sort/paginate - mobile.de: push/update/delete listings, field mapping, retry logic - MobileDeListing model for sync status tracking - Frontend: VehicleList, VehicleForm, VehicleDetail, MobileDeStatus - 73 backend tests (82% coverage), 16 frontend tests
430 lines
15 KiB
Python
430 lines
15 KiB
Python
"""Tests for mobile.de service: field mapping, push, update, delete, status, retry."""
|
|
|
|
import uuid
|
|
from datetime import date, datetime, timezone
|
|
from decimal import Decimal
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.vehicle import MobileDeListing, Vehicle
|
|
from app.services import mobilede_service
|
|
from app.utils.mobilede_mapping import map_fields
|
|
|
|
|
|
def _make_vehicle(**overrides) -> Vehicle:
|
|
"""Create a Vehicle instance with defaults and optional overrides."""
|
|
defaults = {
|
|
"make": "Mercedes-Benz",
|
|
"model": "Actros",
|
|
"fin": "WDB9066351L123456",
|
|
"year": 2020,
|
|
"first_registration": date(2020, 3, 15),
|
|
"power_kw": 300,
|
|
"power_hp": 408,
|
|
"fuel_type": "Diesel",
|
|
"transmission": "Manual",
|
|
"color": "White",
|
|
"condition": "used",
|
|
"location": "Berlin",
|
|
"availability": "available",
|
|
"price": Decimal("45000.00"),
|
|
"vehicle_type": "lkw",
|
|
"lkw_type": "sattelzugmaschine",
|
|
"mileage_km": 120000,
|
|
"description": "Well maintained truck",
|
|
}
|
|
defaults.update(overrides)
|
|
vehicle = Vehicle(**defaults)
|
|
vehicle.id = uuid.uuid4()
|
|
return vehicle
|
|
|
|
|
|
class TestFieldMapping:
|
|
"""Tests for mobile.de field mapping (map_fields)."""
|
|
|
|
def test_map_fields_basic_lkw(self):
|
|
"""map_fields produces correct ad format for LKW."""
|
|
vehicle = _make_vehicle()
|
|
ad = map_fields(vehicle)
|
|
|
|
assert ad["vin"] == "WDB9066351L123456"
|
|
assert ad["make"] == "Mercedes-Benz"
|
|
assert ad["model"] == "Actros"
|
|
assert ad["category"] == "SemiTractor"
|
|
assert ad["price"] == {"amount": 45000.0, "currency": "EUR"}
|
|
assert ad["availabilityStatus"] == "available"
|
|
assert ad["condition"] == "used"
|
|
assert ad["firstRegistration"] == "2020-03"
|
|
assert ad["mileage"] == {"value": 120000, "unit": "km"}
|
|
assert ad["power"] == {"powerKw": 300, "powerHp": 408}
|
|
assert ad["fuelType"] == "Diesel"
|
|
assert ad["transmission"] == "Manual"
|
|
assert ad["color"] == "White"
|
|
assert ad["sellerLocation"] == "Berlin"
|
|
assert ad["description"] == "Well maintained truck"
|
|
|
|
def test_map_fields_baumaschine_with_operating_hours(self):
|
|
"""map_fields maps operating_hours to mileage for baumaschine."""
|
|
vehicle = _make_vehicle(
|
|
vehicle_type="baumaschine",
|
|
machine_type="Bagger",
|
|
operating_hours=Decimal("3500.5"),
|
|
operating_hours_unit="h",
|
|
mileage_km=None,
|
|
lkw_type=None,
|
|
)
|
|
ad = map_fields(vehicle)
|
|
|
|
assert ad["category"] == "ConstructionMachine"
|
|
assert ad["mileage"] == {"value": 3500.5, "unit": "h"}
|
|
assert ad["bodyType"] == "Bagger"
|
|
|
|
def test_map_fields_pkw(self):
|
|
"""map_fields maps PKW correctly."""
|
|
vehicle = _make_vehicle(
|
|
vehicle_type="pkw",
|
|
lkw_type=None,
|
|
body_type="Limousine",
|
|
)
|
|
ad = map_fields(vehicle)
|
|
assert ad["category"] == "Car"
|
|
assert ad["bodyType"] == "Limousine"
|
|
|
|
def test_map_fields_stapler(self):
|
|
"""map_fields maps Stapler correctly."""
|
|
vehicle = _make_vehicle(
|
|
vehicle_type="stapler",
|
|
lkw_type=None,
|
|
operating_hours=Decimal("12000"),
|
|
operating_hours_unit="h",
|
|
mileage_km=None,
|
|
)
|
|
ad = map_fields(vehicle)
|
|
assert ad["category"] == "ForkliftTruck"
|
|
assert ad["mileage"] == {"value": 12000.0, "unit": "h"}
|
|
|
|
def test_map_fields_transporter(self):
|
|
"""map_fields maps Transporter correctly."""
|
|
vehicle = _make_vehicle(
|
|
vehicle_type="transporter",
|
|
lkw_type=None,
|
|
)
|
|
ad = map_fields(vehicle)
|
|
assert ad["category"] == "Van"
|
|
|
|
def test_map_fields_no_optional_fields(self):
|
|
"""map_fields handles vehicle with no optional fields."""
|
|
vehicle = Vehicle(
|
|
make="Test",
|
|
model="Model",
|
|
fin="WDB9066351L123456",
|
|
condition="new",
|
|
availability="available",
|
|
price=Decimal("10000.00"),
|
|
vehicle_type="pkw",
|
|
)
|
|
ad = map_fields(vehicle)
|
|
assert ad["vin"] == "WDB9066351L123456"
|
|
assert ad["make"] == "Test"
|
|
assert "firstRegistration" not in ad
|
|
assert "mileage" not in ad
|
|
assert "power" not in ad
|
|
assert "fuelType" not in ad
|
|
|
|
def test_map_fields_lkw_type_with_prefix(self):
|
|
"""map_fields strips lkw_ prefix for category lookup."""
|
|
vehicle = _make_vehicle(lkw_type="lkw_kipper")
|
|
ad = map_fields(vehicle)
|
|
assert ad["category"] == "Tipper"
|
|
|
|
def test_map_fields_lkw_type_unknown_falls_back(self):
|
|
"""map_fields falls back to base category for unknown lkw_type."""
|
|
vehicle = _make_vehicle(lkw_type="unknown_type")
|
|
ad = map_fields(vehicle)
|
|
assert ad["category"] == "Truck"
|
|
|
|
|
|
class TestPushListing:
|
|
"""Tests for mobilede_service.push_listing."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_push_listing_success(self, db_session):
|
|
"""push_listing creates listing with synced status on success."""
|
|
vehicle = _make_vehicle()
|
|
db_session.add(vehicle)
|
|
await db_session.flush()
|
|
|
|
with patch("app.services.mobilede_service.httpx.AsyncClient") as mock_client_cls:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 201
|
|
mock_response.json.return_value = {"id": "ad-123"}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_client = AsyncMock()
|
|
mock_client.post = AsyncMock(return_value=mock_response)
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=None)
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
listing = await mobilede_service.push_listing(db_session, vehicle)
|
|
|
|
assert listing.sync_status == "synced"
|
|
assert listing.ad_id == "ad-123"
|
|
assert listing.synced_at is not None
|
|
assert listing.error_log is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_push_listing_http_error(self, db_session):
|
|
"""push_listing sets fehler status on HTTP error."""
|
|
import httpx
|
|
|
|
vehicle = _make_vehicle()
|
|
db_session.add(vehicle)
|
|
await db_session.flush()
|
|
|
|
with patch("app.services.mobilede_service.httpx.AsyncClient") as mock_client_cls:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 400
|
|
mock_response.text = "Bad Request"
|
|
mock_response.raise_for_status.side_effect = httpx.HTTPStatusError(
|
|
"Bad Request", request=MagicMock(), response=mock_response
|
|
)
|
|
mock_client = AsyncMock()
|
|
mock_client.post = AsyncMock(return_value=mock_response)
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=None)
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
listing = await mobilede_service.push_listing(db_session, vehicle)
|
|
|
|
assert listing.sync_status == "fehler"
|
|
assert listing.error_log is not None
|
|
assert "400" in listing.error_log
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_push_listing_request_error(self, db_session):
|
|
"""push_listing sets fehler status on request error."""
|
|
import httpx
|
|
|
|
vehicle = _make_vehicle()
|
|
db_session.add(vehicle)
|
|
await db_session.flush()
|
|
|
|
with patch("app.services.mobilede_service.httpx.AsyncClient") as mock_client_cls:
|
|
mock_client = AsyncMock()
|
|
mock_client.post = AsyncMock(side_effect=httpx.ConnectError("Connection refused"))
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=None)
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
listing = await mobilede_service.push_listing(db_session, vehicle)
|
|
|
|
assert listing.sync_status == "fehler"
|
|
assert "Connection refused" in listing.error_log
|
|
|
|
|
|
class TestUpdateListing:
|
|
"""Tests for mobilede_service.update_listing."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_listing_success(self, db_session):
|
|
"""update_listing updates synced status on success."""
|
|
vehicle = _make_vehicle()
|
|
db_session.add(vehicle)
|
|
await db_session.flush()
|
|
|
|
listing = MobileDeListing(
|
|
vehicle_id=vehicle.id,
|
|
ad_id="ad-123",
|
|
sync_status="synced",
|
|
)
|
|
db_session.add(listing)
|
|
await db_session.flush()
|
|
|
|
with patch("app.services.mobilede_service.httpx.AsyncClient") as mock_client_cls:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_client = AsyncMock()
|
|
mock_client.put = AsyncMock(return_value=mock_response)
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=None)
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
result = await mobilede_service.update_listing(db_session, vehicle, listing)
|
|
|
|
assert result.sync_status == "synced"
|
|
assert result.synced_at is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_listing_no_ad_id(self, db_session):
|
|
"""update_listing sets fehler when listing has no ad_id."""
|
|
vehicle = _make_vehicle()
|
|
db_session.add(vehicle)
|
|
await db_session.flush()
|
|
|
|
listing = MobileDeListing(
|
|
vehicle_id=vehicle.id,
|
|
sync_status="pending",
|
|
)
|
|
db_session.add(listing)
|
|
await db_session.flush()
|
|
|
|
result = await mobilede_service.update_listing(db_session, vehicle, listing)
|
|
|
|
assert result.sync_status == "fehler"
|
|
assert "ad_id" in result.error_log
|
|
|
|
|
|
class TestDeleteListing:
|
|
"""Tests for mobilede_service.delete_listing."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_listing_success(self, db_session):
|
|
"""delete_listing sets deleted status on success."""
|
|
vehicle = _make_vehicle()
|
|
db_session.add(vehicle)
|
|
await db_session.flush()
|
|
|
|
listing = MobileDeListing(
|
|
vehicle_id=vehicle.id,
|
|
ad_id="ad-123",
|
|
sync_status="synced",
|
|
)
|
|
db_session.add(listing)
|
|
await db_session.flush()
|
|
|
|
with patch("app.services.mobilede_service.httpx.AsyncClient") as mock_client_cls:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 204
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_client = AsyncMock()
|
|
mock_client.delete = AsyncMock(return_value=mock_response)
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=None)
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
result = await mobilede_service.delete_listing(db_session, listing)
|
|
|
|
assert result.sync_status == "deleted"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_listing_no_ad_id(self, db_session):
|
|
"""delete_listing sets fehler when listing has no ad_id."""
|
|
vehicle = _make_vehicle()
|
|
db_session.add(vehicle)
|
|
await db_session.flush()
|
|
|
|
listing = MobileDeListing(
|
|
vehicle_id=vehicle.id,
|
|
sync_status="pending",
|
|
)
|
|
db_session.add(listing)
|
|
await db_session.flush()
|
|
|
|
result = await mobilede_service.delete_listing(db_session, listing)
|
|
|
|
assert result.sync_status == "fehler"
|
|
assert "ad_id" in result.error_log
|
|
|
|
|
|
class TestGetListingStatus:
|
|
"""Tests for mobilede_service.get_listing_status."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_listing_status_returns_latest(self, db_session):
|
|
"""get_listing_status returns the most recent listing."""
|
|
vehicle = _make_vehicle()
|
|
db_session.add(vehicle)
|
|
await db_session.flush()
|
|
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
listing1 = MobileDeListing(
|
|
vehicle_id=vehicle.id,
|
|
sync_status="fehler",
|
|
error_log="First attempt failed",
|
|
created_at=datetime(2025, 1, 1, 12, 0, 0, tzinfo=timezone.utc),
|
|
)
|
|
db_session.add(listing1)
|
|
await db_session.flush()
|
|
|
|
listing2 = MobileDeListing(
|
|
vehicle_id=vehicle.id,
|
|
ad_id="ad-456",
|
|
sync_status="synced",
|
|
created_at=datetime(2025, 1, 2, 12, 0, 0, tzinfo=timezone.utc),
|
|
)
|
|
db_session.add(listing2)
|
|
await db_session.flush()
|
|
|
|
result = await mobilede_service.get_listing_status(db_session, vehicle.id)
|
|
|
|
assert result is not None
|
|
assert result.sync_status == "synced"
|
|
assert result.ad_id == "ad-456"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_listing_status_returns_none_when_no_listing(self, db_session):
|
|
"""get_listing_status returns None when no listing exists."""
|
|
vehicle_id = uuid.uuid4()
|
|
result = await mobilede_service.get_listing_status(db_session, vehicle_id)
|
|
assert result is None
|
|
|
|
|
|
class TestRetryFailedListing:
|
|
"""Tests for mobilede_service.retry_failed_listing."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retry_succeeds_within_max_retries(self, db_session):
|
|
"""retry_failed_listing re-attempts push when under max retries."""
|
|
vehicle = _make_vehicle()
|
|
db_session.add(vehicle)
|
|
await db_session.flush()
|
|
|
|
listing = MobileDeListing(
|
|
vehicle_id=vehicle.id,
|
|
sync_status="fehler",
|
|
error_log="[retry 1] HTTP 500: Internal Server Error",
|
|
)
|
|
db_session.add(listing)
|
|
await db_session.flush()
|
|
|
|
with patch("app.services.mobilede_service.httpx.AsyncClient") as mock_client_cls:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 201
|
|
mock_response.json.return_value = {"id": "ad-789"}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_client = AsyncMock()
|
|
mock_client.post = AsyncMock(return_value=mock_response)
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=None)
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
result = await mobilede_service.retry_failed_listing(db_session, listing, vehicle)
|
|
|
|
assert result.sync_status == "synced"
|
|
assert result.ad_id == "ad-789"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retry_exceeds_max_retries(self, db_session):
|
|
"""retry_failed_listing marks as permanently failed after max retries."""
|
|
vehicle = _make_vehicle()
|
|
db_session.add(vehicle)
|
|
await db_session.flush()
|
|
|
|
listing = MobileDeListing(
|
|
vehicle_id=vehicle.id,
|
|
sync_status="fehler",
|
|
error_log=f"[retry {mobilede_service.MAX_RETRIES}] Last error",
|
|
)
|
|
db_session.add(listing)
|
|
await db_session.flush()
|
|
|
|
result = await mobilede_service.retry_failed_listing(db_session, listing, vehicle)
|
|
|
|
assert result.sync_status == "fehler"
|
|
assert "Max retries" in result.error_log
|