`extras.webserver` — Web Server¶

Note

Requires the webserver extra. For more information, see our installation guide.

A simple Starlette, web server running on Unicorn. Capable of handling both HTTP and websocket routes.

Note

Errors in side-effects.

Since the server (and handlers for all routes) is running in a separate thread, any errors in side-effects will not be immediately caught by the main thread. Instead, all unhandled errors in routes will be stored by the Webservice and re-raised when some methods of the WebServer, or methods of endpoints linked to the webserver, are called (see HandlerError). These methods are:

WebServer.add_http_endpoint()
WebServer.remove_http_endpoint()
WebServer.patch_http_endpoint()
WebServer.add_ws_endpoint()
WebServer.remove_ws_endpoint()
WebServer.patch_ws_endpoint()
WebServer.stop
WebServer.is_alive
MockHTTPEndpoint.patch()
MockHTTPEndpoint.capture_calls()
MockWSEndpoint.patch()
MockWSEndpoint.capture_calls()

Also, if such an error is encountered, all future calls to any route in the service will return Error Code 500.

class extras.webserver.WebServer(name: str, port: int = 0, **kwargs)[source]¶

A uvicorn-starlette web server that supports on-the-fly adding and removing of routes.

Parameters:

name – The name of the server, used for logging and debugging.
port – The port to bind to when serving, default will bind to an available port.
**kwargs – Additional keyword arguments to pass to the starlette server’s uvicorn configuration.

Note

unless overridden in **kwargs, the following values differ from uvicorn’s defaults:

host: changed to '0.0.0.0'
log_config: Changed to None, to avoid all of uvicorn’s logs.

property port: int[source]¶

The port the server is bound to. If the port was specified in construction, this will be the same. Otherwise, if the server was not started, this will be 0. If the server is started and bound to a port, this will be the port it is bound to. If the server is started but not yet bound to a port, this property will block for at most 1 second, waiting for the binding to complete.

Raises:: RuntimeError – If the binding process takes more than 1 second.

add_http_endpoint(endpoint: MockHTTPEndpoint) → MockHTTPEndpoint[source]¶

add_http_endpoint(methods, rule_string, side_effect, *, auto_read_body=True, forbid_implicit_head_verb=True, name=None) → MockHTTPEndpoint

Add an HTTP endpoint to the server. Can accept either a created endpoint or arguments to create one.

Parameters:: endpoint – The endpoint to add, as returned by http_endpoint().
Other Parameters:: Used to create a new endpoint (forwarded to http_endpoint()).
Returns:: The endpoint that was added, to be used as a decorator.

Example with decorator syntax.¶

@server.add_http_endpoint
@http_endpoint('GET', '/square/{x:int}')
async def square(request):
    return PlainTextResponse(str(request.path_params['a'] ** 2))

assert get(server.local_url() + '/square/12').text == '144'

Example for creating a new endpoint.¶

endpoint = server.add_http_endpoint('GET', 'ping', PlainTextResponse('pong'))

assert get(server.local_url() + '/ping').text == 'pong'

remove_http_endpoint(endpoint: MockHTTPEndpoint)[source]¶

Remove an HTTP endpoint previously added to the server.

Parameters:: endpoint – The endpoint to remove.
Raises:: RuntimeError – If the endpoint is not added to the server.

patch_http_endpoint(endpoint: MockHTTPEndpoint) → AbstractContextManager[MockHTTPEndpoint][source]¶

patch_http_endpoint(methods, rule_string, side_effect, *, auto_read_body=True, forbid_implicit_head_verb=True, name=None) → AbstractContextManager[MockHTTPEndpoint]

Add to, then remove an HTTP endpoint from the server within a context. Can accept either a created endpoint or arguments to create one.

Parameters:: endpoint – The endpoint to add, as returned by http_endpoint().
Other Parameters:: Used to create a new endpoint (forwarded to http_endpoint()).
Returns:: A context manager that adds and yields the endpoint upon entry, and removes it upon exit.
Return type:: ContextManager[MockHTTPEndpoint]

Example¶

@http_endpoint('GET', '/square/{x:int}')
async def square(request):
    return PlainTextResponse(str(request.path_params['a'] ** 2))

with server.patch_http_endpoint(square):
    assert get(server.local_url() + '/square/12').text == '144'

# when the context is exited, the endpoint is removed
assert get(server.local_url() + '/square/12').status_code == 404

add_ws_endpoint(endpoint: MockWSEndpoint) → MockWSEndpoint[source]¶

add_ws_endpoint(rule_string, side_effect, *, name=None, allow_abrupt_disconnect=True) → MockWSEndpoint

Add an HTTP endpoint to the server. Can accept either a created endpoint or arguments to create one.

Parameters:: endpoint – The endpoint to add, as returned by ws_endpoint().
Other Parameters:: Used to create a new endpoint (forwarded to ws_endpoint()).
Returns:: The endpoint that was added, to be used as a decorator.

Example with decorator syntax.¶

@server.add_ws_endpoint
@ws_endpoint('/moria')
async def moria(websocket):
    await websocket.accept()
    await websocket.send_text('Speak, friend, and enter')
    if await websocket.receive_text() == 'Mellon':
        return WS_1000_NORMAL_CLOSURE
    else:
        return WS_1008_POLICY_VIOLATION

ws_client = websocket.create_connection(server.local_url('ws') + '/moria')
assert ws_client.recv() == 'Speak, friend, and enter'
ws_client.send('Mellon')

remove_ws_endpoint(endpoint: MockWSEndpoint)[source]¶

Remove a websocket endpoint previously added to the server.

Parameters:: endpoint – The endpoint to remove.
Raises:: RuntimeError – If the endpoint is not added to the server.

patch_ws_endpoint(endpoint: MockWSEndpoint) → AbstractContextManager[MockWSEndpoint][source]¶

patch_ws_endpoint(rule_string, side_effect, *, name=None, allow_abrupt_disconnect=True) → AbstractContextManager[MockWSEndpoint]

Add to, then remove a websocket endpoint from the server within a context. Can accept either a created endpoint or arguments to create one.

Parameters:: endpoint – The endpoint to add, as returned by ws_endpoint().
Other Parameters:: Used to create a new endpoint (forwarded to ws_endpoint()).
Returns:: A context manager that adds and yields the endpoint upon entry, and removes it upon exit.

local_url(schema: str | None = 'http') → str[source]¶

Get the URL to access the server from the local machine, with the given schema.

Parameters:: schema – The schema to use. On None, returns a URL without a schema.
Returns:: The URL of the server.

container_url(schema: str | None = 'http') → str[source]¶

Get the URL to access the server from a docker container, with the given schema.

Parameters:: schema – The schema to use. On None, returns a URL without a schema.
Returns:: The URL of the server.

extras.webserver.http_endpoint(methods: str | Iterable[str], rule_string: str, side_effect: Response | Callable[[Request], Awaitable[Response]], *, auto_read_body: bool = True, forbid_implicit_head_verb: bool = True, name: str | None = None) → MockHTTPEndpoint[source]¶

extras.webserver.http_endpoint(methods: str | Iterable[str], rule_string: str, *, auto_read_body: bool = True, forbid_implicit_head_verb: bool = True, name: str | None = None) → Callable[[Callable[[Request], Awaitable[Response]]], MockHTTPEndpoint]

Create an HTTP endpoint to link to a Webserver (see WebServer.add_http_endpoint()).

Parameters:

methods – The HTTP method or methods to allow into the endpoint (case insensitive).
rule_string – The URL rule string as specified by Starlette URL rule specs.
side_effect –
The side effect to execute when the endpoint is requested. Can either be a Starlette response, in this case the response will always be returned, or an async callable that accepts a positional Starlette Request and returns a Starlette response. Can be delegated as a decorator.
auto_read_body – By default, Starlette may begin to respond to requests before the request body has fully arrived to the server. This may cause race condition issues on local hosts. This param (enabled by default) ensures that the entire request arrives to the server before a response is returned.
forbid_implicit_head_verb – By default for Starlette routes, if the GET method is allowed for a route , the HEAD method will also be allowed. This param (enabled by default) disables this behavior.
name – The name of the endpoint. If None, the name is inferred from the function name and rule string.

Returns:

The a new HTTP endpoint that can be added to a Webservice.

Note

this function can be used a decorator by omitting side_effect.

Example with decorator syntax.¶

@http_endpoint('GET', '/square/{x:int}')
async def square(request):
    return PlainTextResponse(str(request.path_params['a'] ** 2))

# is equivalent to:

async def square(request):
    return PlainTextResponse(str(request.path_params['a'] ** 2))

square = http_endpoint('GET', '/square/{x:int}', square)

Note

In order to use a “rotating” side effect (i.e. one that returns a different response per request), see iter_side_effects().

extras.webserver.ws_endpoint(rule_string: str, side_effect: Callable[[Websocket], Awaitable[int | None]], *, name: str = None, allow_abrupt_disconnect: bool = True) → MockWSEndpoint[source]¶

extras.webserver.ws_endpoint(rule_string: str, *, name: str = None, allow_abrupt_disconnect: bool = True) → Callable[[Callable[[Websocket], Awaitable[int | None]]], MockWSEndpoint]

Create a WebSocket endpoint to link to a Webserver (see WebServer.add_ws_endpoint()).

Parameters:

rule_string (str) –
The URL rule string as specified by Starlette URL rule specs.
side_effect (async WebSocket → (int | None)) – The side effect to execute when the endpoint is requested. Should be an async callable that accepts a positional Starlette WebSocket. If the callable returns an integer, the connection is closed with that exit code. Can be delegated as a decorator.
name (str | None) – The name of the endpoint. If None, the name is inferred from the function name and rule string.
allow_abrupt_disconnect – Whether to automatically ignore and close websocket connections that the client closes before the server expects.

Returns:

The a new Websocket endpoint that can be added to a Webservice.

Note

this function can be used a decorator by omitting side_effect.

Example with decorator syntax.¶

@ws_endpoint('/square')
async def square(ws: WebSocket):
  await ws.accept()
  x = int(await ws.receive_text())
  await ws.send_text(str(x*x))
  await ws.close()

# is equivalent to:

async def square(ws: WebSocket):
  await ws.accept()
  x = int(await ws.receive_text())
  await ws.send_text(str(x*x))
  await ws.close()

square = ws_endpoint('/square', square)

Note

In order to use a “rotating” side effect (i.e. one that returns a different response per request), see iter_side_effects().

class extras.webserver.MockHTTPEndpoint[source]¶

An HTTP endpoint that can be added to a Webserver (see WebServer.add_http_endpoint()). construct with http_endpoint().

side_effect[source]¶
The current side effect of the endpoint

patch(side_effect: Response | Callable[[Request], Awaitable[Response]]) → AbstractContextManager[...][source]¶

Change the side effect of the endpoint. With the ability to revert it to the original side effect.

Parameters:: side_effect – The new side effect to execute when the endpoint is requested. Accepts the same types as http_endpoint().
Returns:: A context manager that reverts the endpoint’s side effect to the original value, if ever exited.

Example.¶

mock_http_endpoint = http_endpoint('GET', '/ping', PlainTextResponse('pong'))
# the endpoint will return 'pong' if called now
with mock_http_endpoint.patch(PlainTextResponse('pang')):
    # the endpoint will return 'pang' if called now
    ...
# the endpoint will return 'pong' if called now
mock_http_endpoint.patch(PlainTextResponse('powong'))
# the endpoint will return 'powong' if called now

Warning

Because the patch takes effect immediately, but is reversible via context management, using multiple patches out of order can have unexpected results.

mock_http_endpoint = http_endpoint('GET', '/color', PlainTextResponse('red'))
# side effect is now 'red'
patch1 = mock_http_endpoint.patch(PlainTextResponse('green'))
# side effect is now 'green'
patch2 = mock_http_endpoint.patch(PlainTextResponse('blue'))
# side effect is now 'blue'
with patch1:
    # side effect is now *still* 'blue'
    ...
# side effect is now 'red'
with patch2:
    # side effect is now *still* 'red'
    ...
# side effect is now 'green'

Therefore, it is best practice to always either discard the return value of this function, or immediately enter its context

capture_calls() → AbstractContextManager[RecordedHTTPRequests][source]¶

Capture all calls to the endpoint within a context.

Returns:: context manager that begins capturing all calls to endpoint on entry and stops recording on exit, all captured calls are recorded on the yielded http_request_capture.RecordedHTTPRequests.

Example.¶

@server.add_http_endpoint
@http_endpoint('GET', '/square/{x:int}')
async def square(request):
    return PlainTextResponse(str(request.path_params['a'] ** 2))

with square.capture_calls() as calls:
    assert get(server.local_url() + '/square/12').text == '144'
    assert get(server.local_url() + '/square/11').text == '121'

calls.assert_has_requests(
    ExpectedHTTPRequest(path_params={'x': 12}),
    ExpectedHTTPRequest(path_params={'x': 11})
)

Note

auto_read_body must be enabled to capture calls.

class extras.webserver.MockWSEndpoint[source]¶

A websocket endpoint that can be added to a Webserver (see WebServer.add_ws_endpoint()). construct with ws_endpoint().

side_effect[source]¶: The current side effect of the endpoint

patch(side_effect: Callable[[Websocket], Awaitable[int | None]]) → AbstractContextManager[...][source]¶

Change the side effect of the endpoint. With the ability to revert it to the original side effect.

Parameters:: side_effect – The new side effect to execute when the endpoint is requested. Accepts the same types as ws_endpoint().
Returns:: A context manager that reverts the endpoint’s side effect to the original value, if ever exited.

Warning

See the out-of-order warning in MockHTTPEndpoint.patch.

capture_calls() → AbstractContextManager[RecordedWSTranscripts][source]¶

Capture all calls to the endpoint within a context.

Returns:: context manager that begins capturing all calls to endpoint on entry and stops recording on exit, all captured calls are recorded on the yielded ws_request_capture.RecordedWSTranscripts.

extras.webserver.class_http_endpoint(methods, rule_string, side_effect, *, auto_read_body=True, forbid_implicit_head_verb=True, name=None)[source]¶

extras.webserver.class_http_endpoint(methods, rule_string, *, auto_read_body=True, forbid_implicit_head_verb=True, name=None)

extras.webserver.class_ws_endpoint(rule_string, side_effect, *, name=None, allow_abrupt_disconnect=True)[source]¶

extras.webserver.class_ws_endpoint(rule_string, *, name=None, allow_abrupt_disconnect=True)

Create an endpoint template. Declare this in a WebServer subclass body to automatically add an endpoint to all instances of the subclass.

All arguments are the same as http_endpoint() and ws_endpoint().

Example.¶

class MyWebServer(WebServer):
    @class_http_endpoint('GET', '/hello')
    async def hello(self, request):
        return PlainTextResponse('Hello, World!')

    @class_ws_endpoint('/echo')
    async def echo(self, websocket):
        await websocket.accept()
        while True:
            message = await websocket.receive_text()
            await websocket.send_text(message)

server = MyWebServer("my name").start()

assert get(server.local_url() + '/hello').text == 'Hello, World!'

async with websocket_connect(server.local_url() + '/echo') as websocket:
    await websocket.send_text('Hello, World!')
    assert await websocket.receive_text() == 'Hello, World!'

class extras.webserver.ExpectedHTTPRequest(headers: Mapping[str, Collection[str]] = None, headers_submap: Mapping[str, Collection[str]] = None, path: str | Pattern[str] = None, path_params: Mapping[str, ...] = None, path_params_submap: Mapping[str, ...] = None, query_params: Mapping[str, Collection[str]] = None, query_params_submap: Mapping[str, Collection[str]] = None, method: str = None, body: bytes = None, text: str = None, json=..., json_submap: Mapping[str, ...] = None, content_predicate: Callable[[bytes], bool] | tuple[Callable[[bytes], T], T] = None)[source]¶

An expected HTTP request, used for matching a recorded request.

Parameters:

headers – If specified, a recorded request must have the specified headers exactly.
headers_submap – If specified, a recorded request must have at least the specified headers.
path – If specified, a recorded request must have the specified path exactly (if str), or must match the specified pattern fully (if Pattern).
path_params – If specified, a recorded request must have the specified path parameters exactly.
path_params_submap – If specified, a recorded request must have at least the specified path parameters.
query_params – If specified, a recorded request must have the specified query parameters exactly.
query_params_submap – If specified, a recorded request must have at least the specified query parameters.
method – If specified, a recorded request must be of the specified HTTP method (case-insensitive).
body – If specified, a recorded request must have a body equal to the one specified.
text – If specified, a recorded request must have a body equal to the one specified with utf-8 encoding.
json – If specified, a recorded request must have a body equal to the one specified with json encoding.
json_submap – If specified, a recorded request must have a body that is a supermap of the one specified with json encoding.
content_predicate – If specified, may be a callable that accepts a bytes object, in which case the predicate must evaluate to True when called with the request body. Alternatively, the predicate can be a tuple of a callable that returns a value, and a value to compare to, in this case, the callable must return the value specified when called with the request body.

Note

Only one of body, text, json, json_submap, or content_predicate may be specified. Additionally, only a parameter or its *_submap variant may be specified, but not both.

class http_request_capture.RecordedHTTPRequest[source]¶

A recorded HTTP request. Yielded by MockHTTPEndpoint.capture_calls() to record requests.

headers: Mapping[str, Sequence[str]][source]¶: The request headers.

method: str[source]¶: The request method.

path: str[source]¶: The request path.

path_params: Mapping[str, ...][source]¶: The request path parameters.

query_params: Mapping[str, Sequence[str]][source]¶: The request query parameters.

content: bytes[source]¶: The request body.

time_received: datetime[source]¶: The time at which the request began processing. Not timezone-aware.

text(encoding='utf-8') → str[source]¶

Return the request body as a str decoded with the specified encoding.

Parameters:: encoding – The encoding to use. Defaults to 'utf-8'.

json()[source]¶: Return the request body as a json-decoded value.

class http_request_capture.RecordedHTTPRequests[source]¶

A list of http_request_capture.RecordedHTTPRequest. Yielded by MockHTTPEndpoint.capture_calls() to record requests.

assert_not_requested()[source]¶

Assert that no requests were made.

Raises:: AssertionError – If any requests were made.

assert_requested()[source]¶

Assert that at least one request was made.

Raises:: AssertionError – If no requests were made.

assert_requested_once()[source]¶

Assert that exactly one request was made.

Raises:: AssertionError – If no or multiple request were made.

assert_requested_with(expected: ExpectedHTTPRequest)[source]¶

assert_requested_with(**kwargs)

Assert that the latest request was made matches an expected request.

Parameters:

expected – The expected request.
**kwargs – Alternatively, users can skip the expected argument and specify the expected request parameters as keyword arguments.

Raises:

AssertionError – If the last request doesn’t match the expected request, or if there are no requests.

Example¶

recorded: RecordedHTTPRequests = ...

recorded.assert_requested_with(ExpectedHTTPRequest(
    content_predicate = (lambda b:b.decode('utf-7'), 'hi'),
))

# is equivelant to
recorded.assert_requested_with(
    content_predicate = (lambda b:b.decode('utf-7'), 'hi'),
)

assert_requested_once_with(expected: ExpectedHTTPRequest)[source]¶

