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.
    """
    async with database_session(
        db_url=app_globals.application_config.db_url
    ) as session:  # type: ignore
        account = await create_account(session, alice, alice)
        assert account
        await add_permission(session, account, "admin")

    mq_host = urlparse(app_globals.application_config.amqp_url).hostname

    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 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
@then('The "{collection}" collection of "{bob}" does not include "{alice}"')  # type: ignore
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", f"Got result {json.dumps(result)}"
    assert actor not in result.get("orderedItems", []), (
        f"Got result {json.dumps(result)}"
    )

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 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@then('The "{collection}" collection of "{bob}" contains "{alice}"')  # type: ignore
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")

    assert result

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

    assert result.get("type") == "OrderedCollection", f"Got result {json.dumps(result)}"
    assert bob_id in result.get("orderedItems", []), f"Got result {json.dumps(result)}"

cattle_grid.testing.features.steps.drive

cattle_grid.testing.features.steps.fetch

alice_fetches_the_activity_pub_object async 

alice_fetches_the_activity_pub_object(context, alice)

This routine causes the URI stored in context.actiivty_pub_uri to be fetched by the actor corresponding to alice.

Given "Alice" fetches the ActivityPub object

or

When "Alice" fetches the ActivityPub object
Source code in cattle_grid/testing/features/steps/fetch.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@given('"{alice}" fetches the ActivityPub object')  # type: ignore
@when('"{alice}" fetches the ActivityPub object')  # type: ignore
async def alice_fetches_the_activity_pub_object(context, alice):
    """This routine causes the URI stored in
    `context.actiivty_pub_uri` to be fetched by the
    actor corresponding to `alice`.

    ```gherkin
    Given "Alice" fetches the ActivityPub object
    ```

    or

    ```gherkin
    When "Alice" fetches the ActivityPub object
    ```
    """
    result = await fetch_request(context, alice, context.activity_pub_uri)

    assert isinstance(result, dict), json.dumps(result)

    context.fetch_response = result

alices_collection_has_number_of_elements async 

alices_collection_has_number_of_elements(
    context, alice, collection, number
)

Checks that the result from alice_fetches_the_activity_pub_object has a “collection” containing number of elements. Specifying an actor, e.g. “Alice”, is necessary as a fetch request is performed.

Then For "Alice", the "likes" collection contains "no" element
Source code in cattle_grid/testing/features/steps/fetch.py 
 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
106
107
@then('For "{alice}", the "{collection}" collection contains "{number}" element')  # type: ignore
async def alices_collection_has_number_of_elements(context, alice, collection, number):
    """Checks that the result from [alice_fetches_the_activity_pub_object][cattle_grid.testing.features.steps.fetch.alice_fetches_the_activity_pub_object]
    has a "collection" containing number of elements. Specifying an actor, e.g. "Alice",
    is necessary as a fetch request is performed.

    ```gherkin
    Then For "Alice", the "likes" collection contains "no" element
    ```
    """
    collection_uri = context.fetch_response[collection]

    async def fetcher(uri):
        result = await fetch_request(context, alice, uri)
        logger.debug(json.dumps(result, indent=2))
        return result

    items = await all_elements_from_collection(fetcher, collection_uri)

    # items = result.get("orderedItems", [])

    if number == "no":
        assert len(items) == 0, items
    elif number == "one":
        assert len(items) == 1, items
        context.interaction_id = items[0]
    else:
        raise Exception("Unsupported number of elements")

check_fail 

check_fail(context)

Checks that the result in `context.fetch_response indicates failure.

Then The request fails
Source code in cattle_grid/testing/features/steps/fetch.py 
48
49
50
51
52
53
54
55
56
@then("The request fails")  # type: ignore
def check_fail(context):
    """Checks that the result in `context.fetch_response indicates failure.

    ```gherkin
    Then The request fails
    ```
    """
    assert context.fetch_response.get("type") == "Tombstone"

check_response_type 

check_response_type(context, object_type)

Checks that the result in `context.fetch_response is of a predetermined type

