/
test_storage.py
133 lines (116 loc) · 4.41 KB
/
test_storage.py
1
2
3
4
5
6
7
8
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
38
39
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
70
71
72
73
74
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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
"""
Test the storage module.
"""
import enum
import pytest
from sqlalchemy import Column, Enum, Integer, Table
from jobbergate_api.metadata import metadata
from jobbergate_api.storage import build_db_url, database, sort_clause
# Force the async event loop at the app to begin.
# Since this is a time consuming fixture, it is just used were strict necessary.
pytestmark = pytest.mark.usefixtures("startup_event_force")
def test_build_db_url__creates_database_url_from_parts(tweak_settings):
"""
Provide a test ase for the ``build_db_url()`` function.
Tests that the build_db_url function computes a database url value from separate
DATABASE_ settings when the DEPLOY_ENV is not TEST.
"""
with tweak_settings(
DEPLOY_ENV="LOCAL",
DATABASE_USER="built-base-user",
DATABASE_PSWD="built-base-pswd",
DATABASE_HOST="built-base-host",
DATABASE_NAME="built-base-name",
DATABASE_PORT=8888,
TEST_DATABASE_USER="built-test-user",
TEST_DATABASE_PSWD="built-test-pswd",
TEST_DATABASE_HOST="built-test-host",
TEST_DATABASE_NAME="built-test-name",
TEST_DATABASE_PORT=9999,
):
assert build_db_url() == (
"postgresql://built-base-user:built-base-pswd@built-base-host:8888/built-base-name"
)
def test_build_db_url__uses_TEST_prefixed_database_settings_if_passed_the_force_test_flag(tweak_settings):
"""
Provide a test ase for the ``build_db_url()`` function.
Tests that the build_db_url function computes a database url value from separate
TEST_DATABASE_ settings when the DEPLOY_ENV is TEST.
"""
with tweak_settings(
DEPLOY_ENV="TEST",
DATABASE_USER="built-base-user",
DATABASE_PSWD="built-base-pswd",
DATABASE_HOST="built-base-host",
DATABASE_NAME="built-base-name",
DATABASE_PORT=8888,
TEST_DATABASE_USER="built-test-user",
TEST_DATABASE_PSWD="built-test-pswd",
TEST_DATABASE_HOST="built-test-host",
TEST_DATABASE_NAME="built-test-name",
TEST_DATABASE_PORT=9999,
):
assert build_db_url() == (
"postgresql://built-test-user:built-test-pswd@built-test-host:9999/built-test-name"
)
@pytest.mark.asyncio
async def test_sort_clause__auto_sort_enum_column(database_engine):
"""
Provide a test case for the ``sort_clause()`` function.
Test that the sort_clause() function will correctly sort enum columns.
"""
class DummyStatusEnum(str, enum.Enum):
CREATED = "CREATED"
SUBMITTED = "SUBMITTED"
COMPLETED = "COMPLETED"
FAILED = "FAILED"
UNKNOWN = "UNKNOWN"
REJECTED = "REJECTED"
dummy_table = Table(
"dummies",
metadata,
Column("id", Integer, primary_key=True),
Column("status", Enum(DummyStatusEnum, sort_key_function=lambda v: str(v.value)), nullable=False),
)
dummy_sortables = [dummy_table.c.id, dummy_table.c.status]
# Note: do not worry about dropping the table. The pytest fixtures will clean it up for us
dummy_table.create(database_engine)
await database.execute_many(
query=dummy_table.insert(),
values=[
dict(id=44, status="COMPLETED"),
dict(id=45, status="FAILED"),
dict(id=46, status="REJECTED"),
dict(id=47, status="COMPLETED"),
dict(id=48, status="CREATED"),
dict(id=49, status="SUBMITTED"),
dict(id=50, status="REJECTED"),
dict(id=51, status="FAILED"),
],
)
status_sort_clause = sort_clause("status", dummy_sortables, True)
query = dummy_table.select().order_by(status_sort_clause, "id")
results = await database.fetch_all(query)
assert [(d["id"], d["status"]) for d in results] == [
(44, "COMPLETED"),
(47, "COMPLETED"),
(48, "CREATED"),
(45, "FAILED"),
(51, "FAILED"),
(46, "REJECTED"),
(50, "REJECTED"),
(49, "SUBMITTED"),
]
status_sort_clause = sort_clause("status", dummy_sortables, False)
query = dummy_table.select().order_by(status_sort_clause, "id")
results = await database.fetch_all(query)
assert [(d["id"], d["status"]) for d in results] == [
(49, "SUBMITTED"),
(46, "REJECTED"),
(50, "REJECTED"),
(45, "FAILED"),
(51, "FAILED"),
(48, "CREATED"),
(44, "COMPLETED"),
(47, "COMPLETED"),
]