assert_requested_once_with(**kwargs)

Assert that only one request was made, and that it matches an expected request.

Parameters:

expected – The expected request.
**kwargs – Alternatively, users can skip the expected argument and specify the expected request parameters as keyword arguments (see the example in assert_requested_with).

Raises:

AssertionError – If there are more than one or no requests made, or if the only request does not match the expected request.

assert_any_request(expected: ExpectedHTTPRequest)[source]¶

assert_any_request(**kwargs)

Assert that a request was made that matches an expected request.

Parameters:

expected – The expected request.
**kwargs – Alternatively, users can skip the expected argument and specify the expected request parameters as keyword arguments (see the example in assert_requested_with).

Raises:

AssertionError – If no request that matches the expectation was made.

assert_has_requests(*expected_requests: ExpectedHTTPRequest)[source]¶

Assert that a set of expected requests were made, in sequential order.

Parameters:: *expected – The expected requests to match.
Raises:: AssertionError – If the expected requests were not matched in sequential order.

class extras.webserver.Sender[source]¶

An Enum class for two senders in a Websocket connection. Used to create an expected websocket message.

Client[source]¶
Server[source]¶: A sender of a message to the recipient.

__call__(data: str | bytes | Pattern[str | bytes] | ...)[source]¶

Create an expectation of a websocket message from the given data, as sent by the given sender.

Parameters:

data –

The payload of the expected message. Can be one of:

str: A string expected to be the payload.
bytes: A bytes expected to be the payload.
Pattern[bytes | str]: A pattern that the payload is expected to fully match.
Ellipsis: A special value that matches any payload.

class extras.webserver.ExpectedWSTranscript(messages: Sequence[Sender[...] | Ellipsis] = (...,), headers: Mapping[str, Collection[str]] = None, headers_submap: Mapping[str, Collection[str]] = None, path: str | Pattern[str] = None, path_params: Mapping[str, ...] = None, path_params_submap: Mapping[str, ...] = None, query_params: Mapping[str, Collection[str]] = None, query_params_submap: Mapping[str, Collection[str]] = None, close: tuple[Sender, int] = None, accepted: bool = None)[source]¶

An expectation of a websocket transcript. Used to match against a recorded websocket transcript.

Parameters:

messages – The expected messages in the transcript, in order. Create an expected message by calling Sender. The sequence may begin or end with Ellipsis to signify that any number of messages can precede or follow the messages to match.
headers – If specified, a recorded request must have the specified headers exactly.
headers_submap – If specified, a recorded request must have at least the specified headers.
path – If specified, a recorded request must have the specified path exactly (if str), or must match the specified pattern fully (if Pattern).
path_params – If specified, a recorded request must have the specified path parameters exactly.
path_params_submap – If specified, a recorded request must have at least the specified path parameters.
query_params – If specified, a recorded request must have the specified query parameters exactly.
query_params_submap – If specified, a recorded request must have at least the specified query parameters.
close – If specified, the transcript closing must have been done by the specified sender, and with the specified code.
accepted – If specified, the connection must have been accepted by the server (if True), or rejected by the server (if False).

Note

Only a parameter or its *_submap variant may be specified, but not both.

Example usage¶

expected_transcript = ExpectedWSTranscript([
    Sender.Server(b'hi there, what is your name?'),
    Sender.Client(re.compile(b'My name is [A-Z][a-z]+')),
    Sender.Client(b'And I like pie'),
    ...
], close=(Sender.Server, 1000))

# requires that the transcript will begin with the server sending 'hi
# there, what is your name?', then the client should respond with a
# name, then the client should respond with 'And I like pie'. Any
# number of messages can follow after that, but eventually the server
# should close the connection with code 1000.

class ws_request_capture.RecordedWSTranscripts[source]¶

A list of recorded websocket requests. Yielded by MockWSEndpoint.capture_calls() to record transcripts.

assert_not_requested()[source]¶

Assert that no connections were made.

Raises:: AssertionError – If any connections were made.

