109 lines
4.4 KiB
Python
109 lines
4.4 KiB
Python
import asyncio
|
|
import httpx
|
|
from sqlalchemy import select, delete
|
|
from app.db.session import AsyncSessionLocal
|
|
from app.models.identity import User
|
|
from app.models.marketplace.organization import Organization, Branch
|
|
from app.models.vehicle.asset import Asset
|
|
from app.models.vehicle.vehicle_definitions import VehicleModelDefinition
|
|
from app.core.security import create_tokens
|
|
|
|
async def setup_test_data():
|
|
async with AsyncSessionLocal() as db:
|
|
user1_id = 2
|
|
user2_id = 3
|
|
org_id = 1
|
|
|
|
# Clean old test data
|
|
from app.models.vehicle.asset import AssetAssignment
|
|
assets_to_delete = (await db.execute(select(Asset).where(Asset.license_plate.in_(["TEST-000", "DRAFT-000", "OTHER-000"])))).scalars().all()
|
|
for a in assets_to_delete:
|
|
await db.execute(delete(AssetAssignment).where(AssetAssignment.asset_id == a.id))
|
|
await db.execute(delete(Asset).where(Asset.license_plate.in_(["TEST-000", "DRAFT-000", "OTHER-000"])))
|
|
await db.commit()
|
|
|
|
# Ensure main branch exists for org 1
|
|
branch_stmt = select(Branch).where(Branch.organization_id == org_id, Branch.is_main == True)
|
|
branch = (await db.execute(branch_stmt)).scalar()
|
|
|
|
token1, _ = create_tokens(data={"sub": str(user1_id)})
|
|
token2, _ = create_tokens(data={"sub": str(user2_id)})
|
|
|
|
return token1, token2, org_id, branch.id
|
|
|
|
async def run_tests():
|
|
print("--- SETUP ---")
|
|
result = await setup_test_data()
|
|
if not result: return
|
|
token1, token2, org_id, branch_id = result
|
|
print("Test data created successfully.")
|
|
|
|
headers1 = {"Authorization": f"Bearer {token1}"}
|
|
headers2 = {"Authorization": f"Bearer {token2}"}
|
|
|
|
async with httpx.AsyncClient(base_url="http://127.0.0.1:8000") as client:
|
|
print("\n--- TEST 1: CATALOG MATCH & GARAGE TEST ---")
|
|
payload1 = {
|
|
"vin": "TESTVIN1111111111",
|
|
"license_plate": "TEST-000",
|
|
"brand": "Ford",
|
|
"model": "Focus",
|
|
"year_of_manufacture": 2018,
|
|
"organization_id": org_id,
|
|
"vehicle_class": "car",
|
|
"fuel_type": "petrol"
|
|
}
|
|
resp1 = await client.post("/api/v1/assets/vehicles", json=payload1, headers=headers1)
|
|
print(f"Status: {resp1.status_code}")
|
|
try:
|
|
data1 = resp1.json()
|
|
if resp1.status_code == 201:
|
|
assert data1["catalog_id"] > 0, f"Matcher failed: got None"
|
|
assert data1["branch_id"] == str(branch_id), f"Branch failed: expected {branch_id}, got {data1.get('branch_id')}"
|
|
print(f"-> SUCCESS: Matcher assigned catalog ({data1['catalog_id']}) and Main Branch assigned.")
|
|
else:
|
|
print(f"Failed with {data1}")
|
|
except Exception as e:
|
|
print(f"-> FAILED: {e}")
|
|
|
|
print("\n--- TEST 2: DRAFT / INCOMPLETE TEST ---")
|
|
payload2 = {
|
|
"license_plate": "DRAFT-000",
|
|
"brand": "Ismeretlen",
|
|
"organization_id": org_id
|
|
}
|
|
resp2 = await client.post("/api/v1/assets/vehicles", json=payload2, headers=headers1)
|
|
print(f"Status: {resp2.status_code}")
|
|
try:
|
|
data2 = resp2.json()
|
|
if resp2.status_code == 201:
|
|
assert data2["status"] == "draft", f"Status failed: expected 'draft', got {data2.get('status')}"
|
|
assert data2["branch_id"] == str(branch_id), f"Branch failed: expected {branch_id}, got {data2.get('branch_id')}"
|
|
print("-> SUCCESS: Draft created and Main Branch assigned.")
|
|
else:
|
|
print(f"Failed with {data2}")
|
|
except Exception as e:
|
|
print(f"-> FAILED: {e}")
|
|
|
|
print("\n--- TEST 3: OWNERSHIP TRANSFER TEST ---")
|
|
payload3 = {
|
|
"vin": "TESTVIN1111111111",
|
|
"license_plate": "OTHER-000",
|
|
"brand": "Ford",
|
|
"model": "Focus",
|
|
"organization_id": org_id
|
|
}
|
|
resp3 = await client.post("/api/v1/assets/vehicles", json=payload3, headers=headers2)
|
|
print(f"Status: {resp3.status_code}")
|
|
try:
|
|
data3 = resp3.json()
|
|
if resp3.status_code == 202:
|
|
print("-> SUCCESS: VIN collision detected (transfer_pending).")
|
|
else:
|
|
print(f"Failed with {data3}")
|
|
except Exception as e:
|
|
print(f"-> FAILED: {e}")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(run_tests())
|