Testing Adapters¶
Mock Objects from guard-core¶
Guard-core ships mock implementations of both protocols in tests/conftest.py. Use them as references for your own test fixtures, or import them directly in your adapter's test suite.
MockState¶
A mutable namespace that mimics request.state:
class MockState:
def __init__(self) -> None:
self._attrs: dict = {}
def __getattr__(self, name: str) -> object:
if name == "_attrs":
return super().__getattribute__(name)
return self._attrs.get(name)
def __setattr__(self, name: str, value: object) -> None:
if name == "_attrs":
super().__setattr__(name, value)
else:
self._attrs[name] = value
MockGuardRequest¶
A complete GuardRequest implementation for testing:
class MockGuardRequest:
def __init__(
self,
path: str = "/",
method: str = "GET",
headers: dict | None = None,
client_host: str | None = "127.0.0.1",
scheme: str = "https",
query_params: dict | None = None,
body_content: bytes = b"",
scope: dict | None = None,
) -> None:
self._path = path
self._method = method
self._headers = headers or {}
self._client_host = client_host
self._scheme = scheme
self._query_params = query_params or {}
self._body = body_content
self._state = MockState()
self._scope = scope or {}
@property
def url_path(self) -> str:
return self._path
@property
def url_scheme(self) -> str:
return self._scheme
@property
def url_full(self) -> str:
return f"{self._scheme}://test{self._path}"
def url_replace_scheme(self, scheme: str) -> str:
return f"{scheme}://test{self._path}"
@property
def method(self) -> str:
return self._method
@property
def client_host(self) -> str | None:
return self._client_host
@property
def headers(self) -> dict:
return self._headers
@property
def query_params(self) -> dict:
return self._query_params
async def body(self) -> bytes:
return self._body
@property
def state(self) -> MockState:
return self._state
@property
def scope(self) -> dict:
return self._scope
MockGuardResponse¶
class MockGuardResponse:
def __init__(
self,
content: str = "",
status_code: int = 200,
headers: dict | None = None,
) -> None:
self._status_code = status_code
self._headers = headers or {}
self._body = content.encode() if isinstance(content, str) else content
@property
def status_code(self) -> int:
return self._status_code
@property
def headers(self) -> dict:
return self._headers
@property
def body(self) -> bytes:
return self._body
MockGuardResponseFactory¶
class MockGuardResponseFactory:
def create_response(self, content: str, status_code: int) -> MockGuardResponse:
return MockGuardResponse(content, status_code)
def create_redirect_response(
self, url: str, status_code: int
) -> MockGuardResponse:
return MockGuardResponse(
f"Redirect to {url}", status_code, {"Location": url}
)
Pytest Fixtures¶
MockGuardRequest, MockGuardResponse, and MockGuardResponseFactory are plain classes — instantiate them directly in your tests. The only SecurityConfig fixtures conftest.py exposes are security_config (in-memory) and security_config_redis (Redis-backed):
@pytest.fixture
def security_config() -> SecurityConfig:
return SecurityConfig(
enable_redis=False,
whitelist=["127.0.0.1"],
blacklist=["192.168.1.1"],
blocked_user_agents=[r"badbot"],
auto_ban_threshold=3,
auto_ban_duration=300,
)
@pytest.fixture
def security_config_redis(ipinfo_db_path: Path) -> SecurityConfig:
return SecurityConfig(
redis_url=REDIS_URL,
redis_prefix=REDIS_PREFIX,
whitelist=["127.0.0.1"],
blacklist=["192.168.1.1"],
)
And autouse cleanup fixtures that run automatically. reset_state resets the IPBanManager singleton and restores sus_patterns_handler.patterns (along with the cloud and IPInfo handlers); redis_cleanup and reset_rate_limiter clear distributed and rate-limit state:
@pytest.fixture(autouse=True)
async def reset_state() -> AsyncGenerator[None, None]:
IPBanManager._instance = None
original_patterns = sus_patterns_handler.patterns.copy()
yield
sus_patterns_handler.patterns = original_patterns.copy()
IPBanManager._instance = None
These cleanup fixtures are critical. IPBanManager is a singleton and sus_patterns_handler holds mutable pattern state. Without resetting them between tests, state leaks across test cases.
Testing Individual Security Checks¶
Each security check can be tested in isolation by creating a mock middleware object and passing it to the check:
import pytest
from unittest.mock import AsyncMock, MagicMock
from guard_core.core.checks.implementations.ip_security import IpSecurityCheck
from guard_core.models import SecurityConfig
class MockMiddleware:
def __init__(self, config: SecurityConfig) -> None:
self.config = config
self.logger = MagicMock()
self.last_cloud_ip_refresh = 0
self.suspicious_request_counts: dict[str, dict[str, int]] = {}
self.event_bus = MagicMock()
self.event_bus.send_middleware_event = AsyncMock()
self.route_resolver = MagicMock()
self.route_resolver.should_bypass_check.return_value = False
self.response_factory = MagicMock()
self.rate_limit_handler = MagicMock()
self.agent_handler = None
self.geo_ip_handler = None
self.guard_response_factory = MockGuardResponseFactory()
async def create_error_response(
self, status_code: int, default_message: str
) -> MockGuardResponse:
return MockGuardResponse(default_message, status_code)
async def refresh_cloud_ip_ranges(self) -> None:
pass
@pytest.mark.asyncio
async def test_ip_blacklist_blocks():
config = SecurityConfig(
enable_redis=False,
blacklist=["192.168.1.100"],
)
middleware = MockMiddleware(config)
check = IpSecurityCheck(middleware)
request = MockGuardRequest(client_host="192.168.1.100")
request.state.client_ip = "192.168.1.100"
request.state.route_config = None
response = await check.check(request)
assert response is not None
assert response.status_code == 403
Testing Your Adapter's Request Wrapper¶
Verify that your wrapper correctly implements every GuardRequest property:
import pytest
from starlette.testclient import TestClient
from starlette.requests import Request
from starlette.routing import Route
from your_guard.request import StarletteGuardRequest
from guard_core.protocols.request_protocol import GuardRequest
def test_protocol_conformance():
scope = {
"type": "http",
"method": "GET",
"path": "/test",
"query_string": b"key=value",
"headers": [(b"host", b"example.com"), (b"user-agent", b"test")],
"server": ("example.com", 443),
"scheme": "https",
}
request = Request(scope)
wrapped = StarletteGuardRequest(request)
assert isinstance(wrapped, GuardRequest)
def test_url_path():
scope = {"type": "http", "method": "GET", "path": "/api/users"}
request = Request(scope)
wrapped = StarletteGuardRequest(request)
assert wrapped.url_path == "/api/users"
def test_url_scheme():
scope = {"type": "http", "method": "GET", "path": "/", "scheme": "https"}
request = Request(scope)
wrapped = StarletteGuardRequest(request)
assert wrapped.url_scheme == "https"
def test_method():
scope = {"type": "http", "method": "POST", "path": "/"}
request = Request(scope)
wrapped = StarletteGuardRequest(request)
assert wrapped.method == "POST"
def test_client_host():
scope = {"type": "http", "method": "GET", "path": "/", "client": ("10.0.0.1", 8080)}
request = Request(scope)
wrapped = StarletteGuardRequest(request)
assert wrapped.client_host == "10.0.0.1"
def test_client_host_none():
scope = {"type": "http", "method": "GET", "path": "/"}
request = Request(scope)
wrapped = StarletteGuardRequest(request)
assert wrapped.client_host is None
def test_headers():
scope = {
"type": "http",
"method": "GET",
"path": "/",
"headers": [(b"x-custom", b"value")],
}
request = Request(scope)
wrapped = StarletteGuardRequest(request)
assert wrapped.headers.get("x-custom") == "value"
@pytest.mark.asyncio
async def test_body():
async def receive():
return {"type": "http.request", "body": b"test body"}
scope = {"type": "http", "method": "POST", "path": "/"}
request = Request(scope, receive)
wrapped = StarletteGuardRequest(request)
assert await wrapped.body() == b"test body"
def test_state_is_mutable():
scope = {"type": "http", "method": "GET", "path": "/"}
request = Request(scope)
wrapped = StarletteGuardRequest(request)
wrapped.state.custom_value = "test"
assert wrapped.state.custom_value == "test"
def test_url_replace_scheme():
scope = {
"type": "http",
"method": "GET",
"path": "/test",
"scheme": "http",
"server": ("example.com", 80),
}
request = Request(scope)
wrapped = StarletteGuardRequest(request)
replaced = wrapped.url_replace_scheme("https")
assert replaced.startswith("https://")
Testing Your Response Wrapper¶
from your_guard.response import StarletteGuardResponse, StarletteResponseFactory
from guard_core.protocols.response_protocol import GuardResponse, GuardResponseFactory
def test_response_protocol_conformance():
factory = StarletteResponseFactory()
response = factory.create_response("Forbidden", 403)
assert isinstance(response, GuardResponse)
def test_response_factory_protocol_conformance():
factory = StarletteResponseFactory()
assert isinstance(factory, GuardResponseFactory)
def test_create_response():
factory = StarletteResponseFactory()
response = factory.create_response("Not Found", 404)
assert response.status_code == 404
def test_headers_are_mutable():
factory = StarletteResponseFactory()
response = factory.create_response("OK", 200)
response.headers["X-Custom"] = "test"
assert response.headers["X-Custom"] == "test"
def test_create_redirect():
factory = StarletteResponseFactory()
response = factory.create_redirect_response("https://example.com", 301)
assert response.status_code == 301
assert response.headers["Location"] == "https://example.com"
Testing the Pipeline¶
Test the full security pipeline using mock objects:
import pytest
from guard_core.core.checks.pipeline import SecurityCheckPipeline
from guard_core.core.checks.base import SecurityCheck
class AlwaysPassCheck(SecurityCheck):
check_name = "always_pass"
async def check(self, request):
return None
class AlwaysBlockCheck(SecurityCheck):
check_name = "always_block"
async def check(self, request):
return await self.create_error_response(403, "Blocked")
@pytest.mark.asyncio
async def test_pipeline_passes_when_all_checks_pass():
middleware = MockMiddleware(SecurityConfig(enable_redis=False))
pipeline = SecurityCheckPipeline([
AlwaysPassCheck(middleware),
AlwaysPassCheck(middleware),
])
request = MockGuardRequest()
result = await pipeline.execute(request)
assert result is None
@pytest.mark.asyncio
async def test_pipeline_blocks_on_first_failure():
middleware = MockMiddleware(SecurityConfig(enable_redis=False))
pipeline = SecurityCheckPipeline([
AlwaysPassCheck(middleware),
AlwaysBlockCheck(middleware),
AlwaysPassCheck(middleware),
])
request = MockGuardRequest()
result = await pipeline.execute(request)
assert result is not None
assert result.status_code == 403
Integration Testing Patterns¶
Full Middleware Test (FastAPI Example)¶
import pytest
from httpx import ASGITransport, AsyncClient
from fastapi import FastAPI
from your_guard.middleware import SecurityMiddleware
from guard_core.models import SecurityConfig
@pytest.fixture
def app():
config = SecurityConfig(
enable_redis=False,
blacklist=["10.0.0.1"],
)
app = FastAPI()
app.add_middleware(SecurityMiddleware, config=config)
@app.get("/")
async def root():
return {"status": "ok"}
return app
@pytest.fixture
async def client(app):
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as client:
yield client
@pytest.mark.asyncio
async def test_allowed_request(client):
response = await client.get("/", headers={"X-Forwarded-For": "192.168.1.1"})
assert response.status_code == 200
@pytest.mark.asyncio
async def test_blacklisted_ip(client):
response = await client.get("/", headers={"X-Forwarded-For": "10.0.0.1"})
assert response.status_code == 403
Testing with Decorators¶
@pytest.fixture
def decorated_app():
config = SecurityConfig(enable_redis=False)
app = FastAPI()
guard = SecurityDecorator(config)
app.state.guard_decorator = guard
middleware = SecurityMiddleware(app, config=config)
middleware.set_decorator_handler(guard)
app.add_middleware(SecurityMiddleware, config=config)
@app.get("/limited")
@guard.rate_limit(requests=2, window=60)
async def limited():
return {"status": "ok"}
return app
Testing Singleton Cleanup¶
Guard-core uses singletons for IPBanManager, SusPatternsManager, and SecurityHeadersManager. Always reset them in your test suite:
@pytest.fixture(autouse=True)
def cleanup_singletons():
from guard_core.handlers.ipban_handler import IPBanManager
from guard_core.handlers.security_headers_handler import SecurityHeadersManager
from guard_core.handlers.suspatterns_handler import SusPatternsManager
IPBanManager._instance = None
SusPatternsManager._instance = None
SecurityHeadersManager._instance = None
yield
IPBanManager._instance = None
SusPatternsManager._instance = None
SecurityHeadersManager._instance = None
Testing Without Redis¶
Pass enable_redis=False to SecurityConfig for all unit tests that do not require distributed state. This avoids needing a running Redis instance:
For integration tests that require Redis, use a test fixture with a real or mock Redis connection:
import os
@pytest.fixture
def redis_config():
redis_url = os.environ.get("REDIS_URL", "redis://localhost:6379")
return SecurityConfig(
enable_redis=True,
redis_url=redis_url,
redis_prefix="test_guard:",
)
Test Configuration¶
Guard-core uses asyncio_mode = "auto" in pyproject.toml. Your adapter should do the same: