teszt állományok áthelyezése és szelektálása
This commit is contained in:
94
tests/archive/create_vehicle_via_api.py.old
Normal file
94
tests/archive/create_vehicle_via_api.py.old
Normal file
@@ -0,0 +1,94 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user