import pytest
import httpx
import asyncio
import os
import random
import string
from typing import Dict, Any, List
# Configuration for test environment
BASE_URL = "http://localhost:8000" # Adjust if needed
TEST_FILE_DIR = "./test_uploads"
# Utility Functions
def generate_random_string(length: int = 10) -> str:
"""Generate a random string for testing."""
return ''.join(random.choices(string.ascii_letters + string.digits, k=length))
def generate_test_note(tags: List[str] = None, attachments: List[Dict] = None) -> Dict[str, Any]:
"""Generate a test note with optional tags and attachments."""
return {
"title": f"Test Note {generate_random_string()}",
"body": f"This is a test note content {generate_random_string(20)}",
"tags": tags or [generate_random_string(5)],
"attachments": attachments or []
}
@pytest.fixture
async def async_client():
"""Create an async HTTP client for testing."""
async with httpx.AsyncClient(base_url=BASE_URL) as client:
yield client
@pytest.fixture
def test_file():
"""Create a test file for upload."""
os.makedirs(TEST_FILE_DIR, exist_ok=True)
test_file_path = os.path.join(TEST_FILE_DIR, f"test_file_{generate_random_string()}.txt")
with open(test_file_path, 'w') as f:
f.write(f"Test file content {generate_random_string(20)}")
yield test_file_path
# Cleanup
os.remove(test_file_path)
@pytest.mark.asyncio
async def test_health_check(async_client):
"""Test the health check endpoint."""
response = await async_client.get("/api/health")
assert response.status_code == 200
assert response.json() == {"status": "ok"}
@pytest.mark.asyncio
async def test_create_note_basic(async_client):
"""Test creating a basic note."""
note_data = generate_test_note()
response = await async_client.post("/api/notes", json=note_data)
assert response.status_code == 200
created_note = response.json()
assert created_note["title"] == note_data["title"]
assert created_note["body"] == note_data["body"]
assert set(created_note["tags"]) == set(note_data["tags"])
assert "id" in created_note
assert "created_at" in created_note
assert "updated_at" in created_note
@pytest.mark.asyncio
async def test_create_note_with_multiple_tags(async_client):
"""Test creating a note with multiple tags."""
note_data = generate_test_note(
tags=["work", "personal", generate_random_string(5)]
)
response = await async_client.post("/api/notes", json=note_data)
assert response.status_code == 200
created_note = response.json()
assert set(created_note["tags"]) == set(note_data["tags"])
@pytest.mark.asyncio
async def test_update_note(async_client):
"""Test updating an existing note."""
# First create a note
initial_note = generate_test_note()
create_response = await async_client.post("/api/notes", json=initial_note)
created_note = create_response.json()
# Now update the note
update_data = {
"title": f"Updated {created_note['title']}",
"body": f"Updated {created_note['body']}",
"tags": ["updated", "test"]
}
update_response = await async_client.put(f"/api/notes/{created_note['id']}", json=update_data)
assert update_response.status_code == 200
updated_note = update_response.json()
assert updated_note["title"] == update_data["title"]
assert updated_note["body"] == update_data["body"]
assert set(updated_note["tags"]) == set(update_data["tags"])
@pytest.mark.asyncio
async def test_file_upload(async_client, test_file):
"""Test file upload functionality."""
with open(test_file, 'rb') as f:
files = {'file': f}
response = await async_client.post("/api/upload", files=files)
assert response.status_code == 200
upload_result = response.json()
assert "url" in upload_result
assert "type" in upload_result
assert upload_result["type"] in ["file", "image"]
assert "/static/" in upload_result["url"]
@pytest.mark.asyncio
async def test_note_with_attachments(async_client, test_file):
"""Test creating a note with file attachments."""
# First upload a file
with open(test_file, 'rb') as f:
files = {'file': f}
upload_response = await async_client.post("/api/upload", files=files)
file_upload = upload_response.json()
# Create note with attachment
note_data = generate_test_note(
attachments=[{
"url": file_upload["url"],
"type": file_upload["type"]
}]
)
response = await async_client.post("/api/notes", json=note_data)
assert response.status_code == 200
created_note = response.json()
assert len(created_note["attachments"]) == 1
assert created_note["attachments"][0]["url"] == file_upload["url"]
@pytest.mark.asyncio
async def test_list_notes_with_tag_filter(async_client):
"""Test listing notes with tag filtering."""
unique_tag = generate_random_string(5)
# Create multiple notes with the same tag
for _ in range(3):
note_data = generate_test_note(tags=[unique_tag])
await async_client.post("/api/notes", json=note_data)
# List notes with the tag
response = await async_client.get(f"/api/notes?tag={unique_tag}")
assert response.status_code == 200
notes = response.json()
assert len(notes) >= 3
assert all(unique_tag in note["tags"] for note in notes)
@pytest.mark.asyncio
async def test_list_tags(async_client):
"""Test listing all tags."""
# Create some notes with unique tags
unique_tags = [generate_random_string(5) for _ in range(5)]
for tag in unique_tags:
note_data = generate_test_note(tags=[tag])
await async_client.post("/api/notes", json=note_data)
# List tags
response = await async_client.get("/api/tags")
assert response.status_code == 200
tags = response.json()
# Check that all unique tags are present
assert all({"name": tag} in tags for tag in unique_tags)
@pytest.mark.asyncio
async def test_concurrent_note_creation(async_client):
"""Test concurrent note creation to check for race conditions."""
async def create_note():
note_data = generate_test_note()
return await async_client.post("/api/notes", json=note_data)
# Create 10 notes concurrently
responses = await asyncio.gather(*[create_note() for _ in range(10)])
# Check all notes were created successfully
assert all(response.status_code == 200 for response in responses)
# Ensure unique notes
note_ids = [response.json()["id"] for response in responses]
assert len(set(note_ids)) == 10
# Performance and Stress Tests
@pytest.mark.asyncio
async def test_create_many_notes(async_client):
"""Stress test: Create a large number of notes."""
async def create_note():
note_data = generate_test_note()
return await async_client.post("/api/notes", json=note_data)
# Create 100 notes
responses = await asyncio.gather(*[create_note() for _ in range(100)])
assert all(response.status_code == 200 for response in responses)
# Edge Cases and Error Handling
@pytest.mark.asyncio
async def test_update_nonexistent_note(async_client):
"""Test updating a non-existent note."""
non_existent_id = 999999 # Assuming this ID doesn't exist
update_data = generate_test_note()
response = await async_client.put(f"/api/notes/{non_existent_id}", json=update_data)
assert response.status_code == 404
# Bonus: Randomized Testing
@pytest.mark.asyncio
async def test_random_note_operations(async_client):
"""Perform a series of randomized note operations."""
operations = []
# Create some initial notes
for _ in range(5):
note_data = generate_test_note()
response = await async_client.post("/api/notes", json=note_data)
operations.append(("create", response.json()))
# Perform random updates and deletions (simulated)
for _ in range(10):
if operations and random.random() < 0.5:
# Randomly select a note to update
note = random.choice(operations)
if note[0] == "create":
update_data = generate_test_note()
response = await async_client.put(f"/api/notes/{note[1]['id']}", json=update_data)
assert response.status_code == 200
# Configuration for running tests
if __name__ == "__main__":
pytest.main([__file__, "-v", "-s"])
Reference in New Issue View Git Blame Copy Permalink