Generic Tests

The ESG package provides a set of generic tests that can be used to verify the correct implementation of a service. All tests can be found in the test directory. The tests are implemented using the pytest framework.

Below is an example implementation of tests for the basic example service detailed in Example service implementation.

Mock server for third-party services

The example service uses the third-party service Open-Meteo to fetch meteorological data. To avoid the dependency on this service, a mock server is used in the tests. The server is started before the tests are executed and shut down afterwards. It is set up to return predefined data when requested. The data is stored in a python file and loaded into the mock server before the tests are executed (see Preparation of mock data).

import os
from unittest.mock import patch

import pytest

from .data import OPEN_METEO_RESPONSE


class OpenMeteoHttpServer:
    """
    Helper class for tests thad would interact with Open Meteo.
    """

    def __init__(self, httpserver):
        self.httpserver = httpserver

    def add_request(self, lat, lon, past_days):
        expected_request = self.httpserver.expect_oneshot_request(
            "/v1/forecast",
            query_string={
                "latitude": str(lat),
                "longitude": str(lon),
                "minutely_15": (
                    "shortwave_radiation,diffuse_radiation,"
                    "direct_normal_irradiance"
                ),
                "past_days": str(past_days),
                "forecast_days": "1",
            },
        )
        expected_request.respond_with_json(OPEN_METEO_RESPONSE)


@pytest.fixture
def open_meteo_httpserver(httpserver):
    oma_test_url = httpserver.url_for("")[:-1]
    om_httpserver = OpenMeteoHttpServer(httpserver=httpserver)
    with patch.dict(os.environ, {"OPEN_METEO_API_URL": oma_test_url}):
        yield om_httpserver

Preparation of mock data

In addition to the response data for the mocked Open-Meteo request, the data.py file should always contain the following data.

Mock data for testing the /request/ and the /fit-parameters/ endpoints (fooc.py functions in the basic example):
- The expected /request/ inputs, i.e., the RequestArguments as REQUEST_INPUTS_FOOC_TEST
- The expected /fit-parameters/ inputs, i.e., the FitParameterArguments and the Observations as FIT_PARAM_INPUTS_FOOC_TEST consisting of arguments and observations
- The expected outputs, i.e., the RequestOutput and the FittedParameters as REQUEST_OUTPUTS_FOOC_TEST and FIT_PARAM_OUTPUTS_FOOC_TEST
Mock data for testing the input and output data models for the /request/ and the /fit-parameters/ endpoints (data_model.py file in the basic example):
- A valid object of the RequestArguments class as REQUEST_INPUTS_MODEL_TEST and the corresponding valid output object as REQUEST_OUTPUTS_MODEL_TEST
- A valid object of the FitParameterArguments and the Observations class as FIT_PARAM_INPUTS_MODEL_TEST consisting of arguments and observations and the corresponding valid output object as FIT_PARAM_OUTPUTS_MODEL_TEST
- An invalid object of the RequestArguments class as INVALID_REQUEST_INPUTS and the corresponding invalid output as INVALID_REQUEST_OUTPUTS
- An invalid object of the FitParameterArguments and the Observations class as INVALID_FIT_PARAM_INPUTS consisting of arguments and observations and the corresponding invalid output as INVALID_FIT_PARAM_OUTPUTS

Testing the forecast and optimization code

The tests for the forecast and optimization code are implemented in the test_fooc.py file. To test the /request/ and the /fit-parameters/ endpoints, the service specific data models are used as well the GenericFOOCTest, which is included in the generic tests provided by the ESG package. As the code excerpt below shows, implementing the tests is thus fairly simple.

RequestInput = compute_request_input_model(
    RequestArguments=RequestArguments,
    FittedParameters=FittedParameters,
)