assert_requested()[source]¶

Assert that at least one connection was made.

Raises:: AssertionError – If no connections were made.

assert_requested_once()[source]¶

Assert that exactly one connection was made.

Raises:: AssertionError – If no or multiple connections were made.

assert_requested_with(expected: ExpectedWSTranscript)[source]¶

assert_requested_with(**kwargs)

Assert that the latest connections was made matches an expected transcript.

Parameters:

expected – The expected connection.
**kwargs – Alternatively, users can skip the expected argument and specify the expected transcript parameters as keyword arguments.

Raises:

AssertionError – If the last transcript doesn’t match the expected request, or if there are no transcript.

assert_requested_once_with(expected: ExpectedWSTranscript)[source]¶

assert_requested_once_with(**kwargs)

Assert that only one connection was made, and that it matches an expected transcript.

Parameters:

expected – The expected connection.
**kwargs – Alternatively, users can skip the expected argument and specify the expected request parameters as keyword arguments.

Raises:

AssertionError – If there are more than one or no connections made, or if the only transcript does not match the expectation.

assert_any_request(expected: ExpectedWSTranscript)[source]¶

assert_any_request(**kwargs)

Assert that a connection was made that matches an expected transcript.

Parameters:

expected – The expected connection.
**kwargs – Alternatively, users can skip the expected argument and specify the expected request parameters as keyword arguments.

Raises:

AssertionError – If no connection that matches the expectation was made.

extras.webserver.iter_side_effects(side_effects: Iterable) → Callable[source]¶

Combine multiple endpoint side effects into one, so that each subsequent call uses the next side effect.

Parameters:: side_effects – iterable of the side effects to combine.
Returns:: A function that can be used as an endpoint side effect.

Note

This function respects the special case of a side effect being a starlette response.

Warning

If there are less side effects than calls, a StopIteration will be raises within the handler. Which is why it is recommended to use itertools.cycle() to ensure that there are infinite side effects.

Example of infinite side effects¶

side_effect = iter_side_effects(itertools.chain(
    [
        PlainTextResponse('hi'),
        PlainTextResponse('hello'),
        PlainTextResponse('how are you?'),
    ],
    itertools.cycle([PlainTextResponse('im tired now')])
))
endpoint = server.add_http_endpoint('GET', '/', side_effect)
assert get(server.local_url()+'/').text == 'hi'
assert get(server.local_url()+'/').text == 'hello'
assert get(server.local_url()+'/').text == 'how are you?'

assert get(server.local_url()+'/').text == 'im tired now'
assert get(server.local_url()+'/').text == 'im tired now'
...

Note

To reuse an existing side effect, you can use the side_effect attribute.

Example of side effects reuse¶

endpoint = server.add_http_endpoint('GET', '/', PlainTextResponse("Go!"))
side_effect = iter_side_effects(itertools.chain(
    [
        PlainTextResponse('Ready...'),
        PlainTextResponse('Set...'),
    ],
    itertools.cycle([endpoint.side_effect])
))
assert get(server.local_url()+'/').text == 'Ready...'
assert get(server.local_url()+'/').text == 'Set...'

assert get(server.local_url()+'/').text == 'Go!'
assert get(server.local_url()+'/').text == 'Go!'
assert get(server.local_url()+'/').text == 'Go!'
...

extras.webserver.verbose_http_side_effect(side_effect, format_function: Callable[[MockHTTPEndpoint, Request, Response], str] = ..., file: IO[str] = ...) → Callable[source]¶

Wraps an HTTP side effect that prints the request and response to a file.

Parameters:

side_effect – The HTTP side effect to wrap. Can be either a function or a response.
format_function – A function that takes an endpoint, request and response and returns a string. The default function will return a string consisting of the time, webserver and endpoint name, the client address, the HTTP method, the relative path with the query string, the status code and length of the response.
file – The file to write the messages to. defaults to stdout.

extras.webserver — Web Server¶

`extras.webserver` — Web Server¶