Feature testing

Cattle_grid is feature tested using behave. The packages below contain useful general functions. Individual steps are described in Steps.

Use in your project

The tools provided here are meant to be reusable. Generally, the aim is to use almabtrieb to talk to a compatible server. Then one can use the steps and tools provided here to write tests. I have done this a few times, see comments and its feature tests.

Setup

First, the test code reads the following two values from the cattle_grid configuration file:

cattle_grid.toml

amqp_url = "amqp://guest:guest@rabbitmq:5672/"
db_url = "postgresql+asyncpg://postgres:pass@cattle_grid_db/"

The amqp_url is used by almabtrieb. The db_url is currently needed to create and delete accounts to use almabtrieb with.

Project layout

Assuming one locates the feature tests in the features directory, we will need to create the following structure:

features/
├── basic.feature
├── environment.py
└── steps
    ├── custom.py
    └── import.py

where basic.feature is the Gherkin for some test and custom.py represents custom steps. The file environment.py is described here in the behave documentation and should take the form

features/environment.py

from cattle_grid.testing.features.environment import (
    before_all,  
    before_scenario,  
    after_scenario,  
)

similarly import.py makes the steps available:

features/steps/import.py

from cattle_grid.testing.features.steps import *

Custom setup steps

If you have to run custom setup steps, you can do it as follows:

features/environment.py

from cattle_grid.testing.features.environment import (
    before_all,  
    before_scenario as cg_bs,  
    after_scenario,  
)

def before_scenario(context, scenario):
    cg_bs(context, scenario)

    # your setup

Usage

One thing these tools do for you is managing actors for this to work, one needs to create actors using our methods

Given A new user called "Alice"

this ensures that the context.actors["Alice"] and context.connections["Alice"] variables have the appropriate values.

cattle_grid.testing.features

fetch_request async

fetch_request(
    context, username: str, uri: str
) -> dict | None

Sends a fetch request for the uri through the gateway

Parameters:

Name	Type	Description	Default
context		The behave context	required
username	str	username performing the result	required
uri	str	URI being looked up	required

Returns:

Type	Description
dict | None

Source code in cattle_grid/testing/features/__init__.py

async def fetch_request(context, username: str, uri: str) -> dict | None:
    """Sends a fetch request for the uri through the gateway

    :param context: The behave context
    :param username: username performing the result
    :param uri: URI being looked up
    :return:
    """
    connection = context.connections[username]
    actor = context.actors[username].get("id")

    await asyncio.sleep(0.1)

    try:
        data = await connection.fetch(actor, uri)

        assert data.uri == uri

        return data.data
    except ErrorMessageException:
        return None

publish_as async

publish_as(
    context,
    username: str,
    method: str,
    data: dict,
    timeout: float = 0.3,
)

Publishes a message through the gateway

Parameters:

Name	Type	Description	Default
data	dict	The message to be published	required

Source code in cattle_grid/testing/features/__init__.py

async def publish_as(
    context, username: str, method: str, data: dict, timeout: float = 0.3
):
    """Publishes a message through the gateway

    :param data: The message to be published"""
    connection = context.connections[username]

    await connection.trigger(method, data)
    await asyncio.sleep(timeout)

cattle_grid.testing.features.environment

The before_all, _scenario, and after_scenario functions need to be imported in your environment.py file, e.g.

features/environment.py

from cattle_grid.testing.features.environment import (
    before_all,  # noqa
    before_scenario,  # noqa
    after_scenario,  # noqa
)
from cattle_grid.testing.features.reporting import (
    before_step,  # noqa
)

after_scenario

after_scenario(context: Context, scenario: Scenario)

Called in features/environment.py

Deletes the created actors and associated accounts. Closes the aiohttp.ClientSession.

Source code in cattle_grid/testing/features/environment.py

def after_scenario(context: Context, scenario: Scenario):
    """Called in features/environment.py

    Deletes the created actors and associated accounts.
    Closes the aiohttp.ClientSession.
    """

    asyncio.get_event_loop().run_until_complete(
        publish_reporting(
            "scenario_end",
            {},
        )
    )

    if context.session:
        asyncio.get_event_loop().run_until_complete(close_session(context))
        context.session = None

before_all

before_all(context: Context)

Called in features/environment.py

Ensures that default variables, context.session, .actors, .connections exist.

Source code in cattle_grid/testing/features/environment.py

def before_all(context: Context):
    """Called in features/environment.py

    Ensures that default variables, `context.session`, `.actors`, `.connections`
    exist.
    """
    context.session = None

    context.actors = {}
    context.connections = {}

before_scenario

before_scenario(context: Context, scenario: Scenario)

Called in features/environment.py

Opens an aiohttp.ClientSession and sets it to context.session.

Source code in cattle_grid/testing/features/environment.py

def before_scenario(context: Context, scenario: Scenario):
    """Called in features/environment.py

    Opens an [aiohttp.ClientSession][] and sets it to `context.session`.
    """
    asyncio.get_event_loop().run_until_complete(create_session(context))

    context.actors = {}
    context.connections = {}

    asyncio.get_event_loop().run_until_complete(
        publish_reporting(
            "scenario",
            {
                "name": scenario.name,
                "file": scenario.filename,
                "description": scenario.description,
            },
        )
    )