Testing UWS applications¶

UWS applications are arq applications, and therefore should follow the testing advice in Testing applications with an arq queue. This includes testing the frontend and the backend worker separately.

The safir.testing.uws module provides some additional support to make it easier to test the frontend.

Frontend testing fixtures¶

The frontend of a UWS application assumes that arq will execute both jobs and the database worker that recovers the results of a job and stores them in the database. It also assumes Wobbly will be available as an API for storing and retrieving jobs in the database. During testing of the frontend, arq and Wobbly will not be running, and therefore this execution must be simulated. This is done with the MockWobbly and MockUWSJobRunner classes, but it requires some setup.

Mock Wobbly¶

Add a development dependency on respx to the application. Then, add a Wobbly mock fixture to tests/conftest.py:

tests/conftest.py¶

import pytest
import respx
from safir.testing.uws import MockWobbly, patch_wobbly

from example.config import config


@pytest.fixture
def mock_wobbly(respx_mock: respx.Router) -> MockWobbly:
    return patch_wobbly(respx_mock, str(config.wobbly_url))

Change example.config to the config module for your application.

You will need to arrange for wobbly_url to be set in your application configuration during testing. Normally the easiest way to do that is to set APPLICATION_WOBBLY_URL to some reasonable placeholder value such as http://example.com/wobbly in your application’s tox configuration for running tests.

Then, arrange for this fixture to be enabled by the test client for your application. Usually the easiest way to do that is to make the mock_wobbly fixture a parameter to the app fixture that sets up the application for testing.

Mock the request token¶

All of the UWS routes provided by Safir expect to receive username and token information in the incoming request. Normally these headers are generated by Gafaelfawr based on the authentication credentials of the request. Inside the test suite, you will need to provide the headers that Gafaelfawr would have provided when making requests to the UWS routes.

To do this, first create fixtures that define a test username and test service:

tests/conftest.py¶

import pytest


@pytest.fixture
def test_service() -> str:
    return "test-service"


@pytest.fixture
def test_username() -> str:
    return "test-user"

Then, create a fixture that returns a token for the Wobbly mock that encodes that username and service. This allows the mock to recover the intended username and service of a request from the Safir UWS code to Wobbly.

tests/conftest.py¶

import pytest
from safir.testing.uws import MockWobbly


@pytest.fixture
def test_token(test_service: str, test_username: str) -> str:
    return MockWobbly.make_token(test_service, test_username)

Finally, configure the client fixture to send the appropriate headers by default.

tests/conftest.py¶

import pytest
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient


@pytest_asyncio.fixture
async def client(
    app: FastAPI, test_token: str, test_username: str
) -> AsyncIterator[AsyncClient]:
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="https://example.com/",
        headers={
            "X-Auth-Request-Token": test_token,
            "X-Auth-Request-User": test_username,
        },
    ) as client:
        yield client

Any tests that need to know the value of the username or service on whose behalf operations will be performed can use the test_username and test_service fixtures. The token will be required for calls to MockArqQueue (see below) and can be accessed via the test_token fixture.

If a specific test needs to send a request from a different service or username (to test handling of multiple usernames, for instance), it should override the request headers to send a different token and username in X-Auth-Request-Token and X-Auth-Request-User. To generate a token for a different service and username pair, call MockWobbly.make_token.

Mock the arq queue¶

First, the application must be configured to use a MockArqQueue class instead of one based on Redis. This stores all queued jobs in memory and provides some test-only methods to manipulate them.

To do this, first set up a fixture in tests/conftest.py that provides a mock arq queue:

tests/conftest.py¶

import pytest
from safir.arq import MockArqQueue


@pytest.fixture
def arq_queue() -> MockArqQueue:
    return MockArqQueue()

Then, configure the application to use that arq queue instead of the default one in the app fixture.

tests/conftest.py¶

from collections.abc import AsyncIterator

from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from safir.arq import MockArqQueue

