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_url=config.get("db_url")) 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}"')  # 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"
    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
58
59
@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"
    assert bob_id in result.get("orderedItems", [])

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 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
@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 
 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
@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)
        print(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 
45
46
47
48
49
50
51
52
53
@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 
34
35
36
37
38
39
40
41
42
@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 
56
57
58
59
60
61
62
63
64
65
@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')
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

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

    Given "{alice}" fetches the ActivityPub object
    When "{alice}" announces the ActivityPub object
    Then For "{alice}", the "shares" collection contains "one" element
Source code in cattle_grid/testing/features/steps/interactions.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@given('"{alice}" announced the ActivityPub object')  # type: ignore
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 replied in `context.interaction_id.

Source code in cattle_grid/testing/features/steps/interactions.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
@when('"{alice}" deletes her reply to the ActivityPub object')  # type: ignore
async def alice_deletes_reply(context, alice):
    """Deletes the replied 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

    Given "{alice}" fetches the ActivityPub object
    When "{alice}" likes the ActivityPub object
    Then For "{alice}", the "likes" collection contains "one" element
Source code in cattle_grid/testing/features/steps/interactions.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@given('"{alice}" liked the ActivityPub object')  # type: ignore
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 Example 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
32
33
34
35
36
37
@when('"{alice}" likes the ActivityPub object')  # type: ignore
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]
    Example usage

    ```gherkin
    When "Alice" likes the ActivityPub 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

    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
Source code in cattle_grid/testing/features/steps/interactions.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
@given('"{alice}" replied to the ActivityPub object')  # type: ignore
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_replies async 

alice_replies(context, alice, text)

Replies with text to the ActivityPub object

Source code in cattle_grid/testing/features/steps/interactions.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
@when('"{alice}" replies to the ActivityPub object with "{text}"')  # type: ignore
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

Source code in cattle_grid/testing/features/steps/interactions.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@when('"{alice}" announces the ActivityPub object')  # type: ignore
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 Example usage

When "Alice" undoes the interaction the ActivityPub object
Source code in cattle_grid/testing/features/steps/interactions.py 
40
41
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
@when('"{alice}" undoes the interaction the ActivityPub object')  # type: ignore
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]
    Example usage

    ```gherkin
    When "Alice" undoes the interaction the ActivityPub 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)

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.

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_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

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)