Skip to content

Reference .testing

cattle_grid.extensions.testing

account_for_test async 

account_for_test(sql_session) -> Account

Fixture to create an account

Source code in cattle_grid/testing/fixtures.py 
62
63
64
65
66
67
@pytest.fixture
async def account_for_test(sql_session) -> Account:
    """Fixture to create an account"""
    result = await create_account(sql_session, "alice", "alice", permissions=["admin"])
    assert result
    return result

actor_for_test async 

actor_for_test(sql_session) -> Actor

Fixture to create an actor

Source code in cattle_grid/testing/fixtures.py 
70
71
72
73
74
75
@pytest.fixture
async def actor_for_test(sql_session) -> Actor:
    """Fixture to create an actor"""
    actor = await create_actor(sql_session, "http://localhost/ap")

    return actor

actor_with_account async 

actor_with_account(sql_session, account_for_test) -> Actor

Fixture to create an actor with an account

Source code in cattle_grid/testing/fixtures.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
@pytest.fixture
async def actor_with_account(sql_session, account_for_test) -> Actor:
    """Fixture to create an actor with an account"""
    actor = await create_actor(
        sql_session, "http://localhost/ap", preferred_username="test_actor"
    )
    await add_actor_to_account(
        sql_session, account_for_test, actor, name="test_fixture"
    )

    await sql_session.refresh(actor)

    return actor

loaded_config 

loaded_config()

Ensures the configuration variables are loaded

Source code in cattle_grid/testing/fixtures.py 
111
112
113
114
115
@pytest.fixture(autouse=True, scope="session")
def loaded_config():
    """Ensures the configuration variables are loaded"""
    global_container.load_config()
    load(global_container.config)  # type: ignore

sql_engine_for_tests async 

sql_engine_for_tests()

Provides the sql engine (as in memory sqlite) for tests

This fixture has autouse=True, meaning that by importing

from cattle_grid.testing.fixtures import sql_engine_for_tests

it will run automatically. The engine is initialized in the place cattle_grid expects it.

Source code in cattle_grid/testing/fixtures.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
@pytest.fixture(autouse=True)
async def sql_engine_for_tests():
    """Provides the sql engine (as in memory sqlite) for tests

    This fixture has autouse=True, meaning that by importing

    ```python
    from cattle_grid.testing.fixtures import sql_engine_for_tests
    ```

    it will run automatically. The engine is initialized in the
    place cattle_grid expects it.
    """
    async with alchemy_database("sqlite+aiosqlite:///:memory:", echo=False) as engine:
        async with engine.begin() as conn:
            await conn.run_sync(APBase.metadata.create_all)

        yield engine

sql_session async 

sql_session(session_maker_for_tests)

Returns an AsyncSession to be used by tests

Source code in cattle_grid/testing/fixtures.py 
55
56
57
58
59
@pytest.fixture()
async def sql_session(session_maker_for_tests):
    """Returns an [AsyncSession][sqlalchemy.ext.asyncio.AsyncSession] to be used by tests"""
    async with session_maker_for_tests() as session:
        yield session

with_test_broker_for_extension async 

with_test_broker_for_extension(
    extensions, subscribers: dict[str, Callable] = {}
)

Creates a test broker with subscribtions to the routing_keys given in subscriber.

my_mock = AsyncMock()

async with with_test_broker_for_extension([extension], {
    "routing_key_to_listen_to": my_mock
})
Source code in cattle_grid/extensions/testing.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
34
@asynccontextmanager
async def with_test_broker_for_extension(
    extensions, subscribers: dict[str, Callable] = {}
):
    """Creates a test broker with subscribtions to the routing_keys given in subscriber.

    ```python
    my_mock = AsyncMock()

    async with with_test_broker_for_extension([extension], {
        "routing_key_to_listen_to": my_mock
    })
    ```
    """
    broker = RabbitBroker()

    for routing_key, subscriber in subscribers.items():
        broker.subscriber(
            RabbitQueue(routing_key, routing_key=routing_key),
            exchange=app_globals.activity_exchange,
        )(subscriber)

    add_routers_to_broker(broker, extensions)
    async with TestRabbitBroker(broker, with_real=False) as tbr:
        yield tbr