values you provide.

    You can pass the answers to all questions as options to this command.

    """

    dbconf = kallithea.CONFIG['sqlalchemy.url']

    # force_ask should be True (yes), False (no), or None (ask)

    if force_yes:

        force_ask = True

    elif force_no:

        force_ask = False

    else:

        force_ask = None

    cli_args = dict(

            username=user,

            password=password,

            email=email,

            repos_location=repos,

            force_ask=force_ask,

            public_access=public_access,

    )

    dbmanage = DbManage(dbconf=dbconf, root=kallithea.CONFIG['here'],

                        tests=False, cli_args=cli_args)

    repo_root_path = dbmanage.prompt_repo_root_path(None)

    dbmanage.create_settings(repo_root_path)

    dbmanage.create_default_user()

    dbmanage.admin_prompt()

    dbmanage.create_permissions()

    dbmanage.populate_default_permissions()

    Session().commit()

    # initial repository scan

    kallithea.config.application.make_app(

            kallithea.CONFIG.global_conf, **kallithea.CONFIG.local_conf)

    added, _ = kallithea.lib.utils.repo2db_mapper(kallithea.model.scm.ScmModel().repo_scan())

    if added:

        click.echo('Initial repository scan: added following repositories:')

        click.echo('\t%s' % '\n\t'.join(added))

    else:

        click.echo('Initial repository scan: no repositories found.')

    click.echo('Database set up successfully.')

    click.echo("Don't forget to build the front-end using 'kallithea-cli front-end-build'.")

        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

        from kallithea.lib.utils2 import ask_ok

        return ask_ok(msg)

    def init_db(self, SESSION=None):

        if SESSION:

            self.sa = SESSION

        else:

            # init new sessions

            engine = create_engine(self.dburi)

            init_model(engine)

            self.sa = Session()

        """

        Create a auth database

        """

        log.info("Any existing database is going to be destroyed")

        if self.tests:

            destroy = True

        else:

            destroy = self._ask_ok('Are you sure to destroy old database ? [y/n]')

        if not destroy:

            print('Nothing done.')

            sys.exit(0)

        if destroy:

            # drop and re-create old schemas

            url = sqlalchemy.engine.url.make_url(self.dburi)

            database = url.database

            # Some databases enforce foreign key constraints and Base.metadata.drop_all() doesn't work

            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

                url.database = 'postgres'  # connect to the system database (as the real one might not exist)

                engine = sqlalchemy.create_engine(url)

                with engine.connect() as conn:

                    conn.connection.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)

                    conn.execute('DROP DATABASE IF EXISTS "%s"' % database)

                    conn.execute('CREATE DATABASE "%s"' % database)

            else:

                # known to work on SQLite - possibly not on other databases with strong referential integrity

                Base.metadata.drop_all()

        # Create an Alembic configuration and generate the version table,

        # "stamping" it with the most recent Alembic migration revision, to

        # tell Alembic that all the schema upgrades are already in effect.

        alembic_cfg = alembic.config.Config()

        alembic_cfg.set_main_option('script_location', 'kallithea:alembic')

        alembic_cfg.set_main_option('sqlalchemy.url', self.dburi)

        # This command will give an error in an Alembic multi-head scenario,

        # but in practice, such a scenario should not come up during database

        # creation, even during development.

        alembic.command.stamp(alembic_cfg, 'head')

        log.info('Created tables for %s', self.dbname)

    def admin_prompt(self, second=False):

        if not self.tests:

            import getpass

            username = self.cli_args.get('username')

            password = self.cli_args.get('password')

            email = self.cli_args.get('email')

            def get_password():

                password = getpass.getpass('Specify admin password '

        return pull_request.pull_request_id

#==============================================================================

# Global test environment setup

#==============================================================================

def create_test_env(repos_test_path, config):

    """

    Makes a fresh database and

    install test repository into tmp dir

    """

    # 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)

    # 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.admin_prompt()

    dbmanage.create_permissions()

    dbmanage.populate_default_permissions()

    Session().commit()

    # PART TWO make test repo

    log.debug('making test vcs repositories')

    idx_path = config['index_dir']

    data_path = config['cache_dir']

    # clean index and data

    if idx_path and os.path.exists(idx_path):

        log.debug('remove %s', idx_path)

        shutil.rmtree(idx_path)

    if data_path and os.path.exists(data_path):

        log.debug('remove %s', data_path)

        shutil.rmtree(data_path)

    # CREATE DEFAULT TEST REPOS

    tar = tarfile.open(os.path.join(FIXTURES, 'vcs_test_hg.tar.gz'))