# You should have received a copy of the GNU General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import click

import kallithea

import kallithea.bin.kallithea_cli_base as cli_base

from kallithea.lib.db_manage import DbManage

from kallithea.model.meta import Session

@cli_base.register_command(config_file=True)

@click.option('--user', help='Username of administrator account.')

@click.option('--password', help='Password for administrator account.')

@click.option('--email', help='Email address of administrator account.')

@click.option('--repos', help='Absolute path to repositories location.')

@click.option('--force-yes', is_flag=True, help='Answer yes to every question.')

@click.option('--force-no', is_flag=True, help='Answer no to every question.')

@click.option('--public-access/--no-public-access', default=True,

        help='Enable/disable public access on this installation (default: enable)')

def db_create(user, password, email, repos, force_yes, force_no, public_access):

    """Initialize the database.

    Create all required tables in the database specified in the configuration

    file. Create the administrator account. Set certain settings based on

    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'.")

"""

import logging

import os

import sys

import uuid

import alembic.command

import alembic.config

import sqlalchemy

from sqlalchemy.engine import create_engine

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

        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 '

                                           '(min 6 chars):')

                confirm = getpass.getpass('Confirm password:')

                if password != confirm:

                    log.error('passwords mismatch')

                    return False

                if len(password) < 6:

                    log.error('password is to short use at least 6 characters')

                    return False

                return password

            if username is None:

                username = input('Specify admin username:')

            if password is None:

                password = get_password()

                if not password:

                    # second try

                    password = get_password()

                    if not password:

                        sys.exit()

            if email is None:

                email = input('Specify admin email:')

            self.create_user(username, password, email, True)

        else:

                content=content,

                f_path=filename

            )

        return cs

    def review_changeset(self, repo, revision, status, author=TEST_USER_ADMIN_LOGIN):

        comment = ChangesetCommentsModel().create("review comment", repo, author, revision=revision, send_email=False)

        csm = ChangesetStatusModel().set_status(repo, ChangesetStatus.STATUS_APPROVED, author, comment, revision=revision)

        Session().commit()

        return csm

    def create_pullrequest(self, testcontroller, repo_name, pr_src_rev, pr_dst_rev, title='title'):

        org_ref = 'branch:stable:%s' % pr_src_rev

        other_ref = 'branch:default:%s' % pr_dst_rev

        with test_context(testcontroller.app): # needed to be able to mock request user

            org_repo = other_repo = Repository.get_by_repo_name(repo_name)

            owner_user = User.get_by_username(TEST_USER_ADMIN_LOGIN)

            reviewers = [User.get_by_username(TEST_USER_REGULAR_LOGIN)]

            request.authuser = AuthUser(dbuser=owner_user)

            # creating a PR sends a message with an absolute URL - without routing that requires mocking

            with mock.patch.object(helpers, 'url', (lambda arg, qualified=False, **kwargs: ('https://localhost' if qualified else '') + '/fake/' + arg)):

                cmd = CreatePullRequestAction(org_repo, other_repo, org_ref, other_ref, title, 'No description', owner_user, reviewers)

                pull_request = cmd.execute()

            Session().commit()

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

    tar.extractall(os.path.join(TESTS_TMP_PATH, HG_REPO))

    tar.close()

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

    tar.extractall(os.path.join(TESTS_TMP_PATH, GIT_REPO))

    tar.close()

    # LOAD VCS test stuff

    from kallithea.tests.vcs import setup_package

    setup_package()

def create_test_index(repo_location, config, full_index):

    """

    Makes default test index

    """

    from kallithea.lib.indexers.daemon import WhooshIndexingDaemon

    from kallithea.lib.pidlock import DaemonLock

    index_location = os.path.join(config['index_dir'])

    if not os.path.exists(index_location):

        os.makedirs(index_location)