class TestHandleRequest(GenericFOOCTest):
    InputDataModel = RequestInput
    OutputDataModel = RequestOutput
    input_data_jsonable = [m["JSONable"] for m in REQUEST_INPUTS_FOOC_TEST]
    output_data_jsonable = [m["JSONable"] for m in REQUEST_OUTPUTS_FOOC_TEST]

    def get_payload_function(self):
        return handle_request

    @pytest.fixture(autouse=True)
    def open_meteo_httpserver_request(self, open_meteo_httpserver):
        """
        Enable fake Open Meteo endpoint providing data if query parameters
        match the ones expected by request, i.e. requests data for
        the future only.
        """
        open_meteo_httpserver.add_request(
            # NOTE: lat/lon must match items in `REQUEST_INPUTS_FOOC_TEST`
            lat=35.2,
            lon=-110.0,
            past_days=0,
        )


FitParametersInput = compute_fit_parameters_input_model(
    FitParameterArguments=FitParameterArguments,
    Observations=Observations,
)


class TestFitParameters(GenericFOOCTest):
    InputDataModel = FitParametersInput
    OutputDataModel = FittedParameters
    input_data_jsonable = [m["JSONable"] for m in FIT_PARAM_INPUTS_FOOC_TEST]
    output_data_jsonable = [m["JSONable"] for m in FIT_PARAM_OUTPUTS_FOOC_TEST]

    def get_payload_function(self):
        return fit_parameters

    @pytest.fixture(autouse=True)
    def open_meteo_httpserver_fit_parameters(self, open_meteo_httpserver):
        """
        Enable fake Open Meteo endpoint providing data if query parameters
        match the ones expected by fit parameters, i.e. requests data for
        the past 90 days.
        """
        open_meteo_httpserver.add_request(
            # NOTE: lat/lon must match items in `FIT_PARAM_INPUTS_FOOC_TEST`
            lat=35.2,
            lon=-110.0,
            past_days=90,
        )

The tests included in the generic tests only cover basic scenarios. Developers will likely want to add more sophisticated tests for their forecasting or optimization code.

The tests for the third-party service Open-Meteo is omitted as the implementation details are likely to vary from service to service. However, the tests for the Open-Meteo service can be found in the test_fooc.py file as well.

Testing the specified data models

In order to enure that the data models are well defined, the input and output data models for the /request/ and the /fit-parameters/ endpoints are tested. These tests utilize the GenericMessageSerializationTest included in the generic tests provided by the ESG package is used. Due to the generic tests, the implementation is fairly simple.

RequestInput = compute_request_input_model(
    RequestArguments=RequestArguments,
    FittedParameters=FittedParameters,
)


class TestRequestInput(GenericMessageSerializationTest):
    """
    Tests for the `RequestArguments` and `FittedParameters` data models.
    """

    ModelClass = RequestInput
    msgs_as_python = [m["Python"] for m in REQUEST_INPUTS_MODEL_TEST]
    msgs_as_jsonable = [m["JSONable"] for m in REQUEST_INPUTS_MODEL_TEST]
    invalid_msgs_as_jsonable = [m["JSONable"] for m in INVALID_REQUEST_INPUTS]


class TestRequestOutput(GenericMessageSerializationTest):
    """
    Tests for the `RequestArguments` and `FittedParameters` data models.
    """

    ModelClass = RequestOutput
    msgs_as_python = [m["Python"] for m in REQUEST_OUTPUTS_MODEL_TEST]
    msgs_as_jsonable = [m["JSONable"] for m in REQUEST_OUTPUTS_MODEL_TEST]
    invalid_msgs_as_jsonable = [m["JSONable"] for m in INVALID_REQUEST_OUTPUTS]

FitParametersInput = compute_fit_parameters_input_model(
    FitParameterArguments=FitParameterArguments,
    Observations=Observations,
)


class TestFitParametersInput(GenericMessageSerializationTest):
    """
    Tests for the `RequestArguments` and `FittedParameters` data models.
    """

    ModelClass = FitParametersInput
    msgs_as_python = [m["Python"] for m in FIT_PARAM_INPUTS_MODEL_TEST]
    msgs_as_jsonable = [m["JSONable"] for m in FIT_PARAM_INPUTS_MODEL_TEST]
    invalid_msgs_as_jsonable = [m["JSONable"] for m in INVALID_FIT_PARAM_INPUTS]


