227 lines
8.4 KiB
Python
227 lines
8.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Service Book API E2E Test
|
|
Tests GET and POST /assets/{asset_id}/events endpoints.
|
|
Uses existing admin user credentials to avoid complex test data setup.
|
|
"""
|
|
import asyncio
|
|
import httpx
|
|
import uuid
|
|
|
|
|
|
BASE = "http://127.0.0.1:8000/api/v1"
|
|
|
|
|
|
async def run_tests():
|
|
print("=" * 60)
|
|
print("🧪 SERVICE BOOK API E2E TEST")
|
|
print("=" * 60)
|
|
|
|
# Login with admin credentials
|
|
print("\n--- LOGIN ---")
|
|
async with httpx.AsyncClient(base_url=BASE) as client:
|
|
r = await client.post("/auth/login", data={
|
|
"username": "admin@profibot.hu",
|
|
"password": "Admin123!"
|
|
})
|
|
print(f" Status: {r.status_code}")
|
|
if r.status_code != 200:
|
|
print(f" ❌ FAIL: Login failed: {r.text}")
|
|
return
|
|
data = r.json()
|
|
token = data.get("access_token") or data.get("token") or data.get("access")
|
|
print(f" Token obtained: {token[:30]}...")
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
# Get user's vehicles
|
|
print("\n--- GET VEHICLES ---")
|
|
r2 = await client.get("/assets/vehicles", headers=headers)
|
|
print(f" Status: {r2.status_code}")
|
|
if r2.status_code != 200:
|
|
print(f" ❌ FAIL: Could not get vehicles: {r2.text}")
|
|
return
|
|
vehicles = r2.json()
|
|
print(f" Vehicles found: {len(vehicles)}")
|
|
|
|
if not vehicles:
|
|
print("\n ⚠️ No vehicles found. Creating a test vehicle...")
|
|
# Create a vehicle
|
|
r_create = await client.post("/assets/vehicles", headers=headers, json={
|
|
"license_plate": "SERVBK-TEST",
|
|
"brand": "TestBrand",
|
|
"model": "TestModel",
|
|
"year_of_manufacture": 2022,
|
|
"current_mileage": 10000
|
|
})
|
|
print(f" Create vehicle status: {r_create.status_code}")
|
|
if r_create.status_code == 201:
|
|
vehicle = r_create.json()
|
|
vid = vehicle["id"]
|
|
print(f" ✅ Vehicle created: {vid}")
|
|
else:
|
|
print(f" ❌ Could not create vehicle: {r_create.text}")
|
|
return
|
|
else:
|
|
vid = vehicles[0]["id"]
|
|
print(f" Using vehicle: {vid}")
|
|
|
|
# TEST 1: GET events (empty list)
|
|
print("\n--- TEST 1: GET /assets/{asset_id}/events (empty) ---")
|
|
r3 = await client.get(f"/assets/{vid}/events", headers=headers)
|
|
print(f" Status: {r3.status_code}")
|
|
if r3.status_code == 200:
|
|
data = r3.json()
|
|
print(f" Events count: {len(data)}")
|
|
print(" ✅ PASS: Events list returned")
|
|
else:
|
|
print(f" ❌ FAIL: {r3.text}")
|
|
return
|
|
|
|
# TEST 2: POST event (SERVICE type)
|
|
print("\n--- TEST 2: POST /assets/{asset_id}/events (SERVICE) ---")
|
|
r4 = await client.post(
|
|
f"/assets/{vid}/events",
|
|
headers=headers,
|
|
json={
|
|
"event_type": "SERVICE",
|
|
"description": "Test Service Event - Oil Change",
|
|
"odometer_reading": 15000,
|
|
"event_date": "2026-06-15T10:00:00Z"
|
|
}
|
|
)
|
|
print(f" Status: {r4.status_code}")
|
|
if r4.status_code == 201:
|
|
data = r4.json()
|
|
print(f" Event ID: {data.get('id')}")
|
|
print(f" Event type: {data.get('event_type')}")
|
|
print(f" Odometer: {data.get('odometer_reading')}")
|
|
assert data["event_type"] == "SERVICE"
|
|
assert data["odometer_reading"] == 15000
|
|
print(" ✅ PASS: Service event created")
|
|
else:
|
|
print(f" ❌ FAIL: {r4.text}")
|
|
return
|
|
|
|
# TEST 3: POST event (REPAIR type)
|
|
print("\n--- TEST 3: POST /assets/{asset_id}/events (REPAIR) ---")
|
|
r5 = await client.post(
|
|
f"/assets/{vid}/events",
|
|
headers=headers,
|
|
json={
|
|
"event_type": "REPAIR",
|
|
"description": "Test Service Event - Brake replacement",
|
|
"odometer_reading": 15500,
|
|
}
|
|
)
|
|
print(f" Status: {r5.status_code}")
|
|
if r5.status_code == 201:
|
|
data = r5.json()
|
|
print(f" Event ID: {data.get('id')}")
|
|
print(f" Event type: {data.get('event_type')}")
|
|
assert data["event_type"] == "REPAIR"
|
|
print(" ✅ PASS: Repair event created")
|
|
else:
|
|
print(f" ❌ FAIL: {r5.text}")
|
|
return
|
|
|
|
# TEST 4: GET events (should have 2 events)
|
|
print("\n--- TEST 4: GET /assets/{asset_id}/events (2 events) ---")
|
|
r6 = await client.get(f"/assets/{vid}/events", headers=headers)
|
|
print(f" Status: {r6.status_code}")
|
|
if r6.status_code == 200:
|
|
data = r6.json()
|
|
print(f" Events count: {len(data)}")
|
|
assert len(data) >= 2, f"Expected at least 2 events, got {len(data)}"
|
|
print(f" First event type: {data[0]['event_type']}")
|
|
print(f" Second event type: {data[1]['event_type']}")
|
|
print(" ✅ PASS: Events returned, sorted by date")
|
|
else:
|
|
print(f" ❌ FAIL: {r6.text}")
|
|
return
|
|
|
|
# TEST 5: POST event without odometer
|
|
print("\n--- TEST 5: POST event without odometer ---")
|
|
r7 = await client.post(
|
|
f"/assets/{vid}/events",
|
|
headers=headers,
|
|
json={
|
|
"event_type": "INSPECTION",
|
|
"description": "Test Service Event - Annual inspection",
|
|
}
|
|
)
|
|
print(f" Status: {r7.status_code}")
|
|
if r7.status_code == 201:
|
|
data = r7.json()
|
|
print(f" Event type: {data.get('event_type')}")
|
|
print(f" Odometer: {data.get('odometer_reading')}")
|
|
assert data["odometer_reading"] is None
|
|
print(" ✅ PASS: Inspection event created without odometer")
|
|
else:
|
|
print(f" ❌ FAIL: {r7.text}")
|
|
return
|
|
|
|
# TEST 6: GET events with pagination
|
|
print("\n--- TEST 6: GET events with skip=0&limit=2 ---")
|
|
r8 = await client.get(f"/assets/{vid}/events?skip=0&limit=2", headers=headers)
|
|
print(f" Status: {r8.status_code}")
|
|
if r8.status_code == 200:
|
|
data = r8.json()
|
|
print(f" Events count: {len(data)}")
|
|
assert len(data) == 2, f"Expected 2 events, got {len(data)}"
|
|
print(" ✅ PASS: Pagination works correctly")
|
|
else:
|
|
print(f" ❌ FAIL: {r8.text}")
|
|
return
|
|
|
|
# TEST 7: Unauthorized access
|
|
print("\n--- TEST 7: Unauthorized access (no token) ---")
|
|
r9 = await client.get(f"/assets/{vid}/events")
|
|
print(f" Status: {r9.status_code}")
|
|
if r9.status_code == 401:
|
|
print(" ✅ PASS: Unauthorized access rejected")
|
|
else:
|
|
print(f" ❌ FAIL: Expected 401, got {r9.status_code}")
|
|
return
|
|
|
|
# TEST 8: Invalid asset ID
|
|
print("\n--- TEST 8: Invalid asset ID ---")
|
|
fake_id = uuid.uuid4()
|
|
r10 = await client.get(f"/assets/{fake_id}/events", headers=headers)
|
|
print(f" Status: {r10.status_code}")
|
|
if r10.status_code == 404:
|
|
print(" ✅ PASS: Invalid asset ID returns 404")
|
|
else:
|
|
print(f" ❌ FAIL: Expected 404, got {r10.status_code}")
|
|
return
|
|
|
|
# TEST 9: POST with all event types
|
|
print("\n--- TEST 9: POST with all event types ---")
|
|
event_types = ["ACCIDENT", "TIRE_CHANGE", "MAINTENANCE", "UPGRADE", "RECALL"]
|
|
for et in event_types:
|
|
r11 = await client.post(
|
|
f"/assets/{vid}/events",
|
|
headers=headers,
|
|
json={
|
|
"event_type": et,
|
|
"description": f"Test {et} event",
|
|
}
|
|
)
|
|
status = "✅" if r11.status_code == 201 else "❌"
|
|
print(f" {status} {et}: {r11.status_code}")
|
|
|
|
# TEST 10: Final count
|
|
print("\n--- TEST 10: Final events count ---")
|
|
r12 = await client.get(f"/assets/{vid}/events", headers=headers)
|
|
if r12.status_code == 200:
|
|
data = r12.json()
|
|
print(f" Total events: {len(data)}")
|
|
print(" ✅ PASS: All event types accepted")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("🎉 ALL TESTS PASSED!")
|
|
print("=" * 60)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(run_tests())
|