cic-internal-integration/apps/cic-ussd/tests/fixtures/db.py

58 lines
1.8 KiB
Python
Raw Permalink Normal View History

2021-02-06 16:13:47 +01:00
# standard imports
2021-08-06 18:29:01 +02:00
import logging
2021-02-06 16:13:47 +01:00
import os
# third party imports
import alembic
import pytest
from alembic.config import Config as AlembicConfig
# local imports
from cic_ussd.db import dsn_from_config
2021-08-06 18:29:01 +02:00
from cic_ussd.db.models.base import SessionBase, create_engine
from .config import root_directory
logg = logging.getLogger(__name__)
2021-02-06 16:13:47 +01:00
@pytest.fixture(scope='session')
2021-08-06 18:29:01 +02:00
def alembic_engine(load_config):
2021-02-06 16:13:47 +01:00
data_source_name = dsn_from_config(load_config)
2021-08-06 18:29:01 +02:00
return create_engine(data_source_name)
@pytest.fixture(scope='session')
def database_engine(load_config):
2021-02-06 16:13:47 +01:00
if load_config.get('DATABASE_ENGINE') == 'sqlite':
2021-08-06 18:29:01 +02:00
try:
os.unlink(load_config.get('DATABASE_NAME'))
except FileNotFoundError:
pass
SessionBase.transactional = False
SessionBase.poolable = False
dsn = dsn_from_config(load_config)
SessionBase.connect(dsn, debug=load_config.get('DATABASE_DEBUG') is not None)
return dsn
2021-02-06 16:13:47 +01:00
@pytest.fixture(scope='function')
def init_database(load_config, database_engine):
db_directory = os.path.join(root_directory, 'cic_ussd', 'db')
migrations_directory = os.path.join(db_directory, 'migrations', load_config.get('DATABASE_ENGINE'))
if not os.path.isdir(migrations_directory):
migrations_directory = os.path.join(db_directory, 'migrations', 'default')
2021-08-06 18:29:01 +02:00
logg.info(f'using migrations directory {migrations_directory}')
2021-02-06 16:13:47 +01:00
2021-08-06 18:29:01 +02:00
session = SessionBase.create_session()
2021-02-06 16:13:47 +01:00
alembic_config = AlembicConfig(os.path.join(migrations_directory, 'alembic.ini'))
alembic_config.set_main_option('sqlalchemy.url', database_engine)
alembic_config.set_main_option('script_location', migrations_directory)
alembic.command.downgrade(alembic_config, 'base')
alembic.command.upgrade(alembic_config, 'head')
2021-08-06 18:29:01 +02:00
yield session
session.commit()
session.close()