Skip to content

Predefined Steps

In order to use these steps with behave, you have to create a file like

features/steps/import.py
from cattle_grid.testing.features.steps import *  # noqa

in your steps directory.

cattle_grid.testing.features.steps.account

create_account_step async 

create_account_step(context, alice: str)

Creates an account with name alice

Given An account called "Alice"

The connection is stored in context.connections[alice] it is an Almabtrieb object.

Source code in cattle_grid/testing/features/steps/account.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@given('An account called "{alice}"')  # type: ignore
async def create_account_step(context, alice: str):
    """
    Creates an account with name `alice`

    ```gherkin
    Given An account called "Alice"
    ```

    The connection is stored in `context.connections[alice]` it
    is an [Almabtrieb][almabtrieb.Almabtrieb] object.
    """

    config = load_settings()
    async with database_session(db_uri=config.get("db_uri")) as session:  # type: ignore
        account = await create_account(session, alice, alice)
        assert account
        await add_permission(session, account, "admin")

    mq_host = os.environ.get("CATTLE_GRID_MQ", "rabbitmq")

    context.connections[alice] = Almabtrieb.from_connection_string(
        f"amqp://{alice}:{alice}@{mq_host}/", silent=True
    )
    context.connections[alice].task = asyncio.create_task(
        context.connections[alice].run()
    )

    for _ in range(10):
        if context.connections[alice].connected:
            return
        logger.debug("Connecting")
        await asyncio.sleep(0.1)

    raise RuntimeError("Could not connect")

create_actor_step 

create_actor_step(context, alice, alyssa)

Given "Alice" created an actor called "Alyssa"
Source code in cattle_grid/testing/features/steps/account.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
@given('"{alice}" created an actor called "{alyssa}"')  # pyright: ignore[reportCallIssue]
def create_actor_step(context, alice, alyssa):
    """
    ```gherkin
    Given "Alice" created an actor called "Alyssa"
    ```
    """
    hostname = {"alice": "abel", "bob": "banach", "Bob": "banach"}.get(alyssa, "abel")

    context.execute_steps(
        f"""
        Given "{alice}" created an actor on "{hostname}" called "{alyssa}"
        """
    )

cattle_grid.testing.features.steps.block

send_block async 

send_block(context, alice, bob)

When "Alice" blocks "Bob"

The id of the block is stored in context.block_id

Source code in cattle_grid/testing/features/steps/block.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@given('"{alice}" blocks "{bob}"')  # type: ignore
@when('"{alice}" blocks "{bob}"')  # type: ignore
async def send_block(context, alice, bob):
    """
    ```gherkin
    When "Alice" blocks "Bob"
    ```

    The id of the block is stored in `context.block_id`
    """

    actor = context.actors[alice]
    activity_factory, object_factory = factories_for_actor_object(
        actor, id_generator=id_generator_for_actor(actor)
    )

    bob_id = context.actors[bob].get("id")

    activity = activity_factory.custom(type="Block", object=bob_id, to={bob_id}).build()

    context.block_id = activity.get("id")

    await publish_as(
        context,
        alice,
        "send_message",
        send_message_as_actor(actor, activity),
    )

unblock async 

unblock(context, bob, alice)

Sends an Undo Block activity for the id stored in context.block_id.

Usage:

When "Bob" unblocks "Alice"
Source code in cattle_grid/testing/features/steps/block.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
@when('"{bob}" unblocks "{alice}"')  # type: ignore
async def unblock(context, bob, alice):
    """Sends an Undo Block activity for
    the id stored in `context.block_id`.

    Usage:

    ```gherkin
    When "Bob" unblocks "Alice"
    ```
    """
    actor = context.actors[bob]
    activity_factory, _ = factories_for_actor_object(actor)
    alice_id = context.actors[alice].get("id")

    activity = activity_factory.custom(
        object=context.block_id, to={alice_id}, type="Undo"
    ).build()
    activity["id"] = "undo:" + str(uuid4())

    await publish_as(
        context,
        bob,
        "send_message",
        send_message_as_actor(actor, activity),
    )

cattle_grid.testing.features.steps.collection

check_collection async 

check_collection(context, alice, bob, collection)

Used to check if the followers or following collection of the actor bob does not contain the actor alice.

Then The "followers" collection of "bob" does not include "alice"
Source code in cattle_grid/testing/features/steps/collection.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
@then('The "{collection}" collection of "{bob}" does not include "{alice}"')
async def check_collection(context, alice, bob, collection):
    """Used to check if the followers or following collection
    of the actor `bob` does not contain the actor `alice`.

    ```gherkin
    Then The "followers" collection of "bob" does not include "alice"
    ```
    """
    result = await fetch_request(
        context,
        bob,
        context.actors[bob].get(collection),
    )

    assert result

    actor = context.actors[alice].get("id")

    if "raw" in result:
        result = result["raw"]

    assert result.get("type") == "OrderedCollection"
    assert actor not in result.get("orderedItems", [])

check_collection_contains async 

check_collection_contains(context, alice, bob, collection)

Used to check if the followers or following collection of the actor bob contains the actor alice.

Then The "followers" collection of "bob" contains "alice"
Source code in cattle_grid/testing/features/steps/collection.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@then('The "{collection}" collection of "{bob}" contains "{alice}"')
async def check_collection_contains(context, alice, bob, collection):
    """Used to check if the followers or following collection
    of the actor `bob` contains the actor `alice`.

    ```gherkin
    Then The "followers" collection of "bob" contains "alice"
    ```
    """
    result = await fetch_request(
        context,
        bob,
        context.actors[bob].get(collection),
    )

    bob_id = context.actors[alice].get("id")

    if "raw" in result:
        result = result["raw"]

    assert result.get("type") == "OrderedCollection"
    assert bob_id in result.get("orderedItems", [])

cattle_grid.testing.features.steps.drive

cattle_grid.testing.features.steps.follow

accept_follow_request async 

accept_follow_request(context, actor)

Checks that Alice received a follow Activity and then accepts this follow activity

When "Alice" sends an Accept to this Follow Activity
Source code in cattle_grid/testing/features/steps/follow.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@when('"{actor}" sends an Accept to this Follow Activity')
async def accept_follow_request(context, actor):
    """Checks that Alice received a follow Activity and then
    accepts this follow activity

    ```gherkin
    When "Alice" sends an Accept to this Follow Activity
    ```
    """
    result = await context.connections[actor].next_incoming()
    received_activity = result.get("data")
    if "raw" in received_activity:
        received_activity = received_activity["raw"]

    logger.debug("Got follow request:")
    logger.debug(received_activity)

    assert received_activity["type"] == "Follow"

    follow_id = received_activity["id"]
    to_follow = received_activity["actor"]

    alice = context.actors[actor]
    activity_factory, _ = factories_for_actor_object(alice)

    activity = activity_factory.accept(follow_id, to={to_follow}).build()
    activity["id"] = "accept:" + str(uuid4())

    await publish_as(
        context, actor, "send_message", send_message_as_actor(alice, activity)
    )

actor_follows_other 

actor_follows_other(context, bob, alice)

Combination of two steps, i.e.

When "Alice" follows "Bob"

is the same as

When "Alice" sends "Bob" a Follow Activity
And "Bob" sends an Accept to this Follow Activity
Source code in cattle_grid/testing/features/steps/follow.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@given('"{bob}" follows "{alice}"')
def actor_follows_other(context, bob, alice):
    """Combination of two steps, i.e.

    ```gherkin
    When "Alice" follows "Bob"
    ```

    is the same as

    ```gherkin
    When "Alice" sends "Bob" a Follow Activity
    And "Bob" sends an Accept to this Follow Activity
    ```
    """

    logger.debug("Available conections %s", ", ".join(context.connections.keys()))

    if alice not in context.connections:
        context.execute_steps(
            f"""
        When "{bob}" sends "{alice}" a Follow Activity
    """
        )
    else:
        context.execute_steps(
            f"""
        When "{bob}" sends "{alice}" a Follow Activity
        And "{alice}" sends an Accept to this Follow Activity
    """
        )

automatically_accept_followers async 

automatically_accept_followers(context, alice)

FIXME: Should toggle

Source code in cattle_grid/testing/features/steps/follow.py 
133
134
135
136
137
138
139
140
141
142
143
144
@given('"{alice}" automatically accepts followers')
async def automatically_accept_followers(context, alice):
    """FIXME: Should toggle"""

    actor = context.actors[alice]

    await publish_as(
        context,
        alice,
        "update_actor",
        {"actor": actor.get("id"), "autoFollow": True},
    )

send_follow async 

send_follow(context, alice, bob)

Sends a follow Activity. Usage

When "Alice" sends "Bob" a Follow Activity

Stores the follow activity in context.follow_activity

