johnpccd/image_management_api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

cleanup

Browse Source
This commit is contained in:
johnpccd 2025-05-25 17:49:41 +02:00
parent cff511b7bd
commit 19db1dd3b7
8 changed files with 125 additions and 620 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

71
README.md
View File

@ -1,6 +1,6 @@
# SEREACT - Secure Image Management API
# Image Management API
SEREACT is a secure API for storing, organizing, and retrieving images with advanced search capabilities powered by AI-generated embeddings.
A secure API for storing, organizing, and retrieving images with advanced search capabilities powered by AI-generated embeddings.
## Features
@ -21,7 +21,7 @@ SEREACT is a secure API for storing, organizing, and retrieving images with adva
## Architecture
```
sereact/
root/
├── images/ # Sample images for testing
├── deployment/ # Deployment configurations
│ ├── cloud-function/ # **Cloud Function for image processing**
@ -113,7 +113,7 @@ sereact/
### 4. **Search Flow**:
- Search queries processed by FastAPI backend
- Vector similarity search performed against Qdrant VM
- Vector similarity search performed against Qdrant vector database on a VM
- Results combined with metadata from Firestore
## Technology Stack
@ -157,7 +157,7 @@ The system includes a dedicated Google Compute Engine VM running Qdrant vector d
### **AI Embedding Model**
SEREACT uses Google's Vertex AI multimodal embedding model for generating high-quality image embeddings:
Uses Google's Vertex AI multimodal embedding model for generating high-quality image embeddings:
- **Model**: `multimodalembedding@001`
- **Provider**: Google Vertex AI
@ -181,8 +181,8 @@ SEREACT uses Google's Vertex AI multimodal embedding model for generating high-q
1. Clone the repository:
```bash
git clone https://github.com/yourusername/sereact.git
cd sereact
git clone {repo-url}
cd {repo-name}
```
2. Create and activate a virtual environment:
@ -309,26 +309,6 @@ export QDRANT_API_KEY=your-qdrant-api-key
./deploy.sh
```
### **Vector Database Management**
#### **Accessing the Vector Database**
```bash
# SSH into the VM
gcloud compute ssh sereact-vector-db --zone=us-central1-a
# Check Qdrant status
sudo systemctl status qdrant
# View logs
sudo journalctl -u qdrant -f
# Run health check
sudo /opt/qdrant/health_check.sh
# Manual backup
sudo /opt/qdrant/backup.sh
```
#### **Vector Database API Usage**
@ -363,7 +343,6 @@ The API provides the following main endpoints with their authentication and pagi
### 🔓 **Public Endpoints (No Authentication Required)**
#### Authentication & API Key Management
- `/api/v1/auth/bootstrap` - Initial system setup (creates first team, admin user, and API key)
- `/api/v1/auth/api-keys` (POST) - Create new API key (requires `user_id` and `team_id` parameters)
#### Team Management
@ -414,42 +393,11 @@ The API provides the following main endpoints with their authentication and pagi
### 🔑 **Authentication Model**
SEREACT uses a **hybrid authentication model**:
A **hybrid authentication model**:
1. **Public Management Endpoints**: Users, teams, and API key creation are **publicly accessible** for easy integration and setup
2. **Protected Data Endpoints**: Image storage, search, and API key management require **API key authentication**
3. **Bootstrap Process**: Initial setup via `/auth/bootstrap` creates the first team, admin user, and API key
2. **Protected Data Endpoints**: Image storage and search require **API key authentication**
#### **Typical Workflow**:
```bash
# 1. Bootstrap initial setup (public)
POST /api/v1/auth/bootstrap
{
"team_name": "My Team",
"admin_email": "admin@example.com",
"admin_name": "Admin User"
}
# Returns: API key for subsequent authenticated requests
# 2. Create additional users (public)
POST /api/v1/users
{
"name": "John Doe",
"email": "john@example.com",
"team_id": "team_id_from_bootstrap"
}
# 3. Create API keys for users (public)
POST /api/v1/auth/api-keys?user_id={user_id}&team_id={team_id}
{
"name": "John's API Key",
"description": "For image uploads"
}
# 4. Use API key for protected operations
GET /api/v1/images
Headers: X-API-Key: your_api_key_here
```
### **Authentication & Pagination Status**
@ -675,7 +623,6 @@ This modular architecture provides several benefits:
### Low Priority
- [ ] Terraform dependencies
- [ ] Move all auth logic to auth module
- [ ] Remove bootstrap endpoint
- [ ] Move cloud function code to src folder and reuse code with embedding service
- [ ] Thumbnail generation

