tessl/pypi-blacksheep
Fast web framework for Python asyncio
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
testing.mddocs/
Testing Utilities
BlackSheep provides comprehensive testing utilities for unit testing, integration testing, and end-to-end testing of web applications. The testing framework supports async testing, request simulation, and complete application testing without network overhead.
TestClient
The TestClient class is the primary testing utility that allows you to make HTTP requests to your application without starting a server.
Basic Testing Setup
import pytest
import asyncio
from blacksheep.testing import TestClient
from blacksheep import Application, json, Request, FromJSON
# Sample application for testing
app = Application()
@app.get("/")
async def home():
return json({"message": "Hello, World!"})
@app.get("/users/{user_id:int}")
async def get_user(user_id: int):
return json({"id": user_id, "name": f"User {user_id}"})
@app.post("/users")
async def create_user(data: FromJSON[dict]):
return json({"created": True, "data": data.value})
# Basic test example
async def test_basic_endpoints():
client = TestClient(app)
# Test GET endpoint
response = await client.get("/")
assert response.status == 200
data = await response.json()
assert data["message"] == "Hello, World!"
# Test GET with parameters
response = await client.get("/users/123")
assert response.status == 200
data = await response.json()
assert data["id"] == 123
# Test POST with JSON
response = await client.post("/users", json={"name": "Alice"})
assert response.status == 200
data = await response.json()
assert data["created"] is True
# Run test
asyncio.run(test_basic_endpoints())PyTest Integration
import pytest
from blacksheep.testing import TestClient
@pytest.fixture
def client():
"""Create test client fixture"""
return TestClient(app)
@pytest.mark.asyncio
async def test_home_endpoint(client):
response = await client.get("/")
assert response.status == 200
data = await response.json()
assert "message" in data
@pytest.mark.asyncio
async def test_user_creation(client):
user_data = {"name": "Alice", "email": "alice@example.com"}
response = await client.post("/users", json=user_data)
assert response.status == 200
result = await response.json()
assert result["created"] is True
assert result["data"]["name"] == "Alice"
@pytest.mark.asyncio
async def test_not_found(client):
response = await client.get("/nonexistent")
assert response.status == 404Request Methods
from blacksheep.testing import TestClient
from blacksheep import JSONContent, TextContent, FormContent
async def test_http_methods():
client = TestClient(app)
# GET request
response = await client.get("/users")
assert response.status == 200
# GET with query parameters
response = await client.get("/users", params={"page": 1, "limit": 10})
# Equivalent to: /users?page=1&limit=10
# POST request with JSON
response = await client.post("/users", json={"name": "Bob"})
# POST with custom content
json_content = JSONContent({"user": {"name": "Charlie"}})
response = await client.post("/users", content=json_content)
# POST with form data
response = await client.post("/login", data={"username": "user", "password": "pass"})
# PUT request
response = await client.put("/users/123", json={"name": "Updated Name"})
# DELETE request
response = await client.delete("/users/123")
# PATCH request
response = await client.patch("/users/123", json={"email": "new@example.com"})
# HEAD request
response = await client.head("/users/123")
assert response.status == 200
# No response body for HEAD requests
# OPTIONS request
response = await client.options("/users")Headers and Cookies
async def test_headers_and_cookies():
client = TestClient(app)
# Custom headers
headers = {"Authorization": "Bearer token123", "X-API-Version": "1.0"}
response = await client.get("/protected", headers=headers)
# Cookies
cookies = {"session_id": "abc123", "theme": "dark"}
response = await client.get("/dashboard", cookies=cookies)
# Both headers and cookies
response = await client.get(
"/api/data",
headers={"Authorization": "Bearer token"},
cookies={"session": "value"}
)
# Check response headers
assert response.headers.get_first(b"Content-Type") == b"application/json"
# Check response cookies
set_cookie_header = response.headers.get_first(b"Set-Cookie")
if set_cookie_header:
print(f"Server set cookie: {set_cookie_header.decode()}")File Upload Testing
Test file upload endpoints with multipart form data.
File Upload Tests
from blacksheep import MultiPartFormData, FormPart
from io import BytesIO
# Application with file upload
@app.post("/upload")
async def upload_file(files: FromFiles):
uploaded_files = files.value
return json({
"files_received": len(uploaded_files),
"filenames": [f.file_name.decode() if f.file_name else "unknown"
for f in uploaded_files]
})
async def test_file_upload():
client = TestClient(app)
# Create test file content
file_content = b"Test file content"
# Create form parts
text_part = FormPart(b"title", b"My Document")
file_part = FormPart(
name=b"document",
data=file_content,
content_type=b"text/plain",
file_name=b"test.txt"
)
# Create multipart form data
multipart = MultiPartFormData([text_part, file_part])
# Upload file
response = await client.post("/upload", content=multipart)
assert response.status == 200
result = await response.json()
assert result["files_received"] == 1
assert "test.txt" in result["filenames"]
# Alternative: Upload with files parameter
async def test_file_upload_simple():
client = TestClient(app)
# Simple file upload
file_data = BytesIO(b"Simple file content")
response = await client.post(
"/upload",
files={"document": ("simple.txt", file_data, "text/plain")}
)
assert response.status == 200Authentication Testing
Test authentication and authorization features.
Authentication Tests
from blacksheep import auth, allow_anonymous
from blacksheep.server.authentication.jwt import JWTBearerAuthentication
from guardpost import Identity
# Application with auth
auth_app = Application()
auth_strategy = auth_app.use_authentication()
@auth_app.get("/public")
@allow_anonymous
async def public_endpoint():
return json({"public": True})
@auth_app.get("/protected")
@auth()
async def protected_endpoint(request: Request):
identity = request.identity
return json({"user_id": identity.id, "authenticated": True})
async def test_authentication():
client = TestClient(auth_app)
# Test public endpoint
response = await client.get("/public")
assert response.status == 200
# Test protected endpoint without auth (should fail)
response = await client.get("/protected")
assert response.status == 401
# Test protected endpoint with auth
headers = {"Authorization": "Bearer valid-jwt-token"}
response = await client.get("/protected", headers=headers)
# Note: This will depend on your JWT validation logicMock Authentication
from unittest.mock import AsyncMock, patch
async def test_mock_authentication():
client = TestClient(auth_app)
# Mock the authentication handler
mock_identity = Identity({"sub": "user123", "name": "Test User"})
with patch.object(auth_strategy, 'authenticate', return_value=mock_identity):
response = await client.get("/protected")
assert response.status == 200
data = await response.json()
assert data["user_id"] == "user123"
assert data["authenticated"] is TrueDatabase Testing
Test applications that use databases with proper setup and teardown.
Database Test Setup
import pytest
from blacksheep import Application, FromServices
from rodi import Container
# Sample service and repository
class UserRepository:
def __init__(self):
self.users = {}
self.next_id = 1
async def create_user(self, user_data: dict) -> dict:
user = {"id": self.next_id, **user_data}
self.users[self.next_id] = user
self.next_id += 1
return user
async def get_user(self, user_id: int) -> dict:
return self.users.get(user_id)
# Application with database
db_app = Application()
# Configure DI
container = Container()
container.add_singleton(UserRepository)
db_app.services = container
@db_app.post("/users")
async def create_user_endpoint(
data: FromJSON[dict],
repo: FromServices[UserRepository]
):
user = await repo.create_user(data.value)
return json(user)
@db_app.get("/users/{user_id:int}")
async def get_user_endpoint(
user_id: int,
repo: FromServices[UserRepository]
):
user = await repo.get_user(user_id)
if not user:
return Response(404)
return json(user)
@pytest.fixture
async def db_client():
"""Create test client with fresh database"""
# Create fresh application instance for each test
test_app = Application()
test_container = Container()
test_container.add_singleton(UserRepository)
test_app.services = test_container
# Re-register routes (in real app, you'd organize this better)
test_app.post("/users")(create_user_endpoint)
test_app.get("/users/{user_id:int}")(get_user_endpoint)
return TestClient(test_app)
@pytest.mark.asyncio
async def test_user_crud(db_client):
# Create user
user_data = {"name": "Alice", "email": "alice@example.com"}
response = await db_client.post("/users", json=user_data)
assert response.status == 200
created_user = await response.json()
user_id = created_user["id"]
# Get user
response = await db_client.get(f"/users/{user_id}")
assert response.status == 200
retrieved_user = await response.json()
assert retrieved_user["name"] == "Alice"
assert retrieved_user["email"] == "alice@example.com"
# Get non-existent user
response = await db_client.get("/users/999")
assert response.status == 404WebSocket Testing
Test WebSocket endpoints and real-time functionality.
WebSocket Test Setup
from blacksheep.testing import TestClient
from blacksheep import WebSocket, WebSocketState
# WebSocket application
ws_app = Application()
@ws_app.ws("/echo")
async def echo_handler(websocket: WebSocket):
await websocket.accept()
try:
while True:
message = await websocket.receive_text()
await websocket.send_text(f"Echo: {message}")
except WebSocketDisconnectError:
pass
@ws_app.ws("/json-echo")
async def json_echo_handler(websocket: WebSocket):
await websocket.accept()
try:
while True:
data = await websocket.receive_json()
await websocket.send_json({"echo": data})
except WebSocketDisconnectError:
pass
async def test_websocket_echo():
client = TestClient(ws_app)
# Test WebSocket connection
async with client.websocket("/echo") as websocket:
# Send message
await websocket.send_text("Hello WebSocket!")
# Receive echo
response = await websocket.receive_text()
assert response == "Echo: Hello WebSocket!"
# Send another message
await websocket.send_text("Second message")
response = await websocket.receive_text()
assert response == "Echo: Second message"
async def test_websocket_json():
client = TestClient(ws_app)
async with client.websocket("/json-echo") as websocket:
# Send JSON data
test_data = {"message": "Hello", "count": 42}
await websocket.send_json(test_data)
# Receive JSON response
response = await websocket.receive_json()
assert response["echo"]["message"] == "Hello"
assert response["echo"]["count"] == 42Mock Objects
BlackSheep provides mock objects for testing ASGI applications and components.
ASGI Mocks
from blacksheep.testing import MockReceive, MockSend
async def test_asgi_mocks():
# Mock ASGI components
mock_receive = MockReceive()
mock_send = MockSend()
# Add messages to mock receive
mock_receive.add_message({
"type": "http.request",
"body": b'{"test": "data"}',
"more_body": False
})
# Test ASGI application directly
scope = {
"type": "http",
"method": "POST",
"path": "/test",
"headers": [(b"content-type", b"application/json")]
}
# Call application ASGI interface
await app(scope, mock_receive, mock_send)
# Check sent messages
sent_messages = mock_send.messages
assert len(sent_messages) > 0
# Verify response
response_start = sent_messages[0]
assert response_start["type"] == "http.response.start"
assert response_start["status"] == 200Integration Testing
Test complete application workflows and integrations.
Full Application Test
import pytest
from blacksheep.testing import TestClient
# Complete application for integration testing
integration_app = Application()
# Add CORS
integration_app.use_cors(allow_origins="*")
# Add authentication
auth_strategy = integration_app.use_authentication()
# Add routes
@integration_app.get("/")
async def home():
return json({"app": "integration_test", "version": "1.0"})
@integration_app.get("/health")
async def health_check():
# Simulate health check logic
return json({"status": "healthy", "timestamp": time.time()})
@integration_app.post("/api/process")
async def process_data(data: FromJSON[dict]):
# Simulate data processing
result = {
"processed": True,
"input_size": len(str(data.value)),
"output": f"Processed: {data.value}"
}
return json(result)
class TestIntegrationWorkflow:
@pytest.fixture
def client(self):
return TestClient(integration_app)
@pytest.mark.asyncio
async def test_complete_workflow(self, client):
# Test health check
response = await client.get("/health")
assert response.status == 200
health = await response.json()
assert health["status"] == "healthy"
# Test main endpoint
response = await client.get("/")
assert response.status == 200
info = await response.json()
assert info["app"] == "integration_test"
# Test data processing
test_data = {"items": [1, 2, 3], "name": "test"}
response = await client.post("/api/process", json=test_data)
assert response.status == 200
result = await response.json()
assert result["processed"] is True
assert result["input_size"] > 0
@pytest.mark.asyncio
async def test_cors_headers(self, client):
# Test CORS preflight
response = await client.options(
"/api/process",
headers={"Origin": "https://example.com"}
)
# Check CORS headers
cors_origin = response.headers.get_first(b"Access-Control-Allow-Origin")
assert cors_origin == b"*"
@pytest.mark.asyncio
async def test_error_handling(self, client):
# Test 404
response = await client.get("/nonexistent")
assert response.status == 404
# Test invalid JSON
response = await client.post(
"/api/process",
content=TextContent("invalid json"),
headers={"Content-Type": "application/json"}
)
# Should handle parsing error appropriatelyPerformance Testing
Basic performance testing with the test client.
Load Testing
import asyncio
import time
from statistics import mean, stdev
async def performance_test():
client = TestClient(app)
# Warm up
for _ in range(10):
await client.get("/")
# Performance test
iterations = 100
start_times = []
for i in range(iterations):
start = time.time()
response = await client.get("/")
end = time.time()
assert response.status == 200
start_times.append((end - start) * 1000) # Convert to ms
# Calculate statistics
avg_time = mean(start_times)
std_dev = stdev(start_times) if len(start_times) > 1 else 0
min_time = min(start_times)
max_time = max(start_times)
print(f"Performance Results ({iterations} iterations):")
print(f" Average: {avg_time:.2f}ms")
print(f" Std Dev: {std_dev:.2f}ms")
print(f" Min: {min_time:.2f}ms")
print(f" Max: {max_time:.2f}ms")
# Assert performance threshold
assert avg_time < 50 # Under 50ms average
# Concurrent request testing
async def concurrent_test():
client = TestClient(app)
async def make_request():
response = await client.get("/")
return response.status
# Run 20 concurrent requests
tasks = [make_request() for _ in range(20)]
results = await asyncio.gather(*tasks)
# All should succeed
assert all(status == 200 for status in results)
print(f"Concurrent test: {len(results)} requests completed successfully")
asyncio.run(performance_test())
asyncio.run(concurrent_test())Test Configuration
Configure testing environment and fixtures.
Test Configuration
import pytest
import os
from blacksheep import Application
from blacksheep.testing import TestClient
# Test configuration
@pytest.fixture(scope="session")
def test_config():
return {
"DEBUG": True,
"TESTING": True,
"DATABASE_URL": "sqlite:///:memory:",
"SECRET_KEY": "test-secret-key"
}
@pytest.fixture
def app_with_config(test_config):
"""Create application with test configuration"""
app = Application(debug=test_config["DEBUG"])
# Configure for testing
app.configuration = test_config
# Add test routes
@app.get("/config")
async def get_config():
return json({"debug": app.debug})
return app
@pytest.fixture
def client(app_with_config):
"""Test client with configured application"""
return TestClient(app_with_config)
# Environment-specific tests
@pytest.mark.asyncio
async def test_debug_mode(client):
response = await client.get("/config")
data = await response.json()
assert data["debug"] is True
# Skip tests based on conditions
@pytest.mark.skipif(os.getenv("CI") is None, reason="Only run in CI")
async def test_ci_specific():
# Test that only runs in CI environment
passBlackSheep's testing utilities provide comprehensive support for testing all aspects of your web application, from simple unit tests to complex integration scenarios, ensuring your application works correctly and performs well.
Install with Tessl CLI
npx tessl i tessl/pypi-blacksheep