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

@@ @@ -55,25 +55,25 @@ def db_create(user, password, email, rep @@
                 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.')

kallithea/lib/db_manage.py

➞

Show inline comments

@@ @@ -17,24 +17,25 @@ 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
@@ @@ -119,71 +120,54 @@ class DbManage(object): @@
         # "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)

kallithea/tests/fixture.py

➞

Show inline comments

@@ @@ -32,26 +32,27 @@ from kallithea.lib.db_manage import DbMa @@
 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):
@@ @@ -356,30 +357,37 @@ 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)
                         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)

0 comments (0 inline, 0 general)