import os from cachetools import lru_cache import pytest import shutil from flask import Flask, jsonify from flask_login import LoginManager from peewee import SqliteDatabase, savepoint, InternalError from app import app as application from data import model from data.database import close_db_filter, db, configure from data.model.user import LoginWrappedDBUser from endpoints.api import api_bp from endpoints.appr import appr_bp from endpoints.web import web from endpoints.v2 import v2_bp from endpoints.verbs import verbs as verbs_bp from endpoints.webhooks import webhooks from initdb import initialize_database, populate_database from path_converters import APIRepositoryPathConverter, RegexConverter, RepositoryPathConverter from test.testconfig import FakeTransaction INIT_DB_PATH = 0 @pytest.fixture(scope="session") def init_db_path(tmpdir_factory): """ Creates a new database and appropriate configuration. Note that the initial database is created *once* per session. In the non-full-db-test case, the database_uri fixture makes a copy of the SQLite database file on disk and passes a new copy to each test. """ # NOTE: We use a global here because pytest runs this code multiple times, due to the fixture # being imported instead of being in a conftest. Moving to conftest has its own issues, and this # call is quite slow, so we simply cache it here. global INIT_DB_PATH INIT_DB_PATH = INIT_DB_PATH or _init_db_path(tmpdir_factory) return INIT_DB_PATH def _init_db_path(tmpdir_factory): if os.environ.get('TEST_DATABASE_URI'): return _init_db_path_real_db(os.environ.get('TEST_DATABASE_URI')) return _init_db_path_sqlite(tmpdir_factory) def _init_db_path_real_db(db_uri): """ Initializes a real database for testing by populating it from scratch. Note that this does *not* add the tables (merely data). Callers must have migrated the database before calling the test suite. """ configure({ "DB_URI": db_uri, "DB_CONNECTION_ARGS": { 'threadlocals': True, 'autorollback': True, }, "DB_TRANSACTION_FACTORY": _create_transaction, }) populate_database() return db_uri def _init_db_path_sqlite(tmpdir_factory): """ Initializes a SQLite database for testing by populating it from scratch and placing it into a temp directory file. """ sqlitedbfile = str(tmpdir_factory.mktemp("data").join("test.db")) sqlitedb = 'sqlite:///{0}'.format(sqlitedbfile) conf = {"TESTING": True, "DEBUG": True, "DB_URI": sqlitedb} os.environ['DB_URI'] = str(sqlitedb) db.initialize(SqliteDatabase(sqlitedbfile)) application.config.update(conf) application.config.update({"DB_URI": sqlitedb}) initialize_database() db.obj.execute_sql('PRAGMA foreign_keys = ON;') populate_database() close_db_filter(None) return str(sqlitedbfile) @pytest.fixture() def database_uri(monkeypatch, init_db_path, sqlitedb_file): """ Returns the database URI to use for testing. In the SQLite case, a new, distinct copy of the SQLite database is created by copying the initialized database file (sqlitedb_file) on a per-test basis. In the non-SQLite case, a reference to the existing database URI is returned. """ if os.environ.get('TEST_DATABASE_URI'): db_uri = os.environ['TEST_DATABASE_URI'] monkeypatch.setenv("DB_URI", db_uri) return db_uri # Copy the golden database file to a new path. shutil.copy2(init_db_path, sqlitedb_file) # Monkeypatch the DB_URI. db_path = 'sqlite:///{0}'.format(sqlitedb_file) monkeypatch.setenv("DB_URI", db_path) return db_path @pytest.fixture() def sqlitedb_file(tmpdir): """ Returns the path at which the initialized, golden SQLite database file will be placed. """ test_db_file = tmpdir.mkdir("quaydb").join("test.db") return str(test_db_file) def _create_transaction(db): return FakeTransaction() @pytest.fixture() def appconfig(database_uri): """ Returns application configuration for testing that references the proper database URI. """ conf = { "TESTING": True, "DEBUG": True, "DB_URI": database_uri, "SECRET_KEY": 'superdupersecret!!!1', "DB_CONNECTION_ARGS": { 'threadlocals': True, 'autorollback': True, }, "DB_TRANSACTION_FACTORY": _create_transaction, } return conf @pytest.fixture() def initialized_db(appconfig): """ Configures the database for the database found in the appconfig. """ # Configure the database. configure(appconfig) # Initialize caches. model._basequery._lookup_team_roles() model._basequery.get_public_repo_visibility() model.log.get_log_entry_kinds() # If under a test *real* database, setup a savepoint. under_test_real_database = bool(os.environ.get('TEST_DATABASE_URI')) if under_test_real_database: test_savepoint = savepoint(db) test_savepoint.__enter__() yield # Run the test. try: test_savepoint.__exit__(None, None, None) except InternalError: # If postgres fails with an exception (like IntegrityError) mid-transaction, it terminates # it immediately, so when we go to remove the savepoint, it complains. We can safely ignore # this case. pass else: yield @pytest.fixture() def app(appconfig, initialized_db): """ Used by pytest-flask plugin to inject a custom app instance for testing. """ app = Flask(__name__) login_manager = LoginManager(app) @app.errorhandler(model.DataModelException) def handle_dme(ex): response = jsonify({'message': ex.message}) response.status_code = 400 return response @login_manager.user_loader def load_user(user_uuid): return LoginWrappedDBUser(user_uuid) app.url_map.converters['regex'] = RegexConverter app.url_map.converters['apirepopath'] = APIRepositoryPathConverter app.url_map.converters['repopath'] = RepositoryPathConverter app.register_blueprint(api_bp, url_prefix='/api') app.register_blueprint(appr_bp, url_prefix='/cnr') app.register_blueprint(web, url_prefix='/') app.register_blueprint(verbs_bp, url_prefix='/c1') app.register_blueprint(v2_bp, url_prefix='/v2') app.register_blueprint(webhooks, url_prefix='/webhooks') app.config.update(appconfig) return app