Changeset - 3afc6bff84bb
[Not reviewed]
default
0 2 0
Mads Kiilerich (mads) - 5 years ago 2020-10-10 20:04:51
mads@kiilerich.com
Grafted from: d7bd8c2dc13d
db_manage: drop special DbManage mode for tests

There is no need for a special mode - just use force_ask to answer yes to
"destroy old database".
2 files changed with 6 insertions and 8 deletions:
0 comments (0 inline, 0 general)
kallithea/lib/db_manage.py
Show inline comments
 
@@ -40,27 +40,26 @@ from sqlalchemy.engine import create_eng
 
from kallithea.lib.utils2 import ask_ok
 
from kallithea.model.base import init_model
 
from kallithea.model.db import Repository, Setting, Ui, User
 
from kallithea.model.meta import Base, Session
 
from kallithea.model.permission import PermissionModel
 
from kallithea.model.user import UserModel
 

	
 

	
 
log = logging.getLogger(__name__)
 

	
 

	
 
class DbManage(object):
 
    def __init__(self, dbconf, root, tests=False, SESSION=None, cli_args=None):
 
    def __init__(self, dbconf, root, SESSION=None, cli_args=None):
 
        self.dbname = dbconf.split('/')[-1]
 
        self.tests = tests
 
        self.root = root
 
        self.dburi = dbconf
 
        self.cli_args = cli_args or {}
 
        self.init_db(SESSION=SESSION)
 

	
 
    def _ask_ok(self, msg):
 
        """Invoke ask_ok unless the force_ask option provides the answer"""
 
        force_ask = self.cli_args.get('force_ask')
 
        if force_ask is not None:
 
            return force_ask
 
        return ask_ok(msg)
 

	
 
@@ -78,28 +77,27 @@ class DbManage(object):
 
        Create database (optional) and tables.
 
        If reuse_database is false, the database will be dropped (if it exists)
 
        and a new one created. If true, the existing database will be reused
 
        and cleaned for content.
 
        """
 
        url = sqlalchemy.engine.url.make_url(self.dburi)
 
        database = url.database
 
        if reuse_database:
 
            log.info("The content of the database %r will be destroyed and new tables created." % database)
 
        else:
 
            log.info("The existing database %r will be destroyed and a new one created." % database)
 

	
 
        if not self.tests:
 
            if not self._ask_ok('Are you sure to destroy old database? [y/n]'):
 
                print('Nothing done.')
 
                sys.exit(0)
 
        if not self._ask_ok('Are you sure to destroy old database? [y/n]'):
 
            print('Nothing done.')
 
            sys.exit(0)
 

	
 
        if reuse_database:
 
            Base.metadata.drop_all()
 
        else:
 
            if url.drivername == 'mysql':
 
                url.database = None  # don't connect to the database (it might not exist)
 
                engine = sqlalchemy.create_engine(url)
 
                with engine.connect() as conn:
 
                    conn.execute('DROP DATABASE IF EXISTS `%s`' % database)
 
                    conn.execute('CREATE DATABASE `%s` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci' % database)
 
            elif url.drivername == 'postgresql':
 
                from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
 
@@ -188,25 +186,25 @@ class DbManage(object):
 
                log.debug('Skipping option %s', k)
 
                continue
 
            setting = Setting(k, v, t)
 
            self.sa.add(setting)
 

	
 
    def prompt_repo_root_path(self, test_repo_path='', retries=3):
 
        _path = self.cli_args.get('repos_location')
 
        if retries == 3:
 
            log.info('Setting up repositories config')
 

	
 
        if _path is not None:
 
            path = _path
 
        elif not self.tests and not test_repo_path:
 
        elif not test_repo_path:
 
            path = input(
 
                 'Enter a valid absolute path to store repositories. '
 
                 'All repositories in that path will be added automatically:'
 
            )
 
        else:
 
            path = test_repo_path
 
        path_ok = True
 

	
 
        # check proper dir
 
        if not os.path.isdir(path):
 
            path_ok = False
 
            log.error('Given path %s is not a valid directory', path)
kallithea/tests/fixture.py
Show inline comments
 
@@ -357,26 +357,26 @@ def create_test_env(repos_test_path, con
 
    """
 

	
 
    # PART ONE create db
 
    dbconf = config['sqlalchemy.url']
 
    log.debug('making test db %s', dbconf)
 

	
 
    # create test dir if it doesn't exist
 
    if not os.path.isdir(repos_test_path):
 
        log.debug('Creating testdir %s', repos_test_path)
 
        os.makedirs(repos_test_path)
 

	
 
    dbmanage = DbManage(dbconf=dbconf, root=config['here'],
 
                        tests=True,
 
                        cli_args={
 
                            'force_ask': True,
 
                            'username': TEST_USER_ADMIN_LOGIN,
 
                            'password': TEST_USER_ADMIN_PASS,
 
                            'email': TEST_USER_ADMIN_EMAIL,
 
                        })
 
    dbmanage.create_tables(reuse_database=reuse_database)
 
    # for tests dynamically set new root paths based on generated content
 
    dbmanage.create_settings(dbmanage.prompt_repo_root_path(repos_test_path))
 
    dbmanage.create_default_user()
 
    dbmanage.create_admin_user()
 
    dbmanage.create_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS, TEST_USER_REGULAR_EMAIL, False)
 
    dbmanage.create_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS, TEST_USER_REGULAR2_EMAIL, False)
 
    dbmanage.create_permissions()
0 comments (0 inline, 0 general)