Files
service-finder/test_draft_vehicle.py
2026-03-26 07:09:44 +00:00

120 lines
4.7 KiB
Python

#!/usr/bin/env python3
"""
Test script to verify the 2-step asset creation flow.
Tests that draft vehicles can be created without VIN.
"""
import asyncio
import sys
import os
# Add backend to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'backend'))
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from app.db.session import get_db
from app.services.asset_service import AssetService
from app.models.vehicle.asset import Asset
from app.core.config import settings
async def test_draft_vehicle_creation():
"""Test creating a draft vehicle without VIN"""
print("🧪 Testing 2-step asset creation flow...")
# Create async engine
engine = create_async_engine(str(settings.SQLALCHEMY_DATABASE_URI))
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
async with async_session() as db:
try:
# Test 1: Create draft vehicle without VIN
print("1. Testing draft vehicle creation (no VIN)...")
draft_vehicle = await AssetService.create_or_claim_vehicle(
db=db,
user_id=1, # Test user ID
org_id=1, # Test org ID
vin=None, # No VIN for draft
license_plate="DRAFT-001",
catalog_id=None
)
print(f" ✅ Draft vehicle created with ID: {draft_vehicle.id}")
print(f" Status: {draft_vehicle.status}")
print(f" VIN: {draft_vehicle.vin}")
if draft_vehicle.status != "draft":
print(f" ❌ Expected status 'draft', got '{draft_vehicle.status}'")
return False
if draft_vehicle.vin is not None:
print(f" ❌ Expected VIN to be None, got '{draft_vehicle.vin}'")
return False
# Test 2: Create active vehicle with VIN
print("\n2. Testing active vehicle creation (with VIN)...")
active_vehicle = await AssetService.create_or_claim_vehicle(
db=db,
user_id=1,
org_id=1,
vin="WBA12345678901234", # Valid VIN
license_plate="ACTIVE-001",
catalog_id=None
)
print(f" ✅ Active vehicle created with ID: {active_vehicle.id}")
print(f" Status: {active_vehicle.status}")
print(f" VIN: {active_vehicle.vin}")
if active_vehicle.status != "active":
print(f" ❌ Expected status 'active', got '{active_vehicle.status}'")
return False
if active_vehicle.vin != "WBA12345678901234":
print(f" ❌ Expected VIN 'WBA12345678901234', got '{active_vehicle.vin}'")
return False
# Test 3: Create draft vehicle with draft=True parameter
print("\n3. Testing draft vehicle with explicit draft=True...")
explicit_draft = await AssetService.create_or_claim_vehicle(
db=db,
user_id=1,
org_id=1,
vin="WBA99999999999999", # Has VIN but draft=True
license_plate="DRAFT-002",
catalog_id=None,
draft=True
)
print(f" ✅ Explicit draft vehicle created with ID: {explicit_draft.id}")
print(f" Status: {explicit_draft.status}")
print(f" VIN: {explicit_draft.vin}")
if explicit_draft.status != "draft":
print(f" ❌ Expected status 'draft', got '{explicit_draft.status}'")
return False
# VIN should still be stored even for draft
if explicit_draft.vin != "WBA99999999999999":
print(f" ❌ Expected VIN 'WBA99999999999999', got '{explicit_draft.vin}'")
return False
print("\n🎉 All tests passed! 2-step asset creation flow is working correctly.")
print(" - Draft vehicles can be created without VIN")
print(" - Draft vehicles have status='draft'")
print(" - Active vehicles have status='active'")
print(" - Explicit draft=True overrides VIN presence")
return True
except Exception as e:
print(f"❌ Test failed with error: {e}")
import traceback
traceback.print_exc()
return False
finally:
await db.commit()
if __name__ == "__main__":
# Run the test
success = asyncio.run(test_draft_vehicle_creation())
sys.exit(0 if success else 1)