Skip to content

Helpers for BDD

cattle_grid.testing.features

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 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, scenario)

Called in features/environment.py

Source code in cattle_grid/testing/features/__init__.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def after_scenario(context, scenario):
    """Called in features/environment.py"""

    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)

Called in features/environment.py

Source code in cattle_grid/testing/features/__init__.py 
69
70
71
72
73
74
75
76
def before_all(context):
    """Called in features/environment.py"""
    tortoise_logger = logging.getLogger("tortoise")
    tortoise_logger.setLevel(logging.WARNING)
    context.session = None

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

before_scenario 

before_scenario(context, scenario)

Called in features/environment.py

Source code in cattle_grid/testing/features/__init__.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def before_scenario(context, scenario):
    """Called in features/environment.py"""
    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,
            },
        )
    )

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 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
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 
113
114
115
116
117
118
119
120
121
122
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)