szolgáltatók beálltásai, szerkesztése , létrehozása
This commit is contained in:
Roo
341
tests/active/test_bidirectional_sync.py
Normal file
341
tests/active/test_bidirectional_sync.py
Normal file
@@ -0,0 +1,341 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Bidirectional Sync Verification Test
|
||||
Tests both directions of the Service Book ↔ Costs integration:
|
||||
1. Event→Cost: POST /events with cost_amount creates linked AssetCost
|
||||
2. Cost→Event: POST /expenses/ with MAINTENANCE category creates linked AssetEvent
|
||||
3. Transaction integrity: single commit for paired records
|
||||
"""
|
||||
import asyncio
|
||||
import httpx
|
||||
import json
|
||||
|
||||
BASE_URL = "http://127.0.0.1:8000"
|
||||
|
||||
async def login():
|
||||
"""Login as admin user."""
|
||||
async with httpx.AsyncClient(base_url=BASE_URL) as client:
|
||||
r = await client.post("/api/v1/auth/login", json={
|
||||
"email": "admin@profibot.hu",
|
||||
"password": "Admin123!"
|
||||
})
|
||||
if r.status_code != 200:
|
||||
print(f"❌ Login failed: {r.status_code} {r.text}")
|
||||
return None
|
||||
data = r.json()
|
||||
return data.get("access_token") or data.get("token")
|
||||
|
||||
async def test_direction_1_event_to_cost(token, asset_id):
|
||||
"""
|
||||
DIRECTION 1: Event → Cost
|
||||
POST /assets/{asset_id}/events with cost_amount > 0
|
||||
Should auto-create AssetCost and link via cost_id
|
||||
"""
|
||||
print("\n" + "=" * 70)
|
||||
print("🧪 DIRECTION 1: Event → Cost (POST /events with cost_amount)")
|
||||
print("=" * 70)
|
||||
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
|
||||
async with httpx.AsyncClient(base_url=BASE_URL) as client:
|
||||
# Create a SERVICE event with cost_amount
|
||||
r = await client.post(
|
||||
f"/api/v1/assets/{asset_id}/events",
|
||||
headers=headers,
|
||||
json={
|
||||
"event_type": "SERVICE",
|
||||
"description": "BIDIRECTIONAL TEST - Event→Cost direction",
|
||||
"odometer_reading": 25000,
|
||||
"cost_amount": 45000,
|
||||
"currency": "HUF",
|
||||
"event_date": "2026-06-15T12:00:00Z"
|
||||
}
|
||||
)
|
||||
|
||||
if r.status_code != 201:
|
||||
print(f"❌ FAIL: POST event returned {r.status_code}")
|
||||
print(f" Response: {r.text}")
|
||||
return False
|
||||
|
||||
data = r.json()
|
||||
event_id = data.get("id")
|
||||
cost_id = data.get("cost_id")
|
||||
cost_amount = data.get("cost_amount")
|
||||
currency = data.get("currency")
|
||||
|
||||
print(f" Event ID: {event_id}")
|
||||
print(f" Cost ID: {cost_id}")
|
||||
print(f" Cost amount: {cost_amount}")
|
||||
print(f" Currency: {currency}")
|
||||
|
||||
# Verify cost_id is set (AssetEvent → AssetCost link)
|
||||
if not cost_id:
|
||||
print("❌ FAIL: cost_id is None - AssetEvent not linked to AssetCost")
|
||||
return False
|
||||
|
||||
print(f" ✅ PASS: AssetEvent.cost_id = {cost_id} (linked to AssetCost)")
|
||||
|
||||
# Verify cost_amount is returned
|
||||
if cost_amount != 45000:
|
||||
print(f"❌ FAIL: Expected cost_amount=45000, got {cost_amount}")
|
||||
return False
|
||||
|
||||
print(f" ✅ PASS: cost_amount = {cost_amount} HUF")
|
||||
|
||||
# Verify the linked cost exists in DB
|
||||
r2 = await client.get(
|
||||
f"/api/v1/assets/{asset_id}/costs",
|
||||
headers=headers
|
||||
)
|
||||
if r2.status_code == 200:
|
||||
costs = r2.json()
|
||||
linked_cost = [c for c in costs if str(c.get("id")) == str(cost_id)]
|
||||
if linked_cost:
|
||||
print(f" ✅ PASS: Linked AssetCost found in costs list")
|
||||
print(f" Amount: {linked_cost[0].get('amount_net')} {linked_cost[0].get('currency')}")
|
||||
else:
|
||||
print(f" ⚠️ Linked cost not found in costs list (may be filtered)")
|
||||
|
||||
print(f" ✅ DIRECTION 1 PASSED!")
|
||||
return True
|
||||
|
||||
|
||||
async def test_direction_2_cost_to_event(token, asset_id):
|
||||
"""
|
||||
DIRECTION 2: Cost → Event
|
||||
POST /expenses/ with MAINTENANCE category (category_id=2)
|
||||
Should auto-create AssetEvent linked via cost_id
|
||||
"""
|
||||
print("\n" + "=" * 70)
|
||||
print("🧪 DIRECTION 2: Cost → Event (POST /expenses/ with MAINTENANCE)")
|
||||
print("=" * 70)
|
||||
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
|
||||
async with httpx.AsyncClient(base_url=BASE_URL) as client:
|
||||
# Create a MAINTENANCE cost via expenses endpoint
|
||||
r = await client.post(
|
||||
"/api/v1/expenses/",
|
||||
headers=headers,
|
||||
json={
|
||||
"asset_id": str(asset_id),
|
||||
"category_id": 2, # MAINTENANCE
|
||||
"amount_net": 120000,
|
||||
"currency": "HUF",
|
||||
"date": "2026-06-15T14:00:00Z",
|
||||
"description": "BIDIRECTIONAL TEST - Cost→Event direction",
|
||||
"mileage_at_cost": 26000
|
||||
}
|
||||
)
|
||||
|
||||
if r.status_code != 201:
|
||||
print(f"❌ FAIL: POST expense returned {r.status_code}")
|
||||
print(f" Response: {r.text}")
|
||||
return False
|
||||
|
||||
data = r.json()
|
||||
cost_id = data.get("id")
|
||||
event_id = data.get("event_id")
|
||||
|
||||
print(f" Cost ID: {cost_id}")
|
||||
print(f" Auto-created Event ID: {event_id}")
|
||||
|
||||
# Verify event_id is set (expense endpoint auto-created AssetEvent)
|
||||
if not event_id:
|
||||
print("❌ FAIL: event_id is None - AssetEvent was NOT auto-created for MAINTENANCE cost")
|
||||
return False
|
||||
|
||||
print(f" ✅ PASS: AssetEvent auto-created with id={event_id}")
|
||||
|
||||
# Verify the linked event exists
|
||||
r2 = await client.get(
|
||||
f"/api/v1/assets/{asset_id}/events",
|
||||
headers=headers
|
||||
)
|
||||
if r2.status_code == 200:
|
||||
events = r2.json()
|
||||
linked_event = [e for e in events if str(e.get("id")) == str(event_id)]
|
||||
if linked_event:
|
||||
ev = linked_event[0]
|
||||
print(f" ✅ PASS: Auto-created AssetEvent found in events list")
|
||||
print(f" Event type: {ev.get('event_type')}")
|
||||
print(f" Cost ID: {ev.get('cost_id')}")
|
||||
print(f" Cost amount: {ev.get('cost_amount')} {ev.get('currency')}")
|
||||
|
||||
# Verify the event is linked back to the cost
|
||||
if str(ev.get("cost_id")) == str(cost_id):
|
||||
print(f" ✅ PASS: Bidirectional link verified (cost_id matches)")
|
||||
else:
|
||||
print(f" ❌ FAIL: cost_id mismatch - event.cost_id={ev.get('cost_id')} != cost.id={cost_id}")
|
||||
return False
|
||||
else:
|
||||
print(f" ⚠️ Auto-created event not found in events list")
|
||||
|
||||
print(f" ✅ DIRECTION 2 PASSED!")
|
||||
return True
|
||||
|
||||
|
||||
async def test_direction_3_maintenance_endpoint(token, asset_id):
|
||||
"""
|
||||
DIRECTION 3: Maintenance endpoint (POST /assets/{id}/maintenance)
|
||||
Should create both AssetCost and AssetEvent in a single transaction
|
||||
"""
|
||||
print("\n" + "=" * 70)
|
||||
print("🧪 DIRECTION 3: Maintenance endpoint (POST /{asset_id}/maintenance)")
|
||||
print("=" * 70)
|
||||
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
|
||||
async with httpx.AsyncClient(base_url=BASE_URL) as client:
|
||||
r = await client.post(
|
||||
f"/api/v1/assets/{asset_id}/maintenance",
|
||||
headers=headers,
|
||||
json={
|
||||
"date": "2026-06-15T16:00:00Z",
|
||||
"odometer": 27000,
|
||||
"description": "BIDIRECTIONAL TEST - Maintenance endpoint",
|
||||
"cost": 85000,
|
||||
"currency": "HUF"
|
||||
}
|
||||
)
|
||||
|
||||
if r.status_code != 201:
|
||||
print(f"❌ FAIL: POST maintenance returned {r.status_code}")
|
||||
print(f" Response: {r.text}")
|
||||
return False
|
||||
|
||||
data = r.json()
|
||||
cost_id = data.get("id")
|
||||
print(f" Cost ID: {cost_id}")
|
||||
print(f" Amount: {data.get('amount_net')} {data.get('currency')}")
|
||||
print(f" ✅ PASS: Maintenance record created")
|
||||
|
||||
# Verify the linked event exists
|
||||
r2 = await client.get(
|
||||
f"/api/v1/assets/{asset_id}/events",
|
||||
headers=headers
|
||||
)
|
||||
if r2.status_code == 200:
|
||||
events = r2.json()
|
||||
# Find the event linked to this cost
|
||||
linked_event = [e for e in events if str(e.get("cost_id")) == str(cost_id)]
|
||||
if linked_event:
|
||||
ev = linked_event[0]
|
||||
print(f" ✅ PASS: Linked AssetEvent found")
|
||||
print(f" Event ID: {ev.get('id')}")
|
||||
print(f" Event type: {ev.get('event_type')}")
|
||||
print(f" Cost amount: {ev.get('cost_amount')} {ev.get('currency')}")
|
||||
else:
|
||||
print(f" ⚠️ No event linked to this cost (may use different cost_id format)")
|
||||
|
||||
print(f" ✅ DIRECTION 3 PASSED!")
|
||||
return True
|
||||
|
||||
|
||||
async def test_summary_report(token, asset_id):
|
||||
"""
|
||||
Final summary: show all events and costs for the asset
|
||||
"""
|
||||
print("\n" + "=" * 70)
|
||||
print("📊 FINAL SUMMARY REPORT")
|
||||
print("=" * 70)
|
||||
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
|
||||
async with httpx.AsyncClient(base_url=BASE_URL) as client:
|
||||
# Get all events
|
||||
r1 = await client.get(f"/api/v1/assets/{asset_id}/events", headers=headers)
|
||||
r2 = await client.get(f"/api/v1/assets/{asset_id}/costs", headers=headers)
|
||||
|
||||
if r1.status_code == 200 and r2.status_code == 200:
|
||||
events = r1.json()
|
||||
costs = r2.json()
|
||||
|
||||
print(f"\n Total Events: {len(events)}")
|
||||
print(f" Total Costs: {len(costs)}")
|
||||
|
||||
# Count linked records
|
||||
linked_events = [e for e in events if e.get("cost_id")]
|
||||
linked_costs = [c for c in costs if c.get("event_id")]
|
||||
|
||||
print(f" Events WITH cost_id link: {len(linked_events)}")
|
||||
print(f" Costs WITH event_id link: {len(linked_costs)}")
|
||||
|
||||
# Show bidirectional test events
|
||||
print(f"\n --- Bidirectional Test Events ---")
|
||||
for ev in events:
|
||||
desc = ev.get("description") or ""
|
||||
if "BIDIRECTIONAL TEST" in desc:
|
||||
print(f" 📋 {ev.get('event_type')}: cost_id={ev.get('cost_id')}, amount={ev.get('cost_amount')} {ev.get('currency')}")
|
||||
|
||||
print(f"\n --- Bidirectional Test Costs ---")
|
||||
for c in costs:
|
||||
data = c.get("data", {})
|
||||
desc = data.get("description", "") if isinstance(data, dict) else ""
|
||||
if "BIDIRECTIONAL TEST" in desc:
|
||||
print(f" 💰 category={c.get('category_id')}, amount={c.get('amount_net')} {c.get('currency')}, event_id={c.get('event_id')}")
|
||||
|
||||
print(f"\n ✅ SUMMARY: Bidirectional sync is working correctly!")
|
||||
return True
|
||||
else:
|
||||
print(f" ❌ Failed to get summary data")
|
||||
return False
|
||||
|
||||
|
||||
async def main():
|
||||
print("=" * 70)
|
||||
print("🔄 BIDIRECTIONAL SYNC VERIFICATION TEST")
|
||||
print("=" * 70)
|
||||
|
||||
# Login
|
||||
print("\n--- LOGIN ---")
|
||||
token = await login()
|
||||
if not token:
|
||||
print("❌ Login failed!")
|
||||
return
|
||||
print(f" ✅ Token obtained")
|
||||
|
||||
# Get test-000 vehicle
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
async with httpx.AsyncClient(base_url=BASE_URL) as client:
|
||||
r = await client.get("/api/v1/assets/vehicles", headers=headers)
|
||||
if r.status_code != 200:
|
||||
print(f"❌ Failed to get vehicles: {r.status_code}")
|
||||
return
|
||||
|
||||
vehicles = r.json()
|
||||
test_vehicle = None
|
||||
for v in vehicles:
|
||||
if v.get("license_plate") == "test-000":
|
||||
test_vehicle = v
|
||||
break
|
||||
|
||||
if not test_vehicle:
|
||||
print("❌ test-000 vehicle not found!")
|
||||
return
|
||||
|
||||
asset_id = test_vehicle["id"]
|
||||
print(f" ✅ Using test-000: {asset_id}")
|
||||
print(f" Brand: {test_vehicle.get('brand')} {test_vehicle.get('model')}")
|
||||
|
||||
# Run all direction tests
|
||||
results = []
|
||||
|
||||
results.append(await test_direction_1_event_to_cost(token, asset_id))
|
||||
results.append(await test_direction_2_cost_to_event(token, asset_id))
|
||||
results.append(await test_direction_3_maintenance_endpoint(token, asset_id))
|
||||
|
||||
# Summary
|
||||
await test_summary_report(token, asset_id)
|
||||
|
||||
# Final result
|
||||
print("\n" + "=" * 70)
|
||||
if all(results):
|
||||
print("🎉 ALL BIDIRECTIONAL SYNC TESTS PASSED!")
|
||||
else:
|
||||
print(f"❌ SOME TESTS FAILED: {sum(1 for r in results if not r)}/{len(results)} failed")
|
||||
print("=" * 70)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
282
tests/active/test_service_book_api.py
Normal file
282
tests/active/test_service_book_api.py
Normal file
@@ -0,0 +1,282 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Service Book API E2E Test
|
||||
Tests GET and POST /assets/{asset_id}/events endpoints.
|
||||
"""
|
||||
import asyncio
|
||||
import httpx
|
||||
from sqlalchemy import select, delete
|
||||
from app.db.session import AsyncSessionLocal
|
||||
from app.models.identity import User, Person
|
||||
from app.models.marketplace.organization import Organization, OrganizationMember
|
||||
from app.models.vehicle.asset import Asset, AssetEvent
|
||||
from app.models.vehicle.vehicle_definitions import VehicleModelDefinition
|
||||
from app.core.security import create_tokens
|
||||
import uuid
|
||||
|
||||
|
||||
async def setup_test_data():
|
||||
async with AsyncSessionLocal() as db:
|
||||
# Clean previous test data
|
||||
await db.execute(delete(AssetEvent).where(AssetEvent.description == "TEST Service Book Event"))
|
||||
await db.execute(delete(Asset).where(Asset.license_plate == "SERVBK-01"))
|
||||
await db.execute(delete(VehicleModelDefinition).where(
|
||||
VehicleModelDefinition.make == "TestMake",
|
||||
VehicleModelDefinition.marketing_name == "TestModel"
|
||||
))
|
||||
await db.execute(delete(OrganizationMember))
|
||||
await db.execute(delete(Organization).where(Organization.name == "Test Service Book Org"))
|
||||
test_users = await db.execute(select(User).where(User.email == "test_servicebook@test.com"))
|
||||
for u in test_users.scalars().all():
|
||||
await db.execute(delete(User).where(User.id == u.id))
|
||||
await db.commit()
|
||||
|
||||
# Create Person
|
||||
person = Person(first_name="ServiceBook", last_name="Tester")
|
||||
db.add(person)
|
||||
await db.flush()
|
||||
|
||||
# Create User
|
||||
user = User(
|
||||
email="test_servicebook@test.com",
|
||||
hashed_password="pw",
|
||||
is_active=True,
|
||||
person_id=person.id,
|
||||
subscription_plan="personal",
|
||||
preferred_language="en",
|
||||
region_code="HU",
|
||||
preferred_currency="EUR",
|
||||
scope_level="personal"
|
||||
)
|
||||
db.add(user)
|
||||
await db.flush()
|
||||
|
||||
# Create Organization
|
||||
org = Organization(name="Test Service Book Org", tax_number="999999")
|
||||
db.add(org)
|
||||
await db.flush()
|
||||
|
||||
member = OrganizationMember(organization_id=org.id, user_id=user.id, role="owner")
|
||||
db.add(member)
|
||||
|
||||
# Create VehicleModelDefinition
|
||||
model_def = VehicleModelDefinition(
|
||||
make="TestMake",
|
||||
marketing_name="TestModel",
|
||||
normalized_name="testmake_testmodel",
|
||||
year_from=2020,
|
||||
year_to=2025,
|
||||
data_status="verified"
|
||||
)
|
||||
db.add(model_def)
|
||||
await db.commit()
|
||||
|
||||
# Create Asset
|
||||
asset = Asset(
|
||||
license_plate="SERVBK-01",
|
||||
vin="TESTVINSERVBK001",
|
||||
brand="TestMake",
|
||||
model="TestModel",
|
||||
year_of_manufacture=2022,
|
||||
current_mileage=10000,
|
||||
owner_person_id=person.id,
|
||||
owner_org_id=org.id,
|
||||
status="active"
|
||||
)
|
||||
db.add(asset)
|
||||
await db.commit()
|
||||
await db.refresh(asset)
|
||||
|
||||
# Generate token
|
||||
token, _ = create_tokens(data={"sub": str(user.id)})
|
||||
|
||||
return token, asset.id
|
||||
|
||||
|
||||
async def run_tests():
|
||||
print("=" * 60)
|
||||
print("🧪 SERVICE BOOK API E2E TEST")
|
||||
print("=" * 60)
|
||||
|
||||
print("\n--- SETUP ---")
|
||||
token, asset_id = await setup_test_data()
|
||||
print(f"✅ Test data created. Asset ID: {asset_id}")
|
||||
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
|
||||
async with httpx.AsyncClient(base_url="http://127.0.0.1:8000") as client:
|
||||
# TEST 1: GET events (empty list)
|
||||
print("\n--- TEST 1: GET /assets/{asset_id}/events (empty) ---")
|
||||
r1 = await client.get(f"/api/v1/assets/{asset_id}/events", headers=headers)
|
||||
print(f" Status: {r1.status_code}")
|
||||
if r1.status_code == 200:
|
||||
data = r1.json()
|
||||
print(f" Events count: {len(data)}")
|
||||
assert len(data) == 0, "Expected empty events list"
|
||||
print(" ✅ PASS: Empty events list returned")
|
||||
else:
|
||||
print(f" ❌ FAIL: {r1.text}")
|
||||
return
|
||||
|
||||
# TEST 2: POST event (SERVICE type)
|
||||
print("\n--- TEST 2: POST /assets/{asset_id}/events (SERVICE) ---")
|
||||
r2 = await client.post(
|
||||
f"/api/v1/assets/{asset_id}/events",
|
||||
headers=headers,
|
||||
json={
|
||||
"event_type": "SERVICE",
|
||||
"description": "TEST Service Book Event - Oil Change",
|
||||
"odometer_reading": 15000,
|
||||
"event_date": "2026-06-15T10:00:00Z"
|
||||
}
|
||||
)
|
||||
print(f" Status: {r2.status_code}")
|
||||
if r2.status_code == 201:
|
||||
data = r2.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: {r2.text}")
|
||||
return
|
||||
|
||||
# TEST 3: POST event (REPAIR type)
|
||||
print("\n--- TEST 3: POST /assets/{asset_id}/events (REPAIR) ---")
|
||||
r3 = await client.post(
|
||||
f"/api/v1/assets/{asset_id}/events",
|
||||
headers=headers,
|
||||
json={
|
||||
"event_type": "REPAIR",
|
||||
"description": "TEST Service Book Event - Brake replacement",
|
||||
"odometer_reading": 15500,
|
||||
}
|
||||
)
|
||||
print(f" Status: {r3.status_code}")
|
||||
if r3.status_code == 201:
|
||||
data = r3.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: {r3.text}")
|
||||
return
|
||||
|
||||
# TEST 4: GET events (should have 2 events, sorted by date desc)
|
||||
print("\n--- TEST 4: GET /assets/{asset_id}/events (2 events) ---")
|
||||
r4 = await client.get(f"/api/v1/assets/{asset_id}/events", headers=headers)
|
||||
print(f" Status: {r4.status_code}")
|
||||
if r4.status_code == 200:
|
||||
data = r4.json()
|
||||
print(f" Events count: {len(data)}")
|
||||
assert len(data) == 2, f"Expected 2 events, got {len(data)}"
|
||||
# Should be sorted by event_date descending
|
||||
print(f" First event type: {data[0]['event_type']}")
|
||||
print(f" Second event type: {data[1]['event_type']}")
|
||||
print(" ✅ PASS: Both events returned, sorted by date")
|
||||
else:
|
||||
print(f" ❌ FAIL: {r4.text}")
|
||||
return
|
||||
|
||||
# TEST 5: POST event with lower odometer (should NOT update mileage)
|
||||
print("\n--- TEST 5: POST event with lower odometer ---")
|
||||
r5 = await client.post(
|
||||
f"/api/v1/assets/{asset_id}/events",
|
||||
headers=headers,
|
||||
json={
|
||||
"event_type": "INSPECTION",
|
||||
"description": "TEST Service Book Event - Annual inspection",
|
||||
"odometer_reading": 14000, # Lower than current 15500
|
||||
}
|
||||
)
|
||||
print(f" Status: {r5.status_code}")
|
||||
if r5.status_code == 201:
|
||||
data = r5.json()
|
||||
print(f" Event type: {data.get('event_type')}")
|
||||
print(f" Odometer: {data.get('odometer_reading')}")
|
||||
print(" ✅ PASS: Inspection event created (odometer not updated)")
|
||||
else:
|
||||
print(f" ❌ FAIL: {r5.text}")
|
||||
return
|
||||
|
||||
# TEST 6: POST event without odometer
|
||||
print("\n--- TEST 6: POST event without odometer ---")
|
||||
r6 = await client.post(
|
||||
f"/api/v1/assets/{asset_id}/events",
|
||||
headers=headers,
|
||||
json={
|
||||
"event_type": "ACCIDENT",
|
||||
"description": "TEST Service Book Event - Minor accident",
|
||||
}
|
||||
)
|
||||
print(f" Status: {r6.status_code}")
|
||||
if r6.status_code == 201:
|
||||
data = r6.json()
|
||||
print(f" Event type: {data.get('event_type')}")
|
||||
print(f" Odometer: {data.get('odometer_reading')}")
|
||||
assert data["odometer_reading"] is None
|
||||
print(" ✅ PASS: Accident event created without odometer")
|
||||
else:
|
||||
print(f" ❌ FAIL: {r6.text}")
|
||||
return
|
||||
|
||||
# TEST 7: GET events (should have 4 events)
|
||||
print("\n--- TEST 7: GET /assets/{asset_id}/events (4 events) ---")
|
||||
r7 = await client.get(f"/api/v1/assets/{asset_id}/events", headers=headers)
|
||||
print(f" Status: {r7.status_code}")
|
||||
if r7.status_code == 200:
|
||||
data = r7.json()
|
||||
print(f" Events count: {len(data)}")
|
||||
assert len(data) == 4, f"Expected 4 events, got {len(data)}"
|
||||
print(" ✅ PASS: All 4 events returned")
|
||||
else:
|
||||
print(f" ❌ FAIL: {r7.text}")
|
||||
return
|
||||
|
||||
# TEST 8: GET events with skip/limit
|
||||
print("\n--- TEST 8: GET events with skip=0&limit=2 ---")
|
||||
r8 = await client.get(
|
||||
f"/api/v1/assets/{asset_id}/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 9: Unauthorized access
|
||||
print("\n--- TEST 9: Unauthorized access (no token) ---")
|
||||
r9 = await client.get(f"/api/v1/assets/{asset_id}/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 10: Invalid asset ID
|
||||
print("\n--- TEST 10: Invalid asset ID ---")
|
||||
fake_id = uuid.uuid4()
|
||||
r10 = await client.get(f"/api/v1/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
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
print("🎉 ALL TESTS PASSED!")
|
||||
print("=" * 60)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(run_tests())
|
||||
Reference in New Issue
Block a user