Then The response is of type "Page"
Source code in cattle_grid/testing/features/steps/fetch.py 
37
38
39
40
41
42
43
44
45
@then('The response is of type "{object_type}"')  # type: ignore
def check_response_type(context, object_type):
    """Checks that the result in `context.fetch_response is of a predetermined type

    ```gherkin
    Then The response is of type "Page"
    ```
    """
    assert context.fetch_response.get("type") == object_type

object_has_collection 

object_has_collection(context, collection)

Checks that the result from alice_fetches_the_activity_pub_object has a property of type “collection”

Then The ActivityPub object has a "likes" collection
Source code in cattle_grid/testing/features/steps/fetch.py 
59
60
61
62
63
64
65
66
67
68
@then('The ActivityPub object has a "{collection}" collection')  # type: ignore
def object_has_collection(context, collection):
    """Checks that the result from [alice_fetches_the_activity_pub_object][cattle_grid.testing.features.steps.fetch.alice_fetches_the_activity_pub_object]
    has a property of type "collection"

    ```gherkin
    Then The ActivityPub object has a "likes" collection
    ```
    """
    assert collection in context.fetch_response, json.dumps(context.fetch_response)

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')  # type: ignore
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_auto_following_other 

actor_follows_auto_following_other(context, bob, alice)

Combination of two steps, i.e.

When "Alice" follows auto-following "Bob"

is the same as

When "Alice" sends "Bob" a Follow Activity
Then "Bob" receives an activity
And the received activity is of type "Accept"
Source code in cattle_grid/testing/features/steps/follow.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
@given('"{bob}" follows auto-following "{alice}"')  # type: ignore
@when('"{bob}" follows auto-following "{alice}"')  # type: ignore
def actor_follows_auto_following_other(context, bob, alice):
    """Combination of two steps, i.e.

    ```gherkin
    When "Alice" follows auto-following "Bob"
    ```

    is the same as

    ```gherkin
    When "Alice" sends "Bob" a Follow Activity
    Then "Bob" receives an activity
    And the received activity is of type "Accept"
    ```
    """
    context.execute_steps(
        f"""
        When "{bob}" sends "{alice}" a Follow Activity
        Then "{bob}" receives an activity
        And the received activity is of type "Accept"
    """
    )

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
@given('"{bob}" follows "{alice}"')  # type: ignore
@when('"{bob}" follows "{alice}"')  # type: ignore
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
    ```
    """
    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 
150
151
152
153
154
155
156
157
158
159
160
161
@given('"{alice}" automatically accepts followers')  # type: ignore
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')  # type: ignore
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 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
@when('"{alice}" sends "{bob}" a Reject Follow Activity')  # type: ignore
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 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
@when('"{bob}" sends "{alice}" an Undo Follow Activity')  # type: ignore
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

check_response_status_code 

check_response_status_code(context, status_code)

Checks the status code of the response to be status_code.

Source code in cattle_grid/testing/features/steps/http.py 
27
28
29
30
@then('The response code is "{status_code}"')  # type: ignore
def check_response_status_code(context, status_code):
    """Checks the status code of the response to be `status_code`."""
    assert context.response.status == int(status_code)

make_request async 

make_request(context)

Performs the request

Source code in cattle_grid/testing/features/steps/http.py 
17
18
19
20
21
22
23
24
@when("The request is made")  # type: ignore
async def make_request(context):
    """Performs the request"""
    async with context.session.get(
        context.request["url"],
        headers={"accept": context.request["content_type"]},
    ) as response:
        context.response = response

request_set_content_type 

request_set_content_type(context, content_type)

Sets the content type of the request

Source code in cattle_grid/testing/features/steps/http.py 
11
12
13
14
@when('The request requests the content-type "{content_type}"')  # type: ignore
def request_set_content_type(context, content_type):
    """Sets the content type of the request"""
    context.request["content_type"] = content_type

request_url_is_actor 

request_url_is_actor(context, alice)

Sets the propert context.request to alice’s actor uri

Source code in cattle_grid/testing/features/steps/http.py 
4
5
6
7
8
@when('The request URL is the actor object of "{alice}"')  # type: ignore
def request_url_is_actor(context, alice):
    """Sets the propert `context.request` to alice's actor uri"""
    alice_obj = context.actors[alice]
    context.request = {"url": alice_obj.get("id")}

response_is_webpage 

response_is_webpage(context)

Check that the response in context.response has a content-type header starting with text/html

Source code in cattle_grid/testing/features/steps/http.py 
33
34
35
36
37
38
39
40
41
@then("The response is a webpage")  # type: ignore
def response_is_webpage(context):
    """Check that the response in `context.response` has a `content-type` header
    starting with `text/html`"""
    headers = context.response.headers

    assert headers.get("content-type").startswith("text/html"), (
        f"Got content-type {headers.get('content-type')}"
    )

cattle_grid.testing.features.steps.interactions

alice_announced 

alice_announced(context, alice)

alias for

```gherkin
    Given "{alice}" fetches the ActivityPub object
    When "{alice}" announces the ActivityPub object
    Then For "{alice}", the "shares" collection contains "one" element
