# FastAPI Testing
A lightweight, async-first testing framework designed specifically for FastAPI applications. This library provides a simple way to write integration tests for FastAPI applications with proper lifecycle management and async support.
## Features
- Async-first design for modern Python applications
- Automatic port management for test servers
- Clean lifecycle management with context managers
- Built-in HTTP and WebSocket client support
- Proper cleanup of resources after tests
- Support for FastAPI's lifespan events
- Type-safe with full typing support
## Installation
```bash
pip install fastapi-testing
```
## Quick Start
Here's a simple example of how to test a FastAPI endpoint:
```python
import pytest
from fastapi import FastAPI
from fastapi_testing import create_test_server
@pytest.mark.asyncio
async def test_hello_world():
async with create_test_server() as server:
@server.app.get("/hello")
async def hello():
return {"message": "Hello, World!"}
response = await server.client.get("/hello")
await response.expect_status(200)
data = await response.json()
assert data["message"] == "Hello, World!"
```
## Architecture
The following sequence diagram illustrates the lifecycle of a test using this framework:
```mermaid
sequenceDiagram
participant Test
participant AsyncTestServer
participant PortGenerator
participant UvicornServer
participant FastAPI
participant AsyncTestClient
Test->>+AsyncTestServer: create_test_server()
AsyncTestServer->>+PortGenerator: get_port()
PortGenerator-->>-AsyncTestServer: available port
AsyncTestServer->>+UvicornServer: initialize
UvicornServer->>FastAPI: configure
AsyncTestServer->>+UvicornServer: start()
UvicornServer->>FastAPI: startup event
UvicornServer-->>AsyncTestServer: server ready
AsyncTestServer->>+AsyncTestClient: initialize
AsyncTestClient-->>-AsyncTestServer: client ready
AsyncTestServer-->>-Test: server instance
Note over Test,AsyncTestClient: Test execution happens here
Test->>+AsyncTestServer: cleanup (context exit)
AsyncTestServer->>+AsyncTestClient: close()
AsyncTestClient-->>-AsyncTestServer: closed
AsyncTestServer->>+UvicornServer: shutdown
UvicornServer->>FastAPI: shutdown event
UvicornServer-->>-AsyncTestServer: shutdown complete
AsyncTestServer->>+PortGenerator: release_port()
PortGenerator-->>-AsyncTestServer: port released
AsyncTestServer-->>-Test: cleanup complete
```
## Key Components
### AsyncTestServer
The `AsyncTestServer` class is the core component that manages the lifecycle of your test FastAPI application:
```python
from fastapi_testing import AsyncTestServer
server = AsyncTestServer()
await server.start()
# Use server.app to define routes
# Use server.client to make requests
await server.stop()
```
### Context Manager
The recommended way to use the test server is with the async context manager:
```python
from fastapi_testing import create_test_server
async with create_test_server() as server:
# Your test code here
pass # Server automatically starts and stops
```
### AsyncTestClient
The `AsyncTestClient` provides methods for making HTTP requests to your test server:
```python
# Available HTTP methods
await server.client.get("/path")
await server.client.post("/path", json=data)
await server.client.put("/path", json=data)
await server.client.delete("/path")
await server.client.patch("/path", json=data)
```
### Response Assertions
The `AsyncTestResponse` class provides convenient methods for assertions:
```python
response = await server.client.get("/path")
await response.expect_status(200) # Assert status code
data = await response.json() # Get JSON response
text = await response.text() # Get text response
```
### WebSocket Testing
Test WebSocket endpoints with full protocol support:
```python
@pytest.mark.asyncio
async def test_mixed_protocols(test_server):
# Define HTTP endpoint
@test_server.app.get("/api/data")
async def get_data():
return {"status": "ok"}
# Define WebSocket endpoint
@test_server.app.websocket("/ws/echo")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept(subprotocol="test-protocol")
while True:
try:
message = await websocket.receive()
if "text" in message:
data = json.loads(message["text"])
await websocket.send_json(data)
elif "bytes" in message:
await websocket.send_bytes(message["bytes"])
except WebSocketDisconnect:
return
# Test HTTP endpoint
http_response = await test_server.client.get("/api/data")
assert http_response.status_code == 200
# Configure WebSocket
config = WebSocketConfig(
subprotocols=["test-protocol"],
ping_interval=20.0,
ping_timeout=20.0
)
# Test WebSocket endpoint
ws_response = await test_server.client.websocket("/ws/echo", config)
try:
# Test JSON messages
test_json = {"message": "test"}
await test_server.client.ws.send_json(ws_response, test_json)
response = await test_server.client.ws.receive_json(ws_response)
assert response == test_json
# Test binary messages
test_data = b"binary test"
await test_server.client.ws.send_binary(ws_response, test_data)
response = await test_server.client.ws.receive_binary(ws_response)
assert response == test_data
finally:
await ws_response.websocket().close()
```
#### WebSocket Message Operations
The WebSocketHelper provides comprehensive message handling:
```python
# Send Operations
await client.ws.send_text(ws_response, "message")
await client.ws.send_json(ws_response, {"key": "value"})
await client.ws.send_binary(ws_response, b"data")
# Receive Operations
text = await client.ws.receive_text(ws_response)
json_data = await client.ws.receive_json(ws_response)
binary = await client.ws.receive_binary(ws_response)
# Message Expectations
await client.ws.expect_message(
ws_response,
expected="message",
timeout=1.0
)
# Collect Multiple Messages
messages = await client.ws.drain_messages(
ws_response,
timeout=1.0
)
```
#### WebSocket Configuration
Configure connections with various options:
```python
ws_config = WebSocketConfig(
subprotocols=["protocol"], # Supported subprotocols
compression=None, # Compression algorithm
extra_headers={}, # Additional headers
ping_interval=20.0, # Keep-alive interval
ping_timeout=20.0, # Ping timeout
max_size=2 ** 20, # Max message size (1MB)
max_queue=32, # Max queued messages
timeout=30.0 # Connection timeout
)
```
## Advanced Usage
### Advanced Server Configuration
You can customize the server lifecycle using a reusable test fixture:
```python
@pytest.fixture
async def test_server(
test_settings: Settings,
transaction_manager: TransactionManager
) -> AsyncGenerator[AsyncTestServer, None]:
"""Create test server with overridden settings and database connection"""
async def custom_lifespan(app: AppType) -> AsyncGenerator[None, Any]:
# Wire up test-specific dependencies
app.dependency_overrides.update({
get_settings: lambda: test_settings,
get_transaction_manager: lambda: transaction_manager,
get_db_pool: lambda: transaction_manager.pool
})
yield # Server handles requests during this period
# Cleanup after tests complete
await db.cleanup()
async with create_test_server(lifespan=custom_lifespan) as server:
yield server
```
### Testing Routes and Routers
You can test entire routers and complex route configurations:
```python
@pytest.mark.asyncio
async def test_api(test_server: AsyncTestServer):
# Register routes/routers
test_server.app.include_router(your_router)
# Make requests
response = await test_server.client.get("/your-endpoint")
await response.expect_status(200)
# Test concurrent requests
responses = await asyncio.gather(*[
test_server.client.get("/endpoint")
for _ in range(5)
])
for response in responses:
await response.expect_status(200)
```
### Lifecycle Management
You can define setup and cleanup operations using FastAPI's lifespan:
```python
from contextlib import asynccontextmanager
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
# Setup
print("Starting server")
yield
# Cleanup
print("Shutting down server")
async with create_test_server(lifespan=lifespan) as server:
# Your test code here
pass
```
### Concurrent Requests
The framework supports testing concurrent requests:
```python
import asyncio
async with create_test_server() as server:
@server.app.get("/ping")
async def ping():
return {"status": "ok"}
responses = await asyncio.gather(*[
server.client.get("/ping")
for _ in range(5)
])
```
## Configuration
You can customize the server behavior:
```python
server = AsyncTestServer(
startup_timeout=30.0, # Seconds to wait for server startup
shutdown_timeout=10.0, # Seconds to wait for server shutdown
)
```
## Best Practices
1. Always use the async context manager (`create_test_server`) when possible
2. Clean up resources in your tests, especially when managing state
3. Use pytest.mark.asyncio for your test functions
4. Handle exceptions appropriately in your tests
5. Use type hints to catch potential issues early
## Error Handling
The framework provides clear error messages for common issues:
- Server startup timeout
- Port allocation failures
- Connection errors
- Invalid request formats
## Limitations
- Only supports async test cases
- Requires Python 3.11+
- Designed specifically for FastAPI applications
## Contributing
Contributions are welcome! Please feel free to submit a Pull Request.
Raw data
{
"_id": null,
"home_page": "https://github.com/descoped/fastapi-testing",
"name": "fastapi-testing",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.11",
"maintainer_email": null,
"keywords": "fastapi, testing, async, pytest, integration-testing, test-framework",
"author": "Ove Ranheim",
"author_email": "oranheim@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/9a/96/c9ed98672b357572eefa7a675b21961fbb56ab54ac440408ae68fab2db7a/fastapi_testing-0.2.1.tar.gz",
"platform": null,
"description": "# FastAPI Testing\n\nA lightweight, async-first testing framework designed specifically for FastAPI applications. This library provides a simple way to write integration tests for FastAPI applications with proper lifecycle management and async support.\n\n## Features\n\n- Async-first design for modern Python applications\n- Automatic port management for test servers\n- Clean lifecycle management with context managers\n- Built-in HTTP and WebSocket client support\n- Proper cleanup of resources after tests\n- Support for FastAPI's lifespan events\n- Type-safe with full typing support\n\n## Installation\n\n```bash\npip install fastapi-testing\n```\n\n## Quick Start\n\nHere's a simple example of how to test a FastAPI endpoint:\n\n```python\nimport pytest\nfrom fastapi import FastAPI\nfrom fastapi_testing import create_test_server\n\n@pytest.mark.asyncio\nasync def test_hello_world():\n async with create_test_server() as server:\n @server.app.get(\"/hello\")\n async def hello():\n return {\"message\": \"Hello, World!\"}\n \n response = await server.client.get(\"/hello\")\n await response.expect_status(200)\n data = await response.json()\n assert data[\"message\"] == \"Hello, World!\"\n```\n\n## Architecture\n\nThe following sequence diagram illustrates the lifecycle of a test using this framework:\n\n```mermaid\nsequenceDiagram\n participant Test\n participant AsyncTestServer\n participant PortGenerator\n participant UvicornServer\n participant FastAPI\n participant AsyncTestClient\n\n Test->>+AsyncTestServer: create_test_server()\n AsyncTestServer->>+PortGenerator: get_port()\n PortGenerator-->>-AsyncTestServer: available port\n \n AsyncTestServer->>+UvicornServer: initialize\n UvicornServer->>FastAPI: configure\n \n AsyncTestServer->>+UvicornServer: start()\n UvicornServer->>FastAPI: startup event\n UvicornServer-->>AsyncTestServer: server ready\n \n AsyncTestServer->>+AsyncTestClient: initialize\n AsyncTestClient-->>-AsyncTestServer: client ready\n AsyncTestServer-->>-Test: server instance\n \n Note over Test,AsyncTestClient: Test execution happens here\n \n Test->>+AsyncTestServer: cleanup (context exit)\n AsyncTestServer->>+AsyncTestClient: close()\n AsyncTestClient-->>-AsyncTestServer: closed\n \n AsyncTestServer->>+UvicornServer: shutdown\n UvicornServer->>FastAPI: shutdown event\n UvicornServer-->>-AsyncTestServer: shutdown complete\n \n AsyncTestServer->>+PortGenerator: release_port()\n PortGenerator-->>-AsyncTestServer: port released\n AsyncTestServer-->>-Test: cleanup complete\n```\n\n## Key Components\n\n### AsyncTestServer\n\nThe `AsyncTestServer` class is the core component that manages the lifecycle of your test FastAPI application:\n\n```python\nfrom fastapi_testing import AsyncTestServer\n\nserver = AsyncTestServer()\nawait server.start()\n# Use server.app to define routes\n# Use server.client to make requests\nawait server.stop()\n```\n\n### Context Manager\n\nThe recommended way to use the test server is with the async context manager:\n\n```python\nfrom fastapi_testing import create_test_server\n\nasync with create_test_server() as server:\n # Your test code here\n pass # Server automatically starts and stops\n```\n\n### AsyncTestClient\n\nThe `AsyncTestClient` provides methods for making HTTP requests to your test server:\n\n```python\n# Available HTTP methods\nawait server.client.get(\"/path\")\nawait server.client.post(\"/path\", json=data)\nawait server.client.put(\"/path\", json=data)\nawait server.client.delete(\"/path\")\nawait server.client.patch(\"/path\", json=data)\n```\n\n### Response Assertions\n\nThe `AsyncTestResponse` class provides convenient methods for assertions:\n\n```python\nresponse = await server.client.get(\"/path\")\nawait response.expect_status(200) # Assert status code\ndata = await response.json() # Get JSON response\ntext = await response.text() # Get text response\n```\n\n### WebSocket Testing\n\nTest WebSocket endpoints with full protocol support:\n\n```python\n@pytest.mark.asyncio\nasync def test_mixed_protocols(test_server):\n # Define HTTP endpoint\n @test_server.app.get(\"/api/data\")\n async def get_data():\n return {\"status\": \"ok\"}\n\n # Define WebSocket endpoint\n @test_server.app.websocket(\"/ws/echo\")\n async def websocket_endpoint(websocket: WebSocket):\n await websocket.accept(subprotocol=\"test-protocol\")\n while True:\n try:\n message = await websocket.receive()\n if \"text\" in message:\n data = json.loads(message[\"text\"])\n await websocket.send_json(data)\n elif \"bytes\" in message:\n await websocket.send_bytes(message[\"bytes\"])\n except WebSocketDisconnect:\n return\n\n # Test HTTP endpoint\n http_response = await test_server.client.get(\"/api/data\")\n assert http_response.status_code == 200\n\n # Configure WebSocket\n config = WebSocketConfig(\n subprotocols=[\"test-protocol\"],\n ping_interval=20.0,\n ping_timeout=20.0\n )\n\n # Test WebSocket endpoint\n ws_response = await test_server.client.websocket(\"/ws/echo\", config)\n try:\n # Test JSON messages\n test_json = {\"message\": \"test\"}\n await test_server.client.ws.send_json(ws_response, test_json)\n response = await test_server.client.ws.receive_json(ws_response)\n assert response == test_json\n\n # Test binary messages\n test_data = b\"binary test\"\n await test_server.client.ws.send_binary(ws_response, test_data)\n response = await test_server.client.ws.receive_binary(ws_response)\n assert response == test_data\n finally:\n await ws_response.websocket().close()\n```\n\n#### WebSocket Message Operations\n\nThe WebSocketHelper provides comprehensive message handling:\n\n```python\n# Send Operations\nawait client.ws.send_text(ws_response, \"message\")\nawait client.ws.send_json(ws_response, {\"key\": \"value\"})\nawait client.ws.send_binary(ws_response, b\"data\")\n\n# Receive Operations\ntext = await client.ws.receive_text(ws_response)\njson_data = await client.ws.receive_json(ws_response)\nbinary = await client.ws.receive_binary(ws_response)\n\n# Message Expectations\nawait client.ws.expect_message(\n ws_response,\n expected=\"message\",\n timeout=1.0\n)\n\n# Collect Multiple Messages\nmessages = await client.ws.drain_messages(\n ws_response,\n timeout=1.0\n)\n```\n\n#### WebSocket Configuration\n\nConfigure connections with various options:\n\n```python\nws_config = WebSocketConfig(\n subprotocols=[\"protocol\"], # Supported subprotocols\n compression=None, # Compression algorithm\n extra_headers={}, # Additional headers\n ping_interval=20.0, # Keep-alive interval\n ping_timeout=20.0, # Ping timeout\n max_size=2 ** 20, # Max message size (1MB)\n max_queue=32, # Max queued messages\n timeout=30.0 # Connection timeout\n)\n```\n\n## Advanced Usage\n\n### Advanced Server Configuration\n\nYou can customize the server lifecycle using a reusable test fixture:\n\n```python\n@pytest.fixture\nasync def test_server(\n test_settings: Settings,\n transaction_manager: TransactionManager\n) -> AsyncGenerator[AsyncTestServer, None]:\n \"\"\"Create test server with overridden settings and database connection\"\"\"\n\n async def custom_lifespan(app: AppType) -> AsyncGenerator[None, Any]:\n # Wire up test-specific dependencies\n app.dependency_overrides.update({\n get_settings: lambda: test_settings,\n get_transaction_manager: lambda: transaction_manager,\n get_db_pool: lambda: transaction_manager.pool\n })\n\n yield # Server handles requests during this period\n\n # Cleanup after tests complete\n await db.cleanup()\n\n async with create_test_server(lifespan=custom_lifespan) as server:\n yield server\n```\n\n### Testing Routes and Routers\n\nYou can test entire routers and complex route configurations:\n\n```python\n@pytest.mark.asyncio\nasync def test_api(test_server: AsyncTestServer):\n # Register routes/routers\n test_server.app.include_router(your_router)\n\n # Make requests\n response = await test_server.client.get(\"/your-endpoint\")\n await response.expect_status(200)\n \n # Test concurrent requests\n responses = await asyncio.gather(*[\n test_server.client.get(\"/endpoint\")\n for _ in range(5)\n ])\n \n for response in responses:\n await response.expect_status(200)\n```\n\n### Lifecycle Management\n\nYou can define setup and cleanup operations using FastAPI's lifespan:\n\n```python\nfrom contextlib import asynccontextmanager\nfrom fastapi import FastAPI\n\n@asynccontextmanager\nasync def lifespan(app: FastAPI):\n # Setup\n print(\"Starting server\")\n yield\n # Cleanup\n print(\"Shutting down server\")\n\nasync with create_test_server(lifespan=lifespan) as server:\n # Your test code here\n pass\n```\n\n### Concurrent Requests\n\nThe framework supports testing concurrent requests:\n\n```python\nimport asyncio\n\nasync with create_test_server() as server:\n @server.app.get(\"/ping\")\n async def ping():\n return {\"status\": \"ok\"}\n\n responses = await asyncio.gather(*[\n server.client.get(\"/ping\")\n for _ in range(5)\n ])\n```\n\n## Configuration\n\nYou can customize the server behavior:\n\n```python\nserver = AsyncTestServer(\n startup_timeout=30.0, # Seconds to wait for server startup\n shutdown_timeout=10.0, # Seconds to wait for server shutdown\n)\n```\n\n## Best Practices\n\n1. Always use the async context manager (`create_test_server`) when possible\n2. Clean up resources in your tests, especially when managing state\n3. Use pytest.mark.asyncio for your test functions\n4. Handle exceptions appropriately in your tests\n5. Use type hints to catch potential issues early\n\n## Error Handling\n\nThe framework provides clear error messages for common issues:\n\n- Server startup timeout\n- Port allocation failures\n- Connection errors\n- Invalid request formats\n\n## Limitations\n\n- Only supports async test cases\n- Requires Python 3.11+\n- Designed specifically for FastAPI applications\n\n## Contributing\n\nContributions are welcome! Please feel free to submit a Pull Request.\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A lightweight, async-first testing framework for FastAPI applications",
"version": "0.2.1",
"project_urls": {
"Homepage": "https://github.com/descoped/fastapi-testing",
"Repository": "https://github.com/descoped/fastapi-testing"
},
"split_keywords": [
"fastapi",
" testing",
" async",
" pytest",
" integration-testing",
" test-framework"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "bdf63f4b33bc7ad2daf9e07f894aa86a10d469b6e0b1e1cb30304583346d45ae",
"md5": "e2967fe17c2a7ab31f58791718e95502",
"sha256": "3e5003011998290af64cbcc6300bb3313c6df92bf5d6f4c577bb582493a6718e"
},
"downloads": -1,
"filename": "fastapi_testing-0.2.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "e2967fe17c2a7ab31f58791718e95502",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.11",
"size": 10501,
"upload_time": "2025-02-20T15:42:26",
"upload_time_iso_8601": "2025-02-20T15:42:26.899717Z",
"url": "https://files.pythonhosted.org/packages/bd/f6/3f4b33bc7ad2daf9e07f894aa86a10d469b6e0b1e1cb30304583346d45ae/fastapi_testing-0.2.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "9a96c9ed98672b357572eefa7a675b21961fbb56ab54ac440408ae68fab2db7a",
"md5": "075dca9621351bd719bd71d53582f457",
"sha256": "c756dead4ba135eb71280075bfb46cfe9c50f1534d2dd105e3956daf3ebad056"
},
"downloads": -1,
"filename": "fastapi_testing-0.2.1.tar.gz",
"has_sig": false,
"md5_digest": "075dca9621351bd719bd71d53582f457",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.11",
"size": 11980,
"upload_time": "2025-02-20T15:42:28",
"upload_time_iso_8601": "2025-02-20T15:42:28.913438Z",
"url": "https://files.pythonhosted.org/packages/9a/96/c9ed98672b357572eefa7a675b21961fbb56ab54ac440408ae68fab2db7a/fastapi_testing-0.2.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-02-20 15:42:28",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "descoped",
"github_project": "fastapi-testing",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "fastapi-testing"
}
JSON