Source code in cattle_grid/testing/features/steps/follow.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
@when('"{alice}" sends "{bob}" a Follow Activity')
async def send_follow(context, alice, bob):
    """Sends a follow Activity. Usage

    ```gherkin
    When "Alice" sends "Bob" a Follow Activity
    ```

    Stores the follow activity in `context.follow_activity`
    """
    alice_actor = context.actors[alice]
    bob_id = context.actors[bob].get("id")
    activity_factory, _ = factories_for_actor_object(alice_actor)

    context.follow_id = "follow:" + str(uuid4())

    context.follow_activity = activity_factory.follow(
        bob_id, id=context.follow_id
    ).build()

    await publish_as(
        context,
        alice,
        "send_message",
        send_message_as_actor(alice_actor, context.follow_activity),
    )

send_reject_follow async 

send_reject_follow(context, alice, bob)

Sends an Undo Follow activity for the follow activity with id stored in context.follow_activity.

Usage:

When "Alice" sends "Bob" a Reject Follow Activity
Source code in cattle_grid/testing/features/steps/follow.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
@when('"{alice}" sends "{bob}" a Reject Follow Activity')
async def send_reject_follow(context, alice, bob):
    """Sends an Undo Follow activity for the follow activity
    with id stored in `context.follow_activity`.

    Usage:

    ```gherkin
    When "Alice" sends "Bob" a Reject Follow Activity
    ```
    """
    actor = context.actors[alice]
    activity_factory, _ = factories_for_actor_object(actor)

    activity = activity_factory.reject(context.follow_activity).build()
    if isinstance(activity["object"], dict):
        activity["object"] = activity["object"]["id"]

    activity["id"] = "reject:" + str(uuid4())

    await publish_as(
        context, alice, "send_message", send_message_as_actor(actor, activity)
    )

send_undo_follow async 

send_undo_follow(context, bob, alice)

Sends an Undo Follow activity for the follow activity with id stored in context.follow_activity.

Usage:

When "Bob" sends "Alice" an Undo Follow Activity
Source code in cattle_grid/testing/features/steps/follow.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
@when('"{bob}" sends "{alice}" an Undo Follow Activity')
async def send_undo_follow(context, bob, alice):
    """Sends an Undo Follow activity for the follow activity
    with id stored in `context.follow_activity`.

    Usage:

    ```gherkin
    When "Bob" sends "Alice" an Undo Follow Activity
    ```
    """
    actor = context.actors[bob]
    activity_factory, _ = factories_for_actor_object(actor)

    activity = activity_factory.undo(context.follow_activity).build()
    if isinstance(activity["object"], dict):
        activity["object"] = activity["object"]["id"]

    activity["id"] = "undo:" + str(uuid4())

    await publish_as(
        context, bob, "send_message", send_message_as_actor(actor, activity)
    )

cattle_grid.testing.features.steps.http

cattle_grid.testing.features.steps.message_content

Routines to check the content of a message as retrieved by cattle_grid.testing.features.steps.messaging.receive_message

message_contains_url_of_mediatype 

message_contains_url_of_mediatype(context, media_type)

Then The received message contains an URL of mediatype "text/html"

The resolved url is added to context.media_type_link.

Source code in cattle_grid/testing/features/steps/message_content.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
@then('The received message contains an URL of mediatype "{media_type}"')  # pyright: ignore[reportCallIssue]
def message_contains_url_of_mediatype(context, media_type):
    """

    ```gherkin
    Then The received message contains an URL of mediatype "text/html"
    ```

    The resolved url is added to `context.media_type_link`.
    """

    message = context.received_object

    urls = as_list(message.get("url", []))

    html_url = next(
        (x for x in urls if isinstance(x, dict) and x.get("mediaType") == media_type),
        None,
    )
    assert html_url

    assert html_url.get("type") == "Link"
    assert html_url.get("href")

    context.media_type_link = html_url.get("href")

step_impl async 

step_impl(context, text)

Then The URL resolves to a webpage
Source code in cattle_grid/testing/features/steps/message_content.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@then('The URL resolves to a webpage containing "{text}"')  # pyright: ignore[reportCallIssue]
async def step_impl(context, text):
    """
    ```gherkin
    Then The URL resolves to a webpage
    ```
    """

    async with context.session.get(
        context.media_type_link, headers={"accept": "text/html"}
    ) as response:
        assert response.status == 200, (
            f"Status is {response.status} != 200 for {context.media_type_link}"
        )
        content_type = response.headers["content-type"]
        assert content_type.startswith("text/html"), (
            f"Got response content-type {content_type}"
        )

        content = await response.text()

        assert text in content, f"Content {content} does not contain {text}"