Given "{alice}" announced the ActivityPub object

```

Source code in cattle_grid/testing/features/steps/interactions.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@given('"{alice}" announced the ActivityPub object')
def alice_announced(context, alice):
    """alias for

    ```gherkin
        Given "{alice}" fetches the ActivityPub object
        When "{alice}" announces the ActivityPub object
        Then For "{alice}", the "shares" collection contains "one" element
    ```
    """
    context.execute_steps(f"""
        Given "{alice}" fetches the ActivityPub object
        When "{alice}" announces the ActivityPub object
        Then For "{alice}", the "shares" collection contains "one" element
""")

alice_deletes_reply async 

alice_deletes_reply(context, alice)

Deletes the reply in `context.interaction_id.

Usage:

When "{alice}" deletes her reply to the ActivityPub object
Source code in cattle_grid/testing/features/steps/interactions.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
@when('"{alice}" deletes her reply to the ActivityPub object')  # type: ignore
async def alice_deletes_reply(context, alice):
    """Deletes the reply in `context.interaction_id."""
    alice_actor = context.actors[alice]
    activity_factory, _ = factories_for_actor_object(alice_actor)

    activity = (
        activity_factory.delete(
            context.interaction_id,
            to={context.fetch_response.get("attributedTo")},
        )
        .as_public()
        .build()
    )

    await publish_as(
        context,
        alice,
        "publish_activity",
        {"actor": alice_actor.get("id"), "data": activity},
    )

    await asyncio.sleep(0.3)

alice_liked 

alice_liked(context, alice)

alias for

```gherkin
    Given "{alice}" fetches the ActivityPub object
    When "{alice}" likes the ActivityPub object
    Then For "{alice}", the "likes" collection contains "one" element
Given "{alice}" liked the ActivityPub object

```

Source code in cattle_grid/testing/features/steps/interactions.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@given('"{alice}" liked the ActivityPub object')
def alice_liked(context, alice):
    """alias for

    ```gherkin
        Given "{alice}" fetches the ActivityPub object
        When "{alice}" likes the ActivityPub object
        Then For "{alice}", the "likes" collection contains "one" element
    ```
    """
    context.execute_steps(f"""
        Given "{alice}" fetches the ActivityPub object
        When "{alice}" likes the ActivityPub object
        Then For "{alice}", the "likes" collection contains "one" element
""")

alice_likes async 

alice_likes(context, alice)

Alice likes the object given by alice_fetches_the_activity_pub_object

Usage:

When "{alice}" likes the ActivityPub object
Source code in cattle_grid/testing/features/steps/interactions.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
@when('"{alice}" likes the ActivityPub object')
async def alice_likes(context, alice):
    """Alice likes the object given by [alice_fetches_the_activity_pub_object][cattle_grid.testing.features.steps.fetch.alice_fetches_the_activity_pub_object]"""
    alice_actor = context.actors[alice]
    activity_factory, _ = factories_for_actor_object(alice_actor)

    activity = (
        activity_factory.like(
            context.fetch_response.get("id"),
            to={context.fetch_response.get("attributedTo")},
        )
        .as_public()
        .build()
    )

    await publish_as(
        context,
        alice,
        "publish_activity",
        {"actor": alice_actor.get("id"), "data": activity},
    )

    await asyncio.sleep(0.3)

