05 - Add tests
BRANCH: You can start from the branch
all-requests.
git checkout all-requests
Finally, we are going to add some tests.
Add tests for fetch
We are going to use very popular test framework pytest.
1. Create a test directory and test file.
mkdir tests
touch __init__.py
touch test_fetch.py
2. Install pytest, pytest-asyncio and pytest-httpserver.
pip install pytest pytest-httpserver
3. pytest-httpserver allows us to create a local server, which we will use in tests. To be able to use it, we
need to
modify fetch function.
...
async def fetch(delay_seconds: float, url: str = BLOOMREACH_SERVER) -> tuple[Success, dict]:
...
We will pass the URL as parameter, so we can simply modify it.
QUESTION: Do you know how this pattern is called?
Click on me for the answer!
This pattern is called Dependency Injection, where an object or function receives other objects or functions that it depends on.
4. Create fetch tests.
import asyncio
import json
import time
from functools import partial
import pytest
from pytest_httpserver import HTTPServer
from werkzeug import Response
from app import fetch, get_first_successful_request
def slow_response(request, time_to_wait_seconds: float, status: int = 200) -> Response:
time.sleep(time_to_wait_seconds)
return Response(json.dumps({"time": time_to_wait_seconds}), status, content_type="application/json")
@pytest.mark.asyncio
async def test_successful_fetch(httpserver: HTTPServer):
httpserver.expect_request("/foobar").respond_with_handler(partial(slow_response, time_to_wait_seconds=0.2))
success, result = await fetch(0, httpserver.url_for("/foobar"))
assert success
assert {"time": 0.2} == result
@pytest.mark.asyncio
async def test_unsuccessful_fetch(httpserver: HTTPServer):
httpserver.expect_request("/foobar").respond_with_handler(
partial(slow_response, status=400, time_to_wait_seconds=0.2)
)
success, result = await fetch(0, httpserver.url_for("/foobar"))
assert not success
We created a simple handler that returns similar responses to our Bloomreach testing server. In the first test, we
expect that server returns success and correct json. In the second one, we expect that success will be false.
4. Let’s create now test for first successful response. First we need to do a small refactoring.
First move the code, which we would like to test, from app to a separate function.
async def get_first_successful_request(unfinished_tasks: list[asyncio.Task], timeout: None | float):
timeout_remaining = timeout
while not out_of_time(timeout_remaining) and unfinished_tasks:
start = time.monotonic()
finished_tasks, unfinished_tasks = await asyncio.wait(
unfinished_tasks, return_when=asyncio.FIRST_COMPLETED, timeout=timeout_remaining
)
for finished_task in finished_tasks:
success, result = finished_task.result()
if success:
return success, result
end = time.monotonic()
timeout_remaining = timeout_remaining - (end - start) if timeout_remaining is not None else timeout_remaining
for unfinished_task in unfinished_tasks:
unfinished_task.cancel()
if out_of_time(timeout_remaining):
raise RequestTimeout()
return False, {}
Then update the app function, so it uses our new function.
@app.get("/api/smart")
async def smart_api_requester():
timeout_seconds = get_and_validate_timeout()
unfinished_tasks = [
asyncio.create_task(fetch(delay_seconds=0)),
asyncio.create_task(fetch(delay_seconds=WAIT_BEFORE_NEXT_REQUEST_SECONDS)),
asyncio.create_task(fetch(delay_seconds=WAIT_BEFORE_NEXT_REQUEST_SECONDS)),
]
success, result = await get_first_successful_request(unfinished_tasks, timeout_seconds)
return jsonify(success=success, result=result)
5. We cannot use one server as we did with Bloomreach testing server. The reason is that pytest-httpserver
cannot process more than one server. So, we overcome this problem by creating multiple servers. Each running in
different thread.
@pytest.fixture
def httpserver1(httpserver_ssl_context, httpserver_listen_address):
server = HTTPServer(host=HTTPServer.DEFAULT_LISTEN_HOST, port=7001, ssl_context=httpserver_ssl_context)
server.start()
yield server
server.clear()
if server.is_running():
server.stop()
@pytest.fixture
def httpserver2(httpserver_ssl_context, httpserver_listen_address):
server = HTTPServer(host=HTTPServer.DEFAULT_LISTEN_HOST, port=7002, ssl_context=httpserver_ssl_context)
server.start()
yield server
server.clear()
if server.is_running():
server.stop()
6. The last test that we will implement will be getting the first successful response.
@pytest.mark.asyncio
async def test_first_successful_requests(httpserver: HTTPServer, httpserver1, httpserver2):
httpserver.expect_request("/foo").respond_with_handler(partial(slow_response, time_to_wait_seconds=0.8))
httpserver1.expect_request("/foo").respond_with_handler(partial(slow_response, time_to_wait_seconds=0.2))
httpserver2.expect_request("/foo").respond_with_handler(partial(slow_response, time_to_wait_seconds=0.4))
unfinished_tasks = [
asyncio.create_task(fetch(delay_seconds=0, url=httpserver.url_for("/foo"))),
asyncio.create_task(fetch(delay_seconds=0, url=httpserver1.url_for("/foo"))),
asyncio.create_task(fetch(delay_seconds=0, url=httpserver2.url_for("/foo"))),
]
success, result = await get_first_successful_request(unfinished_tasks, timeout=None)
assert success
assert {"time": 0.2} == result
Each http server has same route, but with different sleep time. This means that the server with the lowest sleep time should answer first.
QUESTION: Do you think we need more tests? If yes, what would they test?
Click on me for the answer!
We definitely miss tests that would validate if get_first_successful_request ends always in specified timeout. We also
could add some tests that would validate what happens if the testing server reaches timeout or will respond with invalid
content.