5
client/js/api.js
View File

@ -187,11 +187,6 @@ class ApiClient {
return this.makeRequest('POST', '/search', searchData);
}
// Bootstrap API
async bootstrap(bootstrapData) {
return this.makeRequest('POST', '/auth/bootstrap', bootstrapData);
}
// Health check
async healthCheck() {
this.updateConfig();

76
scripts/create_admin.py
View File

@ -1,76 +0,0 @@
import os
import sys
import asyncio
import logging
from bson import ObjectId
# Add the project root to the Python path
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# Import repositories
from src.db.repositories.team_repository import team_repository
from src.db.repositories.user_repository import user_repository
from src.db.repositories.api_key_repository import api_key_repository
# Import models
from src.models.team import TeamModel
from src.models.user import UserModel
from src.models.api_key import ApiKeyModel
# Import security functions
from src.auth.security import generate_api_key, calculate_expiry_date
async def create_admin():
# Create a new team
print("Creating admin team...")
team = TeamModel(
name="Admin Team",
description="Default admin team for system administration"
)
created_team = await team_repository.create(team)
print(f"Created team with ID: {created_team.id}")
# Create admin user
print("Creating admin user...")
user = UserModel(
name="Admin User",
email="admin@example.com",
team_id=created_team.id,
is_admin=True,
is_active=True
)
created_user = await user_repository.create(user)
print(f"Created admin user with ID: {created_user.id}")
# Generate API key
print("Generating API key...")
raw_key, hashed_key = generate_api_key(str(created_team.id), str(created_user.id))
expiry_date = calculate_expiry_date()
# Create API key in database
api_key = ApiKeyModel(
key_hash=hashed_key,
user_id=created_user.id,
team_id=created_team.id,
name="Admin API Key",
description="Initial API key for admin user",
expiry_date=expiry_date,
is_active=True
)
created_key = await api_key_repository.create(api_key)
print(f"Created API key with ID: {created_key.id}")
print(f"API Key (save this, it won't be shown again): {raw_key}")
return {
"team_id": str(created_team.id),
"user_id": str(created_user.id),
"api_key": raw_key
}
if __name__ == "__main__":
print("Creating admin user and API key...")
result = asyncio.run(create_admin())
print("\nSetup complete! Use the API key to authenticate API calls.")

69
scripts/generate_dev_key.py
View File

@ -1,69 +0,0 @@
import os
import sys
import hmac
import hashlib
import secrets
import string
# Add the project root to the Python path
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# Get the API key secret from environment
from src.config.config import settings
def generate_api_key(team_id="dev-team", user_id="dev-admin"):
"""
Generate a secure API key and its hashed value
Args:
team_id: Team ID for which the key is generated
user_id: User ID for which the key is generated
Returns:
Tuple of (raw_api_key, hashed_api_key)
"""
# Generate a random key prefix (visible part)
prefix = ''.join(secrets.choice(string.ascii_letters + string.digits) for _ in range(8))
# Generate a secure random token for the key
random_part = secrets.token_hex(16)
# Format: prefix.random_part
raw_api_key = f"{prefix}.{random_part}"
# Hash the API key for storage
hashed_api_key = hash_api_key(raw_api_key)
return raw_api_key, hashed_api_key
def hash_api_key(api_key: str) -> str:
"""
Create a secure hash of the API key for storage
Args:
api_key: The raw API key
Returns:
Hashed API key
"""
return hmac.new(
settings.API_KEY_SECRET.encode(),
api_key.encode(),
hashlib.sha256
).hexdigest()
if __name__ == "__main__":
# Generate a development API key
api_key, key_hash = generate_api_key()
print("\n====== DEVELOPMENT API KEY ======")
print(f"API Key: {api_key}")
print(f"Key Hash: {key_hash}")
print("\nCOPY THIS API KEY AND USE IT IN YOUR SWAGGER UI!")
print("Header Name: X-API-Key")
print("Header Value: <the API key value above>")
print("===============================")
print("\nNote: This is a generated key, but since there's no database setup,")
print("you won't be able to use it with the API until the key is added to the database.")
print("This would be useful if you developed a bypass_auth mode for development.")
print("For now, please check with the development team for API key access.")

94
scripts/get_test_api_key.py
View File

@ -1,94 +0,0 @@
#!/usr/bin/env python3
"""
Helper script to get an existing API key for testing purposes.
This script connects to the database and retrieves an active API key
that can be used for E2E testing.
Usage:
python scripts/get_test_api_key.py
"""
import asyncio
import sys
import os
# Add the src directory to the path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src'))
from src.db.repositories.api_key_repository import api_key_repository
from src.db.repositories.user_repository import user_repository
from src.db.repositories.team_repository import team_repository
async def get_test_api_key():
"""Get an existing API key for testing"""
try:
# Get all API keys
api_keys = await api_key_repository.get_all()
if not api_keys:
print("❌ No API keys found in the database")
return None
# Find an active API key
active_keys = [key for key in api_keys if key.is_active]
if not active_keys:
print("❌ No active API keys found in the database")
return None
# Get the first active key
test_key = active_keys[0]
# Get user and team info
user = await user_repository.get_by_id(test_key.user_id)
team = await team_repository.get_by_id(test_key.team_id)
print("✅ Found test API key:")
print(f" Key ID: {test_key.id}")
print(f" Key Name: {test_key.name}")
print(f" User: {user.name} ({user.email})" if user else " User: Not found")
print(f" Team: {team.name}" if team else " Team: Not found")
print(f" Created: {test_key.created_at}")
print(f" Is Admin: {user.is_admin}" if user else " Is Admin: Unknown")
# Note: We can't return the actual key value since it's hashed
print("\n⚠️ Note: The actual API key value is hashed in the database.")
print(" You'll need to use an API key you have access to for testing.")
return {
"key_id": str(test_key.id),
"key_name": test_key.name,
"user_id": str(test_key.user_id),
"team_id": str(test_key.team_id),
"user_name": user.name if user else None,
"user_email": user.email if user else None,
"team_name": team.name if team else None,
"is_admin": user.is_admin if user else None
}
except Exception as e:
print(f"❌ Error getting API key: {e}")
return None
async def main():
"""Main function"""
print("🔍 Looking for existing API keys in the database...")
result = await get_test_api_key()
if result:
print("\n💡 To run E2E tests, you can:")
print(" 1. Use an API key you have access to")
print(" 2. Create a new API key using the bootstrap endpoint (if not already done)")
print(" 3. Set the API key in the test environment")
else:
print("\n💡 To run E2E tests, you may need to:")
print(" 1. Run the bootstrap endpoint to create initial data")
print(" 2. Create API keys manually")
if __name__ == "__main__":
asyncio.run(main())

80
scripts/verify_api_key.py
View File

@ -1,80 +0,0 @@
#!/usr/bin/env python3
"""
Script to verify an API key against the database.
"""
import os
import sys
import asyncio
import logging
import argparse
from datetime import datetime
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
# Add the parent directory to the path so we can import from src
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from src.db.providers.firestore_provider import firestore_db
from src.db.repositories.firestore_api_key_repository import firestore_api_key_repository
from src.db.repositories.firestore_user_repository import firestore_user_repository
from src.auth.security import hash_api_key, verify_api_key
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
)
logger = logging.getLogger(__name__)
async def verify_key(api_key):
"""Verify an API key against the database"""
try:
# Connect to Firestore
firestore_db.connect()
# Hash the API key for database lookup
key_hash = hash_api_key(api_key)
logger.info(f"Looking up key with hash: {key_hash}")
# Get all API keys to test manually
all_keys = await firestore_api_key_repository.get_all()
print(f"Found {len(all_keys)} API keys in the database.")
for idx, key in enumerate(all_keys):
print(f"Key {idx+1}: name={key.name}, hash={key.key_hash}")
is_match = verify_api_key(api_key, key.key_hash)
print(f" Match with input key: {is_match}")
if is_match:
print(f" Found matching key: {key.name} (ID: {key.id})")
user = await firestore_user_repository.get_by_id(key.user_id)
if user:
print(f" User: {user.name} (ID: {user.id})")
print(f" User is admin: {user.is_admin}")
print("\nTesting key hash calculation:")
test_hash = hash_api_key(api_key)
print(f"Input key: {api_key}")
print(f"Calculated hash: {test_hash}")
except Exception as e:
logger.error(f"Error verifying API key: {e}")
raise
finally:
# Disconnect from Firestore
firestore_db.disconnect()
def main():
"""Main entry point"""
parser = argparse.ArgumentParser(description="Verify an API key against the database")
parser.add_argument("api_key", help="API key to verify")
args = parser.parse_args()
asyncio.run(verify_key(args.api_key))
if __name__ == "__main__":
main()

94
src/api/v1/auth.py
View File

@ -20,100 +20,6 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Authentication"], prefix="/auth")
@router.post("/bootstrap", response_model=ApiKeyWithValueResponse, status_code=201)
async def bootstrap_initial_setup(
team_name: str,
admin_email: str,
admin_name: str,
api_key_name: str = "Initial API Key",
request: Request = None
):
"""
Bootstrap the initial setup by creating a team, admin user, and API key.
This endpoint does NOT require authentication and should only be used for initial setup.
For security, this endpoint should be disabled in production after initial setup.
"""
# Check if any teams already exist (prevent multiple bootstrap calls)
existing_teams = await team_repository.get_all()
if existing_teams:
raise HTTPException(
status_code=400,
detail="Bootstrap already completed. Teams already exist in the system."
)
# Check if user with email already exists
existing_user = await user_repository.get_by_email(admin_email)
if existing_user:
raise HTTPException(status_code=400, detail="User with this email already exists")
try:
# 1. Create the team
team = TeamModel(
name=team_name,
description=f"Initial team created during bootstrap"
)
created_team = await team_repository.create(team)
# 2. Create the admin user
user = UserModel(
name=admin_name,
email=admin_email,
team_id=created_team.id,
is_admin=True,
is_active=True
)
created_user = await user_repository.create(user)
# 3. Generate API key
raw_key, hashed_key = generate_api_key(str(created_team.id), str(created_user.id))
expiry_date = calculate_expiry_date()
# 4. Create API key in database
api_key = ApiKeyModel(
key_hash=hashed_key,
user_id=created_user.id,
team_id=created_team.id,
name=api_key_name,
description="Initial API key created during bootstrap",
expiry_date=expiry_date,
is_active=True
)
created_key = await api_key_repository.create(api_key)
logger.info(f"Bootstrap completed: Team '{team_name}', Admin '{admin_email}', API key created")
# Return the API key response
response = ApiKeyWithValueResponse(
id=str(created_key.id),
key=raw_key,
name=created_key.name,
description=created_key.description,
team_id=str(created_key.team_id),
user_id=str(created_key.user_id),
created_at=created_key.created_at,
expiry_date=created_key.expiry_date,
last_used=created_key.last_used,
is_active=created_key.is_active
)
return response
except Exception as e:
logger.error(f"Bootstrap failed: {e}")
# Clean up any partially created resources
try:
if 'created_key' in locals():
await api_key_repository.delete(created_key.id)
if 'created_user' in locals():
await user_repository.delete(created_user.id)
if 'created_team' in locals():
await team_repository.delete(created_team.id)
except:
pass # Best effort cleanup
raise HTTPException(status_code=500, detail=f"Bootstrap failed: {str(e)}")
@router.post("/api-keys", response_model=ApiKeyWithValueResponse, status_code=201)
async def create_api_key(key_data: ApiKeyCreate, request: Request, user_id: str, team_id: str):
"""

256
tests/test_e2e.py
View File

@ -2,7 +2,7 @@
End-to-End Tests for SEREACT API
These tests cover the complete user workflows described in the README:
1. Bootstrap initial setup (team, admin user, API key) with artificial data
1. Use pre-seeded API key for authentication
2. Team creation and management
3. User management within teams
4. API key authentication
@ -16,9 +16,9 @@ These tests cover the complete user workflows described in the README:
12. Real database integration
These tests are completely self-contained:
- Create artificial test data at the start
- Use pre-seeded test data (API key, team, admin user)
- Run all tests against this test data
- Clean up all test data at the end
- Clean up created test data at the end
Run with: pytest tests/test_e2e.py -v
For integration tests: pytest tests/test_e2e.py -v -m integration
@ -42,7 +42,7 @@ from main import app
@pytest.mark.e2e
class TestE2EWorkflows:
"""End-to-end tests that simulate real user workflows with artificial data"""
"""End-to-end tests that simulate real user workflows with pre-seeded data"""
@pytest.fixture(scope="class")
def client(self):
@ -51,78 +51,36 @@ class TestE2EWorkflows:
@pytest.fixture(scope="class")
def test_environment(self, client: TestClient):
"""Create a test environment with team, user, and API key"""
"""Set up test environment using pre-seeded API key"""
# Get the pre-seeded API key from environment variable
api_key = os.getenv("E2E_TEST_API_KEY")
if not api_key:
pytest.skip("E2E_TEST_API_KEY environment variable not set. Please provide a pre-seeded API key.")
headers = {"X-API-Key": api_key}
# Verify the API key works and get user/team info
response = client.get("/api/v1/auth/verify", headers=headers)
if response.status_code != 200:
pytest.skip(f"Pre-seeded API key is invalid or expired: {response.text}")
auth_data = response.json()
unique_suffix = str(uuid.uuid4())[:8]
# Create test environment
async def create_test_environment():
# Create team
team_data = {
"name": f"E2E Test Team {unique_suffix}",
"description": f"Team for E2E testing {unique_suffix}"
}
# Create admin user
admin_data = {
"email": f"e2e-admin-{unique_suffix}@test.com",
"name": f"E2E Admin {unique_suffix}",
"is_admin": True
}
# Create API key
api_key_data = {
"name": f"E2E API Key {unique_suffix}",
"description": "API key for E2E testing"
}
return {
"team_data": team_data,
"admin_data": admin_data,
"api_key_data": api_key_data,
"unique_suffix": unique_suffix
}
# Run the async function
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
env_data = loop.run_until_complete(create_test_environment())
finally:
loop.close()
# Bootstrap the environment
bootstrap_data = {
"team_name": env_data["team_data"]["name"],
"admin_email": env_data["admin_data"]["email"],
"admin_name": env_data["admin_data"]["name"],
"api_key_name": env_data["api_key_data"]["name"]
}
response = client.post("/api/v1/auth/bootstrap", params=bootstrap_data)
# Handle case where team/user already exists
if response.status_code == 400:
# Try with more unique identifiers
bootstrap_data["team_name"] = f"E2E_TEST_{unique_suffix}_{int(time.time())}"
bootstrap_data["admin_email"] = f"e2e-{unique_suffix}-{int(time.time())}@test.com"
response = client.post("/api/v1/auth/bootstrap", params=bootstrap_data)
assert response.status_code == 201, f"Bootstrap failed: {response.text}"
result = response.json()
# Store environment data
env_data.update({
"api_key": result["key"],
"team_id": result["team_id"],
"admin_user_id": result["user_id"],
"headers": {"X-API-Key": result["key"]},
env_data = {
"api_key": api_key,
"team_id": auth_data["team_id"],
"admin_user_id": auth_data["user_id"],
"headers": headers,
"unique_suffix": unique_suffix,
"created_resources": {
"teams": [result["team_id"]],
"users": [result["user_id"]],
"api_keys": [result["api_key_id"]],
"teams": [],
"users": [],
"api_keys": [],
"images": []
}
})
}
yield env_data
@ -179,9 +137,9 @@ class TestE2EWorkflows:
images.append(img_bytes)
return images
def test_bootstrap_and_basic_workflow(self, test_environment, client: TestClient):
"""Test the complete bootstrap process and basic API functionality"""
print(f"🧪 Testing bootstrap and basic workflow with environment {test_environment['unique_suffix']}")
def test_api_key_verification_and_basic_workflow(self, test_environment, client: TestClient):
"""Test API key verification and basic API functionality"""
print(f"🧪 Testing API key verification and basic workflow with environment {test_environment['unique_suffix']}")
env = test_environment
headers = env["headers"]
@ -230,7 +188,7 @@ class TestE2EWorkflows:
response = client.get("/api/v1/auth/api-keys", headers=headers)
assert response.status_code == 200
api_keys = response.json()
assert len(api_keys) >= 1 # Should have at least our bootstrap key
assert len(api_keys) >= 1 # Should have at least our test key
print("✅ API key listing successful")
# Test 7: Basic image operations (placeholder test)
@ -240,7 +198,7 @@ class TestE2EWorkflows:
assert "images" in images or "message" in images # Handle both implemented and placeholder responses
print("✅ Image listing endpoint accessible")
print("🎉 Bootstrap and basic workflow test passed!")
print("🎉 API key verification and basic workflow test passed!")
def test_advanced_search_functionality(self, test_environment, client: TestClient):
"""Test search functionality with fallback for missing services"""
@ -300,7 +258,7 @@ class TestE2EWorkflows:
return img_bytes
def test_user_roles_and_permissions(self, test_environment, client: TestClient):
"""Test user roles and permissions with artificial data"""
"""Test user roles and permissions with pre-seeded data"""
env = test_environment
admin_headers = env["headers"]
@ -352,7 +310,7 @@ class TestE2EWorkflows:
print("✅ Image ownership verification successful")
def test_multi_team_isolation(self, client: TestClient, test_environment, sample_image_file):
"""Test that teams are properly isolated from each other with artificial data"""
"""Test that teams are properly isolated from each other with pre-seeded data"""
env = test_environment
admin_headers = env["headers"]
@ -586,7 +544,7 @@ class TestE2EWorkflows:
print("🎉 Image metadata operations test completed!")
def test_error_handling(self, client: TestClient, test_environment):
"""Test error handling scenarios with artificial data"""
"""Test error handling scenarios with pre-seeded data"""
env = test_environment
headers = env["headers"]
@ -621,7 +579,7 @@ class TestE2EWorkflows:
@pytest.mark.integration
@pytest.mark.e2e
class TestE2EIntegrationWorkflows:
"""End-to-end integration tests that require real services with artificial data"""
"""End-to-end integration tests that require real services with pre-seeded data"""
@pytest.fixture(scope="class")
def client(self):
@ -633,44 +591,71 @@ class TestE2EIntegrationWorkflows:
@pytest.fixture(scope="class")
def integration_environment(self, client: TestClient):
"""Create test environment for integration tests"""
"""Create test environment for integration tests using pre-seeded API key"""
# Get the pre-seeded API key from environment variable
api_key = os.getenv("E2E_TEST_API_KEY")
if not api_key:
pytest.skip("E2E_TEST_API_KEY environment variable not set. Please provide a pre-seeded API key.")
headers = {"X-API-Key": api_key}
# Verify the API key works and get user/team info
response = client.get("/api/v1/auth/verify", headers=headers)
if response.status_code != 200:
pytest.skip(f"Pre-seeded API key is invalid or expired: {response.text}")
auth_data = response.json()
unique_suffix = str(uuid.uuid4())[:8]
bootstrap_data = {
"team_name": f"Integration Test Team {unique_suffix}",
"admin_email": f"integration-admin-{unique_suffix}@test.com",
"admin_name": f"Integration Admin {unique_suffix}",
"api_key_name": f"Integration API Key {unique_suffix}"
}
response = client.post("/api/v1/auth/bootstrap", params=bootstrap_data)
if response.status_code == 400:
# Try with more unique identifiers
bootstrap_data["team_name"] = f"INTEGRATION_TEST_{unique_suffix}_{int(time.time())}"
bootstrap_data["admin_email"] = f"integration-{unique_suffix}-{int(time.time())}@test.com"
response = client.post("/api/v1/auth/bootstrap", params=bootstrap_data)
assert response.status_code == 201
result = response.json()
env_data = {
"api_key": result["key"],
"team_id": result["team_id"],
"admin_user_id": result["user_id"],
"headers": {"X-API-Key": result["key"]},
"unique_suffix": unique_suffix
"api_key": api_key,
"team_id": auth_data["team_id"],
"admin_user_id": auth_data["user_id"],
"headers": headers,
"unique_suffix": unique_suffix,
"created_resources": {
"teams": [],
"users": [],
"api_keys": [],
"images": []
}
}
yield env_data
# Cleanup
try:
client.delete(f"/api/v1/teams/{env_data['team_id']}", headers=env_data["headers"])
except:
pass
# Cleanup - delete created resources
headers = env_data["headers"]
# Delete images first
for image_id in env_data["created_resources"]["images"]:
try:
client.delete(f"/api/v1/images/{image_id}", headers=headers)
except:
pass
# Delete API keys
for api_key_id in env_data["created_resources"]["api_keys"]:
try:
client.delete(f"/api/v1/auth/api-keys/{api_key_id}", headers=headers)
except:
pass
# Delete users
for user_id in env_data["created_resources"]["users"]:
try:
client.delete(f"/api/v1/users/{user_id}", headers=headers)
except:
pass
# Delete teams
for team_id in env_data["created_resources"]["teams"]:
try:
client.delete(f"/api/v1/teams/{team_id}", headers=headers)
except:
pass
def test_real_image_processing_workflow(self, client: TestClient, integration_environment):
"""Test the complete image processing workflow with real services and artificial data"""
"""Test the complete image processing workflow with real services and pre-seeded data"""
env = integration_environment
headers = env["headers"]
@ -689,6 +674,7 @@ class TestE2EIntegrationWorkflows:
assert response.status_code == 201
image = response.json()
image_id = image["id"]
env["created_resources"]["images"].append(image_id)
# Wait for processing to complete (in real scenario, this would be async)
time.sleep(5) # Wait for Cloud Function to process
@ -711,7 +697,7 @@ class TestE2EIntegrationWorkflows:
@pytest.mark.realdb
@pytest.mark.e2e
class TestE2ERealDatabaseWorkflows:
"""End-to-end tests that use real database connections with artificial data"""
"""End-to-end tests that use real database connections with pre-seeded data"""
@pytest.fixture(scope="class")
def client(self):
@ -723,31 +709,27 @@ class TestE2ERealDatabaseWorkflows:
@pytest.fixture(scope="class")
def realdb_environment(self, client: TestClient):
"""Create test environment for real database tests"""
"""Create test environment for real database tests using pre-seeded API key"""
# Get the pre-seeded API key from environment variable
api_key = os.getenv("E2E_TEST_API_KEY")
if not api_key:
pytest.skip("E2E_TEST_API_KEY environment variable not set. Please provide a pre-seeded API key.")
headers = {"X-API-Key": api_key}
# Verify the API key works and get user/team info
response = client.get("/api/v1/auth/verify", headers=headers)
if response.status_code != 200:
pytest.skip(f"Pre-seeded API key is invalid or expired: {response.text}")
auth_data = response.json()
unique_suffix = str(uuid.uuid4())[:8]
bootstrap_data = {
"team_name": f"RealDB Test Team {unique_suffix}",
"admin_email": f"realdb-admin-{unique_suffix}@test.com",
"admin_name": f"RealDB Admin {unique_suffix}",
"api_key_name": f"RealDB API Key {unique_suffix}"
}
response = client.post("/api/v1/auth/bootstrap", params=bootstrap_data)
if response.status_code == 400:
# Try with more unique identifiers
bootstrap_data["team_name"] = f"REALDB_TEST_{unique_suffix}_{int(time.time())}"
bootstrap_data["admin_email"] = f"realdb-{unique_suffix}-{int(time.time())}@test.com"
response = client.post("/api/v1/auth/bootstrap", params=bootstrap_data)
assert response.status_code == 201
result = response.json()
env_data = {
"api_key": result["key"],
"team_id": result["team_id"],
"admin_user_id": result["user_id"],
"headers": {"X-API-Key": result["key"]},
"api_key": api_key,
"team_id": auth_data["team_id"],
"admin_user_id": auth_data["user_id"],
"headers": headers,
"unique_suffix": unique_suffix,
"created_images": []
}
@ -763,15 +745,9 @@ class TestE2ERealDatabaseWorkflows:
client.delete(f"/api/v1/images/{image_id}", headers=headers)
except:
pass
# Delete team (this should cascade delete users and API keys)
try:
client.delete(f"/api/v1/teams/{env_data['team_id']}", headers=headers)
except:
pass
def test_database_performance_and_scalability(self, client: TestClient, realdb_environment):
"""Test database performance with bulk operations and artificial data"""
"""Test database performance with bulk operations and pre-seeded data"""
env = realdb_environment
headers = env["headers"]
@ -828,7 +804,7 @@ class TestE2ERealDatabaseWorkflows:
print("🎉 Database performance test completed!")
def test_data_consistency_and_transactions(self, client: TestClient, realdb_environment):
"""Test data consistency and transaction handling with artificial data"""
"""Test data consistency and transaction handling with pre-seeded data"""
env = realdb_environment
headers = env["headers"]