alice_replied 

alice_replied(context, alice)

alias for

```gherkin
    Given "{alice}" fetches the ActivityPub object
    When "{alice}" replies to the ActivityPub object with "Nice post!"
    Then For "{alice}", the "replies" collection contains "one" element
Given "{alice}" replied to the ActivityPub object

```

Source code in cattle_grid/testing/features/steps/interactions.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
@given('"{alice}" replied to the ActivityPub object')
def alice_replied(context, alice):
    """alias for

    ```gherkin
        Given "{alice}" fetches the ActivityPub object
        When "{alice}" replies to the ActivityPub object with "Nice post!"
        Then For "{alice}", the "replies" collection contains "one" element
    ```
    """
    context.execute_steps(f"""
        Given "{alice}" fetches the ActivityPub object
        When "{alice}" replies to the ActivityPub object with "Nice post!"
        Then For "{alice}", the "replies" collection contains "one" element
""")

alice_replied_with 

alice_replied_with(context, alice, text)

alias for

```gherkin
    Given "{alice}" fetches the ActivityPub object
    When "{alice}" replies to the ActivityPub object with "{text}"
    Then For "{alice}", the "replies" collection contains "one" element
Given "{alice}" replied to the ActivityPub object with "{text}

```

Source code in cattle_grid/testing/features/steps/interactions.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
@given('"{alice}" replied to the ActivityPub object with "{text}')
def alice_replied_with(context, alice, text):
    """alias for

    ```gherkin
        Given "{alice}" fetches the ActivityPub object
        When "{alice}" replies to the ActivityPub object with "{text}"
        Then For "{alice}", the "replies" collection contains "one" element
    ```
    """
    context.execute_steps(f"""
        Given "{alice}" fetches the ActivityPub object
        When "{alice}" replies to the ActivityPub object with "{text}"
        Then For "{alice}", the "replies" collection contains "one" element
""")

alice_replies async 

alice_replies(context, alice, text)

Replies with text to the ActivityPub object

Usage:

When "{alice}" replies to the ActivityPub object with "{text}"
Source code in cattle_grid/testing/features/steps/interactions.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
@when('"{alice}" replies to the ActivityPub object with "{text}"')
async def alice_replies(context, alice, text):
    """Replies with text to the ActivityPub object"""
    alice_actor = context.actors[alice]
    _, object_factory = factories_for_actor_object(alice_actor)

    reply = (
        object_factory.reply(context.fetch_response, content=text).as_public().build()
    )
    reply["type"] = "Note"

    await publish_as(
        context,
        alice,
        "publish_object",
        {"actor": alice_actor.get("id"), "data": reply},
    )

    await asyncio.sleep(0.3)

alice_shares async 

alice_shares(context, alice)

Sends an announce activity

Usage:

When "{alice}" announces the ActivityPub object
Source code in cattle_grid/testing/features/steps/interactions.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
@when('"{alice}" announces the ActivityPub object')
async def alice_shares(context, alice):
    """Sends an announce activity"""
    alice_actor = context.actors[alice]
    activity_factory, _ = factories_for_actor_object(alice_actor)

    activity = (
        activity_factory.announce(
            context.fetch_response.get("id"),
            to={context.fetch_response.get("attributedTo")},
        )
        .as_public()
        .build()
    )

    await publish_as(
        context,
        alice,
        "publish_activity",
        {"actor": alice_actor.get("id"), "data": activity},
    )

    await asyncio.sleep(0.3)

alice_undoes async 

alice_undoes(context, alice)

Alice undoes the last interaction with the object given by alice_fetches_the_activity_pub_object

Usage:

When "{alice}" undoes the interaction the ActivityPub object
Source code in cattle_grid/testing/features/steps/interactions.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
@when('"{alice}" undoes the interaction the ActivityPub object')
async def alice_undoes(context, alice):
    """Alice undoes the last interaction with the object given by
    [alice_fetches_the_activity_pub_object][cattle_grid.testing.features.steps.fetch.alice_fetches_the_activity_pub_object]
    """
    alice_actor = context.actors[alice]
    activity_factory, _ = factories_for_actor_object(alice_actor)

    activity = (
        activity_factory.undo(
            context.interaction_id,
            to={context.fetch_response.get("attributedTo")},
        )
        .as_public()
        .build()
    )

    await publish_as(
        context,
        alice,
        "publish_activity",
        {"actor": alice_actor.get("id"), "data": activity},
    )

    await asyncio.sleep(0.3)

alice_update_post async 

alice_update_post(context, alice, text)

Updates the reply in `context.interaction_id.

Usage:

When "{alice}" updates her reply with "{text}"
Source code in cattle_grid/testing/features/steps/interactions.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
@when('"{alice}" updates her reply with "{text}"')
async def alice_update_post(context, alice, text):
    """Updates the reply in `context.interaction_id."""
    alice_actor = context.actors[alice]
    activity_factory, object_factory = factories_for_actor_object(alice_actor)

    reply = (
        object_factory.reply(context.fetch_response, content=text).as_public().build()
    )
    reply["type"] = "Note"

    reply["id"] = context.interaction_id

    activity = activity_factory.update(reply).build()

    await publish_as(
        context,
        alice,
        "publish_activity",
        {"actor": alice_actor.get("id"), "data": activity},
    )

    await asyncio.sleep(0.3)

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")

url_resolves_to_webpage_containing_text async 

url_resolves_to_webpage_containing_text(context, text)

Then The URL resolves to a webpage containing "I like cows!"
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 url_resolves_to_webpage_containing_text(context, text):
    """
    ```gherkin
    Then The URL resolves to a webpage containing "I like cows!"
    ```
    """

    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.

Usage:

Then the received activity is of type "{activity_type}"
Source code in cattle_grid/testing/features/steps/messaging.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@then('the received activity is of type "{activity_type}"')
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`.
    """

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

    import json

    logger.debug(json.dumps(received, indent=2))

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

check_embedded_object_type 

check_embedded_object_type(context, object_type)

Usage:

Then the embedded object is of type "{object_type}"
Source code in cattle_grid/testing/features/steps/messaging.py 
164
165
166
167
168
169
170
171
172
@then('the embedded object is of type "{object_type}"')
def check_embedded_object_type(context, object_type):
    received = context.activity
    if "raw" in received:
        received = received["raw"]

    obj = received.get("object", {})

    assert obj.get("type") == object_type, f"Embedded object {obj} has the wrong type"

check_lookup_message async 

check_lookup_message(context, bob)

Usage:

Then "{bob}" can lookup this message by id
Source code in cattle_grid/testing/features/steps/messaging.py 
120
121
122
123
124
125
126
127
@then('"{bob}" can lookup this message by id')
async def check_lookup_message(context, bob):
    obj_id = context.received_object.get("id")

    result = await fetch_request(context, bob, obj_id)

    assert isinstance(result, dict), f"got {result}"
    assert result == context.received_object, result

not_receive_activity async 

not_receive_activity(context, actor)

Ensures that no incoming activity was received

Usage:

Then "{actor}" does not receive an activity
Source code in cattle_grid/testing/features/steps/messaging.py 
64
65
66
67
68
69
70
71
72
73
@then('"{actor}" does not receive an activity')
async def not_receive_activity(context, actor):
    """Ensures that no incoming activity was received"""

    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.

Usage:

Then "{actor}" receives an activity
Source code in cattle_grid/testing/features/steps/messaging.py 
50
51
52
53
54
55
56
57
58
59
60
61
@then('"{actor}" receives an activity')
async def receive_activity(context, actor):
    """Ensures that an incoming activity was received
    and stores it in `context.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.

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

Usage:

Then "{actor}" receives a message saying "{text}"
Source code in cattle_grid/testing/features/steps/messaging.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@then('"{actor}" receives a message saying "{text}"')
async def receive_message(context, actor, text):
    """Used to check if the last message received by actor
    is saying the correct thing.

    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)

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

Usage:

When "{actor}" sends "{target}" a message saying "{text}"
Source code in cattle_grid/testing/features/steps/messaging.py 
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
@when('"{actor}" sends "{target}" a message saying "{text}"')
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]
        }
    }
    ```
    """
    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)

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

Usage:

When "{actor}" messages her followers "{text}"
Source code in cattle_grid/testing/features/steps/messaging.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
@when('"{actor}" messages her followers "{text}"')
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]
        }
    }
    ```
    """
    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.property_value

add_property_value async 

add_property_value(context, alice, key, value)

Usage:

When "{alice}" updates the PropertyValue "{key}" with value "{value}"
When "{alice}" adds the PropertyValue "{key}" with value "{value}"
Given "{alice}" has the PropertyValue "{key}" with value "{value}"
Source code in cattle_grid/testing/features/steps/property_value.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
@given('"{alice}" has the PropertyValue "{key}" with value "{value}"')  # type: ignore
@when('"{alice}" adds the PropertyValue "{key}" with value "{value}"')  # type: ignore
@when('"{alice}" updates the PropertyValue "{key}" with value "{value}"')  # type: ignore
async def add_property_value(context, alice, key, value):
    await publish_as(
        context,
        alice,
        "update_actor",
        {
            "actor": context.actors[alice].get("id"),
            "actions": [
                {
                    "action": "update_property_value",
                    "key": key,
                    "value": value,
                }
            ],
        },
    )

check_not_property_value 

check_not_property_value(context, key)

Usage:

Then The profile does not contain the property value "{key}"
Source code in cattle_grid/testing/features/steps/property_value.py 
55
56
57
58
59
60
61
62
@then('The profile does not contain the property value "{key}"')  # type: ignore
def check_not_property_value(context, key):
    attachments = context.result.get("attachment", [])
    assert isinstance(attachments, list)

    filtered = list(filter(lambda x: x["name"] == key, attachments))

    assert len(filtered) == 0

check_property_value 

check_property_value(context, key, value)

Usage:

Then The profile contains the property value "{key}" with value "{value}"
Source code in cattle_grid/testing/features/steps/property_value.py 
44
45
46
47
48
49
50
51
52
@then('The profile contains the property value "{key}" with value "{value}"')  # type: ignore
def check_property_value(context, key, value):
    attachments = context.result.get("attachment", [])
    assert isinstance(attachments, list)

    result = list(filter(lambda x: x["name"] == key, attachments))[0]

    assert result["type"] == "PropertyValue"
    assert result["value"] == value

remove_property_value async 

remove_property_value(context, alice, key)

Usage:

When "{alice}" removes the PropertyValue "{key}"
Source code in cattle_grid/testing/features/steps/property_value.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@when('"{alice}" removes the PropertyValue "{key}"')  # type: ignore
async def remove_property_value(context, alice, key):
    await publish_as(
        context,
        alice,
        "update_actor",
        {
            "actor": context.actors[alice].get("id"),
            "actions": [
                {
                    "action": "remove_property_value",
                    "key": key,
                }
            ],
        },
    )

cattle_grid.testing.features.steps.pull_object

check_result_is_actor 

check_result_is_actor(context, username)

Usage:

Then The retrieved object is the profile of "{username}"
Source code in cattle_grid/testing/features/steps/pull_object.py 
24
25
26
27
@then('The retrieved object is the profile of "{username}"')
def check_result_is_actor(context, username):
    assert context.result.get("type") == "Person"
    assert context.result.get("id") == context.actors[username].get("id")

retrieve_object async 

retrieve_object(context, actor, target)

Usage:

When "{actor}" retrieves the object with the actor id of "{target}"
Source code in cattle_grid/testing/features/steps/pull_object.py 
 6
 7
 8
 9
10
11
12
@when('"{actor}" retrieves the object with the actor id of "{target}"')
async def retrieve_object(context, actor, target):
    context.result = await fetch_request(
        context,
        actor,
        context.actors[target].get("id"),
    )

retrieve_object_by_uri async 

retrieve_object_by_uri(context, actor, uri)

Usage:

When "{actor}" retrieves the object with id "{uri}"
Source code in cattle_grid/testing/features/steps/pull_object.py 
15
16
17
18
19
20
21
@when('"{actor}" retrieves the object with id "{uri}"')
async def retrieve_object_by_uri(context, actor, uri):
    context.result = await fetch_request(
        context,
        actor,
        uri,
    )

cattle_grid.testing.features.steps.user

actor_deletes_themselves async 

actor_deletes_themselves(context, alice)

Usage:

When "{alice}" deletes himself
When "{alice}" deletes herself
Source code in cattle_grid/testing/features/steps/user.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
@when('"{alice}" deletes herself')
@when('"{alice}" deletes himself')
async def actor_deletes_themselves(context, alice):
    alice_id = context.actors[alice].get("id")

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

actor_uri_returned 

actor_uri_returned(context)

Usage:

Then An actor URI is returned
Source code in cattle_grid/testing/features/steps/user.py 
118
119
120
@then("An actor URI is returned")
def actor_uri_returned(context):
    assert context.actor_uri

fetch_profile async 

fetch_profile(context, alice)

The profile is stored in `context.profile`

Usage:

When "{alice}" fetches her profile
Source code in cattle_grid/testing/features/steps/user.py 
59
60
61
62
63
64
65
66
67
68
69
@when('"{alice}" fetches her profile')
async def fetch_profile(context, alice):
    """
    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)

Usage:

Given "{alice}" is in the "{group_name}" group
Source code in cattle_grid/testing/features/steps/user.py 
87
88
89
90
91
92
93
94
95
@given('"{alice}" is in the "{group_name}" group')
async def in_group(context, alice, group_name):
    alice_id = context.actors[alice].get("id")
    config = load_settings()

    async with database_session(db_url=config.get("db_url")) 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. The base_url to use is chosen randomly from the base_urls allowed in the frontend config.

Usage:

Given A new user called "{username}"
Source code in cattle_grid/testing/features/steps/user.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@given('A new user called "{username}"')
def new_user(context, username):
    """Creates a new user. The base_url to use is chosen randomly from
    the base_urls allowed in the frontend config.
    """

    base_urls = app_globals.application_config.frontend_config.base_urls

    hostname = random.choice(base_urls).removeprefix("http://")

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

new_user_on_server 

new_user_on_server(context, username, hostname)

Usage:

Given A new user called "{username}" on "{hostname}"
Source code in cattle_grid/testing/features/steps/user.py 
18
19
20
21
22
23
24
25
@given('A new user called "{username}" on "{hostname}"')
def new_user_on_server(context, username, hostname):
    context.execute_steps(
        f"""
        Given An account called "{username}"
        Given "{username}" created an actor on "{hostname}" called "{username}"
        """
    )

query_webfinger async 

query_webfinger(context, resource)

Usage:

When One queries webfinger for "{resource}"
Source code in cattle_grid/testing/features/steps/user.py 
111
112
113
114
115
@when('One queries webfinger for "{resource}"')
async def query_webfinger(context, resource):
    context.actor_uri, _ = await lookup_uri_with_webfinger(
        context.session, resource, domain="http://abel"
    )

register_boss_actor 

register_boss_actor(context, boss_handle, boss_name)

Usage:

Given the actor "{boss_handle}" is called "{boss_name}"
Source code in cattle_grid/testing/features/steps/user.py 
 98
 99
100
101
102
103
104
105
106
107
108
@given('the actor "{boss_handle}" is called "{boss_name}"')
def register_boss_actor(context, boss_handle, boss_name):
    context.execute_steps(f'When One queries webfinger for "{boss_handle}"')

    actor_name = list(context.connections.keys())[0]

    context.execute_steps(
        f'When "{actor_name}" retrieves the object with id "{context.actor_uri}"'
    )

    context.actors[boss_name] = context.result

update_profile async 

update_profile(context, alice)

Usage:

When "{alice}" updates her profile
Source code in cattle_grid/testing/features/steps/user.py 
45
46
47
48
49
50
51
52
53
54
55
56
@when('"{alice}" updates her profile')
async def update_profile(context, alice):
    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)