chore: auto-commit 2026-03-16 14:51
This commit is contained in:
cclohmar
parent
9a586d0955
commit
f1991c3983
23 changed files with 229 additions and 1875 deletions
|
|
@ -1,4 +0,0 @@
|
|||
DATABASE_URL=sqlite+aiosqlite:///./mockapi.db
|
||||
ADMIN_USERNAME=admin
|
||||
ADMIN_PASSWORD=admin123 # Change this in production
|
||||
SECRET_KEY=your-secret-key-here-change-me
|
||||
76
README.md
76
README.md
|
|
@ -28,53 +28,35 @@ A lightweight, configurable mock API application in Python that allows dynamic e
|
|||
|
||||
```
|
||||
mockapi/
|
||||
├── app.py # FastAPI application factory & lifespan
|
||||
├── config.py # Configuration (Pydantic Settings)
|
||||
├── database.py # SQLAlchemy async database setup
|
||||
├── dependencies.py # FastAPI dependencies
|
||||
├── example_usage.py # Integration test & demonstration script
|
||||
├── middleware/
|
||||
│ └── auth_middleware.py # Admin authentication middleware
|
||||
├── models/
|
||||
│ ├── endpoint_model.py # Endpoint SQLAlchemy model
|
||||
│ └── oauth_models.py # OAuth2 client, token, and user models
|
||||
├── observers/
|
||||
│ └── __init__.py # Observer pattern placeholder
|
||||
├── repositories/
|
||||
│ ├── endpoint_repository.py # Repository pattern for endpoints
|
||||
│ └── oauth2/ # OAuth2 repositories
|
||||
├── run.py # Development runner script (with auto-reload)
|
||||
├── services/
|
||||
│ ├── route_service.py # Dynamic route registration/management
|
||||
│ └── template_service.py # Jinja2 template rendering
|
||||
├── controllers/
|
||||
│ ├── admin_controller.py # Admin UI routes
|
||||
│ └── oauth2/ # OAuth2 controllers and services
|
||||
├── schemas/
|
||||
│ ├── endpoint_schema.py # Pydantic schemas for validation
|
||||
│ └── oauth2/ # OAuth2 schemas
|
||||
├── templates/ # Jinja2 HTML templates
|
||||
│ ├── base.html # Base layout
|
||||
│ └── admin/
|
||||
│ ├── login.html # Login page
|
||||
│ ├── dashboard.html # Admin dashboard
|
||||
│ ├── endpoints.html # Endpoint list
|
||||
│ ├── endpoint_form.html # Create/edit endpoint
|
||||
│ └── oauth/ # OAuth2 management pages
|
||||
├── static/
|
||||
│ └── css/ # Static CSS (optional)
|
||||
├── main.py # Development entry point (uvicorn with reload)
|
||||
├── wsgi.py # Production WSGI entry point (Waitress/Gunicorn)
|
||||
├── app/ # Main application package
|
||||
│ ├── core/ # Core application setup
|
||||
│ │ ├── app.py # FastAPI application factory & lifespan
|
||||
│ │ ├── config.py # Configuration (Pydantic Settings)
|
||||
│ │ ├── database.py # SQLAlchemy async database setup
|
||||
│ │ ├── dependencies.py # FastAPI dependencies
|
||||
│ │ ├── middleware/ # Middleware (authentication, etc.)
|
||||
│ │ └── observers/ # Observer pattern placeholder
|
||||
│ ├── modules/ # Feature modules
|
||||
│ │ ├── admin/ # Admin UI controllers & templates
|
||||
│ │ ├── endpoints/ # Endpoint management (models, repositories, services, schemas)
|
||||
│ │ └── oauth2/ # OAuth2 provider (controllers, models, repositories, schemas)
|
||||
│ ├── static/ # Static assets (CSS, etc.)
|
||||
│ └── templates/ # Jinja2 HTML templates
|
||||
│ ├── base.html # Base layout
|
||||
│ └── admin/ # Admin interface templates
|
||||
├── requirements.txt # Python dependencies
|
||||
├── example.env # Example environment variables
|
||||
├── .env # Local environment variables (create from example.env)
|
||||
├── docs/ # Project documentation
|
||||
├── examples/ # API testing collections and examples
|
||||
├── tests/ # Test suite
|
||||
│ ├── test_admin.py # Admin authentication tests
|
||||
│ ├── test_endpoint_repository.py
|
||||
│ ├── test_route_manager_fix.py
|
||||
│ ├── test_oauth2_controller.py
|
||||
│ └── integration/ # Integration tests
|
||||
├── utils/ # Utility modules
|
||||
│ └── __init__.py
|
||||
├── requirements.txt # Python dependencies
|
||||
├── .env.example # Example environment variables
|
||||
├── .env # Local environment variables (create from .env.example)
|
||||
├── run_example.sh # Script to run the integration test
|
||||
├── LICENSE # MIT License
|
||||
└── README.md # This file
|
||||
```
|
||||
|
|
@ -99,7 +81,7 @@ mockapi/
|
|||
|
||||
4. **Configure environment variables**:
|
||||
```bash
|
||||
cp .env.example .env
|
||||
cp example.env .env
|
||||
# Edit .env with your settings
|
||||
```
|
||||
|
||||
|
|
@ -128,11 +110,11 @@ source venv/bin/activate # Linux/macOS
|
|||
Then run with auto-reload for development:
|
||||
|
||||
```bash
|
||||
# Using run.py (convenience script)
|
||||
python run.py
|
||||
# Using main.py (development entry point)
|
||||
python main.py
|
||||
|
||||
# Or directly with uvicorn
|
||||
uvicorn app:app --reload --host 0.0.0.0 --port 8000
|
||||
uvicorn main:app --reload --host 0.0.0.0 --port 8000
|
||||
```
|
||||
|
||||
### Production (with Waitress)
|
||||
|
|
@ -153,10 +135,10 @@ The server will start on `http://localhost:8000` (or your configured host/port).
|
|||
|
||||
```bash
|
||||
# Development (auto-reload)
|
||||
python run.py
|
||||
python main.py
|
||||
|
||||
# Or directly with uvicorn
|
||||
uvicorn app:app --reload --host 0.0.0.0 --port 8000
|
||||
uvicorn main:app --reload --host 0.0.0.0 --port 8000
|
||||
|
||||
# Production (with Waitress)
|
||||
waitress-serve --host=0.0.0.0 --port=8000 --threads=4 wsgi:wsgi_app
|
||||
|
|
|
|||
|
|
@ -1,98 +0,0 @@
|
|||
import logging
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
from fastapi import FastAPI, Request, status
|
||||
from starlette.middleware.sessions import SessionMiddleware
|
||||
from fastapi.responses import RedirectResponse
|
||||
from starlette.staticfiles import StaticFiles
|
||||
|
||||
from config import settings
|
||||
from database import init_db, AsyncSessionLocal
|
||||
from repositories.endpoint_repository import EndpointRepository
|
||||
from services.route_service import RouteManager
|
||||
from middleware.auth_middleware import AuthMiddleware
|
||||
from controllers.admin_controller import router as admin_router
|
||||
from oauth2 import oauth_router
|
||||
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG if settings.debug else logging.INFO,
|
||||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
"""
|
||||
Lifespan context manager for startup and shutdown events.
|
||||
"""
|
||||
# Startup
|
||||
logger.info("Initializing database...")
|
||||
await init_db()
|
||||
|
||||
# Use the route manager already attached to app.state
|
||||
route_manager = app.state.route_manager
|
||||
logger.info("Refreshing routes...")
|
||||
await route_manager.refresh_routes()
|
||||
|
||||
logger.info("Application startup complete.")
|
||||
yield
|
||||
# Shutdown
|
||||
logger.info("Application shutting down...")
|
||||
|
||||
|
||||
def create_app() -> FastAPI:
|
||||
"""
|
||||
Factory function to create and configure the FastAPI application.
|
||||
"""
|
||||
app = FastAPI(
|
||||
title=settings.title,
|
||||
version=settings.version,
|
||||
debug=settings.debug,
|
||||
lifespan=lifespan,
|
||||
)
|
||||
|
||||
# Attach route manager and session factory to app.state before any request
|
||||
route_manager = RouteManager(app, AsyncSessionLocal)
|
||||
app.state.route_manager = route_manager
|
||||
app.state.session_factory = AsyncSessionLocal
|
||||
|
||||
# Add authentication middleware for admin routes (must be after SessionMiddleware)
|
||||
app.add_middleware(AuthMiddleware)
|
||||
# Add session middleware (must be before AuthMiddleware, but add_middleware prepends)
|
||||
app.add_middleware(
|
||||
SessionMiddleware,
|
||||
secret_key=settings.secret_key,
|
||||
session_cookie=settings.session_cookie_name,
|
||||
max_age=settings.session_max_age,
|
||||
https_only=False,
|
||||
same_site="lax",
|
||||
)
|
||||
|
||||
|
||||
|
||||
# Mount static files (optional, for future)
|
||||
# app.mount("/static", StaticFiles(directory="app/static"), name="static")
|
||||
|
||||
# Add a simple health check endpoint
|
||||
@app.get("/health")
|
||||
async def health_check():
|
||||
return {"status": "healthy", "service": "mock-api"}
|
||||
|
||||
# Redirect root to Swagger documentation
|
||||
@app.get("/")
|
||||
async def root_redirect():
|
||||
"""Redirect the root URL to Swagger documentation."""
|
||||
return RedirectResponse(url="/docs", status_code=status.HTTP_302_FOUND)
|
||||
|
||||
# Include admin controller routes
|
||||
app.include_router(admin_router)
|
||||
# Include OAuth2 routes
|
||||
app.include_router(oauth_router)
|
||||
|
||||
return app
|
||||
|
||||
|
||||
# Create the application instance
|
||||
app = create_app()
|
||||
72
app/modules/admin/templates/base.html
Normal file
72
app/modules/admin/templates/base.html
Normal file
|
|
@ -0,0 +1,72 @@
|
|||
<!DOCTYPE html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="UTF-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||||
<title>{% block title %}Mock API Admin{% endblock %}</title>
|
||||
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css" rel="stylesheet" integrity="sha384-9ndCyUaIbzAi2FUVXJi0CjmCapSmO7SnpJef0486qhLnuZ2cdeRhO02iuK6FUUVM" crossorigin="anonymous">
|
||||
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/bootstrap-icons@1.10.0/font/bootstrap-icons.css">
|
||||
<style>
|
||||
.sidebar {
|
||||
min-height: calc(100vh - 56px);
|
||||
background-color: #f8f9fa;
|
||||
}
|
||||
.content-header {
|
||||
border-bottom: 1px solid #dee2e6;
|
||||
padding-bottom: 0.5rem;
|
||||
margin-bottom: 1rem;
|
||||
}
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<nav class="navbar navbar-expand-lg navbar-dark bg-dark">
|
||||
<div class="container-fluid">
|
||||
<a class="navbar-brand" href="/admin">
|
||||
<i class="bi bi-gear"></i> Mock API Admin
|
||||
</a>
|
||||
<button class="navbar-toggler" type="button" data-bs-toggle="collapse" data-bs-target="#navbarNav">
|
||||
<span class="navbar-toggler-icon"></span>
|
||||
</button>
|
||||
<div class="collapse navbar-collapse" id="navbarNav">
|
||||
<ul class="navbar-nav me-auto">
|
||||
<li class="nav-item">
|
||||
<a class="nav-link {% if request.url.path == '/admin' %}active{% endif %}" href="/admin">
|
||||
<i class="bi bi-speedometer2"></i> Dashboard
|
||||
</a>
|
||||
</li>
|
||||
<li class="nav-item">
|
||||
<a class="nav-link {% if request.url.path == '/admin/endpoints' %}active{% endif %}" href="/admin/endpoints">
|
||||
<i class="bi bi-list-ul"></i> Endpoints
|
||||
</a>
|
||||
</li>
|
||||
</ul>
|
||||
<ul class="navbar-nav">
|
||||
<li class="nav-item">
|
||||
<span class="navbar-text text-light me-3">
|
||||
<i class="bi bi-person-circle"></i> {{ session.get('username', 'Guest') }}
|
||||
</span>
|
||||
</li>
|
||||
<li class="nav-item">
|
||||
<a class="nav-link" href="/admin/logout">
|
||||
<i class="bi bi-box-arrow-right"></i> Logout
|
||||
</a>
|
||||
</li>
|
||||
</ul>
|
||||
</div>
|
||||
</div>
|
||||
</nav>
|
||||
|
||||
<div class="container-fluid">
|
||||
<div class="row">
|
||||
<main class="col-md-12">
|
||||
<div class="container mt-4">
|
||||
{% block content %}{% endblock %}
|
||||
</div>
|
||||
</main>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/js/bootstrap.bundle.min.js" integrity="sha384-geWF76RCwLtnZ8qwWowPQNguL3RmwHVBC9FhGdlKrxdiJJigb/j/68SIy3Te4Bkz" crossorigin="anonymous"></script>
|
||||
{% block extra_scripts %}{% endblock %}
|
||||
</body>
|
||||
</html>
|
||||
|
|
@ -1,173 +0,0 @@
|
|||
# API Testing Collections for MockAPI
|
||||
|
||||
This directory contains ready-to-use API collections for testing the MockAPI application with OAuth2 provider.
|
||||
|
||||
## Available Collections
|
||||
|
||||
### 1. Bruno Collection (`mockapi-collection.bru`)
|
||||
- **Format**: Bruno native format (`.bru`)
|
||||
- **Features**: Scripting support, environment variables, folder organization
|
||||
- **Import**: Drag and drop into Bruno or use "Import Collection"
|
||||
|
||||
### 2. Postman Collection (`mockapi-postman-collection.json`)
|
||||
- **Format**: Postman Collection v2.1
|
||||
- **Features**: Pre-request scripts, tests, environment variables
|
||||
- **Import**: Import into Postman via "Import" button
|
||||
|
||||
### 3. Setup Scripts
|
||||
- `setup-test-client.py`: Creates test OAuth client in database
|
||||
- `setup.sh`: Interactive setup helper
|
||||
|
||||
## Collection Features
|
||||
|
||||
Both collections include:
|
||||
|
||||
### Global Variables
|
||||
- `baseUrl`: Base URL of your MockAPI instance (default: `http://localhost:8000`)
|
||||
- `adminUsername`, `adminPassword`: Admin credentials (default: `admin`/`admin123`)
|
||||
- `clientId`, `clientSecret`: OAuth2 client credentials (use `test_client`/`test_secret` after setup)
|
||||
- `accessToken`, `refreshToken`, `authCode`: OAuth2 tokens (auto-populated)
|
||||
- `endpointId`: ID of created mock endpoints (auto-populated)
|
||||
|
||||
### Request Folders
|
||||
|
||||
1. **Health Check** - Basic health endpoint
|
||||
2. **Admin - Login** - Admin authentication (sets session cookie)
|
||||
3. **Mock Endpoints** - CRUD operations for mock endpoints
|
||||
- Create, list, update, delete endpoints
|
||||
- Call mock endpoints with template variables
|
||||
4. **OAuth2** - Full OAuth2 flow testing
|
||||
- Client credentials grant
|
||||
- Authorization code grant (2-step)
|
||||
- Refresh token grant
|
||||
- Token introspection and revocation
|
||||
- OpenID Connect discovery
|
||||
5. **Admin OAuth Management** - OAuth2 admin UI endpoints
|
||||
6. **Protected Endpoint Example** - Create and test OAuth-protected endpoints
|
||||
|
||||
## Quick Setup
|
||||
|
||||
### 1. Start MockAPI Server
|
||||
```bash
|
||||
# Development (auto-reload)
|
||||
python run.py
|
||||
|
||||
# Or production
|
||||
waitress-serve --host=0.0.0.0 --port=8000 wsgi:wsgi_app
|
||||
```
|
||||
|
||||
### 2. Create Test OAuth Client
|
||||
```bash
|
||||
# Run setup script
|
||||
./examples/setup.sh
|
||||
|
||||
# Or directly
|
||||
python examples/setup-test-client.py
|
||||
```
|
||||
|
||||
This creates an OAuth client with:
|
||||
- **Client ID**: `test_client`
|
||||
- **Client Secret**: `test_secret`
|
||||
- **Grant Types**: `authorization_code`, `client_credentials`, `refresh_token`
|
||||
- **Scopes**: `openid`, `profile`, `email`, `api:read`, `api:write`
|
||||
|
||||
### 3. Import Collection
|
||||
- **Bruno**: Import `mockapi-collection.bru`
|
||||
- **Postman**: Import `mockapi-postman-collection.json`
|
||||
|
||||
### 4. Update Variables (if needed)
|
||||
- Update `baseUrl` if server runs on different host/port
|
||||
- Use `test_client`/`test_secret` for OAuth2 testing
|
||||
|
||||
## Testing Workflow
|
||||
|
||||
### Basic Testing
|
||||
1. Run **Health Check** to verify server is running
|
||||
2. Run **Admin - Login** to authenticate (sets session cookie)
|
||||
3. Use **Mock Endpoints** folder to create and test endpoints
|
||||
|
||||
### OAuth2 Testing
|
||||
1. Ensure test client is created (`test_client`/`test_secret`)
|
||||
2. Run **Client Credentials Grant** to get access token
|
||||
3. Token is automatically saved to `accessToken` variable
|
||||
4. Use token in protected endpoint requests
|
||||
|
||||
### Protected Endpoints
|
||||
1. Create protected endpoint using **Create OAuth-Protected Endpoint**
|
||||
2. Test unauthorized access (should fail with 401/403)
|
||||
3. Test authorized access with saved token
|
||||
|
||||
## Collection-Specific Features
|
||||
|
||||
### Bruno Features
|
||||
- **Scripting**: JavaScript scripts for request/response handling
|
||||
- **Variables**: `{{variable}}` syntax in URLs, headers, body
|
||||
- **`btoa()` function**: Built-in for Basic auth encoding
|
||||
- **Console logs**: Script output in Bruno console
|
||||
|
||||
### Postman Features
|
||||
- **Pre-request scripts**: Setup before requests
|
||||
- **Tests**: JavaScript tests after responses
|
||||
- **Environment variables**: Separate from collection variables
|
||||
- **Basic auth UI**: Built-in authentication helpers
|
||||
|
||||
## Manual Testing with cURL
|
||||
|
||||
For quick manual testing without collections:
|
||||
|
||||
```bash
|
||||
# Health check
|
||||
curl http://localhost:8000/health
|
||||
|
||||
# Create mock endpoint (after admin login)
|
||||
curl -c cookies.txt -X POST http://localhost:8000/admin/login \
|
||||
-H "Content-Type: application/x-www-form-urlencoded" \
|
||||
-d "username=admin&password=admin123"
|
||||
|
||||
curl -b cookies.txt -X POST http://localhost:8000/admin/endpoints \
|
||||
-H "Content-Type: application/x-www-form-urlencoded" \
|
||||
-d "route=/api/test&method=GET&response_body={\"message\":\"test\"}&response_code=200&content_type=application/json&is_active=true"
|
||||
|
||||
# OAuth2 client credentials grant
|
||||
curl -X POST http://localhost:8000/oauth/token \
|
||||
-H "Content-Type: application/x-www-form-urlencoded" \
|
||||
-d "grant_type=client_credentials&client_id=test_client&client_secret=test_secret&scope=api:read"
|
||||
```
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Common Issues
|
||||
|
||||
1. **401 Unauthorized (Admin endpoints)**
|
||||
- Run **Admin - Login** first
|
||||
- Check session cookies are being sent
|
||||
|
||||
2. **403 Forbidden (OAuth2)**
|
||||
- Verify token has required scopes
|
||||
- Check endpoint `oauth_scopes` configuration
|
||||
|
||||
3. **404 Not Found**
|
||||
- Endpoint may not be active (`is_active=true`)
|
||||
- Route may not be registered (refresh routes)
|
||||
|
||||
4. **Invalid OAuth Client**
|
||||
- Run setup script to create test client
|
||||
- Update `clientId`/`clientSecret` variables
|
||||
|
||||
5. **Bruno/Postman Import Errors**
|
||||
- Ensure JSON format is valid
|
||||
- Try re-downloading collection files
|
||||
|
||||
### Debug Tips
|
||||
|
||||
- Enable debug logging in MockAPI: Set `DEBUG=True` in `.env`
|
||||
- Check Bruno/Postman console for script output
|
||||
- Verify variables are set correctly before requests
|
||||
- Test endpoints directly in browser: `http://localhost:8000/docs`
|
||||
|
||||
## Resources
|
||||
|
||||
- [MockAPI Documentation](../README.md)
|
||||
- [Bruno Documentation](https://docs.usebruno.com/)
|
||||
- [Postman Documentation](https://learning.postman.com/docs/getting-started/introduction/)
|
||||
- [OAuth2 RFC 6749](https://tools.ietf.org/html/rfc6749)
|
||||
|
|
@ -1,51 +0,0 @@
|
|||
#!/bin/bash
|
||||
# Setup script for MockAPI API collections
|
||||
|
||||
echo "🔧 Setting up MockAPI API collections..."
|
||||
|
||||
# Check if Python is available
|
||||
if ! command -v python3 &> /dev/null; then
|
||||
echo "❌ Python 3 is not installed or not in PATH"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
# Check if MockAPI server is running (optional)
|
||||
echo "📡 Checking if MockAPI server is running..."
|
||||
if curl -s http://localhost:8000/health > /dev/null 2>&1; then
|
||||
echo "✅ MockAPI server is running"
|
||||
else
|
||||
echo "⚠️ MockAPI server may not be running on http://localhost:8000"
|
||||
echo " Start it with: python run.py"
|
||||
read -p "Continue anyway? (y/n) " -n 1 -r
|
||||
echo
|
||||
if [[ ! $REPLY =~ ^[Yy]$ ]]; then
|
||||
exit 1
|
||||
fi
|
||||
fi
|
||||
|
||||
# Run the test client setup script
|
||||
echo "🔄 Creating test OAuth client..."
|
||||
cd "$(dirname "$0")/.." # Go to project root
|
||||
python3 examples/setup-test-client.py
|
||||
|
||||
echo ""
|
||||
echo "✅ Setup complete!"
|
||||
echo ""
|
||||
echo "📋 Next steps:"
|
||||
echo "1. Choose your API client:"
|
||||
echo " - Bruno: Import 'examples/mockapi-collection.bru'"
|
||||
echo " - Postman: Import 'examples/mockapi-postman-collection.json'"
|
||||
echo "2. Update variables if needed:"
|
||||
echo " - baseUrl: URL of your MockAPI instance"
|
||||
echo " - clientId/clientSecret: Use the values printed above"
|
||||
echo "3. Start testing!"
|
||||
echo ""
|
||||
echo "📚 Collections include:"
|
||||
echo " - Health check"
|
||||
echo " - Admin authentication"
|
||||
echo " - Mock endpoint CRUD operations"
|
||||
echo " - OAuth2 flows (client credentials, auth code, refresh)"
|
||||
echo " - Token introspection and revocation"
|
||||
echo " - Protected endpoint testing"
|
||||
echo ""
|
||||
echo "💡 Tip: Run 'Admin - Login' first to authenticate for admin endpoints"
|
||||
42
main.py
42
main.py
|
|
@ -1,14 +1,11 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Unified entry point for the Mock API application.
|
||||
Development entry point for the Mock API application.
|
||||
|
||||
This module serves as both the development entry point (via uvicorn) and the
|
||||
production WSGI entry point (via Waitress or other WSGI servers).
|
||||
For production WSGI deployment, use wsgi.py instead.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from a2wsgi import ASGIMiddleware
|
||||
from app.core.app import create_app
|
||||
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
|
|
@ -17,45 +14,12 @@ logger = logging.getLogger(__name__)
|
|||
# Create the FastAPI application instance
|
||||
app = create_app()
|
||||
|
||||
|
||||
def create_wsgi_app():
|
||||
"""
|
||||
Create a WSGI application with route refresh on startup.
|
||||
|
||||
This function is intended for production WSGI servers (e.g., Waitress).
|
||||
Since WSGI does not support ASGI lifespan events, we manually refresh
|
||||
routes from the database once when the WSGI app is created.
|
||||
"""
|
||||
loop = None
|
||||
try:
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
route_manager = app.state.route_manager
|
||||
logger.info("Refreshing routes from database (WSGI startup)...")
|
||||
loop.run_until_complete(route_manager.refresh_routes())
|
||||
logger.info(f"Registered {len(route_manager.registered_routes)} routes")
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to refresh routes on startup: {e}")
|
||||
# Continue anyway; routes can be refreshed later via admin interface
|
||||
finally:
|
||||
if loop is not None:
|
||||
loop.close()
|
||||
|
||||
# Wrap FastAPI ASGI app with WSGI adapter
|
||||
wsgi_app = ASGIMiddleware(app)
|
||||
return wsgi_app
|
||||
|
||||
|
||||
# Expose the WSGI application for production servers
|
||||
wsgi_app = create_wsgi_app()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Development entry point: run uvicorn with auto‑reload
|
||||
import uvicorn
|
||||
logger.info("Starting development server on http://0.0.0.0:8000")
|
||||
uvicorn.run(
|
||||
app,
|
||||
"main:app", # Use import string for reload support
|
||||
host="0.0.0.0",
|
||||
port=8000,
|
||||
reload=True,
|
||||
|
|
|
|||
20
run_dev.py
Normal file
20
run_dev.py
Normal file
|
|
@ -0,0 +1,20 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Alternative development runner that avoids the WSGI initialization.
|
||||
"""
|
||||
import uvicorn
|
||||
import logging
|
||||
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
if __name__ == "__main__":
|
||||
logger.info("Starting development server on http://0.0.0.0:8000")
|
||||
# Import and run the app directly without WSGI initialization
|
||||
uvicorn.run(
|
||||
"app.core.app:app", # Import string for reload support
|
||||
host="0.0.0.0",
|
||||
port=8000,
|
||||
reload=True,
|
||||
log_level="info"
|
||||
)
|
||||
|
|
@ -1,22 +0,0 @@
|
|||
import inspect
|
||||
from asgiref.wsgi import WsgiToAsgi
|
||||
from app.core.app import app
|
||||
|
||||
print("app callable?", callable(app))
|
||||
print("app signature:", inspect.signature(app.__call__))
|
||||
wrapper = WsgiToAsgi(app)
|
||||
print("wrapper callable?", callable(wrapper))
|
||||
print("wrapper signature:", inspect.signature(wrapper.__call__))
|
||||
print("wrapper.__class__:", wrapper.__class__)
|
||||
print("wrapper.__class__.__module__:", wrapper.__class__.__module__)
|
||||
# Try to call with dummy environ/start_response
|
||||
def start_response(status, headers):
|
||||
pass
|
||||
environ = {'REQUEST_METHOD': 'GET', 'PATH_INFO': '/'}
|
||||
try:
|
||||
result = wrapper(environ, start_response)
|
||||
print("Success! Result:", result)
|
||||
except Exception as e:
|
||||
print("Error:", e)
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
|
@ -1,145 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Example integration test for the Configurable Mock API.
|
||||
|
||||
This script demonstrates the core functionality:
|
||||
1. Starting the FastAPI app (via TestClient)
|
||||
2. Admin login
|
||||
3. Creating a mock endpoint
|
||||
4. Calling the endpoint and verifying response
|
||||
5. Cleaning up (deleting the endpoint)
|
||||
|
||||
Run with: python example_usage_fixed.py
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import sys
|
||||
import os
|
||||
import json
|
||||
import logging
|
||||
|
||||
# Suppress debug logs for cleaner output
|
||||
logging.basicConfig(level=logging.WARNING)
|
||||
logging.getLogger('sqlalchemy.engine').setLevel(logging.WARNING)
|
||||
logging.getLogger('aiosqlite').setLevel(logging.WARNING)
|
||||
logging.getLogger('httpx').setLevel(logging.WARNING)
|
||||
logging.getLogger('asyncio').setLevel(logging.WARNING)
|
||||
|
||||
# Set environment variables for testing
|
||||
os.environ['DEBUG'] = 'True'
|
||||
os.environ['ADMIN_PASSWORD'] = 'admin123'
|
||||
os.environ['SECRET_KEY'] = 'test-secret-key'
|
||||
|
||||
# Add current directory to path
|
||||
sys.path.insert(0, '.')
|
||||
|
||||
from app.core.app import app
|
||||
from fastapi.testclient import TestClient
|
||||
from app.core.database import init_db
|
||||
|
||||
|
||||
async def setup_database():
|
||||
"""Initialize database tables."""
|
||||
print(" Initializing database...")
|
||||
await init_db()
|
||||
print(" Database initialized")
|
||||
|
||||
|
||||
def main():
|
||||
"""Run the integration test."""
|
||||
print("🚀 Starting Configurable Mock API integration test")
|
||||
print("=" * 60)
|
||||
|
||||
# Initialize database first
|
||||
asyncio.run(setup_database())
|
||||
|
||||
# Create test client
|
||||
client = TestClient(app)
|
||||
|
||||
# 1. Health check
|
||||
print("\n1. Testing health endpoint...")
|
||||
resp = client.get("/health")
|
||||
print(f" Health status: {resp.status_code}")
|
||||
print(f" Response: {resp.json()}")
|
||||
|
||||
if resp.status_code != 200:
|
||||
print(" ❌ Health check failed")
|
||||
return
|
||||
|
||||
print(" ✅ Health check passed")
|
||||
|
||||
# 2. Admin login
|
||||
print("\n2. Admin login...")
|
||||
resp = client.post("/admin/login", data={"username": "admin", "password": "admin123"}, follow_redirects=False)
|
||||
print(f" Login status: {resp.status_code}")
|
||||
print(f" Redirect location: {resp.headers.get('location')}")
|
||||
|
||||
if resp.status_code != 302:
|
||||
print(" ❌ Login failed")
|
||||
return
|
||||
|
||||
# Check session cookie
|
||||
cookies = resp.cookies
|
||||
if "mockapi_session" not in cookies:
|
||||
print(" ❌ Session cookie not set")
|
||||
return
|
||||
|
||||
print(" ✅ Session cookie set")
|
||||
|
||||
# 3. Create a mock endpoint
|
||||
print("\n3. Creating a mock endpoint...")
|
||||
endpoint_data = {
|
||||
"route": "/api/greeting/{name}",
|
||||
"method": "GET",
|
||||
"response_body": '{"message": "Hello, {{ name }}!", "server": "{{ server }}"}',
|
||||
"response_code": 200,
|
||||
"content_type": "application/json",
|
||||
"is_active": True,
|
||||
"variables": '{"server": "mock-api"}',
|
||||
"headers": '{"X-Custom-Header": "test"}',
|
||||
"delay_ms": 0,
|
||||
}
|
||||
|
||||
resp = client.post("/admin/endpoints", data=endpoint_data, follow_redirects=False)
|
||||
print(f" Create endpoint status: {resp.status_code}")
|
||||
|
||||
if resp.status_code != 302:
|
||||
print(f" ❌ Endpoint creation failed: {resp.text}")
|
||||
return
|
||||
|
||||
print(" ✅ Endpoint created (route registered)")
|
||||
|
||||
# 4. Call the mock endpoint
|
||||
print("\n4. Calling the mock endpoint...")
|
||||
resp = client.get("/api/greeting/World")
|
||||
print(f" Mock endpoint status: {resp.status_code}")
|
||||
print(f" Response headers: {{k: v for k, v in resp.headers.items() if k.startswith('X-')}}")
|
||||
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
print(f" Response: {data}")
|
||||
if data.get("message") == "Hello, World!" and data.get("server") == "mock-api":
|
||||
print(" ✅ Mock endpoint works correctly with template variables!")
|
||||
else:
|
||||
print(" ❌ Unexpected response content")
|
||||
else:
|
||||
print(f" ❌ Mock endpoint failed: {resp.text}")
|
||||
|
||||
# 5. Clean up (optional - delete the endpoint)
|
||||
print("\n5. Cleaning up...")
|
||||
# Get endpoint ID from the list page
|
||||
resp = client.get("/admin/endpoints")
|
||||
if resp.status_code == 200:
|
||||
# In a real scenario, we'd parse the HTML to find the ID
|
||||
# For this example, we'll just note that cleanup would happen here
|
||||
print(" (Endpoint cleanup would happen here in a full implementation)")
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
print("🎉 Integration test completed successfully!")
|
||||
print("\nTo test manually:")
|
||||
print("1. Start the server: uvicorn app:app --reload --host 0.0.0.0 --port 8000")
|
||||
print("2. Visit http://localhost:8000/admin/login (admin/admin123)")
|
||||
print("3. Create endpoints and test them at http://localhost:8000/api/...")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -1,22 +0,0 @@
|
|||
#!/bin/bash
|
||||
# Simple script to run the example integration test
|
||||
|
||||
echo "Running integration test for Configurable Mock API..."
|
||||
echo ""
|
||||
|
||||
# Activate virtual environment if exists
|
||||
if [ -d "venv" ]; then
|
||||
echo "Activating virtual environment..."
|
||||
source venv/bin/activate
|
||||
fi
|
||||
|
||||
# Run the example script
|
||||
python example_usage.py
|
||||
|
||||
# Deactivate virtual environment if activated
|
||||
if [ -d "venv" ] && [ "$VIRTUAL_ENV" != "" ]; then
|
||||
deactivate
|
||||
fi
|
||||
|
||||
echo ""
|
||||
echo "Done."
|
||||
|
|
@ -1,121 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Script to create a test OAuth client for Bruno collection testing.
|
||||
Run this script after starting the MockAPI server.
|
||||
"""
|
||||
import asyncio
|
||||
import sys
|
||||
import os
|
||||
import json
|
||||
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy import select, text
|
||||
|
||||
# Add parent directory to path
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
# Import only what we need
|
||||
try:
|
||||
from app.core.middleware.auth_middleware import get_password_hash
|
||||
HAS_AUTH = True
|
||||
except ImportError:
|
||||
# Fallback to simple hash function if middleware not available
|
||||
import hashlib
|
||||
import secrets
|
||||
def get_password_hash(password: str) -> str:
|
||||
# Simple hash for demo purposes (not production-ready)
|
||||
salt = secrets.token_hex(8)
|
||||
return f"{salt}${hashlib.sha256((salt + password).encode()).hexdigest()}"
|
||||
HAS_AUTH = False
|
||||
|
||||
async def create_test_client():
|
||||
"""Create a test OAuth client in the database."""
|
||||
# Use the same database URL as the app
|
||||
database_url = os.getenv("DATABASE_URL", "sqlite+aiosqlite:///./mockapi.db")
|
||||
|
||||
engine = create_async_engine(database_url, echo=False)
|
||||
|
||||
# Create session
|
||||
async_session = sessionmaker(
|
||||
engine, class_=AsyncSession, expire_on_commit=False
|
||||
)
|
||||
|
||||
async with async_session() as session:
|
||||
# Check if test client already exists
|
||||
result = await session.execute(
|
||||
text("SELECT * FROM oauth_clients WHERE client_id = :client_id"),
|
||||
{"client_id": "test_client"}
|
||||
)
|
||||
existing = result.fetchone()
|
||||
|
||||
if existing:
|
||||
print("Test client already exists:")
|
||||
print(f" Client ID: {existing.client_id}")
|
||||
print(f" Client Secret: (hashed, original secret was 'test_secret')")
|
||||
print(f" Name: {existing.name}")
|
||||
print(f" Grant Types: {json.loads(existing.grant_types) if existing.grant_types else []}")
|
||||
print(f" Scopes: {json.loads(existing.scopes) if existing.scopes else []}")
|
||||
return
|
||||
|
||||
# Create new test client
|
||||
client_secret_plain = "test_secret"
|
||||
client_secret_hash = get_password_hash(client_secret_plain)
|
||||
|
||||
# Insert directly using SQL to avoid model import issues
|
||||
await session.execute(
|
||||
text("""
|
||||
INSERT INTO oauth_clients
|
||||
(client_id, client_secret, name, redirect_uris, grant_types, scopes, is_active)
|
||||
VALUES
|
||||
(:client_id, :client_secret, :name, :redirect_uris, :grant_types, :scopes, :is_active)
|
||||
"""),
|
||||
{
|
||||
"client_id": "test_client",
|
||||
"client_secret": client_secret_hash,
|
||||
"name": "Test Client for API Collections",
|
||||
"redirect_uris": json.dumps(["http://localhost:8080/callback"]),
|
||||
"grant_types": json.dumps(["authorization_code", "client_credentials", "refresh_token"]),
|
||||
"scopes": json.dumps(["openid", "profile", "email", "api:read", "api:write"]),
|
||||
"is_active": 1 # SQLite uses 1 for true
|
||||
}
|
||||
)
|
||||
|
||||
await session.commit()
|
||||
|
||||
print("✅ Test OAuth client created successfully!")
|
||||
print()
|
||||
print("Client Details:")
|
||||
print(f" Client ID: test_client")
|
||||
print(f" Client Secret: {client_secret_plain}")
|
||||
print(f" Name: Test Client for API Collections")
|
||||
print(f" Redirect URIs: ['http://localhost:8080/callback']")
|
||||
print(f" Grant Types: ['authorization_code', 'client_credentials', 'refresh_token']")
|
||||
print(f" Scopes: ['openid', 'profile', 'email', 'api:read', 'api:write']")
|
||||
print()
|
||||
print("Update API client variables:")
|
||||
print(" - Set 'clientId' to 'test_client'")
|
||||
print(" - Set 'clientSecret' to 'test_secret'")
|
||||
print()
|
||||
print("Or update the collection file variables directly (.bru for Bruno, .json for Postman).")
|
||||
|
||||
# Get base URL from environment or use default
|
||||
base_url = os.getenv("BASE_URL", "http://localhost:8000")
|
||||
|
||||
print("\nCURL Examples:")
|
||||
print("1. Client Credentials Grant:")
|
||||
print(f' curl -X POST {base_url}/oauth/token \\')
|
||||
print(' -H "Content-Type: application/x-www-form-urlencoded" \\')
|
||||
print(' -d "grant_type=client_credentials&client_id=test_client&client_secret=test_secret&scope=api:read"')
|
||||
|
||||
print("\n2. Create a mock endpoint (after admin login):")
|
||||
print(' # First login (sets session cookie)')
|
||||
print(f' curl -c cookies.txt -X POST {base_url}/admin/login \\')
|
||||
print(' -H "Content-Type: application/x-www-form-urlencoded" \\')
|
||||
print(' -d "username=admin&password=admin123"')
|
||||
print(' # Then create endpoint')
|
||||
print(f' curl -b cookies.txt -X POST {base_url}/admin/endpoints \\')
|
||||
print(' -H "Content-Type: application/x-www-form-urlencoded" \\')
|
||||
print(' -d "route=/api/test&method=GET&response_body={\\"message\\":\\"test\\"}&response_code=200&content_type=application/json&is_active=true"')
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(create_test_client())
|
||||
54
test_run.py
Normal file
54
test_run.py
Normal file
|
|
@ -0,0 +1,54 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test if the application runs correctly.
|
||||
"""
|
||||
import sys
|
||||
import subprocess
|
||||
import time
|
||||
import requests
|
||||
|
||||
def test_app_start():
|
||||
"""Try to start the app and make a request."""
|
||||
print("Testing application startup...")
|
||||
|
||||
# Start the app in a subprocess
|
||||
proc = subprocess.Popen(
|
||||
[sys.executable, "-m", "uvicorn", "app.core.app:app", "--host", "0.0.0.0", "--port", "8001"],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True
|
||||
)
|
||||
|
||||
try:
|
||||
# Give it time to start
|
||||
time.sleep(3)
|
||||
|
||||
# Try to make a request
|
||||
print("Making test request...")
|
||||
try:
|
||||
resp = requests.get("http://localhost:8001/health", timeout=5)
|
||||
print(f"Response: {resp.status_code} - {resp.json()}")
|
||||
if resp.status_code == 200:
|
||||
print("✅ Application starts and responds correctly")
|
||||
return True
|
||||
else:
|
||||
print(f"❌ Unexpected status: {resp.status_code}")
|
||||
except requests.exceptions.ConnectionError:
|
||||
print("❌ Could not connect to server")
|
||||
|
||||
# Check process output
|
||||
stdout, stderr = proc.communicate(timeout=1)
|
||||
print(f"STDOUT: {stdout[:200]}")
|
||||
print(f"STDERR: {stderr[:200]}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error: {e}")
|
||||
finally:
|
||||
proc.terminate()
|
||||
proc.wait()
|
||||
|
||||
return False
|
||||
|
||||
if __name__ == "__main__":
|
||||
success = test_app_start()
|
||||
sys.exit(0 if success else 1)
|
||||
|
|
@ -1,128 +0,0 @@
|
|||
"""
|
||||
Pytest configuration and shared fixtures for integration tests.
|
||||
"""
|
||||
import asyncio
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
from fastapi.testclient import TestClient
|
||||
from app.core import database
|
||||
from app.core.config import settings
|
||||
from app.core.app import create_app
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="function")
|
||||
async def test_db():
|
||||
"""
|
||||
Create a fresh SQLite in-memory database for each test.
|
||||
Returns a tuple (engine, session_factory).
|
||||
"""
|
||||
# Create a new in-memory SQLite engine for this test with shared cache
|
||||
# Using cache=shared allows multiple connections to share the same in-memory database
|
||||
test_engine = create_async_engine(
|
||||
"sqlite+aiosqlite:///:memory:?cache=shared",
|
||||
echo=False,
|
||||
future=True,
|
||||
poolclass=StaticPool, # Use static pool to share in-memory DB across connections
|
||||
connect_args={"check_same_thread": False},
|
||||
)
|
||||
|
||||
# Create tables
|
||||
async with test_engine.begin() as conn:
|
||||
await conn.run_sync(database.Base.metadata.create_all)
|
||||
|
||||
# Create session factory
|
||||
test_session_factory = async_sessionmaker(
|
||||
test_engine,
|
||||
class_=AsyncSession,
|
||||
expire_on_commit=False,
|
||||
)
|
||||
|
||||
yield test_engine, test_session_factory
|
||||
|
||||
# Drop tables after test
|
||||
async with test_engine.begin() as conn:
|
||||
await conn.run_sync(database.Base.metadata.drop_all)
|
||||
|
||||
await test_engine.dispose()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="function")
|
||||
async def test_session(test_db):
|
||||
"""
|
||||
Provide an AsyncSession for database operations in tests.
|
||||
"""
|
||||
_, session_factory = test_db
|
||||
async with session_factory() as session:
|
||||
yield session
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="function")
|
||||
async def test_app(test_db):
|
||||
"""
|
||||
Provide a FastAPI app with a fresh in-memory database.
|
||||
Overrides the database engine and session factory in the app.
|
||||
"""
|
||||
test_engine, test_session_factory = test_db
|
||||
|
||||
# Monkey-patch the database module's engine and AsyncSessionLocal
|
||||
original_engine = database.engine
|
||||
original_session_factory = database.AsyncSessionLocal
|
||||
database.engine = test_engine
|
||||
database.AsyncSessionLocal = test_session_factory
|
||||
|
||||
# Also patch config.settings.database_url to prevent conflicts
|
||||
original_database_url = settings.database_url
|
||||
settings.database_url = "sqlite+aiosqlite:///:memory:?cache=shared"
|
||||
|
||||
# Create app with patched database
|
||||
app = create_app()
|
||||
|
||||
# Override get_db dependency to use our test session
|
||||
from app.core.database import get_db
|
||||
async def override_get_db():
|
||||
async with test_session_factory() as session:
|
||||
yield session
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
|
||||
# Ensure app.state.session_factory uses our test session factory
|
||||
app.state.session_factory = test_session_factory
|
||||
# Ensure route manager uses our test session factory
|
||||
app.state.route_manager.async_session_factory = test_session_factory
|
||||
|
||||
yield app
|
||||
|
||||
# Restore original values
|
||||
database.engine = original_engine
|
||||
database.AsyncSessionLocal = original_session_factory
|
||||
settings.database_url = original_database_url
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="function")
|
||||
async def test_client(test_app):
|
||||
"""
|
||||
Provide a TestClient with a fresh in-memory database.
|
||||
"""
|
||||
with TestClient(test_app) as client:
|
||||
yield client
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="function")
|
||||
async def admin_client(test_client):
|
||||
"""
|
||||
Provide a TestClient with an authenticated admin session.
|
||||
Logs in via POST /admin/login and returns the client with session cookie.
|
||||
"""
|
||||
client = test_client
|
||||
# Perform login
|
||||
response = client.post(
|
||||
"/admin/login",
|
||||
data={"username": "admin", "password": "admin123"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert response.status_code == 302
|
||||
# The session cookie should be set automatically
|
||||
yield client
|
||||
|
|
@ -1,369 +0,0 @@
|
|||
"""
|
||||
Comprehensive integration tests for OAuth2 flows and admin OAuth2 management.
|
||||
"""
|
||||
import pytest
|
||||
from urllib.parse import urlparse, parse_qs
|
||||
from fastapi import status
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.modules.oauth2.repositories import OAuthClientRepository, OAuthTokenRepository
|
||||
from app.modules.oauth2.services import OAuthService
|
||||
from app.modules.oauth2.models import OAuthClient
|
||||
from app.modules.endpoints.services.route_service import RouteManager
|
||||
from app.core.middleware.auth_middleware import get_password_hash
|
||||
from app.modules.endpoints.repositories.endpoint_repository import EndpointRepository
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_admin_oauth_client_creation_via_admin_interface(admin_client):
|
||||
"""
|
||||
Simulate admin login (set session cookie) and create an OAuth client via POST.
|
||||
Verify client is listed and client secret is not exposed after creation.
|
||||
"""
|
||||
client = admin_client
|
||||
|
||||
# Step 1: Navigate to new client form
|
||||
response = client.get("/admin/oauth/clients/new")
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert "Create" in response.text
|
||||
|
||||
# Step 2: Submit client creation form
|
||||
response = client.post(
|
||||
"/admin/oauth/clients",
|
||||
data={
|
||||
"client_name": "Test Integration Client",
|
||||
"redirect_uris": "http://localhost:8080/callback,https://example.com/cb",
|
||||
"grant_types": "authorization_code,client_credentials",
|
||||
"scopes": "api:read,api:write",
|
||||
"is_active": "true",
|
||||
},
|
||||
follow_redirects=False,
|
||||
)
|
||||
# Should redirect to list page
|
||||
assert response.status_code == status.HTTP_302_FOUND
|
||||
assert response.headers["location"] == "/admin/oauth/clients"
|
||||
|
||||
# Step 3: Verify client appears in list (no secret shown)
|
||||
response = client.get("/admin/oauth/clients")
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert "Test Integration Client" in response.text
|
||||
# Client secret should NOT be exposed in HTML
|
||||
assert "client_secret" not in response.text.lower()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_authorization_code_grant_flow(test_client, test_session):
|
||||
"""
|
||||
Complete authorization code grant flow with a real client.
|
||||
"""
|
||||
# Create an OAuth client with authorization_code grant type directly via repository
|
||||
from app.modules.oauth2.repositories import OAuthClientRepository
|
||||
repo = OAuthClientRepository(test_session)
|
||||
client_secret_plain = "test_secret_123"
|
||||
client_secret_hash = get_password_hash(client_secret_plain)
|
||||
client_data = {
|
||||
"client_id": "test_auth_code_client",
|
||||
"client_secret": client_secret_hash,
|
||||
"name": "Auth Code Test Client",
|
||||
"redirect_uris": ["http://localhost:8080/callback"],
|
||||
"grant_types": ["authorization_code"],
|
||||
"scopes": ["api:read", "openid"],
|
||||
"is_active": True,
|
||||
}
|
||||
oauth_client = await repo.create(client_data)
|
||||
assert oauth_client is not None
|
||||
await test_session.commit()
|
||||
|
||||
# Now start the authorization flow
|
||||
response = test_client.get(
|
||||
"/oauth/authorize",
|
||||
params={
|
||||
"response_type": "code",
|
||||
"client_id": "test_auth_code_client",
|
||||
"redirect_uri": "http://localhost:8080/callback",
|
||||
"scope": "api:read",
|
||||
"state": "xyz123",
|
||||
},
|
||||
follow_redirects=False,
|
||||
)
|
||||
# Should redirect with authorization code
|
||||
assert response.status_code == status.HTTP_302_FOUND
|
||||
location = response.headers["location"]
|
||||
assert location.startswith("http://localhost:8080/callback?")
|
||||
# Extract code from URL
|
||||
parsed = urlparse(location)
|
||||
query = parse_qs(parsed.query)
|
||||
assert "code" in query
|
||||
auth_code = query["code"][0]
|
||||
assert "state" in query
|
||||
assert query["state"][0] == "xyz123"
|
||||
|
||||
# Exchange code for tokens
|
||||
response = test_client.post(
|
||||
"/oauth/token",
|
||||
data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": auth_code,
|
||||
"redirect_uri": "http://localhost:8080/callback",
|
||||
"client_id": "test_auth_code_client",
|
||||
"client_secret": client_secret_plain,
|
||||
},
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
token_data = response.json()
|
||||
assert "access_token" in token_data
|
||||
assert "refresh_token" in token_data
|
||||
assert "expires_in" in token_data
|
||||
assert token_data["token_type"] == "Bearer"
|
||||
access_token = token_data["access_token"]
|
||||
refresh_token = token_data["refresh_token"]
|
||||
|
||||
# Verify token exists in database
|
||||
from app.modules.oauth2.repositories import OAuthTokenRepository
|
||||
token_repo = OAuthTokenRepository(test_session)
|
||||
token_record = await token_repo.get_by_access_token(access_token)
|
||||
assert token_record is not None
|
||||
|
||||
# Use access token to call GET /oauth/userinfo
|
||||
response = test_client.get(
|
||||
"/oauth/userinfo",
|
||||
headers={"Authorization": f"Bearer {access_token}"},
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
userinfo = response.json()
|
||||
assert "sub" in userinfo
|
||||
# sub is user_id placeholder (1)
|
||||
assert userinfo["sub"] == "1"
|
||||
assert "client_id" in userinfo
|
||||
|
||||
# Use refresh token to obtain new access token
|
||||
response = test_client.post(
|
||||
"/oauth/token",
|
||||
data={
|
||||
"grant_type": "refresh_token",
|
||||
"refresh_token": refresh_token,
|
||||
"client_id": "test_auth_code_client",
|
||||
"client_secret": client_secret_plain,
|
||||
},
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
new_token_data = response.json()
|
||||
assert "access_token" in new_token_data
|
||||
assert new_token_data["access_token"] != access_token
|
||||
|
||||
# Revoke token
|
||||
response = test_client.post(
|
||||
"/oauth/revoke",
|
||||
data={"token": access_token},
|
||||
auth=("test_auth_code_client", client_secret_plain),
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
# Verify revoked token cannot be used
|
||||
response = test_client.get(
|
||||
"/oauth/userinfo",
|
||||
headers={"Authorization": f"Bearer {access_token}"},
|
||||
)
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_client_credentials_grant_flow(test_client, test_session):
|
||||
"""
|
||||
Client credentials grant flow.
|
||||
"""
|
||||
# Create client with client_credentials grant type
|
||||
repo = OAuthClientRepository(test_session)
|
||||
client_secret_plain = "client_secret_456"
|
||||
client_secret_hash = get_password_hash(client_secret_plain)
|
||||
client_data = {
|
||||
"client_id": "test_client_credentials_client",
|
||||
"client_secret": client_secret_hash,
|
||||
"name": "Client Credentials Test",
|
||||
"redirect_uris": [],
|
||||
"grant_types": ["client_credentials"],
|
||||
"scopes": ["api:read", "api:write"],
|
||||
"is_active": True,
|
||||
}
|
||||
oauth_client = await repo.create(client_data)
|
||||
assert oauth_client is not None
|
||||
await test_session.commit()
|
||||
|
||||
# Obtain token via client credentials
|
||||
response = test_client.post(
|
||||
"/oauth/token",
|
||||
data={
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": "test_client_credentials_client",
|
||||
"client_secret": client_secret_plain,
|
||||
"scope": "api:read",
|
||||
},
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
token_data = response.json()
|
||||
assert "access_token" in token_data
|
||||
assert "token_type" in token_data
|
||||
assert token_data["token_type"] == "Bearer"
|
||||
assert "expires_in" in token_data
|
||||
# No refresh token for client credentials
|
||||
assert "refresh_token" not in token_data
|
||||
|
||||
# Use token to call userinfo (should work? client credentials token has no user)
|
||||
# Actually userinfo expects a token with sub (user). Might fail. Let's skip.
|
||||
# We'll test protected endpoint in another test.
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_protected_endpoint_integration(test_client, test_session, test_app):
|
||||
"""
|
||||
Create a mock endpoint with OAuth protection and test token access.
|
||||
"""
|
||||
# First, create a mock endpoint with requires_oauth=True and scopes
|
||||
endpoint_repo = EndpointRepository(test_session)
|
||||
endpoint_data = {
|
||||
"route": "/api/protected",
|
||||
"method": "GET",
|
||||
"response_body": '{"message": "protected"}',
|
||||
"response_code": 200,
|
||||
"content_type": "application/json",
|
||||
"is_active": True,
|
||||
"requires_oauth": True,
|
||||
"oauth_scopes": ["api:read"],
|
||||
}
|
||||
endpoint = await endpoint_repo.create(endpoint_data)
|
||||
assert endpoint is not None
|
||||
await test_session.commit()
|
||||
# Refresh routes to register the endpoint
|
||||
route_manager = test_app.state.route_manager
|
||||
await route_manager.refresh_routes()
|
||||
|
||||
# Create an OAuth client and token with scope api:read
|
||||
client_repo = OAuthClientRepository(test_session)
|
||||
client_secret_plain = "secret_protected"
|
||||
client_secret_hash = get_password_hash(client_secret_plain)
|
||||
client_data = {
|
||||
"client_id": "protected_client",
|
||||
"client_secret": client_secret_hash,
|
||||
"name": "Protected Endpoint Client",
|
||||
"redirect_uris": ["http://localhost:8080/callback"],
|
||||
"grant_types": ["client_credentials"],
|
||||
"scopes": ["api:read", "api:write"],
|
||||
"is_active": True,
|
||||
}
|
||||
oauth_client = await client_repo.create(client_data)
|
||||
assert oauth_client is not None
|
||||
await test_session.commit()
|
||||
|
||||
# Obtain token with scope api:read
|
||||
response = test_client.post(
|
||||
"/oauth/token",
|
||||
data={
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": "protected_client",
|
||||
"client_secret": client_secret_plain,
|
||||
"scope": "api:read",
|
||||
},
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
token = response.json()["access_token"]
|
||||
|
||||
# Use token to call protected endpoint (should succeed)
|
||||
response = test_client.get(
|
||||
"/api/protected",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["message"] == "protected"
|
||||
|
||||
# Try token without required scope (api:write token, but endpoint requires api:read)
|
||||
response = test_client.post(
|
||||
"/oauth/token",
|
||||
data={
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": "protected_client",
|
||||
"client_secret": client_secret_plain,
|
||||
"scope": "api:write",
|
||||
},
|
||||
)
|
||||
token_write = response.json()["access_token"]
|
||||
response = test_client.get(
|
||||
"/api/protected",
|
||||
headers={"Authorization": f"Bearer {token_write}"},
|
||||
)
|
||||
# Should fail with 403 because missing required scope
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
# Use expired or invalid token (should fail with 401)
|
||||
response = test_client.get(
|
||||
"/api/protected",
|
||||
headers={"Authorization": "Bearer invalid_token"},
|
||||
)
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_admin_oauth_management_pages(test_client, admin_client, test_session):
|
||||
"""
|
||||
Test that admin OAuth pages require authentication, pagination, soft delete, token revocation.
|
||||
"""
|
||||
# 1. Test that /admin/oauth/clients requires authentication (redirect to login)
|
||||
# Create a fresh unauthenticated client (since test_client may be logged in via admin_client fixture)
|
||||
from fastapi.testclient import TestClient
|
||||
with TestClient(test_client.app) as unauth_client:
|
||||
response = unauth_client.get("/admin/oauth/clients", follow_redirects=False)
|
||||
assert response.status_code == status.HTTP_302_FOUND
|
||||
assert response.headers["location"] == "/admin/login"
|
||||
|
||||
# 2. Authenticated admin can access the page
|
||||
response = admin_client.get("/admin/oauth/clients")
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
# 3. Create a few clients to test pagination (we'll create via repository)
|
||||
repo = OAuthClientRepository(test_session)
|
||||
for i in range(25):
|
||||
client_secret_hash = get_password_hash(f"secret_{i}")
|
||||
client_data = {
|
||||
"client_id": f"client_{i}",
|
||||
"client_secret": client_secret_hash,
|
||||
"name": f"Client {i}",
|
||||
"redirect_uris": [],
|
||||
"grant_types": ["client_credentials"],
|
||||
"scopes": ["api:read"],
|
||||
"is_active": True,
|
||||
}
|
||||
await repo.create(client_data)
|
||||
await test_session.commit()
|
||||
|
||||
# First page should show clients
|
||||
response = admin_client.get("/admin/oauth/clients?page=1")
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
# Check that pagination controls appear (next page link)
|
||||
# We'll just assert that page 1 works
|
||||
|
||||
# 4. Test soft delete via admin interface (POST to delete endpoint)
|
||||
# Need a client ID (integer). Let's get the first client.
|
||||
clients = await repo.get_all(limit=1)
|
||||
assert len(clients) > 0
|
||||
client_id = clients[0].id # type: ignore
|
||||
|
||||
# Soft delete (is_active=False) via POST /admin/oauth/clients/{client_id}/delete
|
||||
response = admin_client.post(f"/admin/oauth/clients/{client_id}/delete", follow_redirects=False)
|
||||
assert response.status_code == status.HTTP_302_FOUND
|
||||
# Verify client is inactive
|
||||
# Expire the test session to ensure we get fresh data from database
|
||||
test_session.expire_all()
|
||||
client = await repo.get_by_id(client_id) # type: ignore
|
||||
assert client is not None
|
||||
assert client.is_active == False # type: ignore
|
||||
|
||||
# 5. Test token revocation via admin interface
|
||||
# Create a token first (we can create via service or directly via repository)
|
||||
# For simplicity, we'll skip token creation and just test that the revocation endpoint exists and requires auth.
|
||||
# The endpoint is POST /admin/oauth/tokens/{token_id}/revoke
|
||||
# We'll need a token id. Let's create a token via OAuthService using client credentials.
|
||||
# This is getting long; we can have a separate test for token revocation.
|
||||
# We'll leave as future enhancement.
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
|
|
@ -1,89 +0,0 @@
|
|||
"""
|
||||
Tests for admin interface authentication and endpoints.
|
||||
"""
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from app.core.app import app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client():
|
||||
"""Test client fixture."""
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
def test_admin_login_page(client):
|
||||
"""Login page should be accessible."""
|
||||
response = client.get("/admin/login")
|
||||
assert response.status_code == 200
|
||||
assert "Admin Login" in response.text
|
||||
|
||||
|
||||
def test_admin_dashboard_requires_auth(client):
|
||||
"""Dashboard should redirect to login if not authenticated."""
|
||||
response = client.get("/admin", follow_redirects=False)
|
||||
assert response.status_code == 302
|
||||
assert response.headers["location"] == "/admin/login"
|
||||
|
||||
|
||||
def test_admin_endpoints_requires_auth(client):
|
||||
"""Endpoints list should redirect to login if not authenticated."""
|
||||
response = client.get("/admin/endpoints", follow_redirects=False)
|
||||
assert response.status_code == 302
|
||||
assert response.headers["location"] == "/admin/login"
|
||||
|
||||
|
||||
def test_login_with_valid_credentials(client):
|
||||
"""Successful login should set session and redirect to dashboard."""
|
||||
response = client.post(
|
||||
"/admin/login",
|
||||
data={"username": "admin", "password": "admin123"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert response.status_code == 302
|
||||
assert response.headers["location"] == "/admin"
|
||||
# Check that session cookie is set
|
||||
assert "mockapi_session" in response.cookies
|
||||
|
||||
|
||||
def test_login_with_invalid_credentials(client):
|
||||
"""Invalid credentials should redirect back to login with error."""
|
||||
response = client.post(
|
||||
"/admin/login",
|
||||
data={"username": "admin", "password": "wrong"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert response.status_code == 302
|
||||
assert response.headers["location"] == "/admin/login?error=Invalid+credentials"
|
||||
# No session cookie
|
||||
assert "mockapi_session" not in response.cookies
|
||||
|
||||
|
||||
def test_authenticated_access(client):
|
||||
"""After login, admin routes should be accessible."""
|
||||
# First login
|
||||
login_response = client.post(
|
||||
"/admin/login",
|
||||
data={"username": "admin", "password": "admin123"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert login_response.status_code == 302
|
||||
# Now request dashboard
|
||||
dashboard_response = client.get("/admin")
|
||||
assert dashboard_response.status_code == 200
|
||||
assert "Dashboard" in dashboard_response.text
|
||||
|
||||
|
||||
def test_logout(client):
|
||||
"""Logout should clear session and redirect to login."""
|
||||
# Login first
|
||||
client.post("/admin/login", data={"username": "admin", "password": "admin123"}, follow_redirects=False)
|
||||
# Logout
|
||||
response = client.get("/admin/logout", follow_redirects=False)
|
||||
assert response.status_code == 302
|
||||
assert response.headers["location"] == "/admin/login"
|
||||
# Session cookie should be cleared (or empty)
|
||||
# Actually Starlette SessionMiddleware sets a new empty session
|
||||
# We'll just ensure we can't access dashboard after logout
|
||||
dashboard_response = client.get("/admin", follow_redirects=False)
|
||||
assert dashboard_response.status_code == 302
|
||||
|
|
@ -1,173 +0,0 @@
|
|||
"""
|
||||
Unit tests for AuthorizationCodeStore.
|
||||
"""
|
||||
import asyncio
|
||||
import pytest
|
||||
from datetime import datetime, timedelta
|
||||
from app.modules.oauth2.auth_code_store import AuthorizationCodeStore
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def store():
|
||||
"""Return a fresh AuthorizationCodeStore instance for each test."""
|
||||
return AuthorizationCodeStore(default_expiration=timedelta(seconds=1))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_store_and_retrieve_code(store):
|
||||
"""Store a code and retrieve it before expiration."""
|
||||
code = "test_code_123"
|
||||
data = {
|
||||
"client_id": "test_client",
|
||||
"redirect_uri": "https://example.com/callback",
|
||||
"scopes": ["read", "write"],
|
||||
"user_id": 42,
|
||||
}
|
||||
await store.store_code(code, data)
|
||||
|
||||
retrieved = await store.get_code(code)
|
||||
assert retrieved is not None
|
||||
assert retrieved["client_id"] == data["client_id"]
|
||||
assert retrieved["redirect_uri"] == data["redirect_uri"]
|
||||
assert retrieved["scopes"] == data["scopes"]
|
||||
assert retrieved["user_id"] == data["user_id"]
|
||||
assert "expires_at" in retrieved
|
||||
assert isinstance(retrieved["expires_at"], datetime)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_store_without_expires_at_gets_default(store):
|
||||
"""When expires_at is omitted, the store adds a default expiration."""
|
||||
code = "test_code_no_exp"
|
||||
data = {
|
||||
"client_id": "client1",
|
||||
"redirect_uri": "https://example.com/cb",
|
||||
"scopes": [],
|
||||
}
|
||||
await store.store_code(code, data)
|
||||
retrieved = await store.get_code(code)
|
||||
assert retrieved is not None
|
||||
assert "expires_at" in retrieved
|
||||
# Should be roughly now + default expiration (1 second in test fixture)
|
||||
# Allow small tolerance
|
||||
expected_min = datetime.utcnow() + timedelta(seconds=0.9)
|
||||
expected_max = datetime.utcnow() + timedelta(seconds=1.1)
|
||||
assert expected_min <= retrieved["expires_at"] <= expected_max
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_expired_code_returns_none_and_deletes(store):
|
||||
"""Expired codes are automatically removed on get_code."""
|
||||
code = "expired_code"
|
||||
data = {
|
||||
"client_id": "client",
|
||||
"redirect_uri": "https://example.com/cb",
|
||||
"scopes": [],
|
||||
"expires_at": datetime.utcnow() - timedelta(minutes=5), # already expired
|
||||
}
|
||||
await store.store_code(code, data)
|
||||
# Wait a tiny bit to ensure expiration
|
||||
await asyncio.sleep(0.01)
|
||||
retrieved = await store.get_code(code)
|
||||
assert retrieved is None
|
||||
# Ensure code is removed from store
|
||||
assert store.get_store_size() == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_code(store):
|
||||
"""Explicit deletion removes the code."""
|
||||
code = "to_delete"
|
||||
data = {
|
||||
"client_id": "client",
|
||||
"redirect_uri": "https://example.com/cb",
|
||||
"scopes": [],
|
||||
}
|
||||
await store.store_code(code, data)
|
||||
assert store.get_store_size() == 1
|
||||
await store.delete_code(code)
|
||||
assert store.get_store_size() == 0
|
||||
assert await store.get_code(code) is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_nonexistent_code_is_idempotent(store):
|
||||
"""Deleting a non‑existent code does not raise an error."""
|
||||
await store.delete_code("does_not_exist")
|
||||
# No exception raised
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_prune_expired(store):
|
||||
"""prune_expired removes all expired codes."""
|
||||
# Store one expired and one valid code
|
||||
expired_data = {
|
||||
"client_id": "client1",
|
||||
"redirect_uri": "https://example.com/cb",
|
||||
"scopes": [],
|
||||
"expires_at": datetime.utcnow() - timedelta(seconds=30),
|
||||
}
|
||||
valid_data = {
|
||||
"client_id": "client2",
|
||||
"redirect_uri": "https://example.com/cb",
|
||||
"scopes": [],
|
||||
"expires_at": datetime.utcnow() + timedelta(seconds=30),
|
||||
}
|
||||
await store.store_code("expired", expired_data)
|
||||
await store.store_code("valid", valid_data)
|
||||
assert store.get_store_size() == 2
|
||||
|
||||
removed = await store.prune_expired()
|
||||
assert removed == 1
|
||||
assert store.get_store_size() == 1
|
||||
assert await store.get_code("valid") is not None
|
||||
assert await store.get_code("expired") is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_missing_required_fields_raises_error(store):
|
||||
"""store_code raises ValueError if required fields are missing."""
|
||||
code = "bad_code"
|
||||
incomplete_data = {
|
||||
"client_id": "client",
|
||||
# missing redirect_uri and scopes
|
||||
}
|
||||
with pytest.raises(ValueError) as exc:
|
||||
await store.store_code(code, incomplete_data)
|
||||
assert "Missing required fields" in str(exc.value)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_thread_safety_simulation(store):
|
||||
"""Concurrent access should not raise exceptions (basic safety check)."""
|
||||
codes = [f"code_{i}" for i in range(10)]
|
||||
data = {
|
||||
"client_id": "client",
|
||||
"redirect_uri": "https://example.com/cb",
|
||||
"scopes": [],
|
||||
}
|
||||
# Store concurrently
|
||||
tasks = [store.store_code(code, data) for code in codes]
|
||||
await asyncio.gather(*tasks)
|
||||
assert store.get_store_size() == 10
|
||||
# Retrieve and delete concurrently
|
||||
tasks = [store.get_code(code) for code in codes]
|
||||
results = await asyncio.gather(*tasks)
|
||||
assert all(r is not None for r in results)
|
||||
tasks = [store.delete_code(code) for code in codes]
|
||||
await asyncio.gather(*tasks)
|
||||
assert store.get_store_size() == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_singleton_global_instance():
|
||||
"""The global instance authorization_code_store is a singleton."""
|
||||
from app.modules.oauth2.auth_code_store import authorization_code_store
|
||||
# Import again to ensure it's the same object
|
||||
from app.modules.oauth2.auth_code_store import authorization_code_store as same_instance
|
||||
assert authorization_code_store is same_instance
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Simple standalone test (can be run with python -m pytest)
|
||||
pytest.main([__file__, "-v"])
|
||||
|
|
@ -1,12 +0,0 @@
|
|||
"""
|
||||
Unit tests for EndpointRepository.
|
||||
"""
|
||||
import pytest
|
||||
|
||||
# TODO: Implement tests
|
||||
# from app.modules.endpoints.repositories.endpoint_repository import EndpointRepository
|
||||
|
||||
|
||||
def test_placeholder():
|
||||
"""Placeholder test to ensure test suite runs."""
|
||||
assert True
|
||||
|
|
@ -1,337 +0,0 @@
|
|||
"""
|
||||
Unit tests for OAuth2 controller endpoints.
|
||||
"""
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
from fastapi.testclient import TestClient
|
||||
from fastapi import FastAPI, status
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.modules.oauth2.controller import router as oauth_router
|
||||
|
||||
|
||||
def create_test_app(override_dependency=None) -> FastAPI:
|
||||
"""Create a FastAPI app with OAuth router and optional dependency overrides."""
|
||||
app = FastAPI()
|
||||
app.include_router(oauth_router)
|
||||
if override_dependency:
|
||||
app.dependency_overrides.update(override_dependency)
|
||||
return app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_db_session():
|
||||
"""Mock database session."""
|
||||
session = AsyncMock(spec=AsyncSession)
|
||||
return session
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(mock_db_session):
|
||||
"""Test client with mocked database session."""
|
||||
from app.core.database import get_db
|
||||
def override_get_db():
|
||||
yield mock_db_session
|
||||
app = create_test_app({get_db: override_get_db})
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
# ---------- Authorization Endpoint Tests ----------
|
||||
def test_authorize_missing_parameters(client):
|
||||
"""GET /oauth/authorize without required parameters should return error."""
|
||||
response = client.get("/oauth/authorize")
|
||||
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
|
||||
# FastAPI returns validation error details
|
||||
|
||||
|
||||
def test_authorize_unsupported_response_type(client):
|
||||
"""Only 'code' response_type is supported."""
|
||||
response = client.get(
|
||||
"/oauth/authorize",
|
||||
params={
|
||||
"response_type": "token", # unsupported
|
||||
"client_id": "test_client",
|
||||
"redirect_uri": "https://example.com/callback",
|
||||
}
|
||||
)
|
||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||
data = response.json()
|
||||
assert "detail" in data
|
||||
assert data["detail"]["error"] == "unsupported_response_type"
|
||||
|
||||
|
||||
def test_authorize_success(client, mock_db_session):
|
||||
"""Successful authorization returns redirect with code."""
|
||||
# Mock OAuthService.authorize_code_flow
|
||||
with patch('app.modules.oauth2.controller.OAuthService') as MockOAuthService:
|
||||
mock_service = AsyncMock()
|
||||
mock_service.authorize_code_flow.return_value = {
|
||||
"code": "auth_code_123",
|
||||
"state": "xyz",
|
||||
}
|
||||
MockOAuthService.return_value = mock_service
|
||||
response = client.get(
|
||||
"/oauth/authorize",
|
||||
params={
|
||||
"response_type": "code",
|
||||
"client_id": "test_client",
|
||||
"redirect_uri": "https://example.com/callback",
|
||||
"scope": "read write",
|
||||
"state": "xyz",
|
||||
},
|
||||
follow_redirects=False
|
||||
)
|
||||
assert response.status_code == status.HTTP_302_FOUND
|
||||
assert "location" in response.headers
|
||||
location = response.headers["location"]
|
||||
assert location.startswith("https://example.com/callback?")
|
||||
assert "code=auth_code_123" in location
|
||||
assert "state=xyz" in location
|
||||
# Verify service was called with correct parameters
|
||||
mock_service.authorize_code_flow.assert_called_once_with(
|
||||
client_id="test_client",
|
||||
redirect_uri="https://example.com/callback",
|
||||
scope=["read", "write"],
|
||||
state="xyz",
|
||||
user_id=1, # placeholder
|
||||
)
|
||||
|
||||
|
||||
# ---------- Token Endpoint Tests ----------
|
||||
def test_token_missing_grant_type(client):
|
||||
"""POST /oauth/token without grant_type should error (client auth required)."""
|
||||
response = client.post("/oauth/token", data={})
|
||||
# Client authentication missing -> 401 unauthorized
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
|
||||
def test_token_unsupported_grant_type(client):
|
||||
"""Unsupported grant_type returns error."""
|
||||
response = client.post(
|
||||
"/oauth/token",
|
||||
data={"grant_type": "password"}, # not supported
|
||||
auth=("test_client", "secret")
|
||||
)
|
||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||
data = response.json()
|
||||
assert "detail" in data
|
||||
assert data["detail"]["error"] == "unsupported_grant_type"
|
||||
|
||||
|
||||
def test_token_authorization_code_missing_code(client):
|
||||
"""authorization_code grant requires code."""
|
||||
response = client.post(
|
||||
"/oauth/token",
|
||||
data={
|
||||
"grant_type": "authorization_code",
|
||||
"client_id": "test_client",
|
||||
"client_secret": "secret",
|
||||
"redirect_uri": "https://example.com/callback",
|
||||
}
|
||||
)
|
||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||
data = response.json()
|
||||
assert "detail" in data
|
||||
assert data["detail"]["error"] == "invalid_request"
|
||||
|
||||
|
||||
def test_token_authorization_code_success(client, mock_db_session):
|
||||
"""Successful authorization_code exchange returns tokens."""
|
||||
with patch('app.modules.oauth2.controller.OAuthService') as MockOAuthService:
|
||||
mock_service = AsyncMock()
|
||||
mock_service.exchange_code_for_tokens.return_value = {
|
||||
"access_token": "access_token_123",
|
||||
"token_type": "Bearer",
|
||||
"expires_in": 1800,
|
||||
"refresh_token": "refresh_token_456",
|
||||
"scope": "read write",
|
||||
}
|
||||
MockOAuthService.return_value = mock_service
|
||||
|
||||
response = client.post(
|
||||
"/oauth/token",
|
||||
data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": "auth_code_xyz",
|
||||
"redirect_uri": "https://example.com/callback",
|
||||
"client_id": "test_client",
|
||||
"client_secret": "secret",
|
||||
}
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data["access_token"] == "access_token_123"
|
||||
assert data["token_type"] == "Bearer"
|
||||
assert data["refresh_token"] == "refresh_token_456"
|
||||
mock_service.exchange_code_for_tokens.assert_called_once_with(
|
||||
code="auth_code_xyz",
|
||||
client_id="test_client",
|
||||
redirect_uri="https://example.com/callback",
|
||||
)
|
||||
|
||||
|
||||
def test_token_client_credentials_success(client, mock_db_session):
|
||||
"""Client credentials grant returns access token."""
|
||||
with patch('app.modules.oauth2.controller.OAuthService') as MockOAuthService:
|
||||
mock_service = AsyncMock()
|
||||
mock_service.client_credentials_flow.return_value = {
|
||||
"access_token": "client_token",
|
||||
"token_type": "Bearer",
|
||||
"expires_in": 1800,
|
||||
"scope": "read",
|
||||
}
|
||||
MockOAuthService.return_value = mock_service
|
||||
|
||||
response = client.post(
|
||||
"/oauth/token",
|
||||
data={
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": "test_client",
|
||||
"client_secret": "secret",
|
||||
"scope": "read",
|
||||
}
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data["access_token"] == "client_token"
|
||||
mock_service.client_credentials_flow.assert_called_once_with(
|
||||
client_id="test_client",
|
||||
client_secret="secret",
|
||||
scope=["read"],
|
||||
)
|
||||
|
||||
|
||||
def test_token_refresh_token_success(client, mock_db_session):
|
||||
"""Refresh token grant returns new tokens."""
|
||||
with patch('app.modules.oauth2.controller.OAuthService') as MockOAuthService:
|
||||
mock_service = AsyncMock()
|
||||
mock_service.refresh_token_flow.return_value = {
|
||||
"access_token": "new_access_token",
|
||||
"token_type": "Bearer",
|
||||
"expires_in": 1800,
|
||||
"refresh_token": "new_refresh_token",
|
||||
"scope": "read",
|
||||
}
|
||||
MockOAuthService.return_value = mock_service
|
||||
|
||||
response = client.post(
|
||||
"/oauth/token",
|
||||
data={
|
||||
"grant_type": "refresh_token",
|
||||
"refresh_token": "old_refresh_token",
|
||||
"client_id": "test_client",
|
||||
"client_secret": "secret",
|
||||
}
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
mock_service.refresh_token_flow.assert_called_once_with(
|
||||
refresh_token="old_refresh_token",
|
||||
client_id="test_client",
|
||||
client_secret="secret",
|
||||
scope=[],
|
||||
)
|
||||
|
||||
|
||||
# ---------- UserInfo Endpoint Tests ----------
|
||||
def test_userinfo_missing_token(client):
|
||||
"""UserInfo requires Bearer token."""
|
||||
response = client.get("/oauth/userinfo")
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
|
||||
def test_userinfo_with_valid_token(client, mock_db_session):
|
||||
"""UserInfo returns claims from token payload."""
|
||||
# Mock get_current_token_payload dependency
|
||||
from app.modules.oauth2.dependencies import get_current_token_payload
|
||||
async def mock_payload():
|
||||
return {"sub": "user1", "client_id": "client1", "scopes": ["profile"]}
|
||||
|
||||
app = create_test_app({get_current_token_payload: mock_payload})
|
||||
client_with_auth = TestClient(app)
|
||||
|
||||
response = client_with_auth.get("/oauth/userinfo")
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data["sub"] == "user1"
|
||||
assert data["client_id"] == "client1"
|
||||
assert "scope" in data
|
||||
|
||||
|
||||
# ---------- Introspection Endpoint Tests ----------
|
||||
def test_introspect_missing_authentication(client):
|
||||
"""Introspection requires client credentials."""
|
||||
response = client.post("/oauth/introspect", data={"token": "some_token"})
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
|
||||
def test_introspect_success(client, mock_db_session):
|
||||
"""Introspection returns active token metadata."""
|
||||
# Mock ClientService.validate_client and TokenService.verify_token
|
||||
with patch('app.modules.oauth2.controller.ClientService') as MockClientService, \
|
||||
patch('app.modules.oauth2.controller.TokenService') as MockTokenService:
|
||||
mock_client_service = AsyncMock()
|
||||
mock_client_service.validate_client.return_value = True
|
||||
MockClientService.return_value = mock_client_service
|
||||
|
||||
mock_token_service = AsyncMock()
|
||||
mock_token_service.verify_token.return_value = {
|
||||
"sub": "user1",
|
||||
"client_id": "client1",
|
||||
"scopes": ["read"],
|
||||
"token_type": "Bearer",
|
||||
"exp": 1234567890,
|
||||
"iat": 1234567800,
|
||||
"jti": "jti_123",
|
||||
}
|
||||
MockTokenService.return_value = mock_token_service
|
||||
|
||||
response = client.post(
|
||||
"/oauth/introspect",
|
||||
data={"token": "valid_token"},
|
||||
auth=("test_client", "secret"),
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data["active"] is True
|
||||
assert data["sub"] == "user1"
|
||||
assert data["client_id"] == "client1"
|
||||
|
||||
|
||||
# ---------- Revocation Endpoint Tests ----------
|
||||
def test_revoke_missing_authentication(client):
|
||||
"""Revocation requires client credentials."""
|
||||
response = client.post("/oauth/revoke", data={"token": "some_token"})
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
|
||||
def test_revoke_success(client, mock_db_session):
|
||||
"""Successful revocation returns 200."""
|
||||
with patch('app.modules.oauth2.controller.ClientService') as MockClientService, \
|
||||
patch('app.modules.oauth2.controller.TokenService') as MockTokenService:
|
||||
mock_client_service = AsyncMock()
|
||||
mock_client_service.validate_client.return_value = True
|
||||
MockClientService.return_value = mock_client_service
|
||||
|
||||
mock_token_service = AsyncMock()
|
||||
mock_token_service.revoke_token.return_value = True
|
||||
MockTokenService.return_value = mock_token_service
|
||||
|
||||
response = client.post(
|
||||
"/oauth/revoke",
|
||||
data={"token": "token_to_revoke"},
|
||||
auth=("test_client", "secret"),
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.content == b""
|
||||
|
||||
|
||||
# ---------- OpenID Configuration Endpoint ----------
|
||||
def test_openid_configuration(client):
|
||||
"""Discovery endpoint returns provider metadata."""
|
||||
response = client.get("/oauth/.well-known/openid-configuration")
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert "issuer" in data
|
||||
assert "authorization_endpoint" in data
|
||||
assert "token_endpoint" in data
|
||||
assert "userinfo_endpoint" in data
|
||||
|
|
@ -1,45 +0,0 @@
|
|||
"""
|
||||
Test that route_manager is attached to app.state before first request.
|
||||
"""
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from app.core.app import create_app
|
||||
|
||||
def test_route_manager_attached():
|
||||
"""Ensure route_manager is attached after app creation."""
|
||||
app = create_app()
|
||||
assert hasattr(app.state, 'route_manager')
|
||||
assert hasattr(app.state, 'session_factory')
|
||||
assert app.state.route_manager is not None
|
||||
# Ensure route_manager has app reference
|
||||
assert app.state.route_manager.app is app
|
||||
|
||||
def test_admin_dashboard_with_route_manager():
|
||||
"""Test that admin dashboard can access route_manager dependency."""
|
||||
app = create_app()
|
||||
client = TestClient(app)
|
||||
# Login first
|
||||
resp = client.post("/admin/login", data={"username": "admin", "password": "admin123"})
|
||||
assert resp.status_code in (200, 302, 307)
|
||||
# Request dashboard with trailing slash (correct route)
|
||||
resp = client.get("/admin/", follow_redirects=True)
|
||||
# Should return 200, not 500 AttributeError
|
||||
assert resp.status_code == 200
|
||||
# Ensure route_manager stats are present (optional)
|
||||
# The dashboard template includes stats; we can check for some text
|
||||
assert "Dashboard" in resp.text
|
||||
|
||||
def test_route_manager_dependency():
|
||||
"""Test get_route_manager dependency returns the attached route_manager."""
|
||||
from app.modules.admin.controller import get_route_manager
|
||||
from fastapi import Request
|
||||
from unittest.mock import Mock
|
||||
# Create mock request with app.state.route_manager
|
||||
app = create_app()
|
||||
request = Mock(spec=Request)
|
||||
request.app = app
|
||||
route_manager = get_route_manager(request)
|
||||
assert route_manager is app.state.route_manager
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
51
wsgi.py
Normal file
51
wsgi.py
Normal file
|
|
@ -0,0 +1,51 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
WSGI entry point for production servers (Waitress, Gunicorn, etc.).
|
||||
"""
|
||||
import asyncio
|
||||
import logging
|
||||
from a2wsgi import ASGIMiddleware
|
||||
from app.core.app import create_app
|
||||
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Create the FastAPI application instance
|
||||
app = create_app()
|
||||
|
||||
def create_wsgi_app():
|
||||
"""
|
||||
Create a WSGI application with route refresh on startup.
|
||||
|
||||
This function is intended for production WSGI servers (e.g., Waitress).
|
||||
Since WSGI does not support ASGI lifespan events, we manually refresh
|
||||
routes from the database once when the WSGI app is created.
|
||||
"""
|
||||
loop = None
|
||||
try:
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
route_manager = app.state.route_manager
|
||||
logger.info("Refreshing routes from database (WSGI startup)...")
|
||||
loop.run_until_complete(route_manager.refresh_routes())
|
||||
logger.info(f"Registered {len(route_manager.registered_routes)} routes")
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to refresh routes on startup: {e}")
|
||||
# Continue anyway; routes can be refreshed later via admin interface
|
||||
finally:
|
||||
if loop is not None:
|
||||
loop.close()
|
||||
|
||||
# Wrap FastAPI ASGI app with WSGI adapter
|
||||
wsgi_app = ASGIMiddleware(app)
|
||||
return wsgi_app
|
||||
|
||||
# Create WSGI application
|
||||
wsgi_app = create_wsgi_app()
|
||||
|
||||
if __name__ == "__main__":
|
||||
# For testing WSGI app directly
|
||||
from wsgiref.simple_server import make_server
|
||||
print("Starting WSGI server on http://localhost:8000")
|
||||
server = make_server('localhost', 8000, wsgi_app)
|
||||
server.serve_forever()
|
||||
Loading…
Reference in a new issue