Mocking HTTP Requests
Zapros includes a small, WireMock-style mocking layer you can use in tests to match outgoing requests and return deterministic responses—no real network calls required.
Quickstart
Mocking is implemented with handlers and ensures that all outgoing requests are intercepted and matched against the mock router. It blocks I/O for unmatched requests by default, but you can also configure it to allow unmatched requests to pass through to the network.
Here is a simple example of how you can mock a request with the MockHandler:
import asyncio
from zapros import AsyncClient
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(path("/api")).respond(status=200).mount(router)
async def main():
async with AsyncClient(
handler=MockHandler(router)
) as client:
response = await client.request(
"GET",
"https://api.example.com/api",
)
print(response.status) # 200
asyncio.run(main())from zapros import Client
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(path("/api")).respond(status=200).mount(router)
with Client(handler=MockHandler(router)) as client:
response = client.request(
"GET",
"https://api.example.com/api",
)
print(response.status) # 200or, if you don't have access to the Client and can't easily inject a handler, you can use the mock_http context manager to patch the standard library's HTTP handling:
import asyncio
from zapros import AsyncClient
from zapros.mock import MockRouter
from zapros.matchers import path
from zapros.mock import mock_http
async def main():
async with AsyncClient() as client:
async with mock_http() as router:
Mock.given(path("/api")).respond(
status=200
).mount(router)
response = await client.request(
"GET",
"https://api.example.com/api",
)
assert response.status == 200
asyncio.run(main())from zapros import Client
from zapros.mock import (
MockRouter,
mock_http,
)
from zapros.matchers import path
with Client() as client:
with mock_http() as router:
Mock.given(path("/api")).respond(status=200).mount(
router
)
response = client.request(
"GET",
"https://api.example.com/api",
)
assert response.status == 200Matching Requests
Mocks match requests using matchers. A matcher inspects some part of the request (method, path, headers, etc.).
Path
import asyncio
from zapros import AsyncClient
from zapros.mock import (
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(path("/health")).respond(status=200).mount(
router
)
async def main():
async with AsyncClient(
handler=MockHandler(router)
) as client:
response = await client.request(
"GET",
"https://api.example.com/health",
)
assert response.status == 200
asyncio.run(main())from zapros import Client
from zapros.mock import (
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(path("/health")).respond(status=200).mount(
router
)
with Client(handler=MockHandler(router)) as client:
response = client.request(
"GET",
"https://api.example.com/health",
)
assert response.status == 200Matches requests where the path is /health.
Method
import asyncio
from zapros import AsyncClient
from zapros.mock import (
MockHandler,
MockRouter,
)
from zapros.matchers import method
router = MockRouter()
Mock.given(method("GET")).respond(status=200).mount(router)
async def main():
async with AsyncClient(
handler=MockHandler(router)
) as client:
response = await client.request(
"GET",
"https://api.example.com/",
)
assert response.status == 200
asyncio.run(main())from zapros import Client
from zapros.mock import (
MockHandler,
MockRouter,
)
from zapros.matchers import method
router = MockRouter()
Mock.given(method("GET")).respond(status=200).mount(router)
with Client(handler=MockHandler(router)) as client:
response = client.request(
"GET",
"https://api.example.com/",
)
assert response.status == 200Method matching is case-insensitive.
Host
import asyncio
from zapros import AsyncClient
from zapros.mock import (
MockHandler,
MockRouter,
)
from zapros.matchers import host
router = MockRouter()
Mock.given(host("api.example.com")).respond(
status=200
).mount(router)
async def main():
async with AsyncClient(
handler=MockHandler(router)
) as client:
response = await client.request(
"GET",
"https://api.example.com/",
)
assert response.status == 200
asyncio.run(main())from zapros import Client
from zapros.mock import (
MockHandler,
MockRouter,
)
from zapros.matchers import host
router = MockRouter()
Mock.given(host("api.example.com")).respond(
status=200
).mount(router)
with Client(handler=MockHandler(router)) as client:
response = client.request(
"GET",
"https://api.example.com/",
)
assert response.status == 200Headers
import asyncio
from zapros import AsyncClient
from zapros.mock import (
MockHandler,
MockRouter,
)
from zapros.matchers import header
router = MockRouter()
Mock.given(header("authorization", "Bearer token")).respond(
status=200
).mount(router)
async def main():
async with AsyncClient(
handler=MockHandler(router)
) as client:
response = await client.request(
"GET",
"https://api.example.com/",
headers={"authorization": "Bearer token"},
)
assert response.status == 200
asyncio.run(main())from zapros import Client
from zapros.mock import (
MockHandler,
MockRouter,
)
from zapros.matchers import header
router = MockRouter()
Mock.given(header("authorization", "Bearer token")).respond(
status=200
).mount(router)
with Client(handler=MockHandler(router)) as client:
response = client.request(
"GET",
"https://api.example.com/",
headers={"authorization": "Bearer token"},
)
assert response.status == 200Query Parameters
import asyncio
from zapros import AsyncClient
from zapros.mock import (
MockHandler,
MockRouter,
)
from zapros.matchers import query
router = MockRouter()
Mock.given(query(page="2", limit="10")).respond(
status=200
).mount(router)
async def main():
async with AsyncClient(
handler=MockHandler(router)
) as client:
response = await client.request(
"GET",
"https://api.example.com/items",
params={
"page": "2",
"limit": "10",
},
)
assert response.status == 200
asyncio.run(main())from zapros import Client
from zapros.mock import (
MockHandler,
MockRouter,
)
from zapros.matchers import query
router = MockRouter()
Mock.given(query(page="2", limit="10")).respond(
status=200
).mount(router)
with Client(handler=MockHandler(router)) as client:
response = client.request(
"GET",
"https://api.example.com/items",
params={
"page": "2",
"limit": "10",
},
)
assert response.status == 200JSON Body
import asyncio
from zapros import AsyncClient
from zapros.mock import (
MockHandler,
MockRouter,
)
from zapros.matchers import json
router = MockRouter()
Mock.given(
json(lambda body: body["name"] == "test")
).respond(status=200).mount(router)
async def main():
async with AsyncClient(
handler=MockHandler(router)
) as client:
response = await client.request(
"POST",
"https://api.example.com/items",
json={"name": "test"},
)
assert response.status == 200
asyncio.run(main())from zapros import Client
from zapros.mock import (
MockHandler,
MockRouter,
)
from zapros.matchers import json
router = MockRouter()
Mock.given(
json(lambda body: body["name"] == "test")
).respond(status=200).mount(router)
with Client(handler=MockHandler(router)) as client:
response = client.request(
"POST",
"https://api.example.com/items",
json={"name": "test"},
)
assert response.status == 200The function receives the parsed JSON body.
Combining Matchers
Matchers can be combined using logical helpers.
AND
import asyncio
from zapros import AsyncClient
from zapros.mock import (
MockHandler,
MockRouter,
)
from zapros.matchers import (
and_,
method,
path,
)
router = MockRouter()
Mock.given(and_(method("GET"), path("/health"))).respond(
status=200
).mount(router)
async def main():
async with AsyncClient(
handler=MockHandler(router)
) as client:
response = await client.request(
"GET",
"https://api.example.com/health",
)
assert response.status == 200
asyncio.run(main())from zapros import Client
from zapros.mock import (
MockHandler,
MockRouter,
)
from zapros.matchers import (
and_,
method,
path,
)
router = MockRouter()
Mock.given(and_(method("GET"), path("/health"))).respond(
status=200
).mount(router)
with Client(handler=MockHandler(router)) as client:
response = client.request(
"GET",
"https://api.example.com/health",
)
assert response.status == 200OR
import asyncio
from zapros import AsyncClient
from zapros.mock import (
MockHandler,
MockRouter,
)
from zapros.matchers import or_, path
router = MockRouter()
Mock.given(or_(path("/health"), path("/status"))).respond(
status=200
).mount(router)
async def main():
async with AsyncClient(
handler=MockHandler(router)
) as client:
response = await client.request(
"GET",
"https://api.example.com/status",
)
assert response.status == 200
asyncio.run(main())from zapros import Client
from zapros.mock import (
MockHandler,
MockRouter,
)
from zapros.matchers import or_, path
router = MockRouter()
Mock.given(or_(path("/health"), path("/status"))).respond(
status=200
).mount(router)
with Client(handler=MockHandler(router)) as client:
response = client.request(
"GET",
"https://api.example.com/status",
)
assert response.status == 200NOT
import asyncio
from zapros import AsyncClient
from zapros.mock import (
MockHandler,
MockRouter,
)
from zapros.matchers import method, not_
router = MockRouter()
Mock.given(not_(method("POST"))).respond(status=200).mount(
router
)
async def main():
async with AsyncClient(
handler=MockHandler(router)
) as client:
response = await client.request(
"GET",
"https://api.example.com/",
)
assert response.status == 200
asyncio.run(main())from zapros import Client
from zapros.mock import (
MockHandler,
MockRouter,
)
from zapros.matchers import method, not_
router = MockRouter()
Mock.given(not_(method("POST"))).respond(status=200).mount(
router
)
with Client(handler=MockHandler(router)) as client:
response = client.request(
"GET",
"https://api.example.com/",
)
assert response.status == 200Fluent Matcher API
Matchers can also be chained fluently:
import asyncio
from zapros import AsyncClient
from zapros.mock import (
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(
path("/api/users")
.method("POST")
.header(
"content-type",
"application/json",
)
).respond(status=201).mount(router)
async def main():
async with AsyncClient(
handler=MockHandler(router)
) as client:
response = await client.request(
"POST",
"https://api.example.com/api/users",
headers={"content-type": "application/json"},
)
assert response.status == 201
asyncio.run(main())from zapros import Client
from zapros.mock import (
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(
path("/api/users")
.method("POST")
.header(
"content-type",
"application/json",
)
).respond(status=201).mount(router)
with Client(handler=MockHandler(router)) as client:
response = client.request(
"POST",
"https://api.example.com/api/users",
headers={"content-type": "application/json"},
)
assert response.status == 201Custom Matchers
Any class with a match(self, request: Request) -> bool method satisfies the Matcher protocol. Extending BaseMatcher also gives your matcher the fluent chaining API (.method(), .path(), .header(), etc.).
import asyncio
from zapros import AsyncClient, Request
from zapros.matchers import Matcher
from zapros.mock import (
MockHandler,
MockRouter,
)
class PathPrefixMatcher(Matcher):
def __init__(self, prefix: str) -> None:
self._prefix = prefix
def match(self, request: Request) -> bool:
return request.url.pathname.startswith(self._prefix)
router = MockRouter()
Mock.given(PathPrefixMatcher("/api/v1")).respond(
status=200
).mount(router)
async def main():
async with AsyncClient(
handler=MockHandler(router)
) as client:
assert (
await client.request(
"GET",
"https://api.example.com/api/v1/users",
)
).status == 200
assert (
await client.request(
"GET",
"https://api.example.com/api/v1/orders",
)
).status == 200
asyncio.run(main())from zapros import Client, Request
from zapros.matchers import Matcher
from zapros.mock import (
MockHandler,
MockRouter,
)
class PathPrefixMatcher(Matcher):
def __init__(self, prefix: str) -> None:
self._prefix = prefix
def match(self, request: Request) -> bool:
return request.url.pathname.startswith(self._prefix)
router = MockRouter()
Mock.given(PathPrefixMatcher("/api/v1")).respond(
status=200
).mount(router)
with Client(handler=MockHandler(router)) as client:
assert (
client.request(
"GET",
"https://api.example.com/api/v1/users",
).status
== 200
)
assert (
client.request(
"GET",
"https://api.example.com/api/v1/orders",
).status
== 200
)Custom matchers compose with all built-in helpers:
Mock.given(
PathPrefixMatcher("/api/v1").method("GET")
).respond(status=200).mount(router)Returning Responses
JSON Response
import asyncio
from zapros import AsyncClient
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(path("/data")).respond(
status=200, json={"key": "value"}
).mount(router)
async def main():
async with AsyncClient(
handler=MockHandler(router)
) as client:
response = await client.request(
"GET",
"https://api.example.com/data",
)
assert (await response.ajson()) == {"key": "value"}
asyncio.run(main())from zapros import Client
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(path("/data")).respond(
status=200, json={"key": "value"}
).mount(router)
with Client(handler=MockHandler(router)) as client:
response = client.request(
"GET",
"https://api.example.com/data",
)
assert response.json() == {"key": "value"}Text Response
import asyncio
from zapros import AsyncClient
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(path("/hello")).respond(
status=200, text="Hello World"
).mount(router)
async def main():
async with AsyncClient(
handler=MockHandler(router)
) as client:
response = await client.request(
"GET",
"https://api.example.com/hello",
)
assert await response.atext() == "Hello World"
asyncio.run(main())from zapros import Client
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(path("/hello")).respond(
status=200, text="Hello World"
).mount(router)
with Client(handler=MockHandler(router)) as client:
response = client.request(
"GET",
"https://api.example.com/hello",
)
assert response.text() == "Hello World"Custom Headers
import asyncio
from zapros import AsyncClient
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(path("/custom")).respond(
status=201,
text="Created",
headers={"x-custom": "header"},
).mount(router)
async def main():
async with AsyncClient(
handler=MockHandler(router)
) as client:
response = await client.request(
"GET",
"https://api.example.com/custom",
)
assert response.status == 201
assert response.headers["x-custom"] == "header"
asyncio.run(main())from zapros import Client
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(path("/custom")).respond(
status=201,
text="Created",
headers={"x-custom": "header"},
).mount(router)
with Client(handler=MockHandler(router)) as client:
response = client.request(
"GET",
"https://api.example.com/custom",
)
assert response.status == 201
assert response.headers["x-custom"] == "header"Dynamic Responses
You can generate responses dynamically using a callback:
import asyncio
from zapros import AsyncClient, Response
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import method
router = MockRouter()
def handler(req):
if req.url.pathname == "/notfound":
return Response(status=404)
return Response(status=200)
Mock.given(method("GET")).callback(handler).mount(router)
async def main():
async with AsyncClient(
handler=MockHandler(router)
) as client:
assert (
await client.request(
"GET",
"https://api.example.com/notfound",
)
).status == 404
assert (
await client.request(
"GET",
"https://api.example.com/anything",
)
).status == 200
asyncio.run(main())from zapros import Client, Response
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import method
router = MockRouter()
def handler(req):
if req.url.pathname == "/notfound":
return Response(status=404)
return Response(status=200)
Mock.given(method("GET")).callback(handler).mount(router)
with Client(handler=MockHandler(router)) as client:
assert (
client.request(
"GET",
"https://api.example.com/notfound",
).status
== 404
)
assert (
client.request(
"GET",
"https://api.example.com/anything",
).status
== 200
)Expectations
Mocks can verify how many times they were called.
Exact Count
import asyncio
from zapros import AsyncClient
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(path("/api")).respond(status=200).expect(
2
).mount(router)
async def main():
async with AsyncClient(
handler=MockHandler(router)
) as client:
await client.request(
"GET",
"https://api.example.com/api",
)
await client.request(
"GET",
"https://api.example.com/api",
)
router.verify()
asyncio.run(main())from zapros import Client
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(path("/api")).respond(status=200).expect(
2
).mount(router)
with Client(handler=MockHandler(router)) as client:
client.request(
"GET",
"https://api.example.com/api",
)
client.request(
"GET",
"https://api.example.com/api",
)
router.verify()Once
import asyncio
from zapros import AsyncClient
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(path("/api")).respond(status=200).once().mount(
router
)
async def main():
async with AsyncClient(
handler=MockHandler(router)
) as client:
await client.request(
"GET",
"https://api.example.com/api",
)
router.verify()
asyncio.run(main())from zapros import Client
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(path("/api")).respond(status=200).once().mount(
router
)
with Client(handler=MockHandler(router)) as client:
client.request(
"GET",
"https://api.example.com/api",
)
router.verify()Never
import asyncio
from zapros import AsyncClient
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(path("/api")).respond(status=200).never().mount(
router
)
# No requests made
router.verify()from zapros import Client
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(path("/api")).respond(status=200).never().mount(
router
)
router.verify()Sequences
When a mock has an expected call count set via expect(n), once(), or never(), it stops matching once that count is reached. This lets you register multiple mocks for the same path that fire in order.
import asyncio
from zapros import AsyncClient
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(path("/api")).respond(status=200).once().mount(
router
)
Mock.given(path("/api")).respond(status=500).once().mount(
router
)
async def main():
async with AsyncClient(
handler=MockHandler(router)
) as client:
assert (
await client.request(
"GET",
"https://api.example.com/api",
)
).status == 200
assert (
await client.request(
"GET",
"https://api.example.com/api",
)
).status == 500
asyncio.run(main())from zapros import Client
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(path("/api")).respond(status=200).once().mount(
router
)
Mock.given(path("/api")).respond(status=500).once().mount(
router
)
with Client(handler=MockHandler(router)) as client:
assert (
client.request(
"GET",
"https://api.example.com/api",
).status
== 200
)
assert (
client.request(
"GET",
"https://api.example.com/api",
).status
== 500
)Mocks are matched in registration order. Once a mock is exhausted it is skipped, so the next registered mock with matching matchers takes over.
MockRouter
The MockRouter stores mocks and dispatches requests to the first matching mock.
import asyncio
from zapros import AsyncClient
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(path("/health")).respond(status=200).mount(
router
)
Mock.given(path("/status")).respond(status=204).mount(
router
)
async def main():
async with AsyncClient(
handler=MockHandler(router)
) as client:
assert (
await client.request(
"GET",
"https://api.example.com/health",
)
).status == 200
assert (
await client.request(
"GET",
"https://api.example.com/status",
)
).status == 204
asyncio.run(main())from zapros import Client
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import path
router = MockRouter()
Mock.given(path("/health")).respond(status=200).mount(
router
)
Mock.given(path("/status")).respond(status=204).mount(
router
)
with Client(handler=MockHandler(router)) as client:
assert (
client.request(
"GET",
"https://api.example.com/health",
).status
== 200
)
assert (
client.request(
"GET",
"https://api.example.com/status",
).status
== 204
).mount(router) is the preferred way to register a mock. If you need to add a pre-built Mock object directly, router.add(mock) is also available.
Dispatching happens automatically when used with MockHandler.
Verifying Mocks
At the end of a test you can verify all expectations:
router.verify()If expectations are not met, an AssertionError is raised.
Resetting Mocks
Reset call counts between tests:
router.reset()Async Support
MockHandler also works with async handlers.
handler = MockHandler(router)
response = await handler.ahandle(request)If no mock matches and no fallback handler is configured, a ValueError is raised.
You can optionally provide a fallback handler:
handler = MockHandler(router, fallback=my_handler)Example Test
import asyncio
from zapros import AsyncClient
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import path
async def test_api():
router = MockRouter()
Mock.given(path("/users").method("GET")).respond(
status=200, json=[]
).once().mount(router)
async with AsyncClient(
handler=MockHandler(router)
) as client:
response = await client.request(
"GET",
"https://example.com/users",
)
assert response.status == 200
asyncio.run(test_api())from zapros import Client
from zapros.mock import (
Mock,
MockHandler,
MockRouter,
)
from zapros.matchers import path
def test_api():
router = MockRouter()
Mock.given(path("/users").method("GET")).respond(
status=200, json=[]
).once().mount(router)
with Client(handler=MockHandler(router)) as client:
response = client.request(
"GET",
"https://example.com/users",
)
assert response.status == 200This approach lets you build deterministic HTTP tests without network access.