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
|
import os
import shutil
from pathlib import Path
import alembic
import alembic.config
import alembic.command
from pyramid.paster import get_appsettings
from pyramid.scripting import prepare
from pyramid.testing import DummyRequest, testConfig
import pytest
import transaction
import webtest
from sqlalchemy import delete, select
from fietsboek import main, models
from fietsboek.data import DataManager
from fietsboek.models.meta import Base
def pytest_addoption(parser):
parser.addoption('--ini', action='store', metavar='INI_FILE')
@pytest.fixture(scope='session')
def ini_file(request):
# potentially grab this path from a pytest option
return os.path.abspath(request.config.option.ini or 'testing.ini')
@pytest.fixture(scope='session')
def app_settings(ini_file):
return get_appsettings(ini_file)
@pytest.fixture(scope='session')
def dbengine(app_settings, ini_file):
engine = models.get_engine(app_settings)
alembic_cfg = alembic.config.Config(ini_file)
Base.metadata.drop_all(bind=engine)
alembic.command.stamp(alembic_cfg, None, purge=True)
# run migrations to initialize the database
# depending on how we want to initialize the database from scratch
# we could alternatively call:
# Base.metadata.create_all(bind=engine)
# alembic.command.stamp(alembic_cfg, "head")
alembic.command.upgrade(alembic_cfg, "head")
yield engine
Base.metadata.drop_all(bind=engine)
alembic.command.stamp(alembic_cfg, None, purge=True)
@pytest.fixture
def data_manager(app_settings):
return DataManager(Path(app_settings["fietsboek.data_dir"]))
@pytest.fixture(autouse=True)
def _cleanup_data(app_settings):
yield
engine = models.get_engine(app_settings)
connection = engine.connect()
for table in reversed(Base.metadata.sorted_tables):
connection.execute(table.delete())
data_dir = Path(app_settings["fietsboek.data_dir"])
if (data_dir / "tracks").is_dir():
shutil.rmtree(data_dir / "tracks")
@pytest.fixture(scope='session')
def app(app_settings, dbengine, tmp_path_factory):
app_settings["fietsboek.data_dir"] = str(tmp_path_factory.mktemp("data"))
return main({}, dbengine=dbengine, **app_settings)
@pytest.fixture
def tm():
tm = transaction.TransactionManager(explicit=True)
tm.begin()
tm.doom()
yield tm
tm.abort()
@pytest.fixture
def dbsession(app, tm):
session_factory = app.registry['dbsession_factory']
return models.get_tm_session(session_factory, tm)
@pytest.fixture
def testapp(app, tm, dbsession):
# override request.dbsession and request.tm with our own
# externally-controlled values that are shared across requests but aborted
# at the end
testapp = webtest.TestApp(app, extra_environ={
'HTTP_HOST': 'example.com',
'tm.active': True,
'tm.manager': tm,
'app.dbsession': dbsession,
})
return testapp
@pytest.fixture
def app_request(app, tm, dbsession):
"""
A real request.
This request is almost identical to a real request but it has some
drawbacks in tests as it's harder to mock data and is heavier.
"""
with prepare(registry=app.registry) as env:
request = env['request']
request.host = 'example.com'
# without this, request.dbsession will be joined to the same transaction
# manager but it will be using a different sqlalchemy.orm.Session using
# a separate database transaction
request.dbsession = dbsession
request.tm = tm
yield request
@pytest.fixture
def dummy_request(tm, dbsession):
"""
A lightweight dummy request.
This request is ultra-lightweight and should be used only when the request
itself is not a large focus in the call-stack. It is much easier to mock
and control side-effects using this object, however:
- It does not have request extensions applied.
- Threadlocals are not properly pushed.
"""
request = DummyRequest()
request.host = 'example.com'
request.dbsession = dbsession
request.tm = tm
return request
@pytest.fixture
def dummy_config(dummy_request):
"""
A dummy :class:`pyramid.config.Configurator` object. This allows for
mock configuration, including configuration for ``dummy_request``, as well
as pushing the appropriate threadlocals.
"""
with testConfig(request=dummy_request) as config:
yield config
@pytest.fixture
def route_path(app_request):
"""
A fixture that yields a function to generate route paths.
This is equivalent to calling request.route_path on a request.
"""
def get_route_path(*args, **kwargs):
return app_request.route_path(*args, **kwargs)
return get_route_path
@pytest.fixture()
def logged_in(testapp, route_path, dbsession, tm):
"""
A fixture that represents a logged in state.
This automatically creates a user and returns the created user.
Returns the user that was logged in.
"""
tm.abort()
with tm:
user = models.User(email='foo@barre.com', is_verified=True)
user.set_password("foobar")
dbsession.add(user)
dbsession.flush()
user_id = user.id
tm.begin()
tm.doom()
login = testapp.get(route_path('login'))
form = login.form
form['email'] = 'foo@barre.com'
form['password'] = 'foobar'
response = form.submit()
assert response.status_code == 302
try:
# Make sure to return an object that is not bound to the wrong db
# session by re-fetching it with the proper fixture session:
yield dbsession.execute(select(models.User).filter_by(id=user_id)).scalar_one()
finally:
tm.abort()
with tm:
dbsession.execute(delete(models.User).filter_by(id=user_id))
tm.begin()
tm.doom()
|
