kallithea Changeset - 30aac95e7fd5

Changeset - 30aac95e7fd5

Parent rev.

Child rev.

[Not reviewed]

default

0 3 0

Mads Kiilerich - 5 years ago 2020-10-10 19:59:54
mads@kiilerich.com

Grafted from: 258a2da41849

db_manage: clarify that the purpose of admin_prompt is to create the admin user

Use cli_args to pass the test admin user information so it can be created the
usual way. This removes an unfortunate lib dependency on tests.

3 files changed with 39 insertions and 47 deletions:

kallithea/bin/kallithea_cli_db.py

kallithea/lib/db_manage.py

kallithea/tests/fixture.py

0 comments (0 inline, 0 general)

kallithea/bin/kallithea_cli_db.py

➞

Show inline comments

@@ @@ -43,39 +43,39 @@ def db_create(user, password, email, rep @@
         dbconf = 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=config['here'],
                             tests=False, cli_args=cli_args)
         dbmanage.create_tables(reuse_database=reuse)
         repo_root_path = dbmanage.prompt_repo_root_path(None)
         dbmanage.create_settings(repo_root_path)
         dbmanage.create_default_user()
-        dbmanage.admin_prompt()
+        dbmanage.create_admin_user()
         dbmanage.create_permissions()
         dbmanage.populate_default_permissions()
         Session().commit()
     else:  # then called again after app initialization
         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'.")

kallithea/lib/db_manage.py

➞

Show inline comments

@@ @@ -5,48 +5,49 @@ @@
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.lib.db_manage
 ~~~~~~~~~~~~~~~~~~~~~~~
 Database creation, and setup module for Kallithea. Used for creation
 of database as well as for migration operations
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 10, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import getpass
 import logging
 import os
 import sys
 import uuid
 import alembic.command
 import alembic.config
 import sqlalchemy
 from sqlalchemy.engine import create_engine
 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):
         self.dbname = dbconf.split('/')[-1]
@@ @@ -107,95 +108,78 @@ class DbManage(object): @@
                 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:
                 # Some databases enforce foreign key constraints and Base.metadata.drop_all() doesn't work, but this is
                 # known to work on SQLite - possibly not on other databases with strong referential integrity
                 Base.metadata.drop_all()
         Base.metadata.create_all(checkfirst=False)
         # 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
     def create_admin_user(self):
         username = self.cli_args.get('username')
         password = self.cli_args.get('password')
         email = self.cli_args.get('email')
             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:')
             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
                 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:
             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:
                     # 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:
             log.info('creating admin and regular test users')
             from kallithea.tests.base import (TEST_USER_ADMIN_EMAIL, TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS, TEST_USER_REGULAR2_EMAIL,
                                               TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS, TEST_USER_REGULAR_EMAIL, TEST_USER_REGULAR_LOGIN,
                                               TEST_USER_REGULAR_PASS)
             self.create_user(TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS,
                              TEST_USER_ADMIN_EMAIL, True)
             self.create_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS,
                              TEST_USER_REGULAR_EMAIL, False)
             self.create_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS,
                              TEST_USER_REGULAR2_EMAIL, False)
                     sys.exit()
         if email is None:
             email = input('Specify admin email:')
         self.create_user(username, password, email, True)
     def create_auth_plugin_options(self, skip_existing=False):
         """
         Create default auth plugin settings, and make it active
         :param skip_existing:
         """
         for k, v, t in [('auth_plugins', 'kallithea.lib.auth_modules.auth_internal', 'list'),
                         ('auth_internal_enabled', 'True', 'bool')]:
             if skip_existing and Setting.get_by_name(k) is not None:
                 log.debug('Skipping option %s', k)
                 continue
             setting = Setting(k, v, t)
             self.sa.add(setting)
     def create_default_options(self, skip_existing=False):
         """Creates default settings"""
         for k, v, t in [
             ('default_repo_enable_downloads', False, 'bool'),
             ('default_repo_enable_statistics', False, 'bool'),
             ('default_repo_private', False, 'bool'),
             ('default_repo_type', 'hg', 'unicode')

kallithea/tests/fixture.py

➞

Show inline comments

@@ @@ -20,50 +20,51 @@ import logging @@
 import os
 import shutil
 import tarfile
 from os.path import dirname
 import mock
 from tg import request
 from tg.util.webtest import test_context
 from kallithea.lib import helpers
 from kallithea.lib.auth import AuthUser
 from kallithea.lib.db_manage import DbManage
 from kallithea.lib.vcs.backends.base import EmptyChangeset
 from kallithea.model.changeset_status import ChangesetStatusModel
 from kallithea.model.comment import ChangesetCommentsModel
 from kallithea.model.db import ChangesetStatus, Gist, RepoGroup, Repository, User, UserGroup
 from kallithea.model.gist import GistModel
 from kallithea.model.meta import Session
 from kallithea.model.pull_request import CreatePullRequestAction  # , CreatePullRequestIterationAction, PullRequestModel
 from kallithea.model.repo import RepoModel
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.model.scm import ScmModel
 from kallithea.model.user import UserModel
 from kallithea.model.user_group import UserGroupModel
 from kallithea.tests.base import (GIT_REPO, HG_REPO, IP_ADDR, TEST_USER_ADMIN_EMAIL, TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN, TESTS_TMP_PATH,
                                   invalidate_all_caches)
 from kallithea.tests.base import (GIT_REPO, HG_REPO, IP_ADDR, TEST_USER_ADMIN_EMAIL, TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS, TEST_USER_REGULAR2_EMAIL,
                                   TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS, TEST_USER_REGULAR_EMAIL, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS,
                                   TESTS_TMP_PATH, invalidate_all_caches)
 log = logging.getLogger(__name__)
 FIXTURES = os.path.join(dirname(dirname(os.path.abspath(__file__))), 'tests', 'fixtures')
 def raise_exception(*args, **kwargs):
     raise Exception('raise_exception raised exception')
 class Fixture(object):
     def __init__(self):
         pass
     def anon_access(self, status):
         """
         Context manager for controlling anonymous access.
         Anon access will be set and committed, but restored again when exiting the block.
         Usage:
         fixture = Fixture()
@@ @@ -344,54 +345,61 @@ class Fixture(object): @@
             Session().commit()
         return pull_request.pull_request_id
 #==============================================================================
 # Global test environment setup
 #==============================================================================
 def create_test_env(repos_test_path, config, reuse_database):
     """
     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)
                         tests=True,
                         cli_args={
                             '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.admin_prompt()
     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()
     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'))

0 comments (0 inline, 0 general)