class TestFitParametersOutput(GenericMessageSerializationTest):
    """
    Tests for the `RequestArguments` and `FittedParameters` data models.
    """

    ModelClass = FittedParameters
    msgs_as_python = [m["Python"] for m in FIT_PARAM_OUTPUTS_MODEL_TEST]
    msgs_as_jsonable = [m["JSONable"] for m in FIT_PARAM_OUTPUTS_MODEL_TEST]
    invalid_msgs_as_jsonable = [
        m["JSONable"] for m in INVALID_FIT_PARAM_OUTPUTS
    ]

The code excerpt above can be found in the test_data_model.py file.

Testing the worker component

To ensure worker component is correctly implemented, the GenericWorkerTaskTest included in the generic tests provided by the ESG package is used. The tests for the worker component are implemented in the test_worker.py file. As the code excerpt below shows, the implementation is again quite easy.

class TestRequestTask(GenericWorkerTaskTest):
    task_to_test = request_task
    input_data_jsonable = [m["JSONable"] for m in REQUEST_INPUTS_FOOC_TEST]
    output_data_jsonable = [m["JSONable"] for m in REQUEST_OUTPUTS_FOOC_TEST]

    @pytest.fixture(autouse=True)
    def open_meteo_httpserver_request(self, open_meteo_httpserver):
        """
        Enable fake Open Meteo endpoint providing data if query parameters
        match the ones expected by request, i.e. requests data for
        the future only.
        """
        open_meteo_httpserver.add_request(
            # NOTE: lat/lon must match items in `REQUEST_INPUTS_FOOC_TEST`
            lat=35.2,
            lon=-110.0,
            past_days=0,
        )

class TestFitParametersTask(GenericWorkerTaskTest):
    task_to_test = fit_parameters_task
    input_data_jsonable = [m["JSONable"] for m in FIT_PARAM_INPUTS_FOOC_TEST]
    output_data_jsonable = [m["JSONable"] for m in FIT_PARAM_OUTPUTS_FOOC_TEST]

    @pytest.fixture(autouse=True)
    def open_meteo_httpserver_fit_parameters(self, open_meteo_httpserver):
        """
        Enable fake Open Meteo endpoint providing data if query parameters
        match the ones expected by fit parameters, i.e. requests data for
        the past 90 days.
        """
        open_meteo_httpserver.add_request(
            # NOTE: lat/lon must match items in `FIT_PARAM_INPUTS_FOOC_TEST`
            lat=35.2,
            lon=-110.0,
            past_days=90,
        )

Testing the API component

Finally, the api component needs to be tested for correct implementation. For this purpose, the ESG package’s generic tests provide the GenericAPITest. Again, the implementation is quite simple as demonstrated below.

RequestInput = compute_request_input_model(
    RequestArguments=RequestArguments,
    FittedParameters=FittedParameters,
)


class TestRequestEndpoint(GenericAPITest):
    tested_api = tested_api
    celery_app = app
    endpoint = "request"
    InputDataModel = RequestInput
    OutputDataModel = RequestOutput
    input_data_jsonable = [m["JSONable"] for m in REQUEST_INPUTS_FOOC_TEST]
    output_data_jsonable = [m["JSONable"] for m in REQUEST_OUTPUTS_FOOC_TEST]


FitParametersInput = compute_fit_parameters_input_model(
    FitParameterArguments=FitParameterArguments,
    Observations=Observations,
)


class TestFitParametersEndpoint(GenericAPITest):
    tested_api = tested_api
    celery_app = app
    endpoint = "fit-parameters"
    InputDataModel = FitParametersInput
    OutputDataModel = FittedParameters
    input_data_jsonable = [m["JSONable"] for m in FIT_PARAM_INPUTS_FOOC_TEST]
    output_data_jsonable = [m["JSONable"] for m in FIT_PARAM_OUTPUTS_FOOC_TEST]

The tests for the api component can be found in the test_api.py file.