Leopoldadmin/erp-nutzfahrzeuge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: ruff lint + format fixes in tests, ESLint fixes in frontend

Browse Source
This commit is contained in:
Leopoldadmin
2026-07-17 21:28:58 +02:00
parent 341d0c6f38
commit fbb1b39b57
56 changed files with 1399 additions and 721 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/conftest.py
+3 -1
View File
@@ -5,7 +5,6 @@ from typing import AsyncGenerator
import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy import delete
from sqlalchemy.ext.asyncio import (
AsyncSession,
async_sessionmaker,
@@ -26,6 +25,7 @@ TEST_DATABASE_URL = (
def event_loop():
"""Create a fresh event loop per test for asyncpg compatibility."""
import asyncio
loop = asyncio.new_event_loop()
yield loop
loop.close()
@@ -117,6 +117,7 @@ async def inactive_user(db_session: AsyncSession) -> User:
def _get_test_token(user: User) -> str:
"""Generate an access token for a test user."""
from app.utils.jwt import create_access_token
role_val = user.role.value if isinstance(user.role, UserRole) else str(user.role)
return create_access_token(
user_id=str(user.id),
@@ -141,6 +142,7 @@ async def verkaeufer_token(verkaeufer_user: User) -> str:
@pytest_asyncio.fixture
async def client(test_session_factory) -> AsyncGenerator[AsyncClient, None]:
"""Async HTTP test client with DB session override."""
async def _override_get_db():
async with test_session_factory() as session:
try:
backend/tests/test_auth.py
+56 -30
View File
@@ -10,10 +10,13 @@ from app.utils.jwt import create_access_token, create_refresh_token
@pytest.mark.asyncio
async def test_login_valid_credentials(client: AsyncClient, admin_user: User):
"""POST /api/v1/auth/login with valid credentials returns 200 + JWT."""
response = await client.post("/api/v1/auth/login", json={
"email": "admin@test.com",
"password": "Admin12345!",
})
response = await client.post(
"/api/v1/auth/login",
json={
"email": "admin@test.com",
"password": "Admin12345!",
},
)
assert response.status_code == 200
data = response.json()
assert "access_token" in data
@@ -25,10 +28,13 @@ async def test_login_valid_credentials(client: AsyncClient, admin_user: User):
@pytest.mark.asyncio
async def test_login_invalid_password(client: AsyncClient, admin_user: User):
"""POST /api/v1/auth/login with wrong password returns 401."""
response = await client.post("/api/v1/auth/login", json={
"email": "admin@test.com",
"password": "WrongPassword!",
})
response = await client.post(
"/api/v1/auth/login",
json={
"email": "admin@test.com",
"password": "WrongPassword!",
},
)
assert response.status_code == 401
data = response.json()
assert "error" in data["detail"]
@@ -37,30 +43,39 @@ async def test_login_invalid_password(client: AsyncClient, admin_user: User):
@pytest.mark.asyncio
async def test_login_nonexistent_user(client: AsyncClient):
"""POST /api/v1/auth/login with unknown email returns 401."""
response = await client.post("/api/v1/auth/login", json={
"email": "nobody@test.com",
"password": "SomePassword!",
})
response = await client.post(
"/api/v1/auth/login",
json={
"email": "nobody@test.com",
"password": "SomePassword!",
},
)
assert response.status_code == 401
@pytest.mark.asyncio
async def test_login_inactive_user(client: AsyncClient, inactive_user: User):
"""POST /api/v1/auth/login with deactivated account returns 401."""
response = await client.post("/api/v1/auth/login", json={
"email": "inactive@test.com",
"password": "Inactive123!",
})
response = await client.post(
"/api/v1/auth/login",
json={
"email": "inactive@test.com",
"password": "Inactive123!",
},
)
assert response.status_code == 401
@pytest.mark.asyncio
async def test_login_invalid_email_format(client: AsyncClient):
"""POST /api/v1/auth/login with invalid email format returns 422."""
response = await client.post("/api/v1/auth/login", json={
"email": "not-an-email",
"password": "SomePassword!",
})
response = await client.post(
"/api/v1/auth/login",
json={
"email": "not-an-email",
"password": "SomePassword!",
},
)
assert response.status_code == 422
@@ -73,9 +88,12 @@ async def test_refresh_valid_token(client: AsyncClient, admin_user: User):
email=admin_user.email,
lang=admin_user.language,
)
response = await client.post("/api/v1/auth/refresh", json={
"refresh_token": refresh_token,
})
response = await client.post(
"/api/v1/auth/refresh",
json={
"refresh_token": refresh_token,
},
)
assert response.status_code == 200
data = response.json()
assert "access_token" in data
@@ -86,9 +104,12 @@ async def test_refresh_valid_token(client: AsyncClient, admin_user: User):
@pytest.mark.asyncio
async def test_refresh_invalid_token(client: AsyncClient):
"""POST /api/v1/auth/refresh with invalid token returns 401."""
response = await client.post("/api/v1/auth/refresh", json={
"refresh_token": "invalid.token.here",
})
response = await client.post(
"/api/v1/auth/refresh",
json={
"refresh_token": "invalid.token.here",
},
)
assert response.status_code == 401
@@ -101,14 +122,19 @@ async def test_refresh_access_token_rejected(client: AsyncClient, admin_user: Us
email=admin_user.email,
lang=admin_user.language,
)
response = await client.post("/api/v1/auth/refresh", json={
"refresh_token": access_token,
})
response = await client.post(
"/api/v1/auth/refresh",
json={
"refresh_token": access_token,
},
)
assert response.status_code == 401
@pytest.mark.asyncio
async def test_get_me_with_valid_token(client: AsyncClient, admin_user: User, admin_token: str):
async def test_get_me_with_valid_token(
client: AsyncClient, admin_user: User, admin_token: str
):
"""GET /api/v1/auth/me with valid JWT returns 200 + user object."""
response = await client.get(
"/api/v1/auth/me",
backend/tests/test_auth_service.py
+1
View File
@@ -34,6 +34,7 @@ TEST_DATABASE_URL = (
@pytest.fixture
def event_loop():
import asyncio
loop = asyncio.new_event_loop()
yield loop
loop.close()
backend/tests/test_contacts.py
+169 -55
View File
@@ -1,15 +1,11 @@
"""Tests for contact CRUD, search, filter, and contact person endpoints."""
import uuid
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from app.database import Base, get_db
from app.main import app
from app.models.contact import Contact, ContactPerson
from app.models.contact import Contact
from app.utils.ust_validation import validate_vat_id
@@ -115,7 +111,9 @@ class TestContactList:
"""GET /api/v1/contacts tests."""
@pytest.mark.asyncio
async def test_list_contacts_returns_200_with_pagination(self, admin_client, created_contact):
async def test_list_contacts_returns_200_with_pagination(
self, admin_client, created_contact
):
"""GET /api/v1/contacts returns 200 with paginated list."""
response = await admin_client.get("/api/v1/contacts/?page=1&page_size=20")
assert response.status_code == 200
@@ -136,7 +134,9 @@ class TestContactList:
resp1 = await admin_client.post("/api/v1/contacts/", json=sample_contact_data)
assert resp1.status_code == 201
# Create a beide contact
resp2 = await admin_client.post("/api/v1/contacts/", json=sample_beide_contact_data)
resp2 = await admin_client.post(
"/api/v1/contacts/", json=sample_beide_contact_data
)
assert resp2.status_code == 201
response = await admin_client.get("/api/v1/contacts/?role=kaeufer")
@@ -152,9 +152,13 @@ class TestContactList:
self, admin_client, sample_eu_contact_data, sample_beide_contact_data
):
"""GET /api/v1/contacts?role=verkaeufer returns verkaeufer + beide contacts."""
resp1 = await admin_client.post("/api/v1/contacts/", json=sample_eu_contact_data)
resp1 = await admin_client.post(
"/api/v1/contacts/", json=sample_eu_contact_data
)
assert resp1.status_code == 201
resp2 = await admin_client.post("/api/v1/contacts/", json=sample_beide_contact_data)
resp2 = await admin_client.post(
"/api/v1/contacts/", json=sample_beide_contact_data
)
assert resp2.status_code == 201
response = await admin_client.get("/api/v1/contacts/?role=verkaeufer")
@@ -194,7 +198,9 @@ class TestContactList:
assert item["address_country"] == "DE"
@pytest.mark.asyncio
async def test_list_contacts_search_by_company_name(self, admin_client, created_contact):
async def test_list_contacts_search_by_company_name(
self, admin_client, created_contact
):
"""GET /api/v1/contacts?search=mueller returns matching contacts."""
response = await admin_client.get("/api/v1/contacts/?search=mueller")
assert response.status_code == 200
@@ -220,7 +226,9 @@ class TestContactList:
assert len(data["items"]) == 0
@pytest.mark.asyncio
async def test_list_contacts_sort_by_company_name(self, admin_client, sample_contact_data, sample_eu_contact_data):
async def test_list_contacts_sort_by_company_name(
self, admin_client, sample_contact_data, sample_eu_contact_data
):
"""GET /api/v1/contacts?sort=company_name returns sorted list."""
await admin_client.post("/api/v1/contacts/", json=sample_contact_data)
await admin_client.post("/api/v1/contacts/", json=sample_eu_contact_data)
@@ -251,7 +259,9 @@ class TestContactDetail:
"""GET /api/v1/contacts/:id tests."""
@pytest.mark.asyncio
async def test_get_contact_returns_200_with_detail(self, admin_client, created_contact):
async def test_get_contact_returns_200_with_detail(
self, admin_client, created_contact
):
"""GET /api/v1/contacts/:id returns 200 with contact detail."""
response = await admin_client.get(f"/api/v1/contacts/{created_contact['id']}")
assert response.status_code == 200
@@ -271,9 +281,13 @@ class TestContactDetail:
assert data["detail"]["error"]["code"] == "CONTACT_NOT_FOUND"
@pytest.mark.asyncio
async def test_get_contact_after_soft_delete_returns_404(self, admin_client, created_contact):
async def test_get_contact_after_soft_delete_returns_404(
self, admin_client, created_contact
):
"""GET /api/v1/contacts/:id after soft-delete returns 404."""
del_resp = await admin_client.delete(f"/api/v1/contacts/{created_contact['id']}")
del_resp = await admin_client.delete(
f"/api/v1/contacts/{created_contact['id']}"
)
assert del_resp.status_code == 200
get_resp = await admin_client.get(f"/api/v1/contacts/{created_contact['id']}")
assert get_resp.status_code == 404
@@ -285,7 +299,9 @@ class TestContactCreate:
@pytest.mark.asyncio
async def test_create_contact_returns_201(self, admin_client, sample_contact_data):
"""POST /api/v1/contacts with valid data returns 201."""
response = await admin_client.post("/api/v1/contacts/", json=sample_contact_data)
response = await admin_client.post(
"/api/v1/contacts/", json=sample_contact_data
)
assert response.status_code == 201
data = response.json()
assert data["company_name"] == sample_contact_data["company_name"]
@@ -294,21 +310,27 @@ class TestContactCreate:
assert data["id"] is not None
@pytest.mark.asyncio
async def test_create_contact_with_invalid_vat_id_returns_422(self, admin_client, sample_contact_data):
async def test_create_contact_with_invalid_vat_id_returns_422(
self, admin_client, sample_contact_data
):
"""POST /api/v1/contacts with invalid VAT ID format returns 422."""
data = {**sample_contact_data, "vat_id": "INVALID123"}
response = await admin_client.post("/api/v1/contacts/", json=data)
assert response.status_code == 422
@pytest.mark.asyncio
async def test_create_contact_with_de_vat_too_short_returns_422(self, admin_client, sample_contact_data):
async def test_create_contact_with_de_vat_too_short_returns_422(
self, admin_client, sample_contact_data
):
"""POST /api/v1/contacts with too-short DE VAT ID returns 422."""
data = {**sample_contact_data, "vat_id": "DE12345678"}
response = await admin_client.post("/api/v1/contacts/", json=data)
assert response.status_code == 422
@pytest.mark.asyncio
async def test_create_contact_with_no_vat_id_returns_201(self, admin_client, sample_contact_data):
async def test_create_contact_with_no_vat_id_returns_201(
self, admin_client, sample_contact_data
):
"""POST /api/v1/contacts without VAT ID returns 201."""
data = {**sample_contact_data}
data.pop("vat_id")
@@ -317,7 +339,9 @@ class TestContactCreate:
assert response.json()["vat_id"] is None
@pytest.mark.asyncio
async def test_create_contact_with_contact_persons(self, admin_client, sample_contact_data):
async def test_create_contact_with_contact_persons(
self, admin_client, sample_contact_data
):
"""POST /api/v1/contacts with nested contact persons returns 201."""
data = {
**sample_contact_data,
@@ -337,13 +361,19 @@ class TestContactCreate:
assert contact_data["contact_persons"][0]["name"] == "Hans Müller"
@pytest.mark.asyncio
async def test_create_contact_missing_required_fields_returns_422(self, admin_client):
async def test_create_contact_missing_required_fields_returns_422(
self, admin_client
):
"""POST /api/v1/contacts with missing required fields returns 422."""
response = await admin_client.post("/api/v1/contacts/", json={"address_country": "DE"})
response = await admin_client.post(
"/api/v1/contacts/", json={"address_country": "DE"}
)
assert response.status_code == 422
@pytest.mark.asyncio
async def test_create_contact_invalid_role_returns_422(self, admin_client, sample_contact_data):
async def test_create_contact_invalid_role_returns_422(
self, admin_client, sample_contact_data
):
"""POST /api/v1/contacts with invalid role returns 422."""
data = {**sample_contact_data, "role": "invalid_role"}
response = await admin_client.post("/api/v1/contacts/", json=data)
@@ -376,7 +406,9 @@ class TestContactUpdate:
assert response.status_code == 404
@pytest.mark.asyncio
async def test_update_contact_invalid_vat_id_returns_422(self, admin_client, created_contact):
async def test_update_contact_invalid_vat_id_returns_422(
self, admin_client, created_contact
):
"""PUT /api/v1/contacts/:id with invalid VAT ID returns 422."""
response = await admin_client.put(
f"/api/v1/contacts/{created_contact['id']}",
@@ -385,7 +417,9 @@ class TestContactUpdate:
assert response.status_code == 422
@pytest.mark.asyncio
async def test_update_contact_no_fields_returns_400(self, admin_client, created_contact):
async def test_update_contact_no_fields_returns_400(
self, admin_client, created_contact
):
"""PUT /api/v1/contacts/:id with no fields returns 400."""
response = await admin_client.put(
f"/api/v1/contacts/{created_contact['id']}",
@@ -400,7 +434,9 @@ class TestContactDelete:
@pytest.mark.asyncio
async def test_delete_contact_returns_200(self, admin_client, created_contact):
"""DELETE /api/v1/contacts/:id returns 200 (soft delete)."""
response = await admin_client.delete(f"/api/v1/contacts/{created_contact['id']}")
response = await admin_client.delete(
f"/api/v1/contacts/{created_contact['id']}"
)
assert response.status_code == 200
data = response.json()
assert data["deleted_at"] is not None
@@ -415,7 +451,9 @@ class TestContactDelete:
@pytest.mark.asyncio
async def test_deleted_contact_not_in_list(self, admin_client, created_contact):
"""Soft-deleted contact does not appear in list."""
del_resp = await admin_client.delete(f"/api/v1/contacts/{created_contact['id']}")
del_resp = await admin_client.delete(
f"/api/v1/contacts/{created_contact['id']}"
)
assert del_resp.status_code == 200
list_resp = await admin_client.get("/api/v1/contacts/")
assert list_resp.status_code == 200
@@ -445,7 +483,9 @@ class TestContactPersons:
assert data["id"] is not None
@pytest.mark.asyncio
async def test_add_contact_person_to_nonexistent_contact_returns_404(self, admin_client):
async def test_add_contact_person_to_nonexistent_contact_returns_404(
self, admin_client
):
"""POST /api/v1/contacts/:nonexistent/persons returns 404."""
fake_id = uuid.uuid4()
response = await admin_client.post(
@@ -455,7 +495,9 @@ class TestContactPersons:
assert response.status_code == 404
@pytest.mark.asyncio
async def test_remove_contact_person_returns_204(self, admin_client, created_contact):
async def test_remove_contact_person_returns_204(
self, admin_client, created_contact
):
"""DELETE /api/v1/contacts/:id/persons/:person_id returns 204."""
# First add a person
add_resp = await admin_client.post(
@@ -472,7 +514,9 @@ class TestContactPersons:
assert del_resp.status_code == 204
@pytest.mark.asyncio
async def test_remove_nonexistent_contact_person_returns_404(self, admin_client, created_contact):
async def test_remove_nonexistent_contact_person_returns_404(
self, admin_client, created_contact
):
"""DELETE /api/v1/contacts/:id/persons/:nonexistent returns 404."""
fake_person_id = uuid.uuid4()
response = await admin_client.delete(
@@ -506,25 +550,37 @@ class TestContactRBAC:
assert response.status_code == 401
@pytest.mark.asyncio
async def test_create_contact_as_admin_returns_201(self, admin_client, sample_contact_data):
async def test_create_contact_as_admin_returns_201(
self, admin_client, sample_contact_data
):
"""POST /api/v1/contacts as admin returns 201."""
response = await admin_client.post("/api/v1/contacts/", json=sample_contact_data)
response = await admin_client.post(
"/api/v1/contacts/", json=sample_contact_data
)
assert response.status_code == 201
@pytest.mark.asyncio
async def test_create_contact_as_verkaeufer_returns_201(self, verkaeufer_client, sample_contact_data):
async def test_create_contact_as_verkaeufer_returns_201(
self, verkaeufer_client, sample_contact_data
):
"""POST /api/v1/contacts as verkaeufer returns 201."""
response = await verkaeufer_client.post("/api/v1/contacts/", json=sample_contact_data)
response = await verkaeufer_client.post(
"/api/v1/contacts/", json=sample_contact_data
)
assert response.status_code == 201
@pytest.mark.asyncio
async def test_list_contacts_as_verkaeufer_returns_200(self, verkaeufer_client, created_contact):
async def test_list_contacts_as_verkaeufer_returns_200(
self, verkaeufer_client, created_contact
):
"""GET /api/v1/contacts as verkaeufer returns 200 (read allowed)."""
response = await verkaeufer_client.get("/api/v1/contacts/")
assert response.status_code == 200
@pytest.mark.asyncio
async def test_update_contact_as_verkaeufer_returns_200(self, verkaeufer_client, created_contact):
async def test_update_contact_as_verkaeufer_returns_200(
self, verkaeufer_client, created_contact
):
"""PUT /api/v1/contacts/:id as verkaeufer returns 200."""
response = await verkaeufer_client.put(
f"/api/v1/contacts/{created_contact['id']}",
@@ -533,18 +589,26 @@ class TestContactRBAC:
assert response.status_code == 200
@pytest.mark.asyncio
async def test_delete_contact_as_verkaeufer_returns_200(self, verkaeufer_client, sample_contact_data):
async def test_delete_contact_as_verkaeufer_returns_200(
self, verkaeufer_client, sample_contact_data
):
"""DELETE /api/v1/contacts/:id as verkaeufer returns 200."""
resp = await verkaeufer_client.post("/api/v1/contacts/", json=sample_contact_data)
resp = await verkaeufer_client.post(
"/api/v1/contacts/", json=sample_contact_data
)
assert resp.status_code == 201
contact_id = resp.json()["id"]
del_resp = await verkaeufer_client.delete(f"/api/v1/contacts/{contact_id}")
assert del_resp.status_code == 200
@pytest.mark.asyncio
async def test_add_person_as_verkaeufer_returns_201(self, verkaeufer_client, sample_contact_data):
async def test_add_person_as_verkaeufer_returns_201(
self, verkaeufer_client, sample_contact_data
):
"""POST /api/v1/contacts/:id/persons as verkaeufer returns 201."""
resp = await verkaeufer_client.post("/api/v1/contacts/", json=sample_contact_data)
resp = await verkaeufer_client.post(
"/api/v1/contacts/", json=sample_contact_data
)
assert resp.status_code == 201
contact_id = resp.json()["id"]
person_resp = await verkaeufer_client.post(
@@ -554,9 +618,13 @@ class TestContactRBAC:
assert person_resp.status_code == 201
@pytest.mark.asyncio
async def test_remove_person_as_verkaeufer_returns_204(self, verkaeufer_client, sample_contact_data):
async def test_remove_person_as_verkaeufer_returns_204(
self, verkaeufer_client, sample_contact_data
):
"""DELETE /api/v1/contacts/:id/persons/:pid as verkaeufer returns 204."""
resp = await verkaeufer_client.post("/api/v1/contacts/", json=sample_contact_data)
resp = await verkaeufer_client.post(
"/api/v1/contacts/", json=sample_contact_data
)
contact_id = resp.json()["id"]
person_resp = await verkaeufer_client.post(
f"/api/v1/contacts/{contact_id}/persons",
@@ -569,7 +637,9 @@ class TestContactRBAC:
assert del_resp.status_code == 204
@pytest.mark.asyncio
async def test_update_nonexistent_contact_returns_404_as_verkaeufer(self, verkaeufer_client):
async def test_update_nonexistent_contact_returns_404_as_verkaeufer(
self, verkaeufer_client
):
"""PUT /api/v1/contacts/:nonexistent as verkaeufer returns 404."""
fake_id = uuid.uuid4()
response = await verkaeufer_client.put(
@@ -579,14 +649,18 @@ class TestContactRBAC:
assert response.status_code == 404
@pytest.mark.asyncio
async def test_delete_nonexistent_contact_returns_404_as_verkaeufer(self, verkaeufer_client):
async def test_delete_nonexistent_contact_returns_404_as_verkaeufer(
self, verkaeufer_client
):
"""DELETE /api/v1/contacts/:nonexistent as verkaeufer returns 404."""
fake_id = uuid.uuid4()
response = await verkaeufer_client.delete(f"/api/v1/contacts/{fake_id}")
assert response.status_code == 404
@pytest.mark.asyncio
async def test_add_person_to_nonexistent_contact_returns_404_as_verkaeufer(self, verkaeufer_client):
async def test_add_person_to_nonexistent_contact_returns_404_as_verkaeufer(
self, verkaeufer_client
):
"""POST /api/v1/contacts/:nonexistent/persons as verkaeufer returns 404."""
fake_id = uuid.uuid4()
response = await verkaeufer_client.post(
@@ -596,9 +670,13 @@ class TestContactRBAC:
assert response.status_code == 404
@pytest.mark.asyncio
async def test_remove_nonexistent_person_returns_404_as_verkaeufer(self, verkaeufer_client, sample_contact_data):
async def test_remove_nonexistent_person_returns_404_as_verkaeufer(
self, verkaeufer_client, sample_contact_data
):
"""DELETE /api/v1/contacts/:id/persons/:nonexistent as verkaeufer returns 404."""
resp = await verkaeufer_client.post("/api/v1/contacts/", json=sample_contact_data)
resp = await verkaeufer_client.post(
"/api/v1/contacts/", json=sample_contact_data
)
contact_id = resp.json()["id"]
fake_person_id = uuid.uuid4()
response = await verkaeufer_client.delete(
@@ -614,6 +692,7 @@ class TestContactServiceDirect:
async def test_service_list_contacts_with_all_filters(self, db_session):
"""Test list_contacts with all filter parameters."""
from app.services import contact_service
contact1 = Contact(
company_name="Alpha GmbH",
address_city="Berlin",
@@ -662,22 +741,29 @@ class TestContactServiceDirect:
)
db_session.add(contact4)
await db_session.commit()
results, total = await contact_service.list_contacts(db_session, is_private=True)
results, total = await contact_service.list_contacts(
db_session, is_private=True
)
assert total == 1
# Test sort descending
results, total = await contact_service.list_contacts(db_session, sort="-company_name")
results, total = await contact_service.list_contacts(
db_session, sort="-company_name"
)
names = [r.company_name for r in results]
assert names == sorted(names, reverse=True)
# Test invalid sort field falls back to created_at
results, total = await contact_service.list_contacts(db_session, sort="invalid_field")
results, total = await contact_service.list_contacts(
db_session, sort="invalid_field"
)
assert total >= 4
@pytest.mark.asyncio
async def test_service_get_contact_by_id_not_found(self, db_session):
"""Test get_contact_by_id returns None for nonexistent ID."""
from app.services import contact_service
result = await contact_service.get_contact_by_id(db_session, uuid.uuid4())
assert result is None
@@ -685,6 +771,7 @@ class TestContactServiceDirect:
async def test_service_create_contact_with_persons(self, db_session):
"""Test create_contact with nested contact persons."""
from app.services import contact_service
data = {
"company_name": "Test Service GmbH",
"address_country": "DE",
@@ -702,13 +789,17 @@ class TestContactServiceDirect:
async def test_service_update_contact_not_found(self, db_session):
"""Test update_contact returns None for nonexistent ID."""
from app.services import contact_service
result = await contact_service.update_contact(db_session, uuid.uuid4(), {"company_name": "Test"})
result = await contact_service.update_contact(
db_session, uuid.uuid4(), {"company_name": "Test"}
)
assert result is None
@pytest.mark.asyncio
async def test_service_soft_delete_contact_not_found(self, db_session):
"""Test soft_delete_contact returns None for nonexistent ID."""
from app.services import contact_service
result = await contact_service.soft_delete_contact(db_session, uuid.uuid4())
assert result is None
@@ -716,20 +807,27 @@ class TestContactServiceDirect:
async def test_service_add_contact_person_not_found(self, db_session):
"""Test add_contact_person returns None for nonexistent contact."""
from app.services import contact_service
result = await contact_service.add_contact_person(db_session, uuid.uuid4(), {"name": "Test"})
result = await contact_service.add_contact_person(
db_session, uuid.uuid4(), {"name": "Test"}
)
assert result is None
@pytest.mark.asyncio
async def test_service_remove_contact_person_not_found(self, db_session):
"""Test remove_contact_person returns False for nonexistent person."""
from app.services import contact_service
result = await contact_service.remove_contact_person(db_session, uuid.uuid4(), uuid.uuid4())
result = await contact_service.remove_contact_person(
db_session, uuid.uuid4(), uuid.uuid4()
)
assert result is False
@pytest.mark.asyncio
async def test_service_update_contact_success(self, db_session):
"""Test update_contact successfully updates fields."""
from app.services import contact_service
contact = Contact(
company_name="Original GmbH",
address_country="DE",
@@ -739,7 +837,11 @@ class TestContactServiceDirect:
await db_session.commit()
await db_session.refresh(contact)
updated = await contact_service.update_contact(db_session, contact.id, {"company_name": "Updated GmbH", "phone": "+49 30 999"})
updated = await contact_service.update_contact(
db_session,
contact.id,
{"company_name": "Updated GmbH", "phone": "+49 30 999"},
)
assert updated.company_name == "Updated GmbH"
assert updated.phone == "+49 30 999"
@@ -747,6 +849,7 @@ class TestContactServiceDirect:
async def test_service_soft_delete_contact_success(self, db_session):
"""Test soft_delete_contact sets deleted_at."""
from app.services import contact_service
contact = Contact(
company_name="To Delete GmbH",
address_country="DE",
@@ -763,6 +866,7 @@ class TestContactServiceDirect:
async def test_service_add_and_remove_contact_person(self, db_session):
"""Test add_contact_person and remove_contact_person."""
from app.services import contact_service
contact = Contact(
company_name="Person Test GmbH",
address_country="DE",
@@ -772,17 +876,22 @@ class TestContactServiceDirect:
await db_session.commit()
await db_session.refresh(contact)
person = await contact_service.add_contact_person(db_session, contact.id, {"name": "Test Person", "function": "Manager"})
person = await contact_service.add_contact_person(
db_session, contact.id, {"name": "Test Person", "function": "Manager"}
)
assert person is not None
assert person.name == "Test Person"
removed = await contact_service.remove_contact_person(db_session, contact.id, person.id)
removed = await contact_service.remove_contact_person(
db_session, contact.id, person.id
)
assert removed is True
@pytest.mark.asyncio
async def test_ust_validation_or_raise_valid(self):
"""Test validate_vat_id_or_raise with valid VAT ID."""
from app.utils.ust_validation import validate_vat_id_or_raise
result = validate_vat_id_or_raise("DE123456789")
assert result == "DE123456789"
@@ -790,6 +899,7 @@ class TestContactServiceDirect:
async def test_ust_validation_or_raise_none(self):
"""Test validate_vat_id_or_raise with None."""
from app.utils.ust_validation import validate_vat_id_or_raise
result = validate_vat_id_or_raise(None)
assert result is None
@@ -797,6 +907,7 @@ class TestContactServiceDirect:
async def test_ust_validation_or_raise_empty(self):
"""Test validate_vat_id_or_raise with empty string."""
from app.utils.ust_validation import validate_vat_id_or_raise
result = validate_vat_id_or_raise("")
assert result is None
@@ -804,6 +915,7 @@ class TestContactServiceDirect:
async def test_ust_validation_or_raise_invalid(self):
"""Test validate_vat_id_or_raise raises ValueError for invalid format."""
from app.utils.ust_validation import validate_vat_id_or_raise
with pytest.raises(ValueError, match="Invalid VAT ID format"):
validate_vat_id_or_raise("DE123")
@@ -811,6 +923,7 @@ class TestContactServiceDirect:
async def test_ust_validation_get_country_code(self):
"""Test get_country_code_from_vat_id."""
from app.utils.ust_validation import get_country_code_from_vat_id
assert get_country_code_from_vat_id("DE123456789") == "DE"
assert get_country_code_from_vat_id("at123") == "AT"
assert get_country_code_from_vat_id("") is None
@@ -821,6 +934,7 @@ class TestContactServiceDirect:
async def test_ust_validation_eu_fallback(self):
"""Test EU fallback pattern for countries without specific regex."""
from app.utils.ust_validation import validate_vat_id
# Ireland (IE) is in EU set but has no specific pattern
assert validate_vat_id("IE1234567AB") is True
# Bulgaria (BG) is in EU set but has no specific pattern
backend/tests/test_copilot.py
+49 -35
View File
@@ -3,17 +3,12 @@
import base64
import json
import uuid
from unittest.mock import AsyncMock, MagicMock, patch
from unittest.mock import AsyncMock, patch
import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from app.database import Base, get_db
from app.main import app
from app.models.copilot import CopilotChat, CopilotRole, CopilotSession
from app.models.user import User, UserRole
from app.services.auth_service import hash_password
from app.models.copilot import CopilotChat
@pytest_asyncio.fixture
@@ -37,10 +32,12 @@ async def sample_vehicle_data():
def _mock_chat_json(response_text: str, actions: list | None = None) -> str:
"""Build a JSON response string as the AI would return."""
return json.dumps({
"response": response_text,
"actions": actions or [],
})
return json.dumps(
{
"response": response_text,
"actions": actions or [],
}
)
def _patch_openrouter_chat(content: str):
@@ -188,7 +185,9 @@ class TestCopilotAction:
"""POST /api/v1/copilot/action tests."""
@pytest.mark.asyncio
async def test_action_search_vehicles_returns_200(self, admin_client, sample_vehicle_data):
async def test_action_search_vehicles_returns_200(
self, admin_client, sample_vehicle_data
):
"""POST /copilot/action with search_vehicles returns 200 + results."""
# First create a vehicle
await admin_client.post("/api/v1/vehicles/", json=sample_vehicle_data)
@@ -311,7 +310,7 @@ class TestCopilotHistory:
"/api/v1/copilot/chat",
json={"message": "Session 1 message"},
)
resp2 = await admin_client.post(
await admin_client.post(
"/api/v1/copilot/chat",
json={"message": "Session 2 message"},
)
@@ -332,16 +331,21 @@ class TestCopilotVoice:
"""POST /api/v1/copilot/voice tests."""
@pytest.mark.asyncio
async def test_voice_returns_200_with_transcription_and_response(self, admin_client):
async def test_voice_returns_200_with_transcription_and_response(
self, admin_client
):
"""POST /copilot/voice returns 200 with transcription + chat response."""
mock_content = _mock_chat_json("Ich helfe dir bei der Suche.", [])
audio_b64 = base64.b64encode(b"fake-audio-data").decode("utf-8")
with patch(
"app.services.copilot_service.transcribe_audio",
new_callable=AsyncMock,
return_value="Zeige alle LKWs",
), _patch_openrouter_chat(mock_content):
with (
patch(
"app.services.copilot_service.transcribe_audio",
new_callable=AsyncMock,
return_value="Zeige alle LKWs",
),
_patch_openrouter_chat(mock_content),
):
response = await admin_client.post(
"/api/v1/copilot/voice",
json={"audio": audio_b64, "mime_type": "audio/webm"},
@@ -385,10 +389,12 @@ class TestCopilotServiceUnit:
"""_parse_ai_response correctly parses valid JSON."""
from app.services.copilot_service import _parse_ai_response
raw = json.dumps({
"response": "Ich suche LKWs.",
"actions": [{"type": "search_vehicles", "params": {"type": "lkw"}}],
})
raw = json.dumps(
{
"response": "Ich suche LKWs.",
"actions": [{"type": "search_vehicles", "params": {"type": "lkw"}}],
}
)
result = _parse_ai_response(raw)
assert result["response"] == "Ich suche LKWs."
assert len(result["actions"]) == 1
@@ -398,10 +404,16 @@ class TestCopilotServiceUnit:
"""_parse_ai_response handles markdown code fences."""
from app.services.copilot_service import _parse_ai_response
raw = "```json\n" + json.dumps({
"response": "Test",
"actions": [],
}) + "\n```"
raw = (
"```json\n"
+ json.dumps(
{
"response": "Test",
"actions": [],
}
)
+ "\n```"
)
result = _parse_ai_response(raw)
assert result["response"] == "Test"
assert result["actions"] == []
@@ -419,14 +431,16 @@ class TestCopilotServiceUnit:
"""_parse_ai_response filters out invalid action structures."""
from app.services.copilot_service import _parse_ai_response
raw = json.dumps({
"response": "Test",
"actions": [
{"type": "search_vehicles", "params": {}},
{"invalid": "no type"},
"not a dict",
],
})
raw = json.dumps(
{
"response": "Test",
"actions": [
{"type": "search_vehicles", "params": {}},
{"invalid": "no type"},
"not a dict",
],
}
)
result = _parse_ai_response(raw)
assert len(result["actions"]) == 1
assert result["actions"][0]["type"] == "search_vehicles"
backend/tests/test_copilot_coverage.py
+63 -35
View File
@@ -1,12 +1,9 @@
"""Additional tests to improve coverage for copilot_service functions."""
import json
import uuid
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
import pytest_asyncio
from sqlalchemy import select
from app.models.copilot import CopilotChat, CopilotRole, CopilotSession
from app.services import copilot_service
@@ -18,18 +15,18 @@ class TestCopilotServiceCoverage:
@pytest.mark.asyncio
async def test_chat_calls_openrouter_and_persists(self, db_session, admin_user):
"""chat() calls _call_openrouter_chat and persists both messages."""
mock_content = json.dumps({
"response": "Ich suche LKWs.",
"actions": [{"type": "search_vehicles", "params": {"type": "lkw"}}],
})
mock_content = json.dumps(
{
"response": "Ich suche LKWs.",
"actions": [{"type": "search_vehicles", "params": {"type": "lkw"}}],
}
)
with patch(
"app.services.copilot_service._call_openrouter_chat",
new_callable=AsyncMock,
return_value=mock_content,
) as mock_chat:
result = await copilot_service.chat(
db_session, admin_user.id, "Zeige LKWs"
)
result = await copilot_service.chat(db_session, admin_user.id, "Zeige LKWs")
assert mock_chat.call_count == 1
assert result["response"] == "Ich suche LKWs."
@@ -57,7 +54,9 @@ class TestCopilotServiceCoverage:
assert result["session_id"] == str(session.id)
@pytest.mark.asyncio
async def test_chat_with_invalid_session_id_creates_new(self, db_session, admin_user):
async def test_chat_with_invalid_session_id_creates_new(
self, db_session, admin_user
):
"""chat() with invalid session_id creates a new session."""
mock_content = json.dumps({"response": "OK", "actions": []})
with patch(
@@ -79,9 +78,7 @@ class TestCopilotServiceCoverage:
new_callable=AsyncMock,
return_value="Das ist kein JSON.",
):
result = await copilot_service.chat(
db_session, admin_user.id, "Hallo"
)
result = await copilot_service.chat(db_session, admin_user.id, "Hallo")
assert result["response"] == "Das ist kein JSON."
assert result["actions"] == []
@@ -89,10 +86,18 @@ class TestCopilotServiceCoverage:
@pytest.mark.asyncio
async def test_chat_with_code_fence_response(self, db_session, admin_user):
"""chat() handles markdown code-fenced JSON response."""
content = "```json\n" + json.dumps({
"response": "Test",
"actions": [{"type": "search_contacts", "params": {"search": "Mueller"}}],
}) + "\n```"
content = (
"```json\n"
+ json.dumps(
{
"response": "Test",
"actions": [
{"type": "search_contacts", "params": {"search": "Mueller"}}
],
}
)
+ "\n```"
)
with patch(
"app.services.copilot_service._call_openrouter_chat",
new_callable=AsyncMock,
@@ -157,7 +162,9 @@ class TestCopilotServiceCoverage:
assert msg.session_id == session1.id
@pytest.mark.asyncio
async def test_get_history_with_invalid_session_id_ignores_filter(self, db_session, admin_user):
async def test_get_history_with_invalid_session_id_ignores_filter(
self, db_session, admin_user
):
"""get_history() ignores invalid session_id and returns all."""
session = CopilotSession(user_id=admin_user.id, title="Test")
db_session.add(session)
@@ -268,7 +275,9 @@ class TestCopilotServiceCoverage:
)
@pytest.mark.asyncio
async def test_execute_confirmed_action_create_vehicle_missing_fields(self, db_session):
async def test_execute_confirmed_action_create_vehicle_missing_fields(
self, db_session
):
"""execute_confirmed_action raises ValueError for missing required fields."""
with pytest.raises(ValueError, match="Missing required fields"):
await copilot_service.execute_confirmed_action(
@@ -287,8 +296,14 @@ class TestCopilotServiceCoverage:
@pytest.mark.asyncio
async def test_transcribe_audio_with_api_key_calls_openrouter(self):
"""transcribe_audio calls OpenRouter when API key is set."""
with patch("app.services.copilot_service.settings") as mock_settings, \
patch("app.services.copilot_service._call_openrouter_chat", new_callable=AsyncMock, return_value=" Transkribierter Text ") as mock_chat:
with (
patch("app.services.copilot_service.settings") as mock_settings,
patch(
"app.services.copilot_service._call_openrouter_chat",
new_callable=AsyncMock,
return_value=" Transkribierter Text ",
) as mock_chat,
):
mock_settings.OPENROUTER_API_KEY = "test-key"
result = await copilot_service.transcribe_audio(b"fake-audio", "audio/webm")
@@ -298,8 +313,14 @@ class TestCopilotServiceCoverage:
@pytest.mark.asyncio
async def test_transcribe_audio_api_error_returns_error_message(self):
"""transcribe_audio returns error message on API failure."""
with patch("app.services.copilot_service.settings") as mock_settings, \
patch("app.services.copilot_service._call_openrouter_chat", new_callable=AsyncMock, side_effect=Exception("API error")):
with (
patch("app.services.copilot_service.settings") as mock_settings,
patch(
"app.services.copilot_service._call_openrouter_chat",
new_callable=AsyncMock,
side_effect=Exception("API error"),
),
):
mock_settings.OPENROUTER_API_KEY = "test-key"
result = await copilot_service.transcribe_audio(b"fake-audio")
@@ -313,14 +334,17 @@ class TestCopilotServiceCoverage:
audio_b64 = base64.b64encode(b"fake-audio").decode("utf-8")
mock_content = json.dumps({"response": "Antwort", "actions": []})
with patch(
"app.services.copilot_service.transcribe_audio",
new_callable=AsyncMock,
return_value="Transkription",
), patch(
"app.services.copilot_service._call_openrouter_chat",
new_callable=AsyncMock,
return_value=mock_content,
with (
patch(
"app.services.copilot_service.transcribe_audio",
new_callable=AsyncMock,
return_value="Transkription",
),
patch(
"app.services.copilot_service._call_openrouter_chat",
new_callable=AsyncMock,
return_value=mock_content,
),
):
result = await copilot_service.voice_chat(
db_session, admin_user.id, audio_b64
@@ -337,7 +361,9 @@ class TestCopilotServiceCoverage:
"""_call_openrouter_chat raises ValueError when no API key."""
with patch("app.services.copilot_service.settings") as mock_settings:
mock_settings.OPENROUTER_API_KEY = ""
with pytest.raises(ValueError, match="OPENROUTER_API_KEY is not configured"):
with pytest.raises(
ValueError, match="OPENROUTER_API_KEY is not configured"
):
await copilot_service._call_openrouter_chat([])
@pytest.mark.asyncio
@@ -349,8 +375,10 @@ class TestCopilotServiceCoverage:
}
mock_response.raise_for_status = MagicMock()
with patch("app.services.copilot_service.settings") as mock_settings, \
patch("app.services.copilot_service.httpx.AsyncClient") as mock_client_cls:
with (
patch("app.services.copilot_service.settings") as mock_settings,
patch("app.services.copilot_service.httpx.AsyncClient") as mock_client_cls,
):
mock_settings.OPENROUTER_API_KEY = "test-key"
mock_settings.OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
backend/tests/test_datev.py
+53 -24
View File
@@ -5,9 +5,7 @@ import io
import uuid
from datetime import date
from decimal import Decimal
from unittest.mock import patch
import pytest
import pytest_asyncio
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession
@@ -56,10 +54,14 @@ async def test_buyer_for_datev(db_session: AsyncSession) -> Contact:
@pytest_asyncio.fixture
async def test_completed_sales(db_session: AsyncSession, test_vehicle_for_datev, test_buyer_for_datev) -> list[Sale]:
async def test_completed_sales(
db_session: AsyncSession, test_vehicle_for_datev, test_buyer_for_datev
) -> list[Sale]:
"""Create test completed sales within a date range."""
sales = []
for i, (day, price) in enumerate([(5, Decimal("30000.00")), (10, Decimal("25000.00")), (15, Decimal("40000.00"))]):
for i, (day, price) in enumerate(
[(5, Decimal("30000.00")), (10, Decimal("25000.00")), (15, Decimal("40000.00"))]
):
sale = Sale(
vehicle_id=test_vehicle_for_datev.id,
buyer_contact_id=test_buyer_for_datev.id,
@@ -81,10 +83,13 @@ class TestDATEVExportAPI:
async def test_create_export(self, admin_client: AsyncClient, test_completed_sales):
"""POST /datev/export with valid date range returns 201."""
response = await admin_client.post("/api/v1/datev/export", json={
"start_date": "2025-01-01",
"end_date": "2025-01-31",
})
response = await admin_client.post(
"/api/v1/datev/export",
json={
"start_date": "2025-01-01",
"end_date": "2025-01-31",
},
)
assert response.status_code == 201
data = response.json()
assert data["start_date"] == "2025-01-01"
@@ -94,19 +99,25 @@ class TestDATEVExportAPI:
async def test_create_export_invalid_date_range(self, admin_client: AsyncClient):
"""POST /datev/export with start > end returns 422."""
response = await admin_client.post("/api/v1/datev/export", json={
"start_date": "2025-12-31",
"end_date": "2025-01-01",
})
response = await admin_client.post(
"/api/v1/datev/export",
json={
"start_date": "2025-12-31",
"end_date": "2025-01-01",
},
)
assert response.status_code == 422
async def test_list_exports(self, admin_client: AsyncClient, test_completed_sales):
"""GET /datev/exports returns list of exports."""
# First create an export
await admin_client.post("/api/v1/datev/export", json={
"start_date": "2025-01-01",
"end_date": "2025-01-31",
})
await admin_client.post(
"/api/v1/datev/export",
json={
"start_date": "2025-01-01",
"end_date": "2025-01-31",
},
)
response = await admin_client.get("/api/v1/datev/exports")
assert response.status_code == 200
@@ -116,13 +127,18 @@ class TestDATEVExportAPI:
assert "start_date" in data["items"][0]
assert "end_date" in data["items"][0]
async def test_download_export_csv(self, admin_client: AsyncClient, test_completed_sales):
async def test_download_export_csv(
self, admin_client: AsyncClient, test_completed_sales
):
"""GET /datev/exports/:id/download returns CSV content."""
# Create export
create_response = await admin_client.post("/api/v1/datev/export", json={
"start_date": "2025-01-01",
"end_date": "2025-01-31",
})
create_response = await admin_client.post(
"/api/v1/datev/export",
json={
"start_date": "2025-01-01",
"end_date": "2025-01-31",
},
)
assert create_response.status_code == 201
export_id = create_response.json()["id"]
@@ -145,7 +161,9 @@ class TestDATEVExportAPI:
async def test_download_export_not_found(self, admin_client: AsyncClient):
"""GET /datev/exports/:nonexistent/download returns 404."""
response = await admin_client.get(f"/api/v1/datev/exports/{uuid.uuid4()}/download")
response = await admin_client.get(
f"/api/v1/datev/exports/{uuid.uuid4()}/download"
)
assert response.status_code == 404
@@ -154,7 +172,14 @@ class TestDATEVCSVFormat:
def test_datev_csv_headers(self):
"""DATEV CSV has correct headers."""
assert DATEV_HEADERS == ["Datum", "Konto", "Gegenkonto", "Betrag", "Belegfeld", "Buchungstext"]
assert DATEV_HEADERS == [
"Datum",
"Konto",
"Gegenkonto",
"Betrag",
"Belegfeld",
"Buchungstext",
]
def test_generate_datev_csv_empty(self):
"""Generate DATEV CSV with no sales returns only headers."""
@@ -170,6 +195,7 @@ class TestDATEVCSVFormat:
def test_generate_datev_csv_with_sales(self, test_completed_sales):
"""Generate DATEV CSV with sales produces correct rows."""
# test_completed_sales is a fixture but we need to call it differently for sync test
# Instead, create mock objects
class MockVehicle:
@@ -225,6 +251,7 @@ class TestDATEVCSVFormat:
def test_datev_csv_amount_format(self):
"""DATEV CSV amount uses comma as decimal separator."""
class MockVehicle:
make = "VW"
model = "Crafter"
@@ -271,6 +298,8 @@ class TestDATEVExportService:
end_date=date(2025, 1, 31),
)
exports, total = await datev_service.list_exports(db_session, page=1, page_size=2)
exports, total = await datev_service.list_exports(
db_session, page=1, page_size=2
)
assert total >= 3
assert len(exports) <= 2
backend/tests/test_files.py
+64 -28
View File
@@ -1,7 +1,6 @@
"""Tests for file upload, list, download, delete, MIME validation, size limit, and thumbnail generation."""
import io
import os
import uuid
from pathlib import Path
from unittest.mock import patch
@@ -12,16 +11,14 @@ from httpx import ASGITransport, AsyncClient
from PIL import Image
from app.config import settings
from app.database import Base, get_db
from app.database import get_db
from app.main import app
from app.models.file import File
from app.models.user import User, UserRole
from app.models.vehicle import Vehicle
from app.services.auth_service import hash_password
# ---- Test fixtures ----
@pytest_asyncio.fixture
async def sample_vehicle_data():
"""Valid vehicle data for creation."""
@@ -104,6 +101,7 @@ async def _create_test_vehicle(db_session) -> uuid.UUID:
# ---- Tests: File Upload ----
class TestFileUpload:
"""POST /api/v1/vehicles/:id/files tests."""
@@ -172,7 +170,9 @@ class TestFileUpload:
assert data["thumbnail_path"] is None
@pytest.mark.asyncio
async def test_upload_invalid_mime_type_returns_422(self, admin_client, created_vehicle):
async def test_upload_invalid_mime_type_returns_422(
self, admin_client, created_vehicle
):
"""Upload a file with an unsupported MIME type and verify 422."""
vehicle_id = created_vehicle["id"]
files = {"file": ("malware.exe", b"MZ\x90\x00", "application/x-msdownload")}
@@ -196,7 +196,9 @@ class TestFileUpload:
assert response.status_code == 422, response.text
@pytest.mark.asyncio
async def test_upload_oversized_file_returns_413(self, admin_client, created_vehicle):
async def test_upload_oversized_file_returns_413(
self, admin_client, created_vehicle
):
"""Upload a file larger than 20MB and verify 413."""
vehicle_id = created_vehicle["id"]
# Create a 21MB file (21 * 1024 * 1024 bytes)
@@ -223,7 +225,9 @@ class TestFileUpload:
assert response.status_code == 404, response.text
@pytest.mark.asyncio
async def test_upload_without_auth_returns_401(self, test_session_factory, created_vehicle):
async def test_upload_without_auth_returns_401(
self, test_session_factory, created_vehicle
):
"""Upload without authentication and verify 401."""
vehicle_id = created_vehicle["id"]
image_bytes = _make_image_bytes()
@@ -243,7 +247,9 @@ class TestFileUpload:
app.dependency_overrides[get_db] = _override_get_db
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as unauth_client:
async with AsyncClient(
transport=transport, base_url="http://test"
) as unauth_client:
response = await unauth_client.post(
f"/api/v1/vehicles/{vehicle_id}/files",
files=files,
@@ -254,11 +260,14 @@ class TestFileUpload:
# ---- Tests: File List ----
class TestFileList:
"""GET /api/v1/vehicles/:id/files tests."""
@pytest.mark.asyncio
async def test_list_files_returns_200_with_pagination(self, admin_client, created_vehicle):
async def test_list_files_returns_200_with_pagination(
self, admin_client, created_vehicle
):
"""List files for a vehicle returns 200 with paginated response."""
vehicle_id = created_vehicle["id"]
# Upload a file first
@@ -284,9 +293,7 @@ class TestFileList:
async def test_list_files_empty_returns_200(self, admin_client, created_vehicle):
"""List files for a vehicle with no files returns 200 with empty list."""
vehicle_id = created_vehicle["id"]
response = await admin_client.get(
f"/api/v1/vehicles/{vehicle_id}/files"
)
response = await admin_client.get(f"/api/v1/vehicles/{vehicle_id}/files")
assert response.status_code == 200, response.text
data = response.json()
assert data["total"] == 0
@@ -302,6 +309,7 @@ class TestFileList:
# ---- Tests: File Download ----
class TestFileDownload:
"""GET /api/v1/vehicles/:id/files/:fileId tests."""
@@ -325,7 +333,9 @@ class TestFileDownload:
assert len(response.content) == len(image_bytes)
@pytest.mark.asyncio
async def test_download_nonexistent_file_returns_404(self, admin_client, created_vehicle):
async def test_download_nonexistent_file_returns_404(
self, admin_client, created_vehicle
):
"""Download a non-existent file returns 404."""
vehicle_id = created_vehicle["id"]
fake_file_id = str(uuid.uuid4())
@@ -337,6 +347,7 @@ class TestFileDownload:
# ---- Tests: File Delete ----
class TestFileDelete:
"""DELETE /api/v1/vehicles/:id/files/:fileId tests."""
@@ -361,14 +372,14 @@ class TestFileDelete:
assert data["id"] == file_id
# Verify file is gone from list
list_resp = await admin_client.get(
f"/api/v1/vehicles/{vehicle_id}/files"
)
list_resp = await admin_client.get(f"/api/v1/vehicles/{vehicle_id}/files")
assert list_resp.status_code == 200
assert list_resp.json()["total"] == 0
@pytest.mark.asyncio
async def test_delete_nonexistent_file_returns_404(self, admin_client, created_vehicle):
async def test_delete_nonexistent_file_returns_404(
self, admin_client, created_vehicle
):
"""Delete a non-existent file returns 404."""
vehicle_id = created_vehicle["id"]
fake_file_id = str(uuid.uuid4())
@@ -380,77 +391,96 @@ class TestFileDelete:
# ---- Tests: MIME Type Validation ----
class TestMIMEValidation:
"""Unit tests for MIME type validation."""
def test_validate_jpeg_mime_type(self):
from app.services.file_service import validate_mime_type
assert validate_mime_type("image/jpeg", "photo.jpg") is True
assert validate_mime_type("image/jpeg", "photo.jpeg") is True
def test_validate_png_mime_type(self):
from app.services.file_service import validate_mime_type
assert validate_mime_type("image/png", "photo.png") is True
def test_validate_webp_mime_type(self):
from app.services.file_service import validate_mime_type
assert validate_mime_type("image/webp", "photo.webp") is True
def test_validate_pdf_mime_type(self):
from app.services.file_service import validate_mime_type
assert validate_mime_type("application/pdf", "doc.pdf") is True
def test_validate_doc_mime_type(self):
from app.services.file_service import validate_mime_type
assert validate_mime_type("application/msword", "doc.doc") is True
def test_validate_docx_mime_type(self):
from app.services.file_service import validate_mime_type
assert validate_mime_type(
"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
"doc.docx",
) is True
assert (
validate_mime_type(
"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
"doc.docx",
)
is True
)
def test_reject_exe_mime_type(self):
from app.services.file_service import validate_mime_type
assert validate_mime_type("application/x-msdownload", "malware.exe") is False
def test_reject_text_mime_type(self):
from app.services.file_service import validate_mime_type
assert validate_mime_type("text/plain", "notes.txt") is False
def test_reject_mismatched_extension(self):
"""MIME type image/jpeg with .png extension should fail."""
from app.services.file_service import validate_mime_type
assert validate_mime_type("image/jpeg", "photo.png") is False
# ---- Tests: File Size Validation ----
class TestFileSizeValidation:
"""Unit tests for file size validation."""
def test_validate_small_file_size(self):
from app.services.file_service import validate_file_size
assert validate_file_size(1024, max_size_mb=20) is True
def test_validate_exact_20mb_file_size(self):
from app.services.file_service import validate_file_size
exact_20mb = 20 * 1024 * 1024
assert validate_file_size(exact_20mb, max_size_mb=20) is True
def test_reject_oversized_file(self):
from app.services.file_service import validate_file_size
over_20mb = 20 * 1024 * 1024 + 1
assert validate_file_size(over_20mb, max_size_mb=20) is False
def test_validate_zero_byte_file(self):
from app.services.file_service import validate_file_size
assert validate_file_size(0, max_size_mb=20) is True
# ---- Tests: Thumbnail Generation ----
class TestThumbnailGeneration:
"""Tests for thumbnail generation utility."""
@@ -534,6 +564,7 @@ class TestThumbnailGeneration:
def test_is_image_mime_type(self):
"""Test is_image_mime_type helper."""
from app.utils.thumbnails import is_image_mime_type
assert is_image_mime_type("image/jpeg") is True
assert is_image_mime_type("image/png") is True
assert is_image_mime_type("image/webp") is True
@@ -543,6 +574,7 @@ class TestThumbnailGeneration:
# ---- Tests: File Service Unit Tests ----
class TestFileServiceUnit:
"""Unit tests for file service functions."""
@@ -622,7 +654,9 @@ class TestFileServiceUnit:
file_path = Path(file_record.file_path)
assert file_path.exists()
deleted = await file_service.delete_file(db_session, vehicle_id, file_record.id)
deleted = await file_service.delete_file(
db_session, vehicle_id, file_record.id
)
assert deleted is not None
assert not file_path.exists()
@@ -631,9 +665,7 @@ class TestFileServiceUnit:
"""Test that delete_file returns None for non-existent file."""
from app.services import file_service
result = await file_service.delete_file(
db_session, uuid.uuid4(), uuid.uuid4()
)
result = await file_service.delete_file(db_session, uuid.uuid4(), uuid.uuid4())
assert result is None
@pytest.mark.asyncio
@@ -682,7 +714,9 @@ class TestFileServiceUnit:
mime_type="image/jpeg",
)
retrieved = await file_service.get_file(db_session, vehicle_id, file_record.id)
retrieved = await file_service.get_file(
db_session, vehicle_id, file_record.id
)
assert retrieved is not None
assert retrieved.id == file_record.id
assert retrieved.original_filename == "test.jpg"
@@ -704,5 +738,7 @@ class TestFileServiceUnit:
)
wrong_vehicle_id = uuid.uuid4()
retrieved = await file_service.get_file(db_session, wrong_vehicle_id, file_record.id)
retrieved = await file_service.get_file(
db_session, wrong_vehicle_id, file_record.id
)
assert retrieved is None
backend/tests/test_mobilede.py
+27 -13
View File
@@ -6,8 +6,6 @@ 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
@@ -157,7 +155,9 @@ class TestPushListing:
db_session.add(vehicle)
await db_session.flush()
with patch("app.services.mobilede_service.httpx.AsyncClient") as mock_client_cls:
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"}
@@ -184,7 +184,9 @@ class TestPushListing:
db_session.add(vehicle)
await db_session.flush()
with patch("app.services.mobilede_service.httpx.AsyncClient") as mock_client_cls:
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"
@@ -212,9 +214,13 @@ class TestPushListing:
db_session.add(vehicle)
await db_session.flush()
with patch("app.services.mobilede_service.httpx.AsyncClient") as mock_client_cls:
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.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
@@ -243,7 +249,9 @@ class TestUpdateListing:
db_session.add(listing)
await db_session.flush()
with patch("app.services.mobilede_service.httpx.AsyncClient") as mock_client_cls:
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()
@@ -296,7 +304,9 @@ class TestDeleteListing:
db_session.add(listing)
await db_session.flush()
with patch("app.services.mobilede_service.httpx.AsyncClient") as mock_client_cls:
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()
@@ -340,8 +350,6 @@ class TestGetListingStatus:
db_session.add(vehicle)
await db_session.flush()
from datetime import datetime, timezone, timedelta
listing1 = MobileDeListing(
vehicle_id=vehicle.id,
sync_status="fehler",
@@ -392,7 +400,9 @@ class TestRetryFailedListing:
db_session.add(listing)
await db_session.flush()
with patch("app.services.mobilede_service.httpx.AsyncClient") as mock_client_cls:
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"}
@@ -403,7 +413,9 @@ class TestRetryFailedListing:
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)
result = await mobilede_service.retry_failed_listing(
db_session, listing, vehicle
)
assert result.sync_status == "synced"
assert result.ad_id == "ad-789"
@@ -423,7 +435,9 @@ class TestRetryFailedListing:
db_session.add(listing)
await db_session.flush()
result = await mobilede_service.retry_failed_listing(db_session, listing, vehicle)
result = await mobilede_service.retry_failed_listing(
db_session, listing, vehicle
)
assert result.sync_status == "fehler"
assert "Max retries" in result.error_log
backend/tests/test_ocr.py
+28 -14
View File
@@ -2,7 +2,6 @@
import io
import uuid
from datetime import date
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@@ -11,15 +10,14 @@ from httpx import ASGITransport, AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import Base, get_db
from app.database import get_db
from app.main import app
from app.models.ocr_result import OCRResult, OCRStatus
from app.models.vehicle import Vehicle
from app.services import ocr_service
from app.services.ocr_service import CONFIDENCE_THRESHOLD
# Ensure all models are registered with Base.metadata
from app.models import user, vehicle, ocr_result # noqa: F401
from app.models import user, vehicle as vehicle_model, ocr_result # noqa: F401
@pytest_asyncio.fixture
@@ -227,9 +225,7 @@ class TestOCRService:
assert result.structured_data["vin"] == "WDB9066351L123456"
@pytest.mark.asyncio
async def test_process_ocr_low_confidence(
self, db_session: AsyncSession, tmp_path
):
async def test_process_ocr_low_confidence(self, db_session: AsyncSession, tmp_path):
"""Test OCR processing with low confidence sets status to manual_review."""
with patch.object(ocr_service.settings, "UPLOAD_DIR", str(tmp_path)):
ocr_result = await ocr_service.upload_file(
@@ -359,6 +355,7 @@ class TestOCRRouter:
@pytest_asyncio.fixture
async def ocr_client(self, test_session_factory, admin_token):
"""HTTP client with DB override and admin auth."""
async def _override_get_db():
async with test_session_factory() as session:
try:
@@ -382,10 +379,12 @@ class TestOCRRouter:
"""POST /api/v1/ocr/upload with valid image returns 202."""
with patch.object(ocr_service.settings, "UPLOAD_DIR", str(tmp_path)):
# Patch background task to avoid actual processing
with patch("app.routers.ocr.run_ocr_processing") as mock_task:
with patch("app.routers.ocr.run_ocr_processing"):
response = await ocr_client.post(
"/api/v1/ocr/upload",
files={"file": ("scan.png", io.BytesIO(b"fake-image"), "image/png")},
files={
"file": ("scan.png", io.BytesIO(b"fake-image"), "image/png")
},
)
assert response.status_code == 202
data = response.json()
@@ -418,7 +417,9 @@ class TestOCRRouter:
with patch("app.routers.ocr.run_ocr_processing"):
upload_resp = await ocr_client.post(
"/api/v1/ocr/upload",
files={"file": ("scan.png", io.BytesIO(b"fake-image"), "image/png")},
files={
"file": ("scan.png", io.BytesIO(b"fake-image"), "image/png")
},
)
result_id = upload_resp.json()["ocr_result_id"]
@@ -443,7 +444,13 @@ class TestOCRRouter:
for i in range(3):
await ocr_client.post(
"/api/v1/ocr/upload",
files={"file": (f"scan{i}.png", io.BytesIO(b"fake-image"), "image/png")},
files={
"file": (
f"scan{i}.png",
io.BytesIO(b"fake-image"),
"image/png",
)
},
)
response = await ocr_client.get("/api/v1/ocr/results")
@@ -485,7 +492,9 @@ class TestOCRRouter:
with patch("app.routers.ocr.run_ocr_processing"):
upload_resp = await ocr_client.post(
"/api/v1/ocr/upload",
files={"file": ("scan.png", io.BytesIO(b"fake-image"), "image/png")},
files={
"file": ("scan.png", io.BytesIO(b"fake-image"), "image/png")
},
data={"vehicle_id": str(test_vehicle.id)},
)
result_id = upload_resp.json()["ocr_result_id"]
@@ -493,8 +502,11 @@ class TestOCRRouter:
# Manually set structured data via direct DB session
from tests.conftest import TEST_DATABASE_URL
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
engine = create_async_engine(TEST_DATABASE_URL)
factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
factory = async_sessionmaker(
engine, class_=AsyncSession, expire_on_commit=False
)
async with factory() as session:
stmt = select(OCRResult).where(OCRResult.id == uuid.UUID(result_id))
res = await session.execute(stmt)
@@ -527,7 +539,9 @@ class TestOpenRouterClient:
"""Test parsing a valid JSON response."""
from app.utils.openrouter import _parse_response
raw = '{"brand": "BMW", "model": "X5", "vin": "ABC123", "confidence_score": 0.9}'
raw = (
'{"brand": "BMW", "model": "X5", "vin": "ABC123", "confidence_score": 0.9}'
)
result = _parse_response(raw)
assert result["structured_data"]["brand"] == "BMW"
assert result["confidence_score"] == 0.9
backend/tests/test_retouch.py
+88 -65
View File
@@ -5,12 +5,10 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import Base, get_db
from app.main import app
from app.models.retouch import RetouchResult, RetouchStatus
from app.models.retouch import RetouchStatus
from app.models.vehicle import Vehicle
from app.services import retouch_service, price_compare_service
@@ -83,7 +81,9 @@ class TestRetouchService:
assert "red" in prompt
@pytest.mark.asyncio
async def test_upload_retouch_file_success(self, db_session: AsyncSession, tmp_path):
async def test_upload_retouch_file_success(
self, db_session: AsyncSession, tmp_path
):
with patch.object(retouch_service.settings, "UPLOAD_DIR", str(tmp_path)):
result = await retouch_service.upload_retouch_file(
db=db_session,
@@ -132,12 +132,16 @@ class TestRetouchService:
async def test_list_results_with_data(self, db_session: AsyncSession, tmp_path):
with patch.object(retouch_service.settings, "UPLOAD_DIR", str(tmp_path)):
await retouch_service.upload_retouch_file(
db=db_session, file_bytes=_fake_image(),
file_name="img1.png", mime_type="image/png",
db=db_session,
file_bytes=_fake_image(),
file_name="img1.png",
mime_type="image/png",
)
await retouch_service.upload_retouch_file(
db=db_session, file_bytes=_fake_image(),
file_name="img2.png", mime_type="image/png",
db=db_session,
file_bytes=_fake_image(),
file_name="img2.png",
mime_type="image/png",
)
await db_session.commit()
items, total = await retouch_service.list_results(db_session)
@@ -150,13 +154,17 @@ class TestRetouchService:
):
with patch.object(retouch_service.settings, "UPLOAD_DIR", str(tmp_path)):
await retouch_service.upload_retouch_file(
db=db_session, file_bytes=_fake_image(),
file_name="img1.png", mime_type="image/png",
db=db_session,
file_bytes=_fake_image(),
file_name="img1.png",
mime_type="image/png",
vehicle_id=test_vehicle.id,
)
await retouch_service.upload_retouch_file(
db=db_session, file_bytes=_fake_image(),
file_name="img2.png", mime_type="image/png",
db=db_session,
file_bytes=_fake_image(),
file_name="img2.png",
mime_type="image/png",
)
await db_session.commit()
items, total = await retouch_service.list_results(
@@ -171,20 +179,22 @@ class TestRetouchService:
"""Test process_retouch with mocked Flux.1-Pro call."""
with patch.object(retouch_service.settings, "UPLOAD_DIR", str(tmp_path)):
result = await retouch_service.upload_retouch_file(
db=db_session, file_bytes=_fake_image(),
file_name="truck.png", mime_type="image/png",
db=db_session,
file_bytes=_fake_image(),
file_name="truck.png",
mime_type="image/png",
)
await db_session.commit()
# Mock the _call_flux_pro function to return fake retouched bytes
retouched_bytes = b"\x89PNG\r\n\x1a\n" + b"\xFF" * 1016
retouched_bytes = b"\x89PNG\r\n\x1a\n" + b"\xff" * 1016
with patch.object(
retouch_service, "_call_flux_pro",
new_callable=AsyncMock, return_value=retouched_bytes,
retouch_service,
"_call_flux_pro",
new_callable=AsyncMock,
return_value=retouched_bytes,
):
processed = await retouch_service.process_retouch(
db_session, result.id
)
processed = await retouch_service.process_retouch(db_session, result.id)
await db_session.commit()
assert processed.status == RetouchStatus.completed.value
@@ -197,19 +207,20 @@ class TestRetouchService:
"""Test process_retouch sets failed status on OpenRouter error."""
with patch.object(retouch_service.settings, "UPLOAD_DIR", str(tmp_path)):
result = await retouch_service.upload_retouch_file(
db=db_session, file_bytes=_fake_image(),
file_name="truck.png", mime_type="image/png",
db=db_session,
file_bytes=_fake_image(),
file_name="truck.png",
mime_type="image/png",
)
await db_session.commit()
with patch.object(
retouch_service, "_call_flux_pro",
retouch_service,
"_call_flux_pro",
new_callable=AsyncMock,
side_effect=Exception("OpenRouter unavailable"),
):
processed = await retouch_service.process_retouch(
db_session, result.id
)
processed = await retouch_service.process_retouch(db_session, result.id)
await db_session.commit()
assert processed.status == RetouchStatus.failed.value
@@ -227,7 +238,9 @@ class TestRetouchService:
async def test_call_flux_pro_no_api_key(self):
"""Test _call_flux_pro raises ValueError when no API key configured."""
with patch.object(retouch_service.settings, "OPENROUTER_API_KEY", ""):
with pytest.raises(ValueError, match="OPENROUTER_API_KEY is not configured"):
with pytest.raises(
ValueError, match="OPENROUTER_API_KEY is not configured"
):
await retouch_service._call_flux_pro(
image_bytes=b"fake-image",
mime_type="image/png",
@@ -235,9 +248,12 @@ class TestRetouchService:
)
@pytest.mark.asyncio
async def test_call_flux_pro_success_data_uri(self, db_session: AsyncSession, tmp_path):
async def test_call_flux_pro_success_data_uri(
self, db_session: AsyncSession, tmp_path
):
"""Test _call_flux_pro with mocked httpx returning base64 data URI."""
import base64 as b64mod
retouched = b"\x89PNG\r\n\x1a\nretouched-data"
b64_retouched = b64mod.b64encode(retouched).decode("utf-8")
@@ -245,11 +261,7 @@ class TestRetouchService:
mock_response.raise_for_status = MagicMock()
mock_response.json.return_value = {
"choices": [
{
"message": {
"content": f"data:image/png;base64,{b64_retouched}"
}
}
{"message": {"content": f"data:image/png;base64,{b64_retouched}"}}
]
}
@@ -259,7 +271,10 @@ class TestRetouchService:
mock_client.__aexit__ = AsyncMock(return_value=None)
with patch.object(retouch_service.settings, "OPENROUTER_API_KEY", "test-key"):
with patch("app.services.retouch_service.httpx.AsyncClient", return_value=mock_client):
with patch(
"app.services.retouch_service.httpx.AsyncClient",
return_value=mock_client,
):
result = await retouch_service._call_flux_pro(
image_bytes=b"fake-image",
mime_type="image/png",
@@ -268,9 +283,12 @@ class TestRetouchService:
assert result == retouched
@pytest.mark.asyncio
async def test_call_flux_pro_success_list_content(self, db_session: AsyncSession, tmp_path):
async def test_call_flux_pro_success_list_content(
self, db_session: AsyncSession, tmp_path
):
"""Test _call_flux_pro with content as list of parts."""
import base64 as b64mod
retouched = b"\x89PNG\r\n\x1a\nlist-content"
b64_retouched = b64mod.b64encode(retouched).decode("utf-8")
@@ -282,7 +300,12 @@ class TestRetouchService:
"message": {
"content": [
{"type": "text", "text": "Here is the image"},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64_retouched}"}},
{
"type": "image_url",
"image_url": {
"url": f"data:image/png;base64,{b64_retouched}"
},
},
]
}
}
@@ -295,7 +318,10 @@ class TestRetouchService:
mock_client.__aexit__ = AsyncMock(return_value=None)
with patch.object(retouch_service.settings, "OPENROUTER_API_KEY", "test-key"):
with patch("app.services.retouch_service.httpx.AsyncClient", return_value=mock_client):
with patch(
"app.services.retouch_service.httpx.AsyncClient",
return_value=mock_client,
):
result = await retouch_service._call_flux_pro(
image_bytes=b"fake-image",
mime_type="image/png",
@@ -309,13 +335,7 @@ class TestRetouchService:
mock_response = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_response.json.return_value = {
"choices": [
{
"message": {
"content": "Sorry, I cannot process this image."
}
}
]
"choices": [{"message": {"content": "Sorry, I cannot process this image."}}]
}
mock_client = AsyncMock()
@@ -325,7 +345,10 @@ class TestRetouchService:
original = b"original-image-bytes"
with patch.object(retouch_service.settings, "OPENROUTER_API_KEY", "test-key"):
with patch("app.services.retouch_service.httpx.AsyncClient", return_value=mock_client):
with patch(
"app.services.retouch_service.httpx.AsyncClient",
return_value=mock_client,
):
result = await retouch_service._call_flux_pro(
image_bytes=original,
mime_type="image/png",
@@ -341,9 +364,7 @@ class TestPriceCompareService:
async def test_compare_prices_success(
self, db_session: AsyncSession, test_vehicle: Vehicle
):
result = await price_compare_service.compare_prices(
db_session, test_vehicle.id
)
result = await price_compare_service.compare_prices(db_session, test_vehicle.id)
assert result.vehicle_id == test_vehicle.id
assert len(result.comparable_listings) > 0
assert result.average_price is not None
@@ -355,17 +376,13 @@ class TestPriceCompareService:
@pytest.mark.asyncio
async def test_compare_prices_vehicle_not_found(self, db_session: AsyncSession):
with pytest.raises(ValueError, match="not found"):
await price_compare_service.compare_prices(
db_session, uuid.uuid4()
)
await price_compare_service.compare_prices(db_session, uuid.uuid4())
@pytest.mark.asyncio
async def test_compare_prices_listings_have_correct_make_model(
self, db_session: AsyncSession, test_vehicle: Vehicle
):
result = await price_compare_service.compare_prices(
db_session, test_vehicle.id
)
result = await price_compare_service.compare_prices(db_session, test_vehicle.id)
for listing in result.comparable_listings:
assert listing.make == test_vehicle.make
assert listing.model == test_vehicle.model
@@ -375,10 +392,10 @@ class TestPriceCompareService:
async def test_compare_prices_average_calculation(
self, db_session: AsyncSession, test_vehicle: Vehicle
):
result = await price_compare_service.compare_prices(
db_session, test_vehicle.id
)
expected_avg = sum(l.price for l in result.comparable_listings) / len(result.comparable_listings)
result = await price_compare_service.compare_prices(db_session, test_vehicle.id)
expected_avg = sum(
listing.price for listing in result.comparable_listings
) / len(result.comparable_listings)
assert abs(result.average_price - round(expected_avg, 2)) < 0.01
@@ -447,8 +464,10 @@ class TestRetouchRouter:
"""GET /retouch/results/:id returns 200 with completed status."""
with patch.object(retouch_service.settings, "UPLOAD_DIR", str(tmp_path)):
result = await retouch_service.upload_retouch_file(
db=db_session, file_bytes=_fake_image(),
file_name="truck.png", mime_type="image/png",
db=db_session,
file_bytes=_fake_image(),
file_name="truck.png",
mime_type="image/png",
)
result.status = RetouchStatus.completed.value
result.retouched_file_path = str(tmp_path / "retouched.png")
@@ -468,8 +487,10 @@ class TestRetouchRouter:
"""GET /retouch/results/:id before completion returns 200 with processing status."""
with patch.object(retouch_service.settings, "UPLOAD_DIR", str(tmp_path)):
result = await retouch_service.upload_retouch_file(
db=db_session, file_bytes=_fake_image(),
file_name="truck.png", mime_type="image/png",
db=db_session,
file_bytes=_fake_image(),
file_name="truck.png",
mime_type="image/png",
)
result.status = RetouchStatus.processing.value
await db_session.commit()
@@ -487,8 +508,10 @@ class TestRetouchRouter:
"""GET /retouch/results/:id with failed status returns 200 + error."""
with patch.object(retouch_service.settings, "UPLOAD_DIR", str(tmp_path)):
result = await retouch_service.upload_retouch_file(
db=db_session, file_bytes=_fake_image(),
file_name="truck.png", mime_type="image/png",
db=db_session,
file_bytes=_fake_image(),
file_name="truck.png",
mime_type="image/png",
)
result.status = RetouchStatus.failed.value
result.error_message = "OpenRouter unavailable"
backend/tests/test_sales.py
+89 -45
View File
@@ -3,9 +3,8 @@
import uuid
from datetime import date
from decimal import Decimal
from unittest.mock import patch, MagicMock
from unittest.mock import patch
import pytest
import pytest_asyncio
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession
@@ -14,7 +13,6 @@ from app.models.sale import Sale
from app.models.vehicle import Vehicle
from app.models.contact import Contact
from app.utils.contract_pdf import build_contract_html
from app.utils.datev import generate_datev_csv, validate_datev_csv, DATEV_HEADERS
@pytest_asyncio.fixture
@@ -75,7 +73,9 @@ async def test_seller(db_session: AsyncSession) -> Contact:
@pytest_asyncio.fixture
async def test_sale(db_session: AsyncSession, test_vehicle: Vehicle, test_buyer: Contact) -> Sale:
async def test_sale(
db_session: AsyncSession, test_vehicle: Vehicle, test_buyer: Contact
) -> Sale:
"""Create a test sale."""
sale = Sale(
vehicle_id=test_vehicle.id,
@@ -109,16 +109,21 @@ class TestSaleCRUD:
assert data["page"] == 1
assert data["page_size"] == 20
async def test_create_sale(self, admin_client: AsyncClient, test_vehicle, test_buyer):
async def test_create_sale(
self, admin_client: AsyncClient, test_vehicle, test_buyer
):
"""POST /sales with valid data returns 201."""
response = await admin_client.post("/api/v1/sales/", json={
"vehicle_id": str(test_vehicle.id),
"buyer_contact_id": str(test_buyer.id),
"sale_price": "45000.00",
"sale_date": "2025-01-15",
"status": "draft",
"is_gwg": False,
})
response = await admin_client.post(
"/api/v1/sales/",
json={
"vehicle_id": str(test_vehicle.id),
"buyer_contact_id": str(test_buyer.id),
"sale_price": "45000.00",
"sale_date": "2025-01-15",
"status": "draft",
"is_gwg": False,
},
)
assert response.status_code == 201
data = response.json()
assert data["vehicle_id"] == str(test_vehicle.id)
@@ -127,20 +132,30 @@ class TestSaleCRUD:
assert data["status"] == "draft"
assert data["is_gwg"] is False
async def test_create_sale_without_vehicle_id(self, admin_client: AsyncClient, test_buyer):
async def test_create_sale_without_vehicle_id(
self, admin_client: AsyncClient, test_buyer
):
"""POST /sales without vehicle_id returns 422."""
response = await admin_client.post("/api/v1/sales/", json={
"buyer_contact_id": str(test_buyer.id),
"sale_price": "45000.00",
})
response = await admin_client.post(
"/api/v1/sales/",
json={
"buyer_contact_id": str(test_buyer.id),
"sale_price": "45000.00",
},
)
assert response.status_code == 422
async def test_create_sale_without_buyer_contact_id(self, admin_client: AsyncClient, test_vehicle):
async def test_create_sale_without_buyer_contact_id(
self, admin_client: AsyncClient, test_vehicle
):
"""POST /sales without buyer_contact_id returns 422."""
response = await admin_client.post("/api/v1/sales/", json={
"vehicle_id": str(test_vehicle.id),
"sale_price": "45000.00",
})
response = await admin_client.post(
"/api/v1/sales/",
json={
"vehicle_id": str(test_vehicle.id),
"sale_price": "45000.00",
},
)
assert response.status_code == 422
async def test_get_sale_by_id(self, admin_client: AsyncClient, test_sale):
@@ -159,16 +174,21 @@ class TestSaleCRUD:
async def test_update_sale(self, admin_client: AsyncClient, test_sale):
"""PUT /sales/:id updates sale fields."""
response = await admin_client.put(f"/api/v1/sales/{test_sale.id}", json={
"sale_price": "42000.00",
"status": "completed",
})
response = await admin_client.put(
f"/api/v1/sales/{test_sale.id}",
json={
"sale_price": "42000.00",
"status": "completed",
},
)
assert response.status_code == 200
data = response.json()
assert data["sale_price"] == "42000.00"
assert data["status"] == "completed"
async def test_delete_sale_cancels_and_restores_vehicle(self, admin_client: AsyncClient, test_sale, db_session):
async def test_delete_sale_cancels_and_restores_vehicle(
self, admin_client: AsyncClient, test_sale, db_session
):
"""DELETE /sales/:id cancels sale and restores vehicle status to 'available'."""
# First set vehicle to sold (as create_sale would)
vehicle = await db_session.get(Vehicle, test_sale.vehicle_id)
@@ -193,9 +213,13 @@ class TestSaleCRUD:
for item in data["items"]:
assert item["status"] == "completed"
async def test_list_sales_with_date_filter(self, admin_client: AsyncClient, test_sale):
async def test_list_sales_with_date_filter(
self, admin_client: AsyncClient, test_sale
):
"""GET /sales?date_from=2025-01-01&date_to=2025-12-31 filters by date range."""
response = await admin_client.get("/api/v1/sales/?date_from=2025-01-01&date_to=2025-12-31")
response = await admin_client.get(
"/api/v1/sales/?date_from=2025-01-01&date_to=2025-12-31"
)
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
@@ -204,22 +228,29 @@ class TestSaleCRUD:
class TestSaleVehicleStatus:
"""Test vehicle status changes on sale create/delete."""
async def test_create_sale_sets_vehicle_sold(self, admin_client: AsyncClient, test_vehicle, test_buyer, db_session):
async def test_create_sale_sets_vehicle_sold(
self, admin_client: AsyncClient, test_vehicle, test_buyer, db_session
):
"""Creating a sale sets vehicle availability to 'sold'."""
response = await admin_client.post("/api/v1/sales/", json={
"vehicle_id": str(test_vehicle.id),
"buyer_contact_id": str(test_buyer.id),
"sale_price": "45000.00",
"sale_date": "2025-01-15",
"status": "draft",
})
response = await admin_client.post(
"/api/v1/sales/",
json={
"vehicle_id": str(test_vehicle.id),
"buyer_contact_id": str(test_buyer.id),
"sale_price": "45000.00",
"sale_date": "2025-01-15",
"status": "draft",
},
)
assert response.status_code == 201
# Verify vehicle status
await db_session.refresh(test_vehicle)
assert test_vehicle.availability == "sold"
async def test_cancel_sale_restores_vehicle_available(self, admin_client: AsyncClient, test_sale, db_session):
async def test_cancel_sale_restores_vehicle_available(
self, admin_client: AsyncClient, test_sale, db_session
):
"""Cancelling a sale restores vehicle availability to 'available'."""
# Set vehicle to sold first
vehicle = await db_session.get(Vehicle, test_sale.vehicle_id)
@@ -246,15 +277,20 @@ class TestContractPDF:
assert data["sale_id"] == str(test_sale.id)
assert "contract_pdf_path" in data
async def test_download_contract_not_found(self, admin_client: AsyncClient, test_sale):
async def test_download_contract_not_found(
self, admin_client: AsyncClient, test_sale
):
"""GET /sales/:id/contract returns 404 if no PDF generated."""
response = await admin_client.get(f"/api/v1/sales/{test_sale.id}/contract")
assert response.status_code == 404
async def test_download_contract_pdf(self, admin_client: AsyncClient, test_sale, db_session):
async def test_download_contract_pdf(
self, admin_client: AsyncClient, test_sale, db_session
):
"""GET /sales/:id/contract returns PDF content."""
# Create a fake PDF file
import os
os.makedirs("/tmp/contracts", exist_ok=True)
pdf_path = f"/tmp/contracts/contract_{test_sale.id}.pdf"
with open(pdf_path, "wb") as f:
@@ -283,7 +319,9 @@ class TestContractPDF:
assert "Geringwertige Wirtschaftsgüter" in html
assert "§ 6 Abs. 2 EStG" in html
def test_gwg_clause_not_in_contract_when_price_too_high(self, test_sale, test_vehicle, test_buyer):
def test_gwg_clause_not_in_contract_when_price_too_high(
self, test_sale, test_vehicle, test_buyer
):
"""GwG clause does NOT appear when price > 800 even if is_gwg=true."""
test_sale.is_gwg = True
test_sale.sale_price = Decimal("5000.00")
@@ -293,7 +331,9 @@ class TestContractPDF:
html = build_contract_html(test_sale)
assert "Geringwertige Wirtschaftsgüter" not in html
def test_gwg_clause_not_in_contract_when_not_gwg(self, test_sale, test_vehicle, test_buyer):
def test_gwg_clause_not_in_contract_when_not_gwg(
self, test_sale, test_vehicle, test_buyer
):
"""GwG clause does NOT appear when is_gwg=false."""
test_sale.is_gwg = False
test_sale.sale_price = Decimal("500.00")
@@ -303,7 +343,9 @@ class TestContractPDF:
html = build_contract_html(test_sale)
assert "Geringwertige Wirtschaftsgüter" not in html
def test_contract_html_contains_ust_id_field(self, test_sale, test_vehicle, test_buyer):
def test_contract_html_contains_ust_id_field(
self, test_sale, test_vehicle, test_buyer
):
"""Contract HTML contains USt-IdNr. field."""
test_sale.vehicle = test_vehicle
test_sale.buyer = test_buyer
@@ -318,7 +360,9 @@ class TestUstIdVerification:
async def test_verify_ust_id_disabled(self, admin_client: AsyncClient, test_sale):
"""POST /sales/:id/verify-ust-id returns not verified when BZSt API disabled."""
response = await admin_client.post(f"/api/v1/sales/{test_sale.id}/verify-ust-id")
response = await admin_client.post(
f"/api/v1/sales/{test_sale.id}/verify-ust-id"
)
assert response.status_code == 200
data = response.json()
assert data["verified"] is False
backend/tests/test_users.py
+52 -32
View File
@@ -53,13 +53,16 @@ async def test_list_users_pagination(admin_client: AsyncClient, admin_user: User
@pytest.mark.asyncio
async def test_create_user_as_admin(admin_client: AsyncClient):
"""POST /api/v1/users as admin creates a new user, returns 201."""
response = await admin_client.post("/api/v1/users/", json={
"email": "newuser@test.com",
"password": "NewUser123!",
"full_name": "New User",
"role": "verkaeufer",
"language": "de",
})
response = await admin_client.post(
"/api/v1/users/",
json={
"email": "newuser@test.com",
"password": "NewUser123!",
"full_name": "New User",
"role": "verkaeufer",
"language": "de",
},
)
assert response.status_code == 201
data = response.json()
assert data["email"] == "newuser@test.com"
@@ -73,44 +76,55 @@ async def test_create_user_as_admin(admin_client: AsyncClient):
@pytest.mark.asyncio
async def test_create_user_as_non_admin(verkaeufer_client: AsyncClient):
"""POST /api/v1/users as verkaeufer returns 403."""
response = await verkaeufer_client.post("/api/v1/users/", json={
"email": "forbidden@test.com",
"password": "Forbidden123!",
"full_name": "Forbidden",
"role": "admin",
"language": "de",
})
response = await verkaeufer_client.post(
"/api/v1/users/",
json={
"email": "forbidden@test.com",
"password": "Forbidden123!",
"full_name": "Forbidden",
"role": "admin",
"language": "de",
},
)
assert response.status_code == 403
@pytest.mark.asyncio
async def test_create_user_duplicate_email(admin_client: AsyncClient, admin_user: User):
"""POST /api/v1/users with existing email returns 409."""
response = await admin_client.post("/api/v1/users/", json={
"email": "admin@test.com",
"password": "SomePassword123!",
"full_name": "Duplicate",
"role": "verkaeufer",
"language": "de",
})
response = await admin_client.post(
"/api/v1/users/",
json={
"email": "admin@test.com",
"password": "SomePassword123!",
"full_name": "Duplicate",
"role": "verkaeufer",
"language": "de",
},
)
assert response.status_code == 409
@pytest.mark.asyncio
async def test_create_user_short_password(admin_client: AsyncClient):
"""POST /api/v1/users with password < 8 chars returns 422."""
response = await admin_client.post("/api/v1/users/", json={
"email": "shortpw@test.com",
"password": "short",
"full_name": "Short PW",
"role": "verkaeufer",
"language": "de",
})
response = await admin_client.post(
"/api/v1/users/",
json={
"email": "shortpw@test.com",
"password": "short",
"full_name": "Short PW",
"role": "verkaeufer",
"language": "de",
},
)
assert response.status_code == 422
@pytest.mark.asyncio
async def test_update_user_as_admin(admin_client: AsyncClient, admin_user: User, verkaeufer_user: User):
async def test_update_user_as_admin(
admin_client: AsyncClient, admin_user: User, verkaeufer_user: User
):
"""PUT /api/v1/users/:id as admin updates user fields, returns 200."""
response = await admin_client.put(
f"/api/v1/users/{verkaeufer_user.id}",
@@ -134,7 +148,9 @@ async def test_update_user_nonexistent(admin_client: AsyncClient):
@pytest.mark.asyncio
async def test_delete_user_soft_deactivate(admin_client: AsyncClient, verkaeufer_user: User):
async def test_delete_user_soft_deactivate(
admin_client: AsyncClient, verkaeufer_user: User
):
"""DELETE /api/v1/users/:id soft-deletes (is_active=false), returns 200."""
response = await admin_client.delete(f"/api/v1/users/{verkaeufer_user.id}")
assert response.status_code == 200
@@ -152,14 +168,18 @@ async def test_delete_user_nonexistent(admin_client: AsyncClient):
@pytest.mark.asyncio
async def test_delete_user_as_non_admin(verkaeufer_client: AsyncClient, admin_user: User):
async def test_delete_user_as_non_admin(
verkaeufer_client: AsyncClient, admin_user: User
):
"""DELETE /api/v1/users/:id as verkaeufer returns 403."""
response = await verkaeufer_client.delete(f"/api/v1/users/{admin_user.id}")
assert response.status_code == 403
@pytest.mark.asyncio
async def test_user_response_excludes_password_hash(admin_client: AsyncClient, admin_user: User):
async def test_user_response_excludes_password_hash(
admin_client: AsyncClient, admin_user: User
):
"""User response never includes password_hash or password fields."""
response = await admin_client.get("/api/v1/users/")
assert response.status_code == 200
backend/tests/test_vehicles.py
+78 -33
View File
@@ -1,17 +1,10 @@
"""Tests for vehicle CRUD endpoints and mobile.de integration."""
import uuid
from datetime import date
from decimal import Decimal
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from app.database import Base, get_db
from app.main import app
from app.models.vehicle import MobileDeListing, Vehicle
@pytest_asyncio.fixture
@@ -50,7 +43,9 @@ class TestVehicleList:
"""GET /api/v1/vehicles tests."""
@pytest.mark.asyncio
async def test_list_vehicles_returns_200_with_pagination(self, admin_client, created_vehicle):
async def test_list_vehicles_returns_200_with_pagination(
self, admin_client, created_vehicle
):
"""GET /api/v1/vehicles returns 200 with paginated list."""
response = await admin_client.get("/api/v1/vehicles/?page=1&page_size=20")
assert response.status_code == 200
@@ -74,7 +69,9 @@ class TestVehicleList:
assert item["vehicle_type"] == "lkw"
@pytest.mark.asyncio
async def test_list_vehicles_filter_by_availability(self, admin_client, created_vehicle):
async def test_list_vehicles_filter_by_availability(
self, admin_client, created_vehicle
):
"""GET /api/v1/vehicles?availability=available returns filtered results."""
response = await admin_client.get("/api/v1/vehicles/?availability=available")
assert response.status_code == 200
@@ -92,9 +89,13 @@ class TestVehicleList:
assert data["items"][0]["created_at"] >= data["items"][1]["created_at"]
@pytest.mark.asyncio
async def test_list_vehicles_filter_by_price_range(self, admin_client, created_vehicle):
async def test_list_vehicles_filter_by_price_range(
self, admin_client, created_vehicle
):
"""GET /api/v1/vehicles?min_price=40000&max_price=50000 returns filtered results."""
response = await admin_client.get("/api/v1/vehicles/?min_price=40000&max_price=50000")
response = await admin_client.get(
"/api/v1/vehicles/?min_price=40000&max_price=50000"
)
assert response.status_code == 200
data = response.json()
for item in data["items"]:
@@ -123,7 +124,9 @@ class TestVehicleCreate:
@pytest.mark.asyncio
async def test_create_vehicle_returns_201(self, admin_client, sample_vehicle_data):
"""POST /api/v1/vehicles with valid data returns 201."""
response = await admin_client.post("/api/v1/vehicles/", json=sample_vehicle_data)
response = await admin_client.post(
"/api/v1/vehicles/", json=sample_vehicle_data
)
assert response.status_code == 201
data = response.json()
assert data["make"] == sample_vehicle_data["make"]
@@ -133,38 +136,58 @@ class TestVehicleCreate:
assert data["id"] is not None
@pytest.mark.asyncio
async def test_create_vehicle_missing_make_returns_422(self, admin_client, sample_vehicle_data):
async def test_create_vehicle_missing_make_returns_422(
self, admin_client, sample_vehicle_data
):
"""POST /api/v1/vehicles without make returns 422."""
del sample_vehicle_data["make"]
response = await admin_client.post("/api/v1/vehicles/", json=sample_vehicle_data)
response = await admin_client.post(
"/api/v1/vehicles/", json=sample_vehicle_data
)
assert response.status_code == 422
@pytest.mark.asyncio
async def test_create_vehicle_missing_fin_returns_422(self, admin_client, sample_vehicle_data):
async def test_create_vehicle_missing_fin_returns_422(
self, admin_client, sample_vehicle_data
):
"""POST /api/v1/vehicles without fin returns 422."""
del sample_vehicle_data["fin"]
response = await admin_client.post("/api/v1/vehicles/", json=sample_vehicle_data)
response = await admin_client.post(
"/api/v1/vehicles/", json=sample_vehicle_data
)
assert response.status_code == 422
@pytest.mark.asyncio
async def test_create_vehicle_short_fin_returns_422(self, admin_client, sample_vehicle_data):
async def test_create_vehicle_short_fin_returns_422(
self, admin_client, sample_vehicle_data
):
"""POST /api/v1/vehicles with short FIN returns 422."""
sample_vehicle_data["fin"] = "SHORT"
response = await admin_client.post("/api/v1/vehicles/", json=sample_vehicle_data)
response = await admin_client.post(
"/api/v1/vehicles/", json=sample_vehicle_data
)
assert response.status_code == 422
@pytest.mark.asyncio
async def test_create_vehicle_duplicate_fin_returns_409(self, admin_client, sample_vehicle_data, created_vehicle):
async def test_create_vehicle_duplicate_fin_returns_409(
self, admin_client, sample_vehicle_data, created_vehicle
):
"""POST /api/v1/vehicles with duplicate FIN returns 409."""
response = await admin_client.post("/api/v1/vehicles/", json=sample_vehicle_data)
response = await admin_client.post(
"/api/v1/vehicles/", json=sample_vehicle_data
)
assert response.status_code == 409
@pytest.mark.asyncio
async def test_create_vehicle_auto_computes_power_hp(self, admin_client, sample_vehicle_data):
async def test_create_vehicle_auto_computes_power_hp(
self, admin_client, sample_vehicle_data
):
"""POST /api/v1/vehicles auto-computes power_hp from power_kw."""
sample_vehicle_data["power_kw"] = 100
sample_vehicle_data.pop("power_hp", None)
response = await admin_client.post("/api/v1/vehicles/", json=sample_vehicle_data)
response = await admin_client.post(
"/api/v1/vehicles/", json=sample_vehicle_data
)
assert response.status_code == 201
data = response.json()
assert data["power_hp"] == 136 # 100 * 1.35962 ≈ 136
@@ -218,7 +241,9 @@ class TestVehicleUpdate:
assert response.status_code == 404
@pytest.mark.asyncio
async def test_update_vehicle_no_fields_returns_400(self, admin_client, created_vehicle):
async def test_update_vehicle_no_fields_returns_400(
self, admin_client, created_vehicle
):
"""PUT /api/v1/vehicles/:id with no fields returns 400."""
vehicle_id = created_vehicle["id"]
response = await admin_client.put(
@@ -232,7 +257,9 @@ class TestVehicleDelete:
"""DELETE /api/v1/vehicles/:id tests."""
@pytest.mark.asyncio
async def test_delete_vehicle_returns_200_with_deleted_at(self, admin_client, created_vehicle):
async def test_delete_vehicle_returns_200_with_deleted_at(
self, admin_client, created_vehicle
):
"""DELETE /api/v1/vehicles/:id returns 200 and sets deleted_at."""
vehicle_id = created_vehicle["id"]
response = await admin_client.delete(f"/api/v1/vehicles/{vehicle_id}")
@@ -259,7 +286,9 @@ class TestVehicleDelete:
assert item["id"] != vehicle_id
@pytest.mark.asyncio
async def test_deleted_vehicle_returns_404_on_detail(self, admin_client, created_vehicle):
async def test_deleted_vehicle_returns_404_on_detail(
self, admin_client, created_vehicle
):
"""After soft-delete, GET /api/v1/vehicles/:id returns 404."""
vehicle_id = created_vehicle["id"]
await admin_client.delete(f"/api/v1/vehicles/{vehicle_id}")
@@ -274,7 +303,9 @@ class TestMobileDePush:
async def test_push_returns_202(self, admin_client, created_vehicle):
"""POST /api/v1/vehicles/:id/mobile-de/push returns 202."""
vehicle_id = created_vehicle["id"]
with patch("app.services.mobilede_service.httpx.AsyncClient") as mock_client_cls:
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": "mobile-de-ad-123"}
@@ -285,7 +316,9 @@ class TestMobileDePush:
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client_cls.return_value = mock_client
response = await admin_client.post(f"/api/v1/vehicles/{vehicle_id}/mobile-de/push")
response = await admin_client.post(
f"/api/v1/vehicles/{vehicle_id}/mobile-de/push"
)
assert response.status_code == 202
data = response.json()
@@ -305,20 +338,28 @@ class TestMobileDeStatus:
"""GET /api/v1/vehicles/:id/mobile-de/status tests."""
@pytest.mark.asyncio
async def test_status_returns_200_with_no_listing(self, admin_client, created_vehicle):
async def test_status_returns_200_with_no_listing(
self, admin_client, created_vehicle
):
"""GET /api/v1/vehicles/:id/mobile-de/status returns 200 with pending status when no listing exists."""
vehicle_id = created_vehicle["id"]
response = await admin_client.get(f"/api/v1/vehicles/{vehicle_id}/mobile-de/status")
response = await admin_client.get(
f"/api/v1/vehicles/{vehicle_id}/mobile-de/status"
)
assert response.status_code == 200
data = response.json()
assert data["synced"] is False
assert data["sync_status"] == "pending"
@pytest.mark.asyncio
async def test_status_returns_200_with_synced_listing(self, admin_client, created_vehicle):
async def test_status_returns_200_with_synced_listing(
self, admin_client, created_vehicle
):
"""GET /api/v1/vehicles/:id/mobile-de/status returns 200 with sync info after push."""
vehicle_id = created_vehicle["id"]
with patch("app.services.mobilede_service.httpx.AsyncClient") as mock_client_cls:
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": "mobile-de-ad-456"}
@@ -331,7 +372,9 @@ class TestMobileDeStatus:
await admin_client.post(f"/api/v1/vehicles/{vehicle_id}/mobile-de/push")
response = await admin_client.get(f"/api/v1/vehicles/{vehicle_id}/mobile-de/status")
response = await admin_client.get(
f"/api/v1/vehicles/{vehicle_id}/mobile-de/status"
)
assert response.status_code == 200
data = response.json()
assert data["synced"] is True
@@ -342,5 +385,7 @@ class TestMobileDeStatus:
async def test_status_nonexistent_vehicle_returns_404(self, admin_client):
"""GET /api/v1/vehicles/:id/mobile-de/status with nonexistent ID returns 404."""
fake_id = str(uuid.uuid4())
response = await admin_client.get(f"/api/v1/vehicles/{fake_id}/mobile-de/status")
response = await admin_client.get(
f"/api/v1/vehicles/{fake_id}/mobile-de/status"
)
assert response.status_code == 404
backend/tests/test_vehicles_extra.py
+76 -26
View File
@@ -1,15 +1,14 @@
"""Additional tests for vehicle_service and router to reach 80% coverage."""
import uuid
from datetime import date, datetime, timezone
from datetime import date
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.models.vehicle import Vehicle
from app.services import vehicle_service
@@ -84,25 +83,38 @@ class TestVehicleServiceDirect:
@pytest.mark.asyncio
async def test_list_vehicles_pagination(self, db_session):
"""list_vehicles respects page and page_size."""
fins = ["WDB9066351L123450", "WDB9066351L123451", "WDB9066351L123452",
"WDB9066351L123453", "WDB9066351L123454"]
fins = [
"WDB9066351L123450",
"WDB9066351L123451",
"WDB9066351L123452",
"WDB9066351L123453",
"WDB9066351L123454",
]
for fin in fins:
data = _make_vehicle_data(fin=fin)
vehicle = Vehicle(**data)
db_session.add(vehicle)
await db_session.flush()
vehicles, total = await vehicle_service.list_vehicles(db_session, page=1, page_size=2)
vehicles, total = await vehicle_service.list_vehicles(
db_session, page=1, page_size=2
)
assert len(vehicles) == 2
assert total == 5
vehicles_page2, _ = await vehicle_service.list_vehicles(db_session, page=2, page_size=2)
vehicles_page2, _ = await vehicle_service.list_vehicles(
db_session, page=2, page_size=2
)
assert len(vehicles_page2) == 2
@pytest.mark.asyncio
async def test_list_vehicles_sort_ascending(self, db_session):
"""list_vehicles sorts ascending by make."""
makes_fins = [("Zebra", "WDB9066351L000001"), ("Alpha", "WDB9066351L000002"), ("Mike", "WDB9066351L000003")]
makes_fins = [
("Zebra", "WDB9066351L000001"),
("Alpha", "WDB9066351L000002"),
("Mike", "WDB9066351L000003"),
]
for make, fin in makes_fins:
data = _make_vehicle_data(make=make, fin=fin)
vehicle = Vehicle(**data)
@@ -114,14 +126,18 @@ class TestVehicleServiceDirect:
assert makes == sorted(makes)
@pytest.mark.asyncio
async def test_list_vehicles_sort_invalid_field_defaults_to_created_at(self, db_session):
async def test_list_vehicles_sort_invalid_field_defaults_to_created_at(
self, db_session
):
"""list_vehicles falls back to created_at sort for invalid field."""
data = _make_vehicle_data()
vehicle = Vehicle(**data)
db_session.add(vehicle)
await db_session.flush()
vehicles, total = await vehicle_service.list_vehicles(db_session, sort="invalid_field")
vehicles, total = await vehicle_service.list_vehicles(
db_session, sort="invalid_field"
)
assert total == 1
assert len(vehicles) == 1
@@ -134,7 +150,9 @@ class TestVehicleServiceDirect:
db_session.add(Vehicle(**data2))
await db_session.flush()
vehicles, total = await vehicle_service.list_vehicles(db_session, min_price=50000)
vehicles, total = await vehicle_service.list_vehicles(
db_session, min_price=50000
)
assert total == 1
assert float(vehicles[0].price) >= 50000
@@ -147,7 +165,9 @@ class TestVehicleServiceDirect:
db_session.add(Vehicle(**data2))
await db_session.flush()
vehicles, total = await vehicle_service.list_vehicles(db_session, max_price=40000)
vehicles, total = await vehicle_service.list_vehicles(
db_session, max_price=40000
)
assert total == 1
assert float(vehicles[0].price) <= 40000
@@ -158,7 +178,9 @@ class TestVehicleServiceDirect:
db_session.add(Vehicle(**data))
await db_session.flush()
vehicles, total = await vehicle_service.list_vehicles(db_session, search="123456")
vehicles, total = await vehicle_service.list_vehicles(
db_session, search="123456"
)
assert total == 1
assert "123456" in vehicles[0].fin
@@ -169,7 +191,9 @@ class TestVehicleServiceDirect:
db_session.add(Vehicle(**data))
await db_session.flush()
vehicles, total = await vehicle_service.list_vehicles(db_session, search="Munich")
vehicles, total = await vehicle_service.list_vehicles(
db_session, search="Munich"
)
assert total == 1
assert vehicles[0].location == "Munich"
@@ -181,14 +205,18 @@ class TestVehicleServiceDirect:
db_session.add(vehicle)
await db_session.flush()
result = await vehicle_service.get_vehicle_by_fin(db_session, "WDB9066351L999999")
result = await vehicle_service.get_vehicle_by_fin(
db_session, "WDB9066351L999999"
)
assert result is not None
assert result.fin == "WDB9066351L999999"
@pytest.mark.asyncio
async def test_get_vehicle_by_fin_not_found(self, db_session):
"""get_vehicle_by_fin returns None for nonexistent FIN."""
result = await vehicle_service.get_vehicle_by_fin(db_session, "NONEXISTENT1234567")
result = await vehicle_service.get_vehicle_by_fin(
db_session, "NONEXISTENT1234567"
)
assert result is None
@pytest.mark.asyncio
@@ -302,14 +330,20 @@ class TestRouterAdditionalPaths:
assert data["total"] >= 1
@pytest.mark.asyncio
async def test_create_vehicle_verkaeufer_allowed(self, verkaeufer_client, sample_vehicle_data):
async def test_create_vehicle_verkaeufer_allowed(
self, verkaeufer_client, sample_vehicle_data
):
"""POST /api/v1/vehicles works for verkaeufer role (not admin-only)."""
sample_vehicle_data["fin"] = "WDB9066351L654321"
response = await verkaeufer_client.post("/api/v1/vehicles/", json=sample_vehicle_data)
response = await verkaeufer_client.post(
"/api/v1/vehicles/", json=sample_vehicle_data
)
assert response.status_code == 201
@pytest.mark.asyncio
async def test_update_vehicle_fin_duplicate_returns_409(self, admin_client, sample_vehicle_data):
async def test_update_vehicle_fin_duplicate_returns_409(
self, admin_client, sample_vehicle_data
):
"""PUT /api/v1/vehicles/:id with duplicate FIN returns 409."""
sample_vehicle_data["fin"] = "WDB9066351L111111"
resp1 = await admin_client.post("/api/v1/vehicles/", json=sample_vehicle_data)
@@ -329,7 +363,9 @@ class TestRouterAdditionalPaths:
@pytest.mark.asyncio
async def test_list_vehicles_empty_result(self, admin_client):
"""GET /api/v1/vehicles with filters that match nothing returns empty list."""
response = await admin_client.get("/api/v1/vehicles/?type=baumaschine&min_price=999999")
response = await admin_client.get(
"/api/v1/vehicles/?type=baumaschine&min_price=999999"
)
assert response.status_code == 200
data = response.json()
assert data["total"] == 0
@@ -342,11 +378,16 @@ class TestRouterAdditionalPaths:
assert response.status_code == 422
@pytest.mark.asyncio
async def test_push_to_mobile_de_failure_still_returns_202(self, admin_client, created_vehicle):
async def test_push_to_mobile_de_failure_still_returns_202(
self, admin_client, created_vehicle
):
"""POST /api/v1/vehicles/:id/mobile-de/push returns 202 even when mobile.de API fails."""
import httpx
vehicle_id = created_vehicle["id"]
with patch("app.services.mobilede_service.httpx.AsyncClient") as mock_client_cls:
with patch(
"app.services.mobilede_service.httpx.AsyncClient"
) as mock_client_cls:
mock_response = MagicMock()
mock_response.status_code = 500
mock_response.text = "Internal Server Error"
@@ -359,16 +400,23 @@ class TestRouterAdditionalPaths:
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client_cls.return_value = mock_client
response = await admin_client.post(f"/api/v1/vehicles/{vehicle_id}/mobile-de/push")
response = await admin_client.post(
f"/api/v1/vehicles/{vehicle_id}/mobile-de/push"
)
assert response.status_code == 202
@pytest.mark.asyncio
async def test_mobile_de_status_after_failed_push(self, admin_client, created_vehicle):
async def test_mobile_de_status_after_failed_push(
self, admin_client, created_vehicle
):
"""GET /api/v1/vehicles/:id/mobile-de/status shows fehler after failed push."""
import httpx
vehicle_id = created_vehicle["id"]
with patch("app.services.mobilede_service.httpx.AsyncClient") as mock_client_cls:
with patch(
"app.services.mobilede_service.httpx.AsyncClient"
) as mock_client_cls:
mock_response = MagicMock()
mock_response.status_code = 500
mock_response.text = "Internal Server Error"
@@ -383,7 +431,9 @@ class TestRouterAdditionalPaths:
await admin_client.post(f"/api/v1/vehicles/{vehicle_id}/mobile-de/push")
response = await admin_client.get(f"/api/v1/vehicles/{vehicle_id}/mobile-de/status")
response = await admin_client.get(
f"/api/v1/vehicles/{vehicle_id}/mobile-de/status"
)
assert response.status_code == 200
data = response.json()
assert data["synced"] is False