94 lines
3.0 KiB
Python
94 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Create a vehicle via external API call (from host).
|
|
"""
|
|
import requests
|
|
import json
|
|
import sys
|
|
|
|
BASE_URL = "http://192.168.100.10:8000"
|
|
|
|
def get_token():
|
|
"""Login with user 28 credentials."""
|
|
# We need to know the password for user 28. Let's try to get token via internal script.
|
|
# Instead, we can use a known token from previous runs.
|
|
# Let's try to fetch token via login endpoint with email/password.
|
|
# Need to find user 28 email. Check database via docker exec.
|
|
# For simplicity, we can use the token generation script inside container.
|
|
# We'll run a docker command to get token.
|
|
import subprocess
|
|
result = subprocess.run(
|
|
["docker", "compose", "exec", "-T", "sf_api", "python3", "-c", """
|
|
import asyncio
|
|
from app.db.session import AsyncSessionLocal
|
|
from app.models.identity import User
|
|
from app.core.security import create_tokens
|
|
from sqlalchemy import select
|
|
import sys
|
|
|
|
async def main():
|
|
async with AsyncSessionLocal() as db:
|
|
result = await db.execute(select(User).where(User.id == 28))
|
|
user = result.scalar_one_or_none()
|
|
if not user:
|
|
print("NOTFOUND")
|
|
return
|
|
token_payload = {
|
|
"sub": str(user.id),
|
|
"role": user.role.value if hasattr(user.role, 'value') else user.role,
|
|
"rank": 10,
|
|
"scope_level": user.scope_level.value if hasattr(user.scope_level, 'value') else (user.scope_level or "individual"),
|
|
"scope_id": str(user.scope_id) if user.scope_id else str(user.id)
|
|
}
|
|
access_token, refresh_token = create_tokens(data=token_payload)
|
|
print(access_token)
|
|
|
|
asyncio.run(main())
|
|
"""],
|
|
capture_output=True,
|
|
text=True
|
|
)
|
|
if result.returncode == 0:
|
|
token = result.stdout.strip()
|
|
if token and token != "NOTFOUND":
|
|
return token
|
|
# fallback: maybe there is a test token in env
|
|
return None
|
|
|
|
def main():
|
|
token = get_token()
|
|
if not token:
|
|
print("Failed to obtain token")
|
|
sys.exit(1)
|
|
print(f"Token: {token[:30]}...")
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
payload = {
|
|
"license_plate": "ABC-123",
|
|
"vin": "1HGCM82633A123456",
|
|
"brand": "Toyota",
|
|
"model": "Corolla",
|
|
"vehicle_class": "Passenger Car",
|
|
"fuel_type": "Petrol",
|
|
"organization_id": 34
|
|
}
|
|
|
|
resp = requests.post(f"{BASE_URL}/api/v1/assets/vehicles", headers=headers, json=payload)
|
|
print(f"Status: {resp.status_code}")
|
|
print(f"Response: {resp.text}")
|
|
if resp.status_code == 201:
|
|
data = resp.json()
|
|
print(f"Vehicle created successfully: ID={data.get('id')}")
|
|
print(f"Status: {data.get('status')}")
|
|
print(f"Data Enrichment Status: {data.get('data_status')}")
|
|
return data
|
|
else:
|
|
print("Failed to create vehicle")
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main() |