import alembic import alembic.config import alembic.command import os 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 from fietsboek import models 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(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()