#!/usr/bin/env python3
"""
Create a test API key directly in the database for MCP server testing.
This bypasses the email requirement for quick testing.
"""
import asyncio
import secrets
import sys
import os
# Add backend to path to import modules
sys.path.insert(0, '/Users/jerlitaburanday/clipsense/backend')
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import select
from app.db.models import User, ApiKey, SubscriptionTier
async def create_test_api_key():
"""Create a test API key for e2e testing"""
# Get database URL from environment or use default
database_url = os.getenv('DATABASE_URL', 'postgresql+asyncpg://postgres:postgres@localhost:5432/clipsense')
# Railway uses postgres:// which needs to be replaced with postgresql://
if database_url.startswith('postgres://'):
database_url = database_url.replace('postgres://', 'postgresql+asyncpg://', 1)
elif not database_url.startswith('postgresql+asyncpg://'):
database_url = f'postgresql+asyncpg://{database_url}'
print(f"š Connecting to database...")
engine = create_async_engine(database_url, echo=False)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
async with async_session() as db:
test_email = "test-mcp@clipsense.app"
# Find or create test user
print(f"š¤ Finding or creating test user: {test_email}")
result = await db.execute(select(User).where(User.email == test_email))
user = result.scalar_one_or_none()
if not user:
user = User(
id=f"user_{secrets.token_urlsafe(16)}",
email=test_email,
name="MCP Test User",
subscription_tier=SubscriptionTier.FREE,
monthly_analyses_used=0
)
db.add(user)
await db.commit()
await db.refresh(user)
print(f"ā
Created new user: {user.id}")
else:
print(f"ā
Found existing user: {user.id}")
# Check for existing active API key
result = await db.execute(
select(ApiKey).where(
ApiKey.user_id == user.id,
ApiKey.is_active == True
).limit(1)
)
existing_key = result.scalar_one_or_none()
if existing_key:
print(f"\nš Existing API key found!")
print(f" API Key: {existing_key.key}")
print(f" Tier: {existing_key.tier}")
print(f" Analyses used: {existing_key.monthly_analyses_used}/3")
return existing_key.key
# Create new API key
print(f"š Generating new API key...")
api_key_value = f"cs_sk_{secrets.token_urlsafe(32)}"
api_key = ApiKey(
key=api_key_value,
user_id=user.id,
name="MCP E2E Test Key",
tier=user.subscription_tier,
monthly_analyses_used=0,
is_active=True
)
db.add(api_key)
await db.commit()
await db.refresh(api_key)
print(f"\nā
API key created successfully!")
print(f" User ID: {user.id}")
print(f" Email: {test_email}")
print(f" API Key: {api_key_value}")
print(f" Tier: {user.subscription_tier}")
print(f"\nš” Export this key for testing:")
print(f" export CLIPSENSE_API_KEY=\"{api_key_value}\"")
return api_key_value
if __name__ == "__main__":
try:
api_key = asyncio.run(create_test_api_key())
# Write to file for easy sourcing
with open('/tmp/clipsense_test_key.sh', 'w') as f:
f.write(f'export CLIPSENSE_API_KEY="{api_key}"\n')
print(f"\nš Key also saved to: /tmp/clipsense_test_key.sh")
print(f" Source it with: source /tmp/clipsense_test_key.sh")
except Exception as e:
print(f"\nā Error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