cattle_grid.testing.features.steps.messaging

check_activity_type 

check_activity_type(context, activity_type)

Checks that the received activity from cattle_grid.testing.features.steps.messaging.receive_activity is of type activity_type.

Then the received activity is of type "Update"
Source code in cattle_grid/testing/features/steps/messaging.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
@then('the received activity is of type "{activity_type}"')  # type: ignore
def check_activity_type(context, activity_type):
    """Checks that the received activity from [cattle_grid.testing.features.steps.messaging.receive_activity][]
    is of type `activity_type`.

    ```gherkin
    Then the received activity is of type "Update"
    ```
    """

    received = context.activity
    if "raw" in received:
        received = received["raw"]

    import json

    print(json.dumps(received, indent=2))

    assert received.get("type") == activity_type, (
        f"Activity {received} has the wrong type"
    )

not_receive_activity async 

not_receive_activity(context, actor)

Ensures that no incoming activity was received

Then "bob" does not receive an activity
Source code in cattle_grid/testing/features/steps/messaging.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@then('"{actor}" does not receive an activity')  # type: ignore
async def not_receive_activity(context, actor):
    """Ensures that no incoming activity was received

    ```gherkin
    Then "bob" does not receive an activity
    ```
    """

    try:
        result = await context.connections[actor].next_incoming()

        assert result is None, f"Received activity {json.dumps(result, indent=2)}"
    except StreamNoNewItemException:
        ...

receive_activity async 

receive_activity(context, actor)

Ensures that an incoming activity was received and stores it in context.activity.

Then "bob" receives an activity
Source code in cattle_grid/testing/features/steps/messaging.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
@then('"{actor}" receives an activity')  # type: ignore
async def receive_activity(context, actor):
    """Ensures that an incoming activity was received
    and stores it in `context.activity`.

    ```gherkin
    Then "bob" receives an activity
    ```
    """

    data = await context.connections[actor].next_incoming()
    assert data.get("event_type") == "incoming"

    context.activity = data["data"]["raw"]

    assert context.activity["@context"]

receive_message async 

receive_message(context, actor, text)

Used to check if the last message received by actor is saying the correct thing.

Then "bob" receives a message saying "Got milk?"

The received object is stored in context.received_object.

Source code in cattle_grid/testing/features/steps/messaging.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
@then('"{actor}" receives a message saying "{text}"')  # type: ignore
async def receive_message(context, actor, text):
    """Used to check if the last message received by actor
    is saying the correct thing.

    ```gherkin
    Then "bob" receives a message saying "Got milk?"
    ```

    The received object is stored in `context.received_object`.
    """

    data = await context.connections[actor].next_incoming()

    assert data.get("event_type") == "incoming"
    activity = data.get("data")

    if "raw" in activity:
        activity = activity["raw"]

    assert activity.get("type") == "Create", f"got {activity}"
    assert activity.get("@context"), f"got {activity}"

    obj = activity.get("object", {})
    assert obj.get("content") == text, f"""got {obj.get("content")}"""

    context.received_object = obj

send_message async 

send_message(context, actor, target, text)

Used to send a message. The message has the format (with a lot of stuff omitted)

{
    "type": "Create",
    "object": {
        "type": "Note",
        "content": text,
        "to": [actor_id_of_target]
    }
}

This step can be used as

When "alice" sends "bob" a message saying "You stole my milk!"
Source code in cattle_grid/testing/features/steps/messaging.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
@when('"{actor}" sends "{target}" a message saying "{text}"')  # type: ignore
async def send_message(context, actor, target, text):
    """Used to send a message. The message has the format (with a lot of stuff omitted)

    ```json
    {
        "type": "Create",
        "object": {
            "type": "Note",
            "content": text,
            "to": [actor_id_of_target]
        }
    }
    ```

    This step can be used as

    ```gherkin
    When "alice" sends "bob" a message saying "You stole my milk!"
    ```
    """
    alice = context.actors[actor]
    activity_factory, object_factory = factories_for_actor_object(
        alice, id_generator=id_generator_for_actor(alice)
    )

    bob_id = context.actors[target].get("id")

    note = object_factory.note(content=text, to={bob_id}).build()
    activity = activity_factory.create(note).build()

    await publish_as(
        context,
        actor,
        "send_message",
        send_message_as_actor(alice, activity),
    )