from example import main
from example.config import uws


@pytest_asyncio.fixture
async def app(arq_queue: MockArqQueue) -> AsyncIterator[FastAPI]:
    async with LifespanManager(main.app):
        uws.override_arq_queue(arq_queue)
        yield main.app

Mock Google Cloud Storage¶

The UWS library assumes results are in Google Cloud Storage and creates signed URLs to allow the client to retrieve those results. This support needs to be mocked out during testing. Do this by adding the following fixture to tests/conftest.py:

tests/conftest.py¶

from datetime import timedelta

import pytest
from safir.testing.gcs import MockStorageClient, patch_google_storage


@pytest.fixture(autouse=True)
def mock_google_storage() -> Iterator[MockStorageClient]:
    yield from patch_google_storage(
        expected_expiration=timedelta(minutes=15), bucket_name="some-bucket"
    )

See Testing with mock Google Cloud Storage for more information.

Provide a mock arq queue runner¶

Finally, you can create a fixture that provides a mock arq queue runner. This runner will be used to simulate the execution of the backend worker and the collection of that result and subsequent database updates via Wobbly.

tests/conftest.py¶

import pytest_asyncio
from safir.arq import MockArqQueue
from safir.testing.uws import MockUWSJobRunner

from example.config import config


@pytest_asyncio.fixture
async def runner(arq_queue: MockArqQueue) -> MockUWSJobRunner:
    return MockUWSJobRunner(config.uws_config, arq_queue)

Writing a frontend test¶

Now, all the pieces are in place to write a meaningful test of the frontend. You can use the methods of MockUWSJobRunner to change the state of a mocked backend job and set the results that it returned. Here is an example of a test of a hypothetical cutout service.

tests/handlers/async_test.py¶

import pytest
from httpx import AsyncClient
from safir.testing.uws import MockUWSJobRunner
from safir.uws import JobResult


@pytest.mark.asyncio
async def test_create_job(
    client: AsyncClient, test_token: str, runner: MockUWSJobRunner
) -> None:
    r = await client.post(
        "/api/cutout/jobs",
        data={"ID": "1:2:band:value", "Pos": "CIRCLE 0 1 2"},
    )
    assert r.status_code == 303
    assert r.headers["Location"] == "https://example.com/api/cutout/jobs/1"
    await runner.mark_in_progress(test_token, "1")

    async def run_job() -> None:
        results = [
            JobResult(
                id="cutout",
                url="s3://some-bucket/some/path",
                mime_type="application/fits",
            )
        ]
        await runner.mark_complete(test_token, "1", results, delay=0.2)

    _, r = await asyncio.gather(
        run_job(),
        client.get(
            "/api/cutout/jobs/1", params={"wait": 2, "phase": "EXECUTING"}
        ),
    )
    assert r.status_code == 200
    assert "https://example.com/some/path" in r.text

Note the use of MockUWSJobRunner.mark_complete with a delay argument and asyncio.gather to simulate a job that takes some time to complete so that the client request to wait for job completion can be tested.

A more sophisticated test would check the XML results returned by the API against the UWS XML schema. This can be done using the models provided by vo-models.

Testing the backend worker¶

The backend divides naturally into two pieces: the wrapper code that accepts the arguments in the format passed by the UWS library, handles exceptions, and constructs WorkerResult objects; and the code that performs the underlying operation of the service.

To make testing easier, it’s usually a good idea to separate those two pieces. The wrapper that handles talking to the UWS library and translating exceptions can be included in the source of the application. The underlying code to perform the operation is often best maintained in a library of domain-specific code. For example, for Rubin Observatory, this will usually be a function in a Science Pipelines package with its own separate tests.

If the code is structured this way, there won’t be much to test in the backend worker wrapper and often one can get away with integration tests. If more robust tests are desired, though, the backend worker function is a simple function that can be called and tested directly by the test suite, possibly after mocking out the underlying library function.