diff --git a/.env.example b/.env.example deleted file mode 100644 index 1221d51..0000000 --- a/.env.example +++ /dev/null @@ -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 diff --git a/README.md b/README.md index b76d779..f90a5ce 100644 --- a/README.md +++ b/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 diff --git a/app.py.backup b/app.py.backup deleted file mode 100644 index b256904..0000000 --- a/app.py.backup +++ /dev/null @@ -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() diff --git a/app/modules/admin/templates/base.html b/app/modules/admin/templates/base.html new file mode 100644 index 0000000..2654048 --- /dev/null +++ b/app/modules/admin/templates/base.html @@ -0,0 +1,72 @@ + + + + + + {% block title %}Mock API Admin{% endblock %} + + + + + + + +
+
+
+
+ {% block content %}{% endblock %} +
+
+
+
+ + + {% block extra_scripts %}{% endblock %} + + \ No newline at end of file diff --git a/.env.backup b/example.env similarity index 100% rename from .env.backup rename to example.env diff --git a/examples/README.md b/examples/README.md deleted file mode 100644 index 8879c88..0000000 --- a/examples/README.md +++ /dev/null @@ -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) diff --git a/examples/setup.sh b/examples/setup.sh deleted file mode 100755 index a89865d..0000000 --- a/examples/setup.sh +++ /dev/null @@ -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" diff --git a/main.py b/main.py index 19a8afb..b0d5ad1 100644 --- a/main.py +++ b/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, diff --git a/run_dev.py b/run_dev.py new file mode 100644 index 0000000..900d2bf --- /dev/null +++ b/run_dev.py @@ -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" + ) \ No newline at end of file diff --git a/scripts/debug_wsgi.py b/scripts/debug_wsgi.py deleted file mode 100644 index d593cb5..0000000 --- a/scripts/debug_wsgi.py +++ /dev/null @@ -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() \ No newline at end of file diff --git a/scripts/example_usage.py b/scripts/example_usage.py deleted file mode 100644 index 4c5c3ee..0000000 --- a/scripts/example_usage.py +++ /dev/null @@ -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() diff --git a/scripts/run_example.sh b/scripts/run_example.sh deleted file mode 100755 index eec19b7..0000000 --- a/scripts/run_example.sh +++ /dev/null @@ -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." \ No newline at end of file diff --git a/scripts/setup-test-client.py b/scripts/setup-test-client.py deleted file mode 100755 index 8593c1d..0000000 --- a/scripts/setup-test-client.py +++ /dev/null @@ -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()) diff --git a/test_run.py b/test_run.py new file mode 100644 index 0000000..fe1be35 --- /dev/null +++ b/test_run.py @@ -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) \ No newline at end of file diff --git a/tests/__init__.py b/tests/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/tests/conftest.py b/tests/conftest.py deleted file mode 100644 index ca51eca..0000000 --- a/tests/conftest.py +++ /dev/null @@ -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 \ No newline at end of file diff --git a/tests/integration/test_oauth2_integration.py b/tests/integration/test_oauth2_integration.py deleted file mode 100644 index a029591..0000000 --- a/tests/integration/test_oauth2_integration.py +++ /dev/null @@ -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"]) \ No newline at end of file diff --git a/tests/test_admin.py b/tests/test_admin.py deleted file mode 100644 index 8b39a8d..0000000 --- a/tests/test_admin.py +++ /dev/null @@ -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 \ No newline at end of file diff --git a/tests/test_auth_code_store.py b/tests/test_auth_code_store.py deleted file mode 100644 index f337cec..0000000 --- a/tests/test_auth_code_store.py +++ /dev/null @@ -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"]) \ No newline at end of file diff --git a/tests/test_endpoint_repository.py b/tests/test_endpoint_repository.py deleted file mode 100644 index 84477f4..0000000 --- a/tests/test_endpoint_repository.py +++ /dev/null @@ -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 \ No newline at end of file diff --git a/tests/test_oauth2_controller.py b/tests/test_oauth2_controller.py deleted file mode 100644 index fffbc23..0000000 --- a/tests/test_oauth2_controller.py +++ /dev/null @@ -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 \ No newline at end of file diff --git a/tests/test_route_manager_fix.py b/tests/test_route_manager_fix.py deleted file mode 100644 index b2b5722..0000000 --- a/tests/test_route_manager_fix.py +++ /dev/null @@ -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"]) \ No newline at end of file diff --git a/wsgi.py b/wsgi.py new file mode 100644 index 0000000..6fa9c5c --- /dev/null +++ b/wsgi.py @@ -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() \ No newline at end of file