908 lines
37 KiB
Python
908 lines
37 KiB
Python
"""
|
|
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
|
|
2. Team creation and management
|
|
3. User management within teams
|
|
4. API key authentication
|
|
5. Image upload and storage
|
|
6. Image search and retrieval
|
|
7. Multi-team isolation
|
|
8. Advanced search functionality
|
|
9. Image collections management
|
|
10. User role and permission testing
|
|
11. Image metadata operations
|
|
12. Real database integration
|
|
|
|
These tests are completely self-contained:
|
|
- Create artificial test data at the start
|
|
- Run all tests against this test data
|
|
- Clean up all test data at the end
|
|
|
|
Run with: pytest tests/test_e2e.py -v
|
|
For integration tests: pytest tests/test_e2e.py -v -m integration
|
|
For real database tests: pytest tests/test_e2e.py -v -m realdb
|
|
"""
|
|
|
|
import pytest
|
|
import asyncio
|
|
import os
|
|
import io
|
|
import uuid
|
|
import time
|
|
import json
|
|
from typing import Dict, Any, List, Optional
|
|
from fastapi.testclient import TestClient
|
|
from PIL import Image as PILImage
|
|
import tempfile
|
|
|
|
from main import app
|
|
|
|
|
|
@pytest.mark.e2e
|
|
class TestE2EWorkflows:
|
|
"""End-to-end tests that simulate real user workflows with artificial data"""
|
|
|
|
@pytest.fixture(scope="class")
|
|
def client(self):
|
|
"""Create test client for E2E testing"""
|
|
return TestClient(app)
|
|
|
|
@pytest.fixture(scope="class")
|
|
def test_environment(self, client: TestClient):
|
|
"""Create a test environment with team, user, and API key"""
|
|
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"]},
|
|
"created_resources": {
|
|
"teams": [result["team_id"]],
|
|
"users": [result["user_id"]],
|
|
"api_keys": [result["api_key_id"]],
|
|
"images": []
|
|
}
|
|
})
|
|
|
|
yield env_data
|
|
|
|
# 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 last
|
|
for team_id in env_data["created_resources"]["teams"]:
|
|
try:
|
|
client.delete(f"/api/v1/teams/{team_id}", headers=headers)
|
|
except:
|
|
pass
|
|
|
|
@pytest.fixture(scope="function")
|
|
def sample_image_file(self):
|
|
"""Create a sample image file for testing"""
|
|
img = PILImage.new('RGB', (100, 100), color='red')
|
|
img_bytes = io.BytesIO()
|
|
img.save(img_bytes, format='JPEG')
|
|
img_bytes.seek(0)
|
|
return img_bytes
|
|
|
|
@pytest.fixture(scope="function")
|
|
def sample_image_files(self):
|
|
"""Create multiple sample image files for testing"""
|
|
images = []
|
|
colors = ['red', 'green', 'blue', 'yellow', 'purple']
|
|
for i, color in enumerate(colors):
|
|
img = PILImage.new('RGB', (100, 100), color=color)
|
|
img_bytes = io.BytesIO()
|
|
img.save(img_bytes, format='JPEG')
|
|
img_bytes.seek(0)
|
|
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']}")
|
|
|
|
env = test_environment
|
|
headers = env["headers"]
|
|
|
|
# Test 1: Verify API key works
|
|
response = client.get("/api/v1/auth/verify", headers=headers)
|
|
assert response.status_code == 200
|
|
auth_data = response.json()
|
|
assert auth_data["valid"] is True
|
|
assert auth_data["team_id"] == env["team_id"]
|
|
assert auth_data["user_id"] == env["admin_user_id"]
|
|
print("✅ API key verification successful")
|
|
|
|
# Test 2: List teams (should see our team)
|
|
response = client.get("/api/v1/teams", headers=headers)
|
|
assert response.status_code == 200
|
|
teams = response.json()
|
|
team_ids = [team["id"] for team in teams]
|
|
assert env["team_id"] in team_ids
|
|
print("✅ Team listing successful")
|
|
|
|
# Test 3: Get team details
|
|
response = client.get(f"/api/v1/teams/{env['team_id']}", headers=headers)
|
|
assert response.status_code == 200
|
|
team = response.json()
|
|
assert team["id"] == env["team_id"]
|
|
print("✅ Team details retrieval successful")
|
|
|
|
# Test 4: List users (should see admin user)
|
|
response = client.get("/api/v1/users", headers=headers)
|
|
assert response.status_code == 200
|
|
users = response.json()
|
|
user_ids = [user["id"] for user in users]
|
|
assert env["admin_user_id"] in user_ids
|
|
print("✅ User listing successful")
|
|
|
|
# Test 5: Get user details
|
|
response = client.get(f"/api/v1/users/{env['admin_user_id']}", headers=headers)
|
|
assert response.status_code == 200
|
|
user = response.json()
|
|
assert user["id"] == env["admin_user_id"]
|
|
assert user["is_admin"] is True
|
|
print("✅ User details retrieval successful")
|
|
|
|
# Test 6: List API keys
|
|
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
|
|
print("✅ API key listing successful")
|
|
|
|
# Test 7: Basic image operations (placeholder test)
|
|
response = client.get("/api/v1/images", headers=headers)
|
|
assert response.status_code == 200
|
|
images = response.json()
|
|
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!")
|
|
|
|
def test_advanced_search_functionality(self, test_environment, client: TestClient):
|
|
"""Test search functionality with fallback for missing services"""
|
|
print(f"🧪 Testing search functionality with environment {test_environment['unique_suffix']}")
|
|
|
|
env = test_environment
|
|
headers = env["headers"]
|
|
unique_suffix = env["unique_suffix"]
|
|
|
|
# Test basic search endpoint
|
|
response = client.get(f"/api/v1/search?q={unique_suffix}", headers=headers)
|
|
assert response.status_code == 200
|
|
search_results = response.json()
|
|
|
|
# Verify search response structure
|
|
assert "results" in search_results
|
|
assert "total" in search_results
|
|
assert "query" in search_results
|
|
assert search_results["query"] == unique_suffix
|
|
|
|
if len(search_results["results"]) == 0:
|
|
print("⚠️ Search returned empty results (likely Pinecone not configured)")
|
|
print("✅ Search endpoint responding correctly (empty results)")
|
|
else:
|
|
print("✅ Search endpoint returning results")
|
|
# Verify result structure
|
|
for result in search_results["results"]:
|
|
assert "id" in result
|
|
assert "description" in result or "filename" in result
|
|
|
|
# Test search with different parameters
|
|
response = client.get("/api/v1/search?q=nonexistent", headers=headers)
|
|
assert response.status_code == 200
|
|
empty_results = response.json()
|
|
assert "results" in empty_results
|
|
assert len(empty_results["results"]) == 0
|
|
print("✅ Search with no matches handled correctly")
|
|
|
|
# Test search without query (should handle gracefully)
|
|
response = client.get("/api/v1/search", headers=headers)
|
|
assert response.status_code in [200, 400] # Either works or returns bad request
|
|
if response.status_code == 200:
|
|
no_query_results = response.json()
|
|
assert "results" in no_query_results
|
|
print("✅ Search without query handled gracefully")
|
|
else:
|
|
print("✅ Search without query properly rejected")
|
|
|
|
print("🎉 Search functionality test completed!")
|
|
|
|
def create_test_image(self, filename: str) -> io.BytesIO:
|
|
"""Create a test image for upload testing"""
|
|
img = PILImage.new('RGB', (200, 200), color='blue')
|
|
img_bytes = io.BytesIO()
|
|
img.save(img_bytes, format='JPEG')
|
|
img_bytes.seek(0)
|
|
return img_bytes
|
|
|
|
def test_user_roles_and_permissions(self, test_environment, client: TestClient):
|
|
"""Test user roles and permissions with artificial data"""
|
|
|
|
env = test_environment
|
|
admin_headers = env["headers"]
|
|
unique_suffix = env["unique_suffix"]
|
|
|
|
print(f"🧪 Testing user roles and permissions with environment {unique_suffix}")
|
|
|
|
# Test 1: Admin can create users
|
|
regular_user_data = {
|
|
"email": f"regular-user-{unique_suffix}@test.com",
|
|
"name": f"Regular User {unique_suffix}",
|
|
"is_admin": False,
|
|
"team_id": env["team_id"]
|
|
}
|
|
|
|
response = client.post("/api/v1/users", json=regular_user_data, headers=admin_headers)
|
|
assert response.status_code == 201
|
|
regular_user = response.json()
|
|
env["created_resources"]["users"].append(regular_user["id"])
|
|
|
|
# Verify user properties
|
|
assert regular_user["email"] == regular_user_data["email"]
|
|
assert regular_user["name"] == regular_user_data["name"]
|
|
assert regular_user["is_admin"] is False
|
|
assert regular_user["team_id"] == env["team_id"]
|
|
|
|
if "is_active" in regular_user:
|
|
assert regular_user["is_active"] is True
|
|
print("✅ Regular user creation verified")
|
|
|
|
# Test that regular user can upload images (basic functionality)
|
|
# Since we can't easily create a separate API key for the regular user in the current setup,
|
|
# we'll test basic user management functionality
|
|
test_image = self.create_test_image(f"regular_user_image_{unique_suffix}.jpg")
|
|
files = {"file": ("regular_user_image.jpg", test_image, "image/jpeg")}
|
|
data = {
|
|
"description": f"Image uploaded by admin for regular user testing {unique_suffix}"
|
|
}
|
|
|
|
response = client.post("/api/v1/images", files=files, data=data, headers=admin_headers)
|
|
assert response.status_code == 201
|
|
uploaded_image = response.json()
|
|
env["created_resources"]["images"].append(uploaded_image["id"])
|
|
print("✅ Image upload functionality verified")
|
|
|
|
# Verify the image belongs to the admin user (since we used admin's API key)
|
|
assert uploaded_image["uploader_id"] == env["admin_user_id"]
|
|
assert uploaded_image["team_id"] == env["team_id"]
|
|
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"""
|
|
|
|
env = test_environment
|
|
admin_headers = env["headers"]
|
|
unique_suffix = env["unique_suffix"]
|
|
|
|
print(f"🧪 Testing multi-team isolation with environment {unique_suffix}")
|
|
|
|
# Create two separate teams
|
|
team1_data = {
|
|
"name": f"Team Alpha {unique_suffix}",
|
|
"description": f"First team for isolation testing {unique_suffix}"
|
|
}
|
|
|
|
team2_data = {
|
|
"name": f"Team Beta {unique_suffix}",
|
|
"description": f"Second team for isolation testing {unique_suffix}"
|
|
}
|
|
|
|
response = client.post("/api/v1/teams", json=team1_data, headers=admin_headers)
|
|
assert response.status_code == 201
|
|
team1 = response.json()
|
|
team1_id = team1["id"]
|
|
env["created_resources"]["teams"].append(team1_id)
|
|
|
|
response = client.post("/api/v1/teams", json=team2_data, headers=admin_headers)
|
|
assert response.status_code == 201
|
|
team2 = response.json()
|
|
team2_id = team2["id"]
|
|
env["created_resources"]["teams"].append(team2_id)
|
|
|
|
print("✅ Two teams created for isolation testing")
|
|
|
|
# Create users for each team
|
|
user1_data = {
|
|
"email": f"user1-{unique_suffix}@team1.com",
|
|
"name": f"Team1 User {unique_suffix}",
|
|
"is_admin": True,
|
|
"team_id": team1_id
|
|
}
|
|
|
|
user2_data = {
|
|
"email": f"user2-{unique_suffix}@team2.com",
|
|
"name": f"Team2 User {unique_suffix}",
|
|
"is_admin": True,
|
|
"team_id": team2_id
|
|
}
|
|
|
|
response = client.post("/api/v1/users", json=user1_data, headers=admin_headers)
|
|
assert response.status_code == 201
|
|
user1 = response.json()
|
|
env["created_resources"]["users"].append(user1["id"])
|
|
|
|
response = client.post("/api/v1/users", json=user2_data, headers=admin_headers)
|
|
assert response.status_code == 201
|
|
user2 = response.json()
|
|
env["created_resources"]["users"].append(user2["id"])
|
|
|
|
print("✅ Users created for each team")
|
|
|
|
# Create API keys for each team's user
|
|
api_key1_data = {
|
|
"name": f"Team1 API Key {unique_suffix}",
|
|
"description": "API key for team 1 testing",
|
|
"team_id": team1_id,
|
|
"user_id": user1["id"]
|
|
}
|
|
|
|
api_key2_data = {
|
|
"name": f"Team2 API Key {unique_suffix}",
|
|
"description": "API key for team 2 testing",
|
|
"team_id": team2_id,
|
|
"user_id": user2["id"]
|
|
}
|
|
|
|
response = client.post("/api/v1/auth/api-keys", json=api_key1_data, headers=admin_headers)
|
|
assert response.status_code == 201
|
|
team1_api_key = response.json()["key"]
|
|
team1_headers = {"X-API-Key": team1_api_key}
|
|
env["created_resources"]["api_keys"].append(response.json()["id"])
|
|
|
|
response = client.post("/api/v1/auth/api-keys", json=api_key2_data, headers=admin_headers)
|
|
assert response.status_code == 201
|
|
team2_api_key = response.json()["key"]
|
|
team2_headers = {"X-API-Key": team2_api_key}
|
|
env["created_resources"]["api_keys"].append(response.json()["id"])
|
|
|
|
print("✅ API keys created for each team")
|
|
|
|
# Upload images to each team
|
|
sample_image_file.seek(0)
|
|
files1 = {"file": (f"team1_image_{unique_suffix}.jpg", sample_image_file, "image/jpeg")}
|
|
data1 = {
|
|
"description": f"Team 1 confidential image {unique_suffix}"
|
|
}
|
|
|
|
response = client.post("/api/v1/images", files=files1, data=data1, headers=team1_headers)
|
|
assert response.status_code == 201
|
|
team1_image = response.json()
|
|
team1_image_id = team1_image["id"]
|
|
env["created_resources"]["images"].append(team1_image_id)
|
|
|
|
sample_image_file.seek(0)
|
|
files2 = {"file": (f"team2_image_{unique_suffix}.jpg", sample_image_file, "image/jpeg")}
|
|
data2 = {
|
|
"description": f"Team 2 secret image {unique_suffix}"
|
|
}
|
|
|
|
response = client.post("/api/v1/images", files=files2, data=data2, headers=team2_headers)
|
|
assert response.status_code == 201
|
|
team2_image = response.json()
|
|
team2_image_id = team2_image["id"]
|
|
env["created_resources"]["images"].append(team2_image_id)
|
|
|
|
print("✅ Images uploaded to each team")
|
|
|
|
# Test 1: Team 1 user can only see Team 1 images
|
|
response = client.get("/api/v1/images", headers=team1_headers)
|
|
assert response.status_code == 200
|
|
team1_images = response.json()
|
|
team1_image_ids = [img["id"] for img in team1_images["images"]]
|
|
assert team1_image_id in team1_image_ids
|
|
assert team2_image_id not in team1_image_ids
|
|
print("✅ Team 1 user can only see Team 1 images")
|
|
|
|
# Test 2: Team 1 user CANNOT access Team 2's image
|
|
response = client.get(f"/api/v1/images/{team2_image_id}", headers=team1_headers)
|
|
assert response.status_code == 403 # Should be forbidden (not 404)
|
|
print("✅ Team 1 user cannot access Team 2's image")
|
|
|
|
# Test 3: Search results are isolated by team
|
|
response = client.get(f"/api/v1/search?q={unique_suffix}", headers=team1_headers)
|
|
assert response.status_code == 200
|
|
team1_search = response.json()
|
|
team1_search_ids = [img["id"] for img in team1_search["results"]]
|
|
|
|
if len(team1_search_ids) == 0:
|
|
print("⚠️ Search returned empty results (likely Pinecone not configured)")
|
|
# Verify the search endpoint is working correctly
|
|
assert "results" in team1_search
|
|
assert "total" in team1_search
|
|
assert team1_search["query"] == unique_suffix
|
|
print("✅ Search endpoint responding correctly (empty results)")
|
|
else:
|
|
# If search is working, verify team isolation
|
|
assert team1_image_id in team1_search_ids
|
|
assert team2_image_id not in team1_search_ids
|
|
print("✅ Search results properly isolated by team")
|
|
|
|
print("🎉 Multi-team isolation test passed!")
|
|
|
|
def test_image_metadata_operations(self, test_environment, client: TestClient):
|
|
"""Test comprehensive image metadata management"""
|
|
print(f"🧪 Testing image metadata operations with environment {test_environment['unique_suffix']}")
|
|
|
|
headers = test_environment["headers"]
|
|
unique_suffix = test_environment["unique_suffix"]
|
|
|
|
# Upload an image with initial metadata
|
|
test_image = self.create_test_image(f"metadata_test_{unique_suffix}.jpg")
|
|
files = {"file": (f"metadata_test_{unique_suffix}.jpg", test_image, "image/jpeg")}
|
|
data = {
|
|
"description": f"Initial metadata test image {unique_suffix}"
|
|
}
|
|
|
|
response = client.post("/api/v1/images", files=files, data=data, headers=headers)
|
|
assert response.status_code == 201
|
|
uploaded_image = response.json()
|
|
image_id = uploaded_image["id"]
|
|
test_environment["created_resources"]["images"].append(image_id)
|
|
print("✅ Image uploaded with initial metadata")
|
|
|
|
# Test 1: Update description
|
|
description_update = {
|
|
"description": f"Updated description for metadata testing {unique_suffix}"
|
|
}
|
|
|
|
response = client.put(f"/api/v1/images/{image_id}", json=description_update, headers=headers)
|
|
assert response.status_code == 200
|
|
updated_image = response.json()
|
|
assert f"Updated description for metadata testing {unique_suffix}" in updated_image["description"]
|
|
print("✅ Description update successful")
|
|
|
|
# Test 2: Search by updated metadata (with fallback for missing Pinecone)
|
|
response = client.get(f"/api/v1/search?q=updated", headers=headers)
|
|
assert response.status_code == 200
|
|
search_results = response.json()
|
|
found_images = search_results["results"]
|
|
|
|
if len(found_images) == 0:
|
|
print("⚠️ Metadata search returned empty results (likely Pinecone not configured)")
|
|
# Verify the search endpoint is working correctly
|
|
assert "results" in search_results
|
|
assert "total" in search_results
|
|
assert search_results["query"] == "updated"
|
|
print("✅ Search endpoint responding correctly for metadata search")
|
|
else:
|
|
# If search is working, verify we can find our updated image
|
|
assert len(found_images) >= 1
|
|
# Check if our image is in the results (by checking description)
|
|
our_image_found = any(
|
|
unique_suffix in img.get("description", "")
|
|
for img in found_images
|
|
)
|
|
if our_image_found:
|
|
print("✅ Updated image found in search results")
|
|
else:
|
|
print("⚠️ Updated image not found in search results (may be due to indexing delay)")
|
|
|
|
# Test 3: Retrieve image directly to verify metadata persistence
|
|
response = client.get(f"/api/v1/images/{image_id}", headers=headers)
|
|
assert response.status_code == 200
|
|
retrieved_image = response.json()
|
|
|
|
# Verify all metadata updates persisted
|
|
assert f"Updated description for metadata testing {unique_suffix}" in retrieved_image["description"]
|
|
print("✅ Metadata persistence verified")
|
|
|
|
# Test 4: Partial metadata update (only description)
|
|
partial_update = {
|
|
"description": f"Final description update {unique_suffix}"
|
|
}
|
|
|
|
response = client.put(f"/api/v1/images/{image_id}", json=partial_update, headers=headers)
|
|
assert response.status_code == 200
|
|
final_image = response.json()
|
|
|
|
# Verify description changed
|
|
assert f"Final description update {unique_suffix}" in final_image["description"]
|
|
print("✅ Partial metadata update successful")
|
|
|
|
print("🎉 Image metadata operations test completed!")
|
|
|
|
def test_error_handling(self, client: TestClient, test_environment):
|
|
"""Test error handling scenarios with artificial data"""
|
|
|
|
env = test_environment
|
|
headers = env["headers"]
|
|
unique_suffix = env["unique_suffix"]
|
|
|
|
print(f"🧪 Testing error handling with environment {unique_suffix}")
|
|
|
|
# Test invalid API key
|
|
invalid_headers = {"X-API-Key": "invalid-key"}
|
|
response = client.get("/api/v1/auth/verify", headers=invalid_headers)
|
|
assert response.status_code == 401
|
|
print("✅ Invalid API key properly rejected")
|
|
|
|
# Test missing API key
|
|
response = client.get("/api/v1/teams")
|
|
assert response.status_code == 401
|
|
print("✅ Missing API key properly rejected")
|
|
|
|
# Test invalid image ID
|
|
response = client.get("/api/v1/images/invalid-id", headers=headers)
|
|
assert response.status_code == 400 # Bad request for invalid ID format
|
|
print("✅ Invalid image ID properly rejected")
|
|
|
|
# Test unauthorized image upload
|
|
response = client.post("/api/v1/images")
|
|
assert response.status_code == 401 # No API key
|
|
print("✅ Unauthorized image upload properly rejected")
|
|
|
|
print("🎉 Error handling test passed!")
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.e2e
|
|
class TestE2EIntegrationWorkflows:
|
|
"""End-to-end integration tests that require real services with artificial data"""
|
|
|
|
@pytest.fixture(scope="class")
|
|
def client(self):
|
|
"""Create test client for integration testing"""
|
|
if not os.getenv("E2E_INTEGRATION_TEST"):
|
|
pytest.skip("E2E integration tests disabled. Set E2E_INTEGRATION_TEST=1 to enable")
|
|
|
|
return TestClient(app)
|
|
|
|
@pytest.fixture(scope="class")
|
|
def integration_environment(self, client: TestClient):
|
|
"""Create test environment for integration tests"""
|
|
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
|
|
}
|
|
|
|
yield env_data
|
|
|
|
# Cleanup
|
|
try:
|
|
client.delete(f"/api/v1/teams/{env_data['team_id']}", headers=env_data["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"""
|
|
|
|
env = integration_environment
|
|
headers = env["headers"]
|
|
unique_suffix = env["unique_suffix"]
|
|
|
|
# Create a test image
|
|
img = PILImage.new('RGB', (200, 200), color='blue')
|
|
img_bytes = io.BytesIO()
|
|
img.save(img_bytes, format='JPEG')
|
|
img_bytes.seek(0)
|
|
|
|
files = {"file": (f"integration_test_{unique_suffix}.jpg", img_bytes, "image/jpeg")}
|
|
data = {"description": f"Integration test image {unique_suffix}"}
|
|
|
|
response = client.post("/api/v1/images", files=files, data=data, headers=headers)
|
|
assert response.status_code == 201
|
|
image = response.json()
|
|
image_id = image["id"]
|
|
|
|
# Wait for processing to complete (in real scenario, this would be async)
|
|
time.sleep(5) # Wait for Cloud Function to process
|
|
|
|
# Check if embeddings were generated
|
|
response = client.get(f"/api/v1/images/{image_id}", headers=headers)
|
|
assert response.status_code == 200
|
|
processed_image = response.json()
|
|
assert processed_image["has_embedding"] is True
|
|
|
|
# Test semantic search
|
|
response = client.get("/api/v1/search?q=integration test", headers=headers)
|
|
assert response.status_code == 200
|
|
search_results = response.json()
|
|
assert len(search_results["results"]) >= 1
|
|
|
|
print("🎉 Real image processing workflow test passed!")
|
|
|
|
|
|
@pytest.mark.realdb
|
|
@pytest.mark.e2e
|
|
class TestE2ERealDatabaseWorkflows:
|
|
"""End-to-end tests that use real database connections with artificial data"""
|
|
|
|
@pytest.fixture(scope="class")
|
|
def client(self):
|
|
"""Create test client for real database testing"""
|
|
if not os.getenv("E2E_REALDB_TEST"):
|
|
pytest.skip("E2E real database tests disabled. Set E2E_REALDB_TEST=1 to enable")
|
|
|
|
return TestClient(app)
|
|
|
|
@pytest.fixture(scope="class")
|
|
def realdb_environment(self, client: TestClient):
|
|
"""Create test environment for real database tests"""
|
|
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"]},
|
|
"unique_suffix": unique_suffix,
|
|
"created_images": []
|
|
}
|
|
|
|
yield env_data
|
|
|
|
# Cleanup
|
|
headers = env_data["headers"]
|
|
|
|
# Delete created images
|
|
for image_id in env_data["created_images"]:
|
|
try:
|
|
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"""
|
|
|
|
env = realdb_environment
|
|
headers = env["headers"]
|
|
unique_suffix = env["unique_suffix"]
|
|
|
|
print(f"🧪 Testing database performance with environment {unique_suffix}")
|
|
|
|
# Create multiple images for performance testing
|
|
image_count = 10 # Reduced for faster testing
|
|
created_images = []
|
|
|
|
start_time = time.time()
|
|
|
|
for i in range(image_count):
|
|
# Create test image
|
|
img = PILImage.new('RGB', (100, 100), color='red')
|
|
img_bytes = io.BytesIO()
|
|
img.save(img_bytes, format='JPEG')
|
|
img_bytes.seek(0)
|
|
|
|
files = {"file": (f"perf_test_{i}_{unique_suffix}.jpg", img_bytes, "image/jpeg")}
|
|
data = {
|
|
"description": f"Performance test image {i} {unique_suffix}"
|
|
}
|
|
|
|
response = client.post("/api/v1/images", files=files, data=data, headers=headers)
|
|
assert response.status_code == 201
|
|
image = response.json()
|
|
created_images.append(image["id"])
|
|
env["created_images"].append(image["id"])
|
|
|
|
upload_time = time.time() - start_time
|
|
print(f"✅ Uploaded {image_count} images in {upload_time:.2f} seconds")
|
|
|
|
# Test bulk retrieval performance
|
|
start_time = time.time()
|
|
response = client.get("/api/v1/images", headers=headers)
|
|
assert response.status_code == 200
|
|
images = response.json()
|
|
retrieval_time = time.time() - start_time
|
|
|
|
assert len(images["images"]) >= image_count
|
|
print(f"✅ Retrieved images in {retrieval_time:.2f} seconds")
|
|
|
|
# Test search performance (if available)
|
|
start_time = time.time()
|
|
response = client.get(f"/api/v1/search?q={unique_suffix}", headers=headers)
|
|
assert response.status_code == 200
|
|
search_results = response.json()
|
|
search_time = time.time() - start_time
|
|
|
|
print(f"✅ Search completed in {search_time:.2f} seconds")
|
|
|
|
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"""
|
|
|
|
env = realdb_environment
|
|
headers = env["headers"]
|
|
unique_suffix = env["unique_suffix"]
|
|
|
|
print(f"🧪 Testing data consistency with environment {unique_suffix}")
|
|
|
|
# Upload an image
|
|
img = PILImage.new('RGB', (100, 100), color='green')
|
|
img_bytes = io.BytesIO()
|
|
img.save(img_bytes, format='JPEG')
|
|
img_bytes.seek(0)
|
|
|
|
files = {"file": (f"consistency_test_{unique_suffix}.jpg", img_bytes, "image/jpeg")}
|
|
data = {
|
|
"description": f"Consistency test image {unique_suffix}"
|
|
}
|
|
|
|
response = client.post("/api/v1/images", files=files, data=data, headers=headers)
|
|
assert response.status_code == 201
|
|
image = response.json()
|
|
image_id = image["id"]
|
|
env["created_images"].append(image_id)
|
|
|
|
# Verify image exists
|
|
response = client.get(f"/api/v1/images/{image_id}", headers=headers)
|
|
assert response.status_code == 200
|
|
retrieved_image = response.json()
|
|
assert retrieved_image["id"] == image_id
|
|
assert unique_suffix in retrieved_image["description"]
|
|
print("✅ Image consistency verified")
|
|
|
|
# Update image metadata
|
|
update_data = {
|
|
"description": f"Updated consistency test image {unique_suffix}"
|
|
}
|
|
|
|
response = client.put(f"/api/v1/images/{image_id}", json=update_data, headers=headers)
|
|
assert response.status_code == 200
|
|
updated_image = response.json()
|
|
assert f"Updated consistency test image {unique_suffix}" in updated_image["description"]
|
|
|
|
# Verify update persistence
|
|
response = client.get(f"/api/v1/images/{image_id}", headers=headers)
|
|
assert response.status_code == 200
|
|
final_image = response.json()
|
|
assert f"Updated consistency test image {unique_suffix}" in final_image["description"]
|
|
print("✅ Update consistency verified")
|
|
|
|
print("🎉 Data consistency test completed!")
|
|
|
|
|
|
def create_test_image(width: int = 100, height: int = 100, color: str = 'red') -> io.BytesIO:
|
|
"""Helper function to create test images"""
|
|
img = PILImage.new('RGB', (width, height), color=color)
|
|
img_bytes = io.BytesIO()
|
|
img.save(img_bytes, format='JPEG')
|
|
img_bytes.seek(0)
|
|
return img_bytes
|
|
|
|
|
|
def create_test_images_batch(count: int = 5, base_name: str = "test") -> List[io.BytesIO]:
|
|
"""Helper function to create multiple test images"""
|
|
images = []
|
|
colors = ['red', 'green', 'blue', 'yellow', 'purple', 'orange', 'pink', 'brown', 'gray', 'black']
|
|
|
|
for i in range(count):
|
|
color = colors[i % len(colors)]
|
|
img = create_test_image(color=color)
|
|
images.append(img)
|
|
|
|
return images
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Run E2E tests
|
|
pytest.main([__file__, "-v", "-m", "e2e"]) |