138 lines
5.2 KiB
Python
138 lines
5.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Create integration_session.json with test identity credentials.
|
|
Run with: docker compose exec sf_api python /app/create_integration_session.py
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
import json
|
|
import os
|
|
from datetime import datetime, timezone
|
|
import uuid
|
|
|
|
sys.path.insert(0, '/app')
|
|
|
|
async def main():
|
|
from app.db.session import AsyncSessionLocal
|
|
from app.models.identity import User
|
|
from app.models.marketplace.organization import OrganizationMember
|
|
from app.models import Asset, VehicleModelDefinition
|
|
from app.services.auth_service import AuthService
|
|
from app.core.security import create_tokens, get_password_hash
|
|
from app.core.config import settings
|
|
from sqlalchemy import select
|
|
|
|
TEST_EMAIL = "tester_pro@profibot.hu"
|
|
TEST_PASSWORD = "TestPassword123!"
|
|
|
|
async with AsyncSessionLocal() as db:
|
|
# Get the admin user
|
|
result = await db.execute(select(User).where(User.email == TEST_EMAIL))
|
|
user = result.scalar_one_or_none()
|
|
if not user:
|
|
print(f"User {TEST_EMAIL} not found, creating...")
|
|
# We would need to create user, but skip for now
|
|
print("Cannot proceed")
|
|
return
|
|
|
|
print(f"Found user: {user.email}, ID: {user.id}, Role: {user.role}")
|
|
|
|
# Ensure password is set
|
|
if not user.hashed_password or not await AuthService.authenticate(db, TEST_EMAIL, TEST_PASSWORD):
|
|
user.hashed_password = get_password_hash(TEST_PASSWORD)
|
|
await db.commit()
|
|
print("Password updated")
|
|
|
|
# Generate token
|
|
auth_user = await AuthService.authenticate(db, TEST_EMAIL, TEST_PASSWORD)
|
|
if not auth_user:
|
|
print("Authentication failed after password update")
|
|
return
|
|
|
|
ranks = await settings.get_db_setting(db, "rbac_rank_matrix", default={})
|
|
role_key = auth_user.role.value.upper()
|
|
token_payload = {
|
|
"sub": str(auth_user.id),
|
|
"role": auth_user.role.value,
|
|
"rank": ranks.get(role_key, 10),
|
|
"scope_level": auth_user.scope_level or "individual",
|
|
"scope_id": str(auth_user.scope_id) if auth_user.scope_id else str(auth_user.id)
|
|
}
|
|
access_token, refresh_token = create_tokens(data=token_payload)
|
|
print(f"Token generated: {access_token[:50]}...")
|
|
|
|
# Get organization ID if any
|
|
result = await db.execute(
|
|
select(OrganizationMember.organization_id)
|
|
.where(OrganizationMember.user_id == user.id)
|
|
.limit(1)
|
|
)
|
|
org_member = result.scalar_one_or_none()
|
|
org_id = org_member.organization_id if org_member else None
|
|
|
|
# Get a test vehicle ID
|
|
result = await db.execute(
|
|
select(Asset.id)
|
|
.where(Asset.owner_user_id == user.id)
|
|
.limit(1)
|
|
)
|
|
vehicle = result.scalar_one_or_none()
|
|
vehicle_id = vehicle.id if vehicle else None
|
|
|
|
# If no vehicle, create one
|
|
if not vehicle_id:
|
|
result = await db.execute(select(VehicleModelDefinition.id).limit(1))
|
|
catalog_id = result.scalar_one_or_none()
|
|
if catalog_id:
|
|
vehicle = Asset(
|
|
catalog_id=catalog_id,
|
|
license_plate=f"TEST-{uuid.uuid4().hex[:4]}".upper(),
|
|
vin=f"VIN{uuid.uuid4().hex[:10]}".upper(),
|
|
nickname="Integration Test Vehicle",
|
|
owner_user_id=user.id,
|
|
status="DRAFT",
|
|
created_at=datetime.now(timezone.utc)
|
|
)
|
|
db.add(vehicle)
|
|
await db.commit()
|
|
await db.refresh(vehicle)
|
|
vehicle_id = vehicle.id
|
|
print(f"Created test vehicle with ID {vehicle_id}")
|
|
else:
|
|
print("No catalog entries found, skipping vehicle creation")
|
|
|
|
# Prepare session data
|
|
session_data = {
|
|
"email": TEST_EMAIL,
|
|
"password": TEST_PASSWORD,
|
|
"test_token": access_token,
|
|
"user_id": user.id,
|
|
"role": user.role.value,
|
|
"organization_id": org_id,
|
|
"test_vehicle_id": vehicle_id
|
|
}
|
|
|
|
# Write to file
|
|
output_path = "/opt/docker/dev/service_finder/tests/integration_session.json"
|
|
os.makedirs(os.path.dirname(output_path), exist_ok=True)
|
|
with open(output_path, 'w') as f:
|
|
json.dump(session_data, f, indent=2)
|
|
|
|
print("\n" + "="*60)
|
|
print("TEST IDENTITY SETUP COMPLETE")
|
|
print("="*60)
|
|
print(f"Email: {session_data['email']}")
|
|
print(f"Password: {session_data['password']}")
|
|
print(f"Token: {session_data['test_token'][:50]}...")
|
|
print(f"User ID: {session_data['user_id']}")
|
|
print(f"Role: {session_data['role']}")
|
|
print(f"Organization ID: {session_data['organization_id']}")
|
|
print(f"Test Vehicle ID: {session_data['test_vehicle_id']}")
|
|
print(f"Session saved to: {output_path}")
|
|
print("="*60)
|
|
|
|
return session_data
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main()) |