From 96be1c8020ae78c4abf993e1f57283776e80e8fa Mon Sep 17 00:00:00 2001 From: Claude Code Date: Fri, 19 Dec 2025 20:29:51 +0000 Subject: [PATCH 1/3] Add async programming patterns and project structure documentation - Added POST /api/v1/users/bulk endpoint using asyncio.gather() for concurrent user creation - Created py_tests/test_async_example.py with async test patterns and fixtures - Added pytest asyncio configuration (asyncio_mode = auto) - Added comprehensive project structure section to README - Documented Rust tests location and purpose - Updated Tier 1 requirements to include async programming (1 test) Candidates should follow the async patterns shown in test_async_example.py to implement their bulk user creation tests. --- README.md | 58 ++++++++++++++++++++++-- app/main.py | 73 ++++++++++++++++++++++++++++++ py_tests/test_async_example.py | 82 ++++++++++++++++++++++++++++++++++ pytest.ini | 4 ++ 4 files changed, 214 insertions(+), 3 deletions(-) create mode 100644 py_tests/test_async_example.py diff --git a/README.md b/README.md index 391770b..077b1ab 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,7 @@ POST /auth/logout - Invalidate token ``` GET /api/v1/users - List users (paginated) POST /api/v1/users - Create user +POST /api/v1/users/bulk - Create multiple users (async, uses asyncio.gather) GET /api/v1/users/{id} - Get user details PUT /api/v1/users/{id} - Update user DELETE /api/v1/users/{id} - Soft delete user @@ -51,6 +52,45 @@ GET /api/v1/admin/tenants - List all tenants GET /api/v1/admin/stats - System statistics ``` +## Project Structure + +``` +sample-api/ +├── app/ # FastAPI application code +│ ├── main.py # API endpoints and business logic +│ ├── auth.py # JWT authentication, OAuth2 patterns +│ ├── config.py # Environment configuration +│ └── __init__.py +│ +├── py_tests/ # Python test suite (YOUR WORK GOES HERE) +│ ├── conftest.py # Shared pytest fixtures +│ ├── test_health.py # Example starter test +│ ├── test_async_example.py # Async test patterns +│ └── (your tests here) # Create test_auth.py, test_users.py, etc. +│ +├── rust_tests/ # Rust integration tests (OPTIONAL BONUS) +│ ├── Cargo.toml # Rust project config +│ └── tests/ +│ └── integration_tests.rs # HTTP client tests from external process +│ +├── pytest.ini # Pytest configuration (markers, coverage) +├── requirements.txt # Python dependencies +└── README.md # This file +``` + +**Where to add your tests:** +- Create test files in `py_tests/` directory (e.g., `test_auth.py`, `test_users.py`, `test_files.py`) +- Use provided fixtures from `conftest.py` +- Follow patterns in `test_async_example.py` for async operations +- See example fixture patterns below + +**Rust tests (Optional):** +- Located in `rust_tests/tests/integration_tests.rs` +- Tests API from external HTTP client perspective +- Run with: `cd rust_tests && cargo test` +- Demonstrates polyglot testing capability +- **Completely optional** - focus on Python tests first + ## Requirements (Tiered Approach) ### 🎯 Tier 1: Core Requirements (MUST COMPLETE) @@ -69,6 +109,11 @@ GET /api/v1/admin/stats - System statistics - Update user (authenticated) → 200 success - Duplicate username → 409 conflict +**Async Programming (1 test)** +- Bulk user creation using async patterns → tests `POST /api/v1/users/bulk` +- Must use `@pytest.mark.asyncio` and async/await patterns +- See `py_tests/test_async_example.py` for patterns + **Basic Tenant Isolation (2 tests)** - Tenant A cannot access Tenant B's user → 404 - List users only shows current tenant's data @@ -107,6 +152,12 @@ GET /api/v1/admin/stats - System statistics - Invalid/expired token handling - Cross-tenant file access prevention +**Async & Concurrency:** +- Concurrent user operations using `asyncio.gather()` +- Concurrent file uploads +- Performance testing (async vs sequential) +- Race condition handling + **Performance & Limits:** - Rate limiting enforcement (429 responses) - File type validation (415 unsupported media) @@ -127,13 +178,14 @@ GET /api/v1/admin/stats - System statistics - Parametrized tests for multi-scenario coverage - Test markers (`@pytest.mark.auth`, `@pytest.mark.tenant_isolation`, etc.) - Proper setup/teardown for isolation -- Environment configuration support +- **Async test patterns** - At least one test using `@pytest.mark.asyncio` +- AsyncClient for testing bulk operations **Bonus:** -- Async test patterns - Test data factories (factory_boy, Faker) - Custom pytest plugins -- Load/performance testing +- Performance comparison (async vs sync operations) +- Concurrent operation testing with `asyncio.gather()` - Mock external services ### Rust Integration Tests (Optional Bonus) diff --git a/app/main.py b/app/main.py index 76fcc3d..78cdb79 100644 --- a/app/main.py +++ b/app/main.py @@ -9,6 +9,7 @@ from datetime import datetime, UTC import uuid import io +import asyncio from collections import defaultdict from time import time @@ -514,6 +515,78 @@ async def delete_user( user["updated_at"] = datetime.now(UTC) +@app.post("/api/v1/users/bulk", response_model=List[User], status_code=status.HTTP_201_CREATED) +async def create_users_bulk( + request: Request, + users_data: List[UserCreate], + current_user: TokenData = Depends(get_current_user), +): + """ + Create multiple users in parallel (demonstrates async patterns). + + This endpoint uses asyncio.gather() to simulate concurrent operations. + Tests should use @pytest.mark.asyncio to properly test async behavior. + """ + check_rate_limit(request, current_user.user_id) + + if len(users_data) > 50: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Cannot create more than 50 users at once" + ) + + async def create_single_user(user_data: UserCreate): + """Async helper to simulate I/O-bound user creation""" + # Check for duplicates + for user in users_db.values(): + if user["username"] == user_data.username: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=f"Username '{user_data.username}' already exists" + ) + if user["email"] == user_data.email: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=f"Email '{user_data.email}' already exists" + ) + + # Simulate async operation (e.g., database write, external API call) + await asyncio.sleep(0.01) # Simulate I/O delay + + user_id = str(uuid.uuid4()) + now = datetime.now(UTC) + + new_user = { + "id": user_id, + "tenant_id": current_user.tenant_id, + "username": user_data.username, + "email": user_data.email, + "full_name": user_data.full_name, + "password_hash": hash_password("TempPassword123!"), + "role": user_data.role, + "created_at": now, + "updated_at": now, + "is_active": True, + } + + users_db[user_id] = new_user + return User(**{k: v for k, v in new_user.items() if k != "password_hash"}) + + # Execute all user creations concurrently + try: + created_users = await asyncio.gather( + *[create_single_user(user_data) for user_data in users_data] + ) + return created_users + except HTTPException: + raise + except Exception as e: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Bulk user creation failed: {str(e)}" + ) + + # ============================================================================ # FILE MANAGEMENT ENDPOINTS (Simulating S3) # ============================================================================ diff --git a/py_tests/test_async_example.py b/py_tests/test_async_example.py new file mode 100644 index 0000000..5ee9ba4 --- /dev/null +++ b/py_tests/test_async_example.py @@ -0,0 +1,82 @@ +""" +Example async test patterns for testing async endpoints + +This file demonstrates how to test async operations properly. +You should implement similar patterns for bulk operations and +concurrent scenarios. +""" +import pytest +import asyncio +from httpx import AsyncClient +from fastapi.testclient import TestClient +from app.main import app + + +# Fixture for async HTTP client +@pytest.fixture(scope="function") +async def async_client(): + """Async HTTP client for testing async endpoints""" + async with AsyncClient(app=app, base_url="http://test") as client: + yield client + + +# Example: Basic async test +@pytest.mark.asyncio +async def test_health_check_async(async_client): + """Test health endpoint using async client""" + response = await async_client.get("/health") + assert response.status_code == 200 + data = response.json() + assert data["status"] == "healthy" + + +# TODO: Implement async test for bulk user creation +@pytest.mark.asyncio +async def test_bulk_user_creation(): + """ + Test POST /api/v1/users/bulk endpoint. + + Requirements: + 1. Authenticate first (register tenant, login, get token) + 2. Create list of UserCreate objects (3-5 users) + 3. POST to /api/v1/users/bulk with auth token + 4. Verify all users created successfully + 5. Verify response time is reasonable (async should be faster) + 6. Test error case: duplicate username in batch + + Use @pytest.mark.asyncio and async/await patterns. + """ + pytest.skip("TODO: Implement bulk user creation test") + + +# TODO: Test concurrent operations +@pytest.mark.asyncio +async def test_concurrent_user_operations(): + """ + Test multiple operations happening concurrently. + + Use asyncio.gather() to: + 1. Create multiple users in parallel + 2. Update users concurrently + 3. Verify no race conditions + 4. Verify tenant isolation under concurrent load + + This tests real-world async behavior. + """ + pytest.skip("TODO: Implement concurrent operations test") + + +# TODO: Test async file operations +@pytest.mark.asyncio +async def test_concurrent_file_uploads(): + """ + Test uploading multiple files concurrently. + + Requirements: + 1. Authenticate + 2. Upload 3-5 files using asyncio.gather() + 3. Verify all uploads succeed + 4. Verify files are tenant-scoped + 5. Measure performance difference vs sequential + """ + pytest.skip("TODO: Implement concurrent file upload test") diff --git a/pytest.ini b/pytest.ini index eeb7708..6969b6c 100644 --- a/pytest.ini +++ b/pytest.ini @@ -16,8 +16,12 @@ addopts = # Test paths testpaths = py_tests +# Async configuration +asyncio_mode = auto + # Markers for test categorization markers = + asyncio: Async operation tests auth: Authentication and authorization tests tenant_isolation: Multi-tenant isolation and security tests users: User management tests From bf15920633ba333e72375f049ec7df84921742d5 Mon Sep 17 00:00:00 2001 From: Claude Code Date: Fri, 19 Dec 2025 20:31:00 +0000 Subject: [PATCH 2/3] Remove leftover files from previous version - Removed app/main_original.py (old simple API with internal references) - Removed app/events.py (unused module) - Removed py_tests/test_users_original.py (references wrong imports) These files were from before the py/ to app/ migration and would confuse candidates. --- app/events.py | 103 -------------------- app/main_original.py | 160 -------------------------------- py_tests/test_users_original.py | 66 ------------- 3 files changed, 329 deletions(-) delete mode 100644 app/events.py delete mode 100644 app/main_original.py delete mode 100644 py_tests/test_users_original.py diff --git a/app/events.py b/app/events.py deleted file mode 100644 index e10bdba..0000000 --- a/app/events.py +++ /dev/null @@ -1,103 +0,0 @@ -""" -Event publishing module - Separation of concerns for event handling - -This module handles all event publishing logic, keeping it separate from -the main API endpoints. -""" -from typing import Dict, Any, Optional -from datetime import datetime -import json -from collections import defaultdict - - -class EventPublisher: - """ - Simple event publisher that can be easily extended or mocked for testing. - - In production, this would publish to Kafka, SNS, SQS, etc. - For this assessment, it's an in-memory implementation. - """ - - def __init__(self): - self.events: list[Dict[str, Any]] = [] - self.enabled = True - - def publish(self, event_type: str, user_id: str, data: Dict[str, Any]) -> bool: - """ - Publish an event. Returns True if successful, False otherwise. - - Args: - event_type: Type of event (e.g., "user.created") - user_id: ID of the user - data: Event payload data - - Returns: - bool: True if published successfully - """ - if not self.enabled: - return False - - event = { - "event_type": event_type, - "user_id": user_id, - "timestamp": datetime.utcnow().isoformat(), - "data": data - } - - try: - self.events.append(event) - return True - except Exception: - # In production, log this error - return False - - def get_events(self) -> list[Dict[str, Any]]: - """Get all published events (for testing/debugging)""" - return self.events.copy() - - def clear(self): - """Clear all events (for testing)""" - self.events.clear() - - def disable(self): - """Disable event publishing (simulates outage)""" - self.enabled = False - - def enable(self): - """Enable event publishing""" - self.enabled = True - - -class MetricsCollector: - """ - Metrics collection - Separation of concerns for observability. - - Tracks application metrics independently of business logic. - """ - - def __init__(self): - self.counters = defaultdict(int) - self.start_time = datetime.utcnow() - - def increment(self, metric_name: str, amount: int = 1): - """Increment a counter metric""" - self.counters[metric_name] += amount - - def get_metrics(self) -> Dict[str, Any]: - """Get all metrics as a dictionary""" - uptime = (datetime.utcnow() - self.start_time).total_seconds() - - return { - "counters": dict(self.counters), - "uptime_seconds": int(uptime) - } - - def reset(self): - """Reset all metrics (for testing)""" - self.counters.clear() - self.start_time = datetime.utcnow() - - -# Global instances (in production, these would be dependency-injected) -event_publisher = EventPublisher() -metrics_collector = MetricsCollector() diff --git a/app/main_original.py b/app/main_original.py deleted file mode 100644 index 43727b8..0000000 --- a/app/main_original.py +++ /dev/null @@ -1,160 +0,0 @@ -""" -Simple User Management API for pytest Assessment -Mimics RAD team's FastAPI patterns (similar to VizioGram API) -""" -from fastapi import FastAPI, HTTPException, status -from pydantic import BaseModel, EmailStr, Field -from typing import Dict, List, Optional -from datetime import datetime -import uuid - -app = FastAPI(title="User Management API", version="1.0.0") - -# In-memory storage (simulating database) -users_db: Dict[str, dict] = {} - - -class UserCreate(BaseModel): - username: str = Field(..., min_length=3, max_length=50) - email: EmailStr - full_name: str = Field(..., min_length=1, max_length=100) - - -class UserUpdate(BaseModel): - email: Optional[EmailStr] = None - full_name: Optional[str] = Field(None, min_length=1, max_length=100) - - -class User(BaseModel): - id: str - username: str - email: str - full_name: str - created_at: datetime - updated_at: datetime - is_active: bool = True - - -@app.get("/health") -async def health_check(): - """Health check endpoint""" - return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()} - - -@app.post("/users", response_model=User, status_code=status.HTTP_201_CREATED) -async def create_user(user_data: UserCreate): - """Create a new user""" - # Check if username already exists - for user in users_db.values(): - if user["username"] == user_data.username: - raise HTTPException( - status_code=status.HTTP_409_CONFLICT, - detail=f"Username '{user_data.username}' already exists" - ) - if user["email"] == user_data.email: - raise HTTPException( - status_code=status.HTTP_409_CONFLICT, - detail=f"Email '{user_data.email}' already exists" - ) - - user_id = str(uuid.uuid4()) - now = datetime.utcnow() - - user = { - "id": user_id, - "username": user_data.username, - "email": user_data.email, - "full_name": user_data.full_name, - "created_at": now, - "updated_at": now, - "is_active": True - } - - users_db[user_id] = user - return user - - -@app.get("/users", response_model=List[User]) -async def list_users(active_only: bool = True): - """List all users, optionally filter by active status""" - if not active_only: - return [user for user in users_db.values() if user["is_active"]] - return list(users_db.values()) - - -@app.get("/users/{user_id}", response_model=User) -async def get_user(user_id: str): - """Get a specific user by ID""" - if user_id not in users_db: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"User with id '{user_id}' not found" - ) - return users_db[user_id] - - -@app.put("/users/{user_id}", response_model=User) -async def update_user(user_id: str, user_data: UserUpdate): - """Update a user's information""" - if user_id not in users_db: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"User with id '{user_id}' not found" - ) - - user = users_db[user_id] - - # Check for email conflicts if email is being updated - if user_data.email and user_data.email != user["email"]: - for uid, u in users_db.items(): - if uid != user_id and u["email"] == user_data.email: - raise HTTPException( - status_code=status.HTTP_409_CONFLICT, - detail=f"Email '{user_data.email}' already exists" - ) - - # Update fields - if user_data.email: - user["email"] = user_data.email - if user_data.full_name is not None: - user["full_name"] = user_data.full_name - - user["updated_at"] = datetime.utcnow() - users_db[user_id] = user - - return user - - -@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT) -async def delete_user(user_id: str): - """Soft delete a user (sets is_active to False)""" - if user_id not in users_db: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"User with id '{user_id}' not found" - ) - - users_db[user_id]["is_active"] = False - users_db[user_id]["updated_at"] = datetime.utcnow() - - -@app.delete("/users/{user_id}/permanent", status_code=status.HTTP_204_NO_CONTENT) -async def permanently_delete_user(user_id: str): - """Permanently delete a user from the database""" - if user_id not in users_db: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"User with id '{user_id}' not found" - ) - - user = users_db[user_id] - del users_db[user_id] - return {"deleted": user["username"]} - - -# Test helper endpoint - clear all data -@app.post("/test/reset") -async def reset_database(): - """Reset the database (for testing purposes only)""" - users_db.clear() - return {"message": "Database reset successfully"} diff --git a/py_tests/test_users_original.py b/py_tests/test_users_original.py deleted file mode 100644 index 08d8ec9..0000000 --- a/py_tests/test_users_original.py +++ /dev/null @@ -1,66 +0,0 @@ -""" -Example pytest test file for User Management API - -This file contains ONE example test to get you started. -Your task is to add comprehensive test coverage for all endpoints. - -pytest basics: -- Test functions must start with 'test_' -- Use assert statements for validation -- Use fixtures for setup/teardown -- Run with: pytest -v -""" -import pytest -from fastapi.testclient import TestClient -from py.main import app, users_db - - -# Fixture example - this runs before each test -@pytest.fixture -def client(): - """Create a test client for the FastAPI app""" - return TestClient(app) - - -@pytest.fixture(autouse=True) -def reset_db(): - """Automatically reset database before each test""" - users_db.clear() - yield # Test runs here - users_db.clear() # Cleanup after test - - -# Example test - THIS IS YOUR STARTING POINT -def test_health_check(client): - """Test that the health check endpoint returns 200""" - response = client.get("/health") - assert response.status_code == 200 - assert response.json()["status"] == "healthy" - assert "timestamp" in response.json() - - -# TODO: Add more tests below -# Suggested tests to implement: -# -# 1. test_create_user_success - Create a valid user -# 2. test_create_user_duplicate_username - Try to create user with existing username -# 3. test_create_user_duplicate_email - Try to create user with existing email -# 4. test_create_user_invalid_email - Test email validation -# 5. test_create_user_invalid_username_too_short - Test username length validation -# 6. test_list_users_empty - List users when database is empty -# 7. test_list_users_with_data - List users with multiple users -# 8. test_list_users_active_only - Test the active_only filter -# 9. test_get_user_success - Get a specific user by ID -# 10. test_get_user_not_found - Try to get non-existent user -# 11. test_update_user_success - Update user information -# 12. test_update_user_not_found - Try to update non-existent user -# 13. test_update_user_email_conflict - Try to update to existing email -# 14. test_delete_user_soft_delete - Test soft delete (is_active=False) -# 15. test_delete_user_permanent - Test permanent deletion -# 16. test_delete_user_not_found - Try to delete non-existent user -# -# BONUS challenges: -# - Use pytest.mark.parametrize for testing multiple inputs -# - Create custom fixtures for common test data -# - Test error messages match expected format -# - Test timestamp updates on user modifications From 052e4029bd8a3bdf0f13e3f149daa708444584a9 Mon Sep 17 00:00:00 2001 From: Claude Code Date: Fri, 19 Dec 2025 20:32:29 +0000 Subject: [PATCH 3/3] Remove candidate name from code comments Make docstring generic for external use. --- app/main.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/app/main.py b/app/main.py index 78cdb79..e8531f5 100644 --- a/app/main.py +++ b/app/main.py @@ -1,6 +1,6 @@ """ Advanced Multi-Tenant User & File Management API with Authentication -For Senior QA Automation Assessment - Ezequiel Nams +For Senior QA Automation Assessment """ from fastapi import FastAPI, HTTPException, status, Depends, File, UploadFile, Request from fastapi.responses import StreamingResponse