620 lines
18 KiB
Python
620 lines
18 KiB
Python
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from bson import ObjectId
|
|
from datetime import datetime
|
|
|
|
from src.models.team import TeamModel
|
|
from src.models.user import UserModel
|
|
from src.models.image import ImageModel
|
|
|
|
|
|
class CollectionModel:
|
|
"""Mock collection model for testing"""
|
|
def __init__(self, **kwargs):
|
|
self.id = kwargs.get('id', ObjectId())
|
|
self.name = kwargs.get('name')
|
|
self.description = kwargs.get('description')
|
|
self.team_id = kwargs.get('team_id')
|
|
self.created_by = kwargs.get('created_by')
|
|
self.created_at = kwargs.get('created_at', datetime.utcnow())
|
|
self.updated_at = kwargs.get('updated_at', datetime.utcnow())
|
|
self.metadata = kwargs.get('metadata', {})
|
|
self.image_count = kwargs.get('image_count', 0)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_collection(client: TestClient, admin_api_key: tuple, test_team: TeamModel):
|
|
"""Test creating a new image collection"""
|
|
raw_key, _ = admin_api_key
|
|
|
|
# Set up the headers with the admin API key
|
|
headers = {"X-API-Key": raw_key}
|
|
|
|
# Create a new collection
|
|
response = client.post(
|
|
"/api/v1/collections",
|
|
headers=headers,
|
|
json={
|
|
"name": "Test Collection",
|
|
"description": "A collection for testing images",
|
|
"metadata": {
|
|
"category": "test",
|
|
"project": "contoso"
|
|
}
|
|
}
|
|
)
|
|
|
|
# Check response
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert "id" in data
|
|
assert data["name"] == "Test Collection"
|
|
assert data["description"] == "A collection for testing images"
|
|
assert data["team_id"] == str(test_team.id)
|
|
assert "created_at" in data
|
|
assert data["image_count"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_collection_non_admin(client: TestClient, user_api_key: tuple):
|
|
"""Test that regular users can create collections"""
|
|
raw_key, _ = user_api_key
|
|
|
|
# Set up the headers with a regular user API key
|
|
headers = {"X-API-Key": raw_key}
|
|
|
|
# Create a new collection
|
|
response = client.post(
|
|
"/api/v1/collections",
|
|
headers=headers,
|
|
json={
|
|
"name": "User Collection",
|
|
"description": "A collection created by a regular user"
|
|
}
|
|
)
|
|
|
|
# Check response - regular users should be able to create collections
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["name"] == "User Collection"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_collections(client: TestClient, admin_api_key: tuple, test_team: TeamModel):
|
|
"""Test listing collections for a team"""
|
|
raw_key, _ = admin_api_key
|
|
|
|
# Set up the headers
|
|
headers = {"X-API-Key": raw_key}
|
|
|
|
# List collections
|
|
response = client.get(
|
|
"/api/v1/collections",
|
|
headers=headers
|
|
)
|
|
|
|
# Check response
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "collections" in data
|
|
assert "total" in data
|
|
assert isinstance(data["collections"], list)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_collection_by_id(client: TestClient, admin_api_key: tuple):
|
|
"""Test getting a specific collection by ID"""
|
|
raw_key, _ = admin_api_key
|
|
|
|
# First create a collection
|
|
headers = {"X-API-Key": raw_key}
|
|
|
|
create_response = client.post(
|
|
"/api/v1/collections",
|
|
headers=headers,
|
|
json={
|
|
"name": "Get Test Collection",
|
|
"description": "Collection for get testing"
|
|
}
|
|
)
|
|
|
|
assert create_response.status_code == 201
|
|
collection_id = create_response.json()["id"]
|
|
|
|
# Get the collection
|
|
response = client.get(
|
|
f"/api/v1/collections/{collection_id}",
|
|
headers=headers
|
|
)
|
|
|
|
# Check response
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["id"] == collection_id
|
|
assert data["name"] == "Get Test Collection"
|
|
assert data["description"] == "Collection for get testing"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_collection(client: TestClient, admin_api_key: tuple):
|
|
"""Test updating a collection"""
|
|
raw_key, _ = admin_api_key
|
|
|
|
# First create a collection
|
|
headers = {"X-API-Key": raw_key}
|
|
|
|
create_response = client.post(
|
|
"/api/v1/collections",
|
|
headers=headers,
|
|
json={
|
|
"name": "Update Test Collection",
|
|
"description": "Collection for update testing"
|
|
}
|
|
)
|
|
|
|
assert create_response.status_code == 201
|
|
collection_id = create_response.json()["id"]
|
|
|
|
# Update the collection
|
|
response = client.put(
|
|
f"/api/v1/collections/{collection_id}",
|
|
headers=headers,
|
|
json={
|
|
"name": "Updated Collection Name",
|
|
"description": "This collection has been updated",
|
|
"metadata": {
|
|
"updated": True,
|
|
"version": 2
|
|
}
|
|
}
|
|
)
|
|
|
|
# Check response
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["id"] == collection_id
|
|
assert data["name"] == "Updated Collection Name"
|
|
assert data["description"] == "This collection has been updated"
|
|
assert data["metadata"]["updated"] is True
|
|
assert "updated_at" in data
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_collection(client: TestClient, admin_api_key: tuple):
|
|
"""Test deleting a collection"""
|
|
raw_key, _ = admin_api_key
|
|
|
|
# First create a collection
|
|
headers = {"X-API-Key": raw_key}
|
|
|
|
create_response = client.post(
|
|
"/api/v1/collections",
|
|
headers=headers,
|
|
json={
|
|
"name": "Delete Test Collection",
|
|
"description": "Collection for delete testing"
|
|
}
|
|
)
|
|
|
|
assert create_response.status_code == 201
|
|
collection_id = create_response.json()["id"]
|
|
|
|
# Delete the collection
|
|
response = client.delete(
|
|
f"/api/v1/collections/{collection_id}",
|
|
headers=headers
|
|
)
|
|
|
|
# Check response
|
|
assert response.status_code == 204
|
|
|
|
# Verify collection is deleted
|
|
get_response = client.get(
|
|
f"/api/v1/collections/{collection_id}",
|
|
headers=headers
|
|
)
|
|
assert get_response.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_image_to_collection(client: TestClient, admin_api_key: tuple):
|
|
"""Test adding an image to a collection"""
|
|
raw_key, _ = admin_api_key
|
|
headers = {"X-API-Key": raw_key}
|
|
|
|
# Create a collection
|
|
collection_response = client.post(
|
|
"/api/v1/collections",
|
|
headers=headers,
|
|
json={
|
|
"name": "Image Collection",
|
|
"description": "Collection for image testing"
|
|
}
|
|
)
|
|
collection_id = collection_response.json()["id"]
|
|
|
|
# Mock image ID (in real implementation, this would be from uploaded image)
|
|
image_id = str(ObjectId())
|
|
|
|
# Add image to collection
|
|
response = client.post(
|
|
f"/api/v1/collections/{collection_id}/images",
|
|
headers=headers,
|
|
json={
|
|
"image_id": image_id
|
|
}
|
|
)
|
|
|
|
# Check response
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "success" in data
|
|
assert data["success"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remove_image_from_collection(client: TestClient, admin_api_key: tuple):
|
|
"""Test removing an image from a collection"""
|
|
raw_key, _ = admin_api_key
|
|
headers = {"X-API-Key": raw_key}
|
|
|
|
# Create a collection
|
|
collection_response = client.post(
|
|
"/api/v1/collections",
|
|
headers=headers,
|
|
json={
|
|
"name": "Remove Image Collection",
|
|
"description": "Collection for image removal testing"
|
|
}
|
|
)
|
|
collection_id = collection_response.json()["id"]
|
|
|
|
# Mock image ID
|
|
image_id = str(ObjectId())
|
|
|
|
# Add image to collection first
|
|
add_response = client.post(
|
|
f"/api/v1/collections/{collection_id}/images",
|
|
headers=headers,
|
|
json={"image_id": image_id}
|
|
)
|
|
assert add_response.status_code == 200
|
|
|
|
# Remove image from collection
|
|
response = client.delete(
|
|
f"/api/v1/collections/{collection_id}/images/{image_id}",
|
|
headers=headers
|
|
)
|
|
|
|
# Check response
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_images_in_collection(client: TestClient, admin_api_key: tuple):
|
|
"""Test listing images in a collection"""
|
|
raw_key, _ = admin_api_key
|
|
headers = {"X-API-Key": raw_key}
|
|
|
|
# Create a collection
|
|
collection_response = client.post(
|
|
"/api/v1/collections",
|
|
headers=headers,
|
|
json={
|
|
"name": "List Images Collection",
|
|
"description": "Collection for listing images"
|
|
}
|
|
)
|
|
collection_id = collection_response.json()["id"]
|
|
|
|
# List images in collection
|
|
response = client.get(
|
|
f"/api/v1/collections/{collection_id}/images",
|
|
headers=headers
|
|
)
|
|
|
|
# Check response
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "images" in data
|
|
assert "total" in data
|
|
assert isinstance(data["images"], list)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collection_access_control(client: TestClient, user_api_key: tuple):
|
|
"""Test that users can only access collections from their team"""
|
|
raw_key, _ = user_api_key
|
|
headers = {"X-API-Key": raw_key}
|
|
|
|
# Try to access a collection from another team
|
|
other_collection_id = str(ObjectId())
|
|
|
|
response = client.get(
|
|
f"/api/v1/collections/{other_collection_id}",
|
|
headers=headers
|
|
)
|
|
|
|
# Should return 403 or 404 depending on implementation
|
|
assert response.status_code in [403, 404]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collection_pagination(client: TestClient, admin_api_key: tuple):
|
|
"""Test collection listing with pagination"""
|
|
raw_key, _ = admin_api_key
|
|
headers = {"X-API-Key": raw_key}
|
|
|
|
# Test pagination
|
|
response = client.get(
|
|
"/api/v1/collections?page=1&limit=10",
|
|
headers=headers
|
|
)
|
|
|
|
# Check response
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "collections" in data
|
|
assert "pagination" in data
|
|
assert "total" in data["pagination"]
|
|
assert "page" in data["pagination"]
|
|
assert "pages" in data["pagination"]
|
|
assert len(data["collections"]) <= 10
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collection_search(client: TestClient, admin_api_key: tuple):
|
|
"""Test searching collections by name or description"""
|
|
raw_key, _ = admin_api_key
|
|
headers = {"X-API-Key": raw_key}
|
|
|
|
# Create a collection with searchable content
|
|
create_response = client.post(
|
|
"/api/v1/collections",
|
|
headers=headers,
|
|
json={
|
|
"name": "Searchable Collection",
|
|
"description": "This collection contains vacation photos"
|
|
}
|
|
)
|
|
assert create_response.status_code == 201
|
|
|
|
# Search for collections
|
|
response = client.get(
|
|
"/api/v1/collections/search?query=vacation",
|
|
headers=headers
|
|
)
|
|
|
|
# Check response
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "collections" in data
|
|
assert "total" in data
|
|
|
|
# Results should contain collections matching the query
|
|
for collection in data["collections"]:
|
|
assert ("vacation" in collection["name"].lower() or
|
|
"vacation" in collection["description"].lower())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collection_statistics(client: TestClient, admin_api_key: tuple):
|
|
"""Test getting collection statistics"""
|
|
raw_key, _ = admin_api_key
|
|
headers = {"X-API-Key": raw_key}
|
|
|
|
# Create a collection
|
|
collection_response = client.post(
|
|
"/api/v1/collections",
|
|
headers=headers,
|
|
json={
|
|
"name": "Stats Collection",
|
|
"description": "Collection for statistics testing"
|
|
}
|
|
)
|
|
collection_id = collection_response.json()["id"]
|
|
|
|
# Get collection statistics
|
|
response = client.get(
|
|
f"/api/v1/collections/{collection_id}/stats",
|
|
headers=headers
|
|
)
|
|
|
|
# Check response
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "image_count" in data
|
|
assert "total_size" in data
|
|
assert "created_at" in data
|
|
assert "last_updated" in data
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_add_images_to_collection(client: TestClient, admin_api_key: tuple):
|
|
"""Test adding multiple images to a collection at once"""
|
|
raw_key, _ = admin_api_key
|
|
headers = {"X-API-Key": raw_key}
|
|
|
|
# Create a collection
|
|
collection_response = client.post(
|
|
"/api/v1/collections",
|
|
headers=headers,
|
|
json={
|
|
"name": "Bulk Add Collection",
|
|
"description": "Collection for bulk operations"
|
|
}
|
|
)
|
|
collection_id = collection_response.json()["id"]
|
|
|
|
# Mock multiple image IDs
|
|
image_ids = [str(ObjectId()) for _ in range(5)]
|
|
|
|
# Bulk add images to collection
|
|
response = client.post(
|
|
f"/api/v1/collections/{collection_id}/images/bulk",
|
|
headers=headers,
|
|
json={
|
|
"image_ids": image_ids
|
|
}
|
|
)
|
|
|
|
# Check response
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "added_count" in data
|
|
assert data["added_count"] == 5
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collection_export(client: TestClient, admin_api_key: tuple):
|
|
"""Test exporting collection metadata"""
|
|
raw_key, _ = admin_api_key
|
|
headers = {"X-API-Key": raw_key}
|
|
|
|
# Create a collection
|
|
collection_response = client.post(
|
|
"/api/v1/collections",
|
|
headers=headers,
|
|
json={
|
|
"name": "Export Collection",
|
|
"description": "Collection for export testing",
|
|
"metadata": {"category": "test", "project": "contoso"}
|
|
}
|
|
)
|
|
collection_id = collection_response.json()["id"]
|
|
|
|
# Export collection
|
|
response = client.get(
|
|
f"/api/v1/collections/{collection_id}/export",
|
|
headers=headers
|
|
)
|
|
|
|
# Check response
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "collection" in data
|
|
assert "images" in data
|
|
assert data["collection"]["name"] == "Export Collection"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collection_duplicate_name_validation(client: TestClient, admin_api_key: tuple):
|
|
"""Test that duplicate collection names within a team are handled"""
|
|
raw_key, _ = admin_api_key
|
|
headers = {"X-API-Key": raw_key}
|
|
|
|
# Create first collection
|
|
response1 = client.post(
|
|
"/api/v1/collections",
|
|
headers=headers,
|
|
json={
|
|
"name": "Duplicate Name Collection",
|
|
"description": "First collection with this name"
|
|
}
|
|
)
|
|
assert response1.status_code == 201
|
|
|
|
# Try to create second collection with same name
|
|
response2 = client.post(
|
|
"/api/v1/collections",
|
|
headers=headers,
|
|
json={
|
|
"name": "Duplicate Name Collection",
|
|
"description": "Second collection with same name"
|
|
}
|
|
)
|
|
|
|
# Should either allow it or reject it depending on business rules
|
|
# For now, let's assume it's allowed but with different handling
|
|
assert response2.status_code in [201, 400]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collection_metadata_validation(client: TestClient, admin_api_key: tuple):
|
|
"""Test validation of collection metadata"""
|
|
raw_key, _ = admin_api_key
|
|
headers = {"X-API-Key": raw_key}
|
|
|
|
# Test with invalid metadata structure
|
|
response = client.post(
|
|
"/api/v1/collections",
|
|
headers=headers,
|
|
json={
|
|
"name": "Invalid Metadata Collection",
|
|
"description": "Collection with invalid metadata",
|
|
"metadata": "invalid_metadata_type" # Should be dict
|
|
}
|
|
)
|
|
|
|
# Should validate metadata structure
|
|
assert response.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collection_ownership_transfer(client: TestClient, admin_api_key: tuple):
|
|
"""Test transferring collection ownership (admin only)"""
|
|
raw_key, _ = admin_api_key
|
|
headers = {"X-API-Key": raw_key}
|
|
|
|
# Create a collection
|
|
collection_response = client.post(
|
|
"/api/v1/collections",
|
|
headers=headers,
|
|
json={
|
|
"name": "Transfer Collection",
|
|
"description": "Collection for ownership transfer"
|
|
}
|
|
)
|
|
collection_id = collection_response.json()["id"]
|
|
|
|
# Transfer ownership to another user
|
|
new_owner_id = str(ObjectId())
|
|
response = client.put(
|
|
f"/api/v1/collections/{collection_id}/owner",
|
|
headers=headers,
|
|
json={
|
|
"new_owner_id": new_owner_id
|
|
}
|
|
)
|
|
|
|
# Check response
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["created_by"] == new_owner_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_collection_id(client: TestClient, admin_api_key: tuple):
|
|
"""Test handling of invalid collection IDs"""
|
|
raw_key, _ = admin_api_key
|
|
headers = {"X-API-Key": raw_key}
|
|
|
|
# Try to get collection with invalid ID
|
|
response = client.get(
|
|
"/api/v1/collections/invalid-id",
|
|
headers=headers
|
|
)
|
|
|
|
# Check that the request returns 400 Bad Request
|
|
assert response.status_code == 400
|
|
assert "detail" in response.json()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nonexistent_collection(client: TestClient, admin_api_key: tuple):
|
|
"""Test handling of nonexistent collection IDs"""
|
|
raw_key, _ = admin_api_key
|
|
headers = {"X-API-Key": raw_key}
|
|
|
|
# Try to get nonexistent collection
|
|
nonexistent_id = str(ObjectId())
|
|
response = client.get(
|
|
f"/api/v1/collections/{nonexistent_id}",
|
|
headers=headers
|
|
)
|
|
|
|
# Check that the request returns 404 Not Found
|
|
assert response.status_code == 404
|
|
assert "detail" in response.json() |