Skip to content

bovine_store.utils.test

bovine_admin_store(store) async

Fixture for a bovine admin store, uses the same database as the store fixture.

Source code in bovine_store/bovine_store/utils/test.py 
81
82
83
84
85
86
87
88
89
@pytest.fixture
async def bovine_admin_store(store) -> AsyncGenerator[BovineAdminStore, None]:
    """Fixture for a bovine admin store, uses the same database as the store fixture."""
    endpoint_mock = EndpointMock()

    yield BovineAdminStore(
        determine_db_url(),
        endpoint_path_function=endpoint_mock.generate,
    )

bovine_store_actor(store, bovine_admin_store) async

Creates and returns a BovineStoreActor

Source code in bovine_store/bovine_store/utils/test.py 
92
93
94
95
96
97
98
99
@pytest.fixture
async def bovine_store_actor(
    store, bovine_admin_store
) -> AsyncGenerator[BovineStoreActor, None]:
    """Creates and returns a BovineStoreActor"""
    bovine_name = await bovine_admin_store.register("MoonJumpingCow")

    yield await store.actor_for_name(bovine_name)

store() async

pytest fixture for a BovineStore if the environment variable BOVINE_DB_URL the db from it is used. The schema is created and dropped on each test. If the environment variable is not set a sqlite3 database is used, whose file is deleted after the test is finished.

Source code in bovine_store/bovine_store/utils/test.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
70
71
72
73
74
75
76
77
78
@pytest.fixture
async def store() -> AsyncGenerator[BovineStore, None]:
    """pytest fixture for a BovineStore if the environment variable
    `BOVINE_DB_URL` the db from it is used. The schema is created
    and dropped on each test. If the environment variable is not set
    a sqlite3 database is used, whose file is deleted after the test
    is finished."""
    response_mock = AsyncMock()
    response_mock.raise_for_status = lambda: 1
    session_mock = AsyncMock()
    session_mock.get.return_value = response_mock

    db_url = determine_db_url()
    store = BovineStore(
        db_url,
        session=session_mock,
    )
    await init_connection(db_url)

    yield store

    if os.environ.get("BOVINE_DB_URL"):
        connection = connections.get("default")
        for table in [
            "visibleto",
            "bovineactorendpoint",
            "bovineactorkeypair",
            "collectionitem",
            "storedjsonobject",
            "bovineactor",
        ]:
            await connection.execute_query(f"DROP TABLE IF EXISTS {table}")
        await close_connection()
    else:
        await close_connection()