forked from oxen-io/session-pysogs
-
Notifications
You must be signed in to change notification settings - Fork 0
/
conftest.py
280 lines (196 loc) · 7.33 KB
/
conftest.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
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
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
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
import pytest
from sogs import config
config.DB_URL = 'defer-init'
config.REQUIRE_BLIND_KEYS = False
from sogs import web # noqa: E402
from sogs.model.room import Room # noqa: E402
from sogs.model.user import SystemUser # noqa: E402
import sogs.omq # noqa: E402
import sqlalchemy # noqa: E402
import atexit, shutil, tempfile # noqa: E402 E401
_tempdirs = set()
sogs.omq.test_suite = True
def pgsql_url(arg):
if not arg.startswith('postgresql:'):
raise ValueError("Invalid postgresql url; see SQLAlchemy postgresql docs")
return arg
def pytest_addoption(parser):
parser.addoption(
"--sql-tracing", action="store_true", default=False, help="Log all SQL queries"
)
parser.addoption(
"--pgsql", type=pgsql_url, help="Use the given postgresql database url for testing"
)
parser.addoption(
"--pgsql-no-drop-schema",
action="store_true",
default=False,
help="Don't clean up the final test schema; typically used with --maxfail=1",
)
@pytest.fixture
def client():
"""Yields an flask test client for the app that can be used to make test requests"""
with web.app.test_client() as client:
yield client
db_counter_ = 0
# Under postgresql & sqlalchemy 1.3.x we have to hack around a bit to get sqlalchemy to properly use
# the pgsql search_path, most notably passing it in the connect string, but that fails if the schema
# doesn't already exist; thus we establish an extra connection to the database to create it at
# session startup and tear it down at session end.
#
# Under higher sqlalchemy this simply provides the pgsql url, and under sqlite this returns None.
@pytest.fixture(scope="session")
def pgsql(request):
pgsql = request.config.getoption("--pgsql")
if pgsql:
if sqlalchemy.__version__.startswith("1.3."):
import psycopg2
conn = psycopg2.connect(pgsql)
with conn.cursor() as cur:
cur.execute("CREATE SCHEMA IF NOT EXISTS sogs_tests")
conn.commit()
pgsql_w_schema = pgsql + ('&' if '?' in pgsql else '?')
pgsql_w_schema += 'options=--search_path%3Dsogs_tests%2Cpublic'
yield pgsql_w_schema
if not request.config.getoption("--pgsql-no-drop-schema"):
print("DROPPING SCHEMA")
with conn.cursor() as cur:
cur.execute("DROP SCHEMA sogs_tests CASCADE")
conn.commit()
conn.close()
else:
yield pgsql
else:
yield None
@pytest.fixture
def db(request, pgsql):
"""
Import this fixture to get a wiped, re-initialized database for db.engine. The actual fixture
value is the db module itself (so typically you don't import it at all but instead get it
through this fixture, which also creates an empty db for you).
"""
d = tempfile.mkdtemp(prefix='tmp_pysogs')
_tempdirs.add(d)
config.UPLOAD_PATH = d
trace = request.config.getoption("--sql-tracing")
from sogs import db as db_
global db_counter_
db_counter_ += 1
if pgsql:
web.app.logger.warning(f"using postgresql {pgsql}")
first = True
def pg_setup_schema():
# Run everything in a separate schema that we can easily drop when done
@sqlalchemy.event.listens_for(db_.engine, "connect", insert=True)
def setup_schema(dbapi_connection, connection_record):
existing_autocommit = dbapi_connection.autocommit
dbapi_connection.autocommit = True
cursor = dbapi_connection.cursor()
nonlocal first
if first:
cursor.execute("DROP SCHEMA IF EXISTS sogs_tests CASCADE")
first = False
cursor.execute("CREATE SCHEMA IF NOT EXISTS sogs_tests")
cursor.execute("SET search_path TO sogs_tests, public")
cursor.close()
dbapi_connection.autocommit = existing_autocommit
db_.init_engine(pgsql, echo=trace, sogs_preinit=pg_setup_schema)
else:
sqlite_uri = f'file:sogs_testdb{db_counter_}?mode=memory&cache=shared'
web.app.logger.warning(f"using sqlite {sqlite_uri}")
def sqlite_connect():
import sqlite3
web.app.logger.warning(f"connecting to {sqlite_uri}")
return sqlite3.connect(sqlite_uri, uri=True)
db_.init_engine("sqlite://", creator=sqlite_connect, echo=trace)
db_.database_init()
web.appdb = db_.get_conn()
yield db_
web.app.logger.warning("closing db")
if (
pgsql
and not sqlalchemy.__version__.startswith("1.3.")
and not request.config.getoption("--pgsql-no-drop-schema")
):
web.app.logger.critical("DROPPING SCHEMA")
db_.query("DROP SCHEMA sogs_tests CASCADE")
web.appdb.close()
@pytest.fixture
def room(db):
"""
Creates a basic test room, because many, many tests need this. (Also implicitly pulls in the db
fixture for a fresh database).
"""
return Room.create('test-room', name='Test room', description='Test suite testing room')
@pytest.fixture
def room2(db):
"""
Creates a test room, typically used with `room` when two separate rooms are needed. Note that
`mod` and `admin` (if used) are only a mod and admin of `room`, not `room2`
"""
return Room.create('room2', name='Room 2', description='Test suite testing room2')
@pytest.fixture
def user(db):
"""
Generates an ordinary user without any special privileges. Returns a subclass of a model.User
that also has key signing methods.
"""
import user
return user.User()
@pytest.fixture
def user2(db):
"""Same as user; used (along with user) when you want two distinct regular users"""
import user
return user.User()
@pytest.fixture
def mod(room):
"""Creates a user who has moderator (but not admin) permission in `room`"""
import user
u = user.User()
room.set_moderator(u, added_by=SystemUser())
return u
@pytest.fixture
def admin(room):
"""Creates a user who has admin permission in `room`"""
import user
u = user.User()
room.set_moderator(u, added_by=SystemUser(), admin=True)
return u
@pytest.fixture
def global_admin(db):
"""Creates a user who has (hidden) global admin permissions"""
import user
u = user.User()
u.set_moderator(added_by=SystemUser(), admin=True)
return u
@pytest.fixture
def global_mod(db):
"""Creates a user who has (hidden) global moderator permissions"""
import user
u = user.User()
u.set_moderator(added_by=SystemUser())
return u
@pytest.fixture
def banned_user(db):
import user
u = user.User()
u.ban(banned_by=SystemUser())
return u
@pytest.fixture
def blind_user(db):
import user
return user.User(blinded=True)
@pytest.fixture
def blind_user2(db):
import user
return user.User(blinded=True)
@pytest.fixture
def no_rate_limit():
"""Disables post rate limiting for the test"""
import sogs.model.room as mroom
saved = (mroom.rate_limit_size, mroom.rate_limit_interval)
mroom.rate_limit_size, mroom.rate_limit_interval = None, None
yield None
mroom.rate_limit_size, mroom.rate_limit_interval = saved
web.app.config.update({'TESTING': True})
atexit.register(lambda: [shutil.rmtree(d) for d in _tempdirs])