send_message_followers async 

send_message_followers(context, actor, text)

Used to send a message to the followers. The message has the format (with a lot of stuff omitted)

{
    "type": "Create",
    "object": {
        "type": "Note",
        "content": text,
        "to": [followers_collection_of_actor]
    }
}

This step can be used as

When "alice" messages her followers "Got milk?"
Source code in cattle_grid/testing/features/steps/messaging.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
@when('"{actor}" messages her followers "{text}"')  # type: ignore
async def send_message_followers(context, actor, text):
    """Used to send a message to the followers. The message has the format (with a lot of stuff omitted)

    ```json
    {
        "type": "Create",
        "object": {
            "type": "Note",
            "content": text,
            "to": [followers_collection_of_actor]
        }
    }
    ```

    This step can be used as

    ```gherkin
    When "alice" messages her followers "Got milk?"
    ```
    """
    for connection in context.connections.values():
        await connection.clear_incoming()

    alice = context.actors[actor]

    activity_factory, object_factory = factories_for_actor_object(
        alice, id_generator=id_generator_for_actor(alice)
    )
    note = object_factory.note(content=text).as_followers().build()
    activity = activity_factory.create(note).build()

    await publish_as(
        context,
        actor,
        "send_message",
        send_message_as_actor(alice, activity),
    )

cattle_grid.testing.features.steps.user

actor_deletes_themselves async 

actor_deletes_themselves(context, alice)

When "Alice" deletes herself
When "Bob" deletes himself
Source code in cattle_grid/testing/features/steps/user.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
@when('"{alice}" deletes herself')  # pyright: ignore[reportCallIssue]
@when('"{alice}" deletes himself')  # pyright: ignore[reportCallIssue]
async def actor_deletes_themselves(context, alice):
    """
    ```gherkin
    When "Alice" deletes herself
    When "Bob" deletes himself
    ```
    """
    alice_id = context.actors[alice].get("id")

    await publish_as(
        context,
        alice,
        "delete_actor",
        {
            "actor": alice_id,
        },
    )

fetch_profile async 

fetch_profile(context, alice)

When "Alice" fetches her profile

The profile is stored in context.profile

Source code in cattle_grid/testing/features/steps/user.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
@when('"{alice}" fetches her profile')  # pyright: ignore[reportCallIssue]
async def fetch_profile(context, alice):
    """
    ```gherkin
    When "Alice" fetches her profile
    ```

    The profile is stored in `context.profile`
    """

    alice_id = context.actors[alice].get("id")

    context.profile = await fetch_request(context, alice, alice_id)

    assert isinstance(context.profile, dict)

in_group async 

in_group(context, alice, group_name)

And "Alice" is in the "html" group
Source code in cattle_grid/testing/features/steps/user.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@given('"{alice}" is in the "{group_name}" group')  # pyright: ignore[reportCallIssue]
async def in_group(context, alice, group_name):
    """
    ```gherkin
    And "Alice" is in the "html" group
    ```
    """
    alice_id = context.actors[alice].get("id")
    config = load_settings()

    async with database_session(db_uri=config.get("db_uri")) as session:  # type: ignore
        manager = ActorManager(session=session, actor_id=alice_id)

        await manager.add_to_group(group_name)

new_user 

new_user(context, username)

Creates a new user

Usage example:

Given A new user called "Alice"
Source code in cattle_grid/testing/features/steps/user.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
@given('A new user called "{username}"')  # pyright: ignore[reportCallIssue]
def new_user(context, username):
    """Creates a new user

    Usage example:

    ```gherkin
    Given A new user called "Alice"
    ```
    """
    hostname = {"alice": "abel", "bob": "banach", "Bob": "banach"}.get(username, "abel")

    context.execute_steps(
        f"""
        Given A new user called "{username}" on "{hostname}"
        """
    )

update_profile async 

update_profile(context, alice)

When "Alice" updates her profile
Source code in cattle_grid/testing/features/steps/user.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
@when('"{alice}" updates her profile')  # pyright: ignore[reportCallIssue]
async def update_profile(context, alice):
    """
    ```gherkin
    When "Alice" updates her profile
    ```
    """

    for connection in context.connections.values():
        await connection.clear_incoming()

    alice_id = context.actors[alice].get("id")

    msg = UpdateActorMessage(
        actor=alice_id, profile={"summary": "I love cows"}
    ).model_dump()

    await publish_as(context, alice, "update_actor", msg)