50 lines
1.4 KiB
Python
50 lines
1.4 KiB
Python
|
"""Test-only endpoints for e2e test isolation."""
|
||
|
|
|
||
|
|
import os
|
||
|
|
|
||
|
|
from fastapi import APIRouter, Depends, HTTPException
|
||
|
|
from sqlalchemy import text
|
||
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
|
|
||
|
|
from database import get_db
|
||
|
|
|
||
|
|
router = APIRouter(prefix="/api/test", tags=["test"])
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/reset")
|
||
|
|
async def reset_database(db: AsyncSession = Depends(get_db)):
|
||
|
|
"""
|
||
|
|
Truncate all tables and re-seed base data.
|
||
|
|
Only available when E2E_MODE environment variable is set.
|
||
|
|
"""
|
||
|
|
# Safety check - only allow in e2e mode
|
||
|
|
if not os.getenv("E2E_MODE"):
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=403, detail="This endpoint is only available in E2E_MODE"
|
||
|
|
)
|
||
|
|
|
||
|
|
# Get all table names from the database
|
||
|
|
result = await db.execute(
|
||
|
|
text("""
|
||
|
|
SELECT tablename
|
||
|
|
FROM pg_tables
|
||
|
|
WHERE schemaname = 'public'
|
||
|
|
AND tablename != 'alembic_version'
|
||
|
|
ORDER BY tablename
|
||
|
|
""")
|
||
|
|
)
|
||
|
|
tables = [row[0] for row in result]
|
||
|
|
|
||
|
|
# Truncate all tables (CASCADE handles foreign keys)
|
||
|
|
if tables:
|
||
|
|
table_list = ", ".join(f'"{table}"' for table in tables)
|
||
|
|
await db.execute(text(f"TRUNCATE TABLE {table_list} CASCADE"))
|
||
|
|
await db.commit()
|
||
|
|
|
||
|
|
# Re-seed essential data
|
||
|
|
from seed_e2e import seed_base_data
|
||
|
|
|
||
|
|
await seed_base_data(db)
|
||
|
|
|
||
|
|
return {"status": "reset", "tables_truncated": len(tables)}
|