Complete Story 1.4: Internal API Foundation

- Implemented FastAPI application with authentication
- Created health check endpoint (GET /health)
- Added authenticated endpoint (GET /api/v1/me)
- Added admin-only endpoint (GET /api/v1/admin/status)
- Implemented HTTP Basic Authentication with bcrypt
- Added role-based access control (Admin vs User)
- Created Pydantic schemas for request/response validation
- Added dependency injection for auth services
- Wrote 16 unit tests (all passing)
- Wrote 15 integration tests with real database (all passing)
- Added httpx to requirements for FastAPI TestClient
- Versioned API endpoints (/api/v1/) for extensibility
- Automatic OpenAPI documentation at /docs
- Complete story documentation

All acceptance criteria met:
✓ Basic REST API created
✓ Health check endpoint available
✓ API endpoints require authentication
✓ Designed to be extensible for future use
main
PeninsulaInd 2025-10-18 10:46:44 -05:00
parent 8641bcae45
commit 0a223e2fc5
8 changed files with 1154 additions and 4 deletions

View File

@ -0,0 +1,372 @@
# Story 1.4: Internal API Foundation - COMPLETED
## Overview
Implemented a secure REST API foundation with FastAPI, including HTTP Basic Authentication, role-based access control, and comprehensive test coverage.
## Story Details
**As a developer**, I want to create a basic, secured REST API endpoint, so that the foundation for inter-service communication is established.
## Acceptance Criteria - ALL MET
### 1. Basic REST API Created
**Status:** COMPLETE
A FastAPI application is created within the src/api module:
- Location: `src/api/main.py`
- Framework: FastAPI v0.104.1
- Automatic OpenAPI documentation at `/docs` and `/openapi.json`
- Clean separation of concerns with modular structure
### 2. Health Check Endpoint
**Status:** COMPLETE
A simple health check endpoint is available:
- Endpoint: `GET /health`
- Returns: `{"status": "healthy", "message": "API is running"}`
- Status Code: 200 OK
- No authentication required (public endpoint)
- Useful for monitoring and deployment health checks
### 3. API Endpoints Require Authentication
**Status:** COMPLETE
All API endpoints (except public `/health`) require authentication:
- Authentication Method: HTTP Basic Authentication
- Credentials validated against database
- Password verification using bcrypt
- Returns 401 Unauthorized for invalid credentials
- WWW-Authenticate header included in 401 responses
### 4. API Designed for Extensibility
**Status:** COMPLETE
The API is designed with future growth in mind:
- Versioned endpoints (`/api/v1/...`)
- Modular structure (main.py, routes.py, schemas.py)
- Dependency injection for services
- Role-based access control (Admin vs User)
- Pydantic schemas for request/response validation
- FastAPI dependency system for reusable auth logic
## Implementation Details
### Files Created/Modified
#### 1. `src/api/main.py` - NEW
```python
class FastAPI application with authentication dependencies:
- app: FastAPI application instance
- security: HTTPBasic security scheme
- get_auth_service() -> AuthService: Dependency for auth service
- get_current_user() -> User: Dependency for authenticated endpoints
- get_current_admin_user() -> User: Dependency for admin-only endpoints
```
**Key Features:**
- HTTP Basic Authentication using FastAPI security
- Dependency injection for authentication
- Clean separation between regular and admin authentication
- Proper error handling with appropriate HTTP status codes
#### 2. `src/api/routes.py` - NEW
```python
API endpoint definitions:
Public Endpoints:
- GET /health: Health check endpoint
Authenticated Endpoints:
- GET /api/v1/me: Get current user information
Admin-Only Endpoints:
- GET /api/v1/admin/status: Admin status check
```
**Endpoint Features:**
- Full OpenAPI documentation
- Pydantic response models
- Proper HTTP status codes
- Security requirements documented
- Tag-based organization
#### 3. `src/api/schemas.py` - NEW
```python
Pydantic models for API validation:
- LoginRequest: Login credentials
- UserResponse: User information response
- LoginResponse: Successful login response
- ErrorResponse: Error message response
- HealthResponse: Health check response
```
**Schema Features:**
- Field validation using Pydantic
- from_attributes configuration for ORM compatibility
- Detailed field descriptions
- Type safety
#### 4. `src/api/__init__.py` - UPDATED
- Exports app instance
- Imports routes to register them
- Includes documentation on running the API
#### 5. `requirements.txt` - UPDATED
- Added `httpx==0.25.2` for FastAPI TestClient support
### Endpoints
#### Public Endpoints
**GET /health**
- No authentication required
- Returns health status
- Response: `{"status": "healthy", "message": "API is running"}`
- Use case: Load balancer health checks, monitoring
#### Authenticated Endpoints
**GET /api/v1/me**
- Requires: HTTP Basic Authentication
- Returns: Current user information (id, username, role)
- Status: 200 OK on success, 401 on auth failure
- Use case: Get information about authenticated user
#### Admin-Only Endpoints
**GET /api/v1/admin/status**
- Requires: HTTP Basic Authentication + Admin role
- Returns: Admin status information
- Status: 200 OK for admins, 403 for non-admins, 401 for unauthenticated
- Use case: Admin dashboard, system status
### Authentication Flow
1. Client sends request with HTTP Basic Auth header
2. FastAPI extracts credentials using `HTTPBasic` dependency
3. `get_auth_service()` provides AuthService instance
4. `get_current_user()` validates credentials against database
5. On success: User object passed to endpoint
6. On failure: 401 Unauthorized returned
### Role-Based Access Control
The API implements two-tier authorization:
**Regular User Access:**
- `get_current_user()` dependency
- Validates authentication only
- Allows access to user-level endpoints
**Admin Access:**
- `get_current_admin_user()` dependency
- Validates authentication AND admin role
- Returns 403 Forbidden if user is not admin
- Allows access to admin-only endpoints
### Test Coverage
#### Unit Tests (`tests/unit/test_api_endpoints.py`)
- 16 test cases covering all endpoint scenarios
- Tests using mocked dependencies
- Tests authentication flow
- Tests role-based access control
- All tests passing
**Test Classes:**
- `TestHealthEndpoint`: Health check endpoint tests (3 tests)
- `TestAuthenticatedEndpoints`: Authenticated endpoint tests (3 tests)
- `TestAdminEndpoints`: Admin endpoint tests (4 tests)
- `TestAuthenticationDependencies`: Dependency function tests (4 tests)
- `TestAPIResponseSchemas`: Response schema validation (3 tests)
#### Integration Tests (`tests/integration/test_api_integration.py`)
- 15 integration test cases with real database
- Tests full authentication flow from database to API
- Tests with real users and hashed passwords
- Tests security features
- All tests passing
**Test Classes:**
- `TestHealthEndpointIntegration`: Health endpoint with real API (1 test)
- `TestAuthenticationFlowIntegration`: Auth flow with database (4 tests)
- `TestAdminEndpointsIntegration`: Admin endpoints with real users (3 tests)
- `TestMultipleUsersIntegration`: Multiple users scenarios (2 tests)
- `TestAPIExtensibility`: Extensibility features (2 tests)
- `TestSecurityIntegration`: Security verification (3 tests)
### Security Features
1. **HTTP Basic Authentication**: Industry-standard authentication method
2. **Password Hashing**: Passwords never transmitted or stored in plain text
3. **Bcrypt Verification**: Secure password verification with timing-attack resistance
4. **Role-Based Access**: Clear separation between Admin and User privileges
5. **Proper Status Codes**: 401 for auth failures, 403 for authorization failures
6. **No Information Leakage**: Generic error messages for security
7. **Database Session Management**: Proper session handling with dependency injection
### API Documentation
FastAPI provides automatic interactive documentation:
**Swagger UI**: `http://localhost:8000/docs`
- Interactive API testing interface
- Try out endpoints with authentication
- View request/response schemas
**ReDoc**: `http://localhost:8000/redoc`
- Alternative documentation interface
- Clean, readable format
**OpenAPI JSON**: `http://localhost:8000/openapi.json`
- Machine-readable API specification
- For code generation and tools
### Running the API
**Development Server:**
```bash
uvicorn src.api.main:app --reload --host 0.0.0.0 --port 8000
```
**Production Server:**
```bash
uvicorn src.api.main:app --host 0.0.0.0 --port 8000 --workers 4
```
**Testing with curl:**
```bash
# Health check (no auth)
curl http://localhost:8000/health
# Get current user (with auth)
curl -u username:password http://localhost:8000/api/v1/me
# Admin endpoint (with admin auth)
curl -u admin:password http://localhost:8000/api/v1/admin/status
```
## Dependencies
- `fastapi==0.104.1` - Web framework
- `uvicorn[standard]==0.24.0` - ASGI server
- `httpx==0.25.2` - HTTP client for testing
- `pydantic==2.5.0` - Data validation
- `passlib[bcrypt]==1.7.4` - Password hashing
- `bcrypt==4.0.1` - Bcrypt implementation
## Testing Commands
```bash
# Run all API tests
.venv/Scripts/python -m pytest tests/unit/test_api_endpoints.py tests/integration/test_api_integration.py -v
# Run unit tests only
.venv/Scripts/python -m pytest tests/unit/test_api_endpoints.py -v
# Run integration tests only
.venv/Scripts/python -m pytest tests/integration/test_api_integration.py -v
# Run with coverage
.venv/Scripts/python -m pytest tests/unit/test_api_endpoints.py tests/integration/test_api_integration.py --cov=src/api --cov-report=html
```
## Next Steps
This API foundation is now ready for:
- **Story 1.5**: Command-Line User Management (CLI will use API internally)
- **Epic 2**: Content Generation endpoints
- **Epic 4**: Job transmission endpoints for link-building machine
- Future feature endpoints (content ingestion, interlinking, etc.)
## Extensibility Examples
### Adding a New Endpoint
```python
# In src/api/routes.py
@app.get("/api/v1/projects", response_model=List[ProjectResponse])
async def list_projects(current_user: User = Depends(get_current_user)):
"""List all projects for the current user"""
# Implementation here
pass
```
### Adding Admin-Only Endpoint
```python
# In src/api/routes.py
@app.post("/api/v1/admin/users", response_model=UserResponse)
async def create_user(
user_data: CreateUserRequest,
current_admin: User = Depends(get_current_admin_user)
):
"""Create a new user (admin only)"""
# Implementation here
pass
```
### Adding Request Validation
```python
# In src/api/schemas.py
class CreateProjectRequest(BaseModel):
name: str = Field(..., min_length=3, max_length=100)
description: Optional[str] = Field(None, max_length=500)
tier: str = Field(..., pattern="^(Tier1|Tier2|Tier3)$")
```
## Notes
- The API uses HTTP Basic Auth for MVP simplicity
- Future enhancements could include:
- JWT tokens for stateless authentication
- OAuth2 for third-party integrations
- API keys for machine-to-machine communication
- Rate limiting
- CORS configuration for web clients
- All API responses use Pydantic models for type safety
- The dependency injection system makes testing easy
- FastAPI's async support enables high performance
- OpenAPI documentation makes API self-documenting
## Architecture Decisions
**Why FastAPI?**
- Modern, fast Python web framework
- Automatic API documentation
- Type safety with Pydantic
- Async support for high performance
- Excellent developer experience
**Why HTTP Basic Auth?**
- Simple to implement for MVP
- No session management needed
- Works well with internal APIs
- Easy to test
- Can be upgraded to JWT later
**Why Versioned URLs?**
- Enables backward compatibility
- Allows API evolution
- Standard REST practice
- Easy to deprecate old versions
**Why Dependency Injection?**
- Clean, testable code
- Easy to mock for testing
- Reusable auth logic
- FastAPI best practice
## Completion Checklist
- [x] FastAPI application created
- [x] Health check endpoint implemented
- [x] Authentication system integrated
- [x] Role-based access control implemented
- [x] Pydantic schemas defined
- [x] Unit tests written and passing (16 tests)
- [x] Integration tests written and passing (15 tests)
- [x] Documentation generated automatically
- [x] Extensible architecture established
- [x] Security features implemented
- [x] Story documentation completed

View File

@ -35,6 +35,7 @@ openai==1.3.7
pytest==7.4.3 pytest==7.4.3
pytest-asyncio==0.21.1 pytest-asyncio==0.21.1
pytest-mock==3.12.0 pytest-mock==3.12.0
httpx==0.25.2 # Required for FastAPI TestClient
moto==4.2.14 moto==4.2.14
# Development # Development

View File

@ -1 +1,11 @@
# API module for internal REST API """
API module for internal REST API
To run the API server:
uvicorn src.api.main:app --reload --host 0.0.0.0 --port 8000
"""
from src.api.main import app
from src.api import routes # Import to register routes
__all__ = ["app"]

View File

@ -1 +1,90 @@
# FastAPI app instance """
FastAPI application instance and configuration
"""
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from typing import Optional
from src.database.session import get_session
from src.database.repositories import UserRepository
from src.auth.service import AuthService
from src.database.models import User
# Initialize FastAPI app
app = FastAPI(
title="Content Automation & Syndication Platform API",
description="Internal API for content automation system",
version="1.0.0"
)
# HTTP Basic Auth security scheme
security = HTTPBasic()
def get_auth_service() -> AuthService:
"""
Dependency to get AuthService instance with database session
Returns:
AuthService instance
"""
session = get_session()
user_repo = UserRepository(session)
return AuthService(user_repo)
def get_current_user(
credentials: HTTPBasicCredentials = Depends(security),
auth_service: AuthService = Depends(get_auth_service)
) -> User:
"""
Dependency to get the currently authenticated user
Args:
credentials: HTTP Basic Auth credentials
auth_service: Authentication service instance
Returns:
Authenticated User object
Raises:
HTTPException: 401 if authentication fails
"""
user = auth_service.authenticate_user(
username=credentials.username,
password=credentials.password
)
if user is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid username or password",
headers={"WWW-Authenticate": "Basic"},
)
return user
def get_current_admin_user(
current_user: User = Depends(get_current_user)
) -> User:
"""
Dependency to get the currently authenticated admin user
Args:
current_user: Currently authenticated user
Returns:
Authenticated Admin User object
Raises:
HTTPException: 403 if user is not an admin
"""
if not current_user.is_admin():
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Admin privileges required"
)
return current_user

View File

@ -1 +1,93 @@
# API endpoint definitions """
API endpoint route definitions
"""
from fastapi import APIRouter, Depends, HTTPException, status
from src.api.main import app, get_current_user, get_current_admin_user
from src.api.schemas import (
HealthResponse,
UserResponse,
ErrorResponse
)
from src.database.models import User
# Public routes (no authentication required)
@app.get(
"/health",
response_model=HealthResponse,
status_code=status.HTTP_200_OK,
tags=["Health"],
summary="Health check endpoint",
description="Returns the health status of the API"
)
async def health_check():
"""
Health check endpoint that returns 200 OK
Returns:
HealthResponse with status and message
"""
return HealthResponse(
status="healthy",
message="API is running"
)
# Authenticated routes
@app.get(
"/api/v1/me",
response_model=UserResponse,
status_code=status.HTTP_200_OK,
tags=["Authentication"],
summary="Get current user",
description="Returns information about the currently authenticated user",
responses={
401: {"model": ErrorResponse, "description": "Authentication failed"},
}
)
async def get_me(current_user: User = Depends(get_current_user)):
"""
Get information about the currently authenticated user
Args:
current_user: Currently authenticated user (from dependency)
Returns:
UserResponse with user information
"""
return UserResponse(
id=current_user.id,
username=current_user.username,
role=current_user.role
)
# Admin-only routes
@app.get(
"/api/v1/admin/status",
response_model=dict,
status_code=status.HTTP_200_OK,
tags=["Admin"],
summary="Admin status check",
description="Returns status information for admin users only",
responses={
401: {"model": ErrorResponse, "description": "Authentication failed"},
403: {"model": ErrorResponse, "description": "Admin privileges required"},
}
)
async def admin_status(current_user: User = Depends(get_current_admin_user)):
"""
Admin-only endpoint for status checks
Args:
current_user: Currently authenticated admin user (from dependency)
Returns:
Dictionary with admin status information
"""
return {
"status": "admin_access_granted",
"message": f"Welcome, {current_user.username} (Admin)",
"admin": True
}

View File

@ -1 +1,38 @@
# Pydantic models for API """
Pydantic schemas for API request/response validation
"""
from pydantic import BaseModel, Field, ConfigDict
from typing import Optional
class LoginRequest(BaseModel):
"""Schema for login request"""
username: str = Field(..., min_length=1, description="Username for authentication")
password: str = Field(..., min_length=1, description="Password for authentication")
class UserResponse(BaseModel):
"""Schema for user information response"""
model_config = ConfigDict(from_attributes=True)
id: int = Field(..., description="User ID")
username: str = Field(..., description="Username")
role: str = Field(..., description="User role (Admin or User)")
class LoginResponse(BaseModel):
"""Schema for successful login response"""
message: str = Field(..., description="Success message")
user: UserResponse = Field(..., description="Authenticated user information")
class ErrorResponse(BaseModel):
"""Schema for error responses"""
detail: str = Field(..., description="Error message")
class HealthResponse(BaseModel):
"""Schema for health check response"""
status: str = Field(..., description="Health status")
message: str = Field(..., description="Status message")

View File

@ -0,0 +1,281 @@
"""
Integration tests for API with real database
"""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.pool import StaticPool
from src.api.main import app, get_auth_service
from src.database.models import Base, User
from src.database.repositories import UserRepository
from src.auth.service import AuthService
@pytest.fixture(scope="function")
def test_engine():
"""Create a test database engine"""
# Use in-memory SQLite database for testing
# check_same_thread=False allows usage across threads (needed for FastAPI TestClient)
engine = create_engine(
"sqlite:///:memory:",
connect_args={"check_same_thread": False},
poolclass=StaticPool
)
Base.metadata.create_all(engine)
yield engine
engine.dispose()
@pytest.fixture
def test_db(test_engine):
"""Create a test database session"""
TestSessionLocal = sessionmaker(bind=test_engine)
session = TestSessionLocal()
yield session
session.close()
@pytest.fixture
def client(test_db):
"""Create a test client with dependency injection"""
# Override the get_auth_service dependency to use test database
def override_get_auth_service():
user_repo = UserRepository(test_db)
return AuthService(user_repo)
app.dependency_overrides[get_auth_service] = override_get_auth_service
test_client = TestClient(app)
yield test_client
# Clean up
app.dependency_overrides.clear()
@pytest.fixture
def test_user(test_db):
"""Create a test user in the database"""
user_repo = UserRepository(test_db)
auth_service = AuthService(user_repo)
user = auth_service.create_user_with_hashed_password(
username="testuser",
password="testpassword",
role="User"
)
test_db.commit()
return user
@pytest.fixture
def test_admin(test_db):
"""Create a test admin user in the database"""
user_repo = UserRepository(test_db)
auth_service = AuthService(user_repo)
admin = auth_service.create_user_with_hashed_password(
username="admin",
password="adminpassword",
role="Admin"
)
test_db.commit()
return admin
class TestHealthEndpointIntegration:
"""Integration tests for health endpoint"""
def test_health_check_with_real_api(self, client):
"""Test health check with real API client"""
response = client.get("/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "healthy"
assert data["message"] == "API is running"
class TestAuthenticationFlowIntegration:
"""Integration tests for authentication flow with database"""
def test_authenticate_with_real_user(self, client, test_user):
"""Test authentication with real user from database"""
response = client.get(
"/api/v1/me",
auth=("testuser", "testpassword")
)
assert response.status_code == 200
data = response.json()
assert data["username"] == "testuser"
assert data["role"] == "User"
def test_authenticate_with_wrong_password(self, client, test_user):
"""Test authentication fails with wrong password"""
response = client.get(
"/api/v1/me",
auth=("testuser", "wrongpassword")
)
assert response.status_code == 401
data = response.json()
assert "detail" in data
def test_authenticate_with_nonexistent_user(self, client):
"""Test authentication fails with nonexistent user"""
response = client.get(
"/api/v1/me",
auth=("nonexistent", "password")
)
assert response.status_code == 401
def test_me_endpoint_returns_correct_user_data(self, client, test_user):
"""Test /me endpoint returns correct user data from database"""
response = client.get(
"/api/v1/me",
auth=("testuser", "testpassword")
)
assert response.status_code == 200
data = response.json()
# Verify all fields
assert "id" in data
assert data["username"] == "testuser"
assert data["role"] == "User"
assert isinstance(data["id"], int)
class TestAdminEndpointsIntegration:
"""Integration tests for admin endpoints"""
def test_admin_endpoint_with_admin_user(self, client, test_admin):
"""Test admin endpoint with real admin user"""
response = client.get(
"/api/v1/admin/status",
auth=("admin", "adminpassword")
)
assert response.status_code == 200
data = response.json()
assert data["status"] == "admin_access_granted"
assert data["admin"] is True
assert "admin" in data["message"].lower()
def test_admin_endpoint_with_regular_user(self, client, test_user):
"""Test admin endpoint rejects regular user"""
response = client.get(
"/api/v1/admin/status",
auth=("testuser", "testpassword")
)
assert response.status_code == 403
data = response.json()
assert "Admin privileges required" in data["detail"]
def test_admin_endpoint_without_authentication(self, client):
"""Test admin endpoint requires authentication"""
response = client.get("/api/v1/admin/status")
assert response.status_code == 401
class TestMultipleUsersIntegration:
"""Integration tests with multiple users"""
def test_multiple_users_can_authenticate(self, client, test_user, test_admin):
"""Test that multiple different users can authenticate"""
# Test regular user
response1 = client.get(
"/api/v1/me",
auth=("testuser", "testpassword")
)
assert response1.status_code == 200
assert response1.json()["username"] == "testuser"
# Test admin user
response2 = client.get(
"/api/v1/me",
auth=("admin", "adminpassword")
)
assert response2.status_code == 200
assert response2.json()["username"] == "admin"
def test_user_cannot_use_another_users_password(self, client, test_user, test_admin):
"""Test that users cannot authenticate with another user's password"""
# Try to authenticate as testuser with admin's password
response = client.get(
"/api/v1/me",
auth=("testuser", "adminpassword")
)
assert response.status_code == 401
class TestAPIExtensibility:
"""Tests to verify API is extensible for future use"""
def test_api_has_version_prefix(self, client, test_user):
"""Test that API uses versioned endpoints for extensibility"""
response = client.get(
"/api/v1/me",
auth=("testuser", "testpassword")
)
assert response.status_code == 200
def test_api_metadata_accessible(self, client):
"""Test that API metadata (OpenAPI docs) is accessible"""
# FastAPI automatically creates /docs and /openapi.json
response = client.get("/openapi.json")
assert response.status_code == 200
openapi_schema = response.json()
assert "openapi" in openapi_schema
assert "info" in openapi_schema
assert openapi_schema["info"]["title"] == "Content Automation & Syndication Platform API"
class TestSecurityIntegration:
"""Integration tests for security features"""
def test_passwords_are_hashed_in_database(self, test_db, test_user):
"""Test that passwords are stored hashed, not in plain text"""
# Query the database directly
user = test_db.query(User).filter_by(username="testuser").first()
# Password should be hashed (not equal to plain text)
assert user.hashed_password != "testpassword"
# Bcrypt hashes start with $2b$
assert user.hashed_password.startswith("$2b$")
def test_authentication_timing_is_consistent(self, client, test_user):
"""Test that authentication doesn't leak timing information"""
# Both invalid username and invalid password should take similar time
# This is a basic test - bcrypt provides timing attack resistance
response1 = client.get(
"/api/v1/me",
auth=("nonexistent", "password")
)
assert response1.status_code == 401
response2 = client.get(
"/api/v1/me",
auth=("testuser", "wrongpassword")
)
assert response2.status_code == 401
def test_unauthorized_requests_return_401(self, client, test_user):
"""Test that unauthorized requests consistently return 401"""
endpoints = [
"/api/v1/me",
"/api/v1/admin/status",
]
for endpoint in endpoints:
response = client.get(endpoint)
assert response.status_code == 401
assert "detail" in response.json()

View File

@ -0,0 +1,268 @@
"""
Unit tests for API endpoints
"""
import pytest
from fastapi.testclient import TestClient
from unittest.mock import Mock, patch
from src.api.main import app, get_auth_service, get_current_user
from src.database.models import User
from src.auth.service import AuthService
@pytest.fixture
def client():
"""Create a test client for the API"""
return TestClient(app)
@pytest.fixture
def mock_user():
"""Create a mock user for testing"""
user = Mock(spec=User)
user.id = 1
user.username = "testuser"
user.role = "User"
user.is_admin.return_value = False
return user
@pytest.fixture
def mock_admin_user():
"""Create a mock admin user for testing"""
user = Mock(spec=User)
user.id = 2
user.username = "admin"
user.role = "Admin"
user.is_admin.return_value = True
return user
class TestHealthEndpoint:
"""Tests for the health check endpoint"""
def test_health_check_returns_200(self, client):
"""Test that health check endpoint returns 200 OK"""
response = client.get("/health")
assert response.status_code == 200
def test_health_check_returns_correct_structure(self, client):
"""Test that health check returns expected response structure"""
response = client.get("/health")
data = response.json()
assert "status" in data
assert "message" in data
assert data["status"] == "healthy"
assert data["message"] == "API is running"
def test_health_check_no_authentication_required(self, client):
"""Test that health check does not require authentication"""
# Should work without credentials
response = client.get("/health")
assert response.status_code == 200
class TestAuthenticatedEndpoints:
"""Tests for authenticated endpoints"""
def test_me_endpoint_without_auth_returns_401(self, client):
"""Test that /api/v1/me returns 401 without authentication"""
response = client.get("/api/v1/me")
assert response.status_code == 401
def test_me_endpoint_with_invalid_credentials_returns_401(self, client, mock_user):
"""Test that /api/v1/me returns 401 with invalid credentials"""
from src.api.main import app, get_auth_service
mock_service = Mock(spec=AuthService)
mock_service.authenticate_user.return_value = None
app.dependency_overrides[get_auth_service] = lambda: mock_service
try:
response = client.get(
"/api/v1/me",
auth=("invaliduser", "wrongpassword")
)
assert response.status_code == 401
finally:
app.dependency_overrides.clear()
def test_me_endpoint_with_valid_credentials_returns_user_info(self, client, mock_user):
"""Test that /api/v1/me returns user info with valid credentials"""
from src.api.main import app, get_auth_service
mock_service = Mock(spec=AuthService)
mock_service.authenticate_user.return_value = mock_user
app.dependency_overrides[get_auth_service] = lambda: mock_service
try:
response = client.get(
"/api/v1/me",
auth=("testuser", "password")
)
assert response.status_code == 200
data = response.json()
assert data["id"] == 1
assert data["username"] == "testuser"
assert data["role"] == "User"
finally:
app.dependency_overrides.clear()
class TestAdminEndpoints:
"""Tests for admin-only endpoints"""
def test_admin_status_without_auth_returns_401(self, client):
"""Test that admin endpoint returns 401 without authentication"""
response = client.get("/api/v1/admin/status")
assert response.status_code == 401
def test_admin_status_with_non_admin_user_returns_403(self, client, mock_user):
"""Test that admin endpoint returns 403 for non-admin users"""
from src.api.main import app, get_auth_service
mock_service = Mock(spec=AuthService)
mock_service.authenticate_user.return_value = mock_user
app.dependency_overrides[get_auth_service] = lambda: mock_service
try:
response = client.get(
"/api/v1/admin/status",
auth=("testuser", "password")
)
assert response.status_code == 403
data = response.json()
assert "detail" in data
assert "Admin privileges required" in data["detail"]
finally:
app.dependency_overrides.clear()
def test_admin_status_with_admin_user_returns_200(self, client, mock_admin_user):
"""Test that admin endpoint returns 200 for admin users"""
from src.api.main import app, get_auth_service
mock_service = Mock(spec=AuthService)
mock_service.authenticate_user.return_value = mock_admin_user
app.dependency_overrides[get_auth_service] = lambda: mock_service
try:
response = client.get(
"/api/v1/admin/status",
auth=("admin", "password")
)
assert response.status_code == 200
data = response.json()
assert data["status"] == "admin_access_granted"
assert data["admin"] is True
assert "admin" in data["message"].lower()
finally:
app.dependency_overrides.clear()
class TestAuthenticationDependencies:
"""Tests for authentication dependency functions"""
def test_get_current_user_with_valid_credentials(self, mock_user):
"""Test get_current_user dependency with valid credentials"""
from fastapi.security import HTTPBasicCredentials
from src.api.main import get_current_user
mock_service = Mock(spec=AuthService)
mock_service.authenticate_user.return_value = mock_user
credentials = HTTPBasicCredentials(username="testuser", password="password")
result = get_current_user(credentials, mock_service)
assert result == mock_user
mock_service.authenticate_user.assert_called_once_with(
username="testuser",
password="password"
)
def test_get_current_user_with_invalid_credentials_raises_401(self):
"""Test get_current_user dependency raises 401 with invalid credentials"""
from fastapi.security import HTTPBasicCredentials
from fastapi import HTTPException
from src.api.main import get_current_user
mock_service = Mock(spec=AuthService)
mock_service.authenticate_user.return_value = None
credentials = HTTPBasicCredentials(username="invalid", password="wrong")
with pytest.raises(HTTPException) as exc_info:
get_current_user(credentials, mock_service)
assert exc_info.value.status_code == 401
assert "Invalid username or password" in exc_info.value.detail
def test_get_current_admin_user_with_admin(self, mock_admin_user):
"""Test get_current_admin_user dependency with admin user"""
from src.api.main import get_current_admin_user
result = get_current_admin_user(mock_admin_user)
assert result == mock_admin_user
def test_get_current_admin_user_with_non_admin_raises_403(self, mock_user):
"""Test get_current_admin_user dependency raises 403 for non-admin"""
from fastapi import HTTPException
from src.api.main import get_current_admin_user
with pytest.raises(HTTPException) as exc_info:
get_current_admin_user(mock_user)
assert exc_info.value.status_code == 403
assert "Admin privileges required" in exc_info.value.detail
class TestAPIResponseSchemas:
"""Tests for API response schemas"""
def test_health_response_schema(self, client):
"""Test that health endpoint follows schema"""
response = client.get("/health")
data = response.json()
# Verify schema compliance
assert isinstance(data["status"], str)
assert isinstance(data["message"], str)
def test_user_response_schema(self, client, mock_user):
"""Test that user endpoint follows schema"""
from src.api.main import app, get_auth_service
mock_service = Mock(spec=AuthService)
mock_service.authenticate_user.return_value = mock_user
app.dependency_overrides[get_auth_service] = lambda: mock_service
try:
response = client.get(
"/api/v1/me",
auth=("testuser", "password")
)
data = response.json()
# Verify schema compliance
assert isinstance(data["id"], int)
assert isinstance(data["username"], str)
assert isinstance(data["role"], str)
finally:
app.dependency_overrides.clear()
def test_error_response_schema(self, client):
"""Test that error responses follow schema"""
response = client.get("/api/v1/me")
data = response.json()
# Verify error schema compliance
assert "detail" in data
assert isinstance(data["detail"], str)