kallithea Changeset - 089ef495e9dd

Changeset - 089ef495e9dd

Parent rev.

Child rev.

[Not reviewed]

beta

0 1 0

Marcin Kuzminski - 14 years ago 2011-10-24 23:28:04
marcin@python-works.com

fixed test_hg_push operations

1 file changed with 22 insertions and 33 deletions:

rhodecode/tests/test_hg_operations.py

0 comments (0 inline, 0 general)

rhodecode/tests/test_hg_operations.py

➞

Show inline comments

@@ @@ -39,36 +39,40 @@ from pylons import config @@
 from sqlalchemy import engine_from_config
 from rhodecode.lib.utils import add_cache
 from rhodecode.model import init_model
 from rhodecode.model import meta
 from rhodecode.model.db import User, Repository, UserLog
 from rhodecode.lib.auth import get_crypt_password
 from rhodecode.tests import TESTS_TMP_PATH, NEW_HG_REPO, HG_REPO
 from rhodecode.config.environment import load_environment
 rel_path = dn(dn(dn(os.path.abspath(__file__))))
 conf = appconfig('config:development.ini', relative_to=rel_path)
 conf = appconfig('config:%s' % sys.argv[1], relative_to=rel_path)
 load_environment(conf.global_conf, conf.local_conf)
 add_cache(conf)
 USER = 'test_admin'
 PASS = 'test12'
 HOST = '127.0.0.1:5000'
-DEBUG = True if sys.argv[1:] else False
 DEBUG = False
 print 'DEBUG:', DEBUG
 log = logging.getLogger(__name__)
 engine = engine_from_config(conf, 'sqlalchemy.db1.')
 init_model(engine)
 sa = meta.Session
 class Command(object):
     def __init__(self, cwd):
         self.cwd = cwd
     def execute(self, cmd, *args):
         """Runs command on the system with given ``args``.
         """
         command = cmd + ' ' + ' '.join(args)
         log.debug('Executing %s' % command)
@@ @@ -87,100 +91,87 @@ def test_wrapp(func): @@
         print '>>>%s' % func.__name__
         try:
             res = func(*args, **kwargs)
         except Exception, e:
             print ('###############\n-'
                    '--%s failed %s--\n'
                    '###############\n' % (func.__name__, e))
             sys.exit()
         print '++OK++'
         return res
     return __wrapp
 def get_session():
     engine = engine_from_config(conf, 'sqlalchemy.db1.')
     init_model(engine)
     sa = meta.Session
     return sa
 def create_test_user(force=True):
     print '\tcreating test user'
     sa = get_session()
-    user = sa.query(User).filter(User.username == USER).scalar()
+    user = User.get_by_username(USER)
     if force and user is not None:
         print '\tremoving current user'
-        for repo in sa.query(Repository).filter(Repository.user == user).all():
+        for repo in Repository.query().filter(Repository.user == user).all():
             sa.delete(repo)
         sa.delete(user)
         sa.commit()
     if user is None or force:
         print '\tcreating new one'
         new_usr = User()
         new_usr.username = USER
         new_usr.password = get_crypt_password(PASS)
         new_usr.email = 'mail@mail.com'
         new_usr.name = 'test'
         new_usr.lastname = 'lasttestname'
         new_usr.active = True
         new_usr.admin = True
         sa.add(new_usr)
         sa.commit()
     print '\tdone'
 def create_test_repo(force=True):
     from rhodecode.model.repo import RepoModel
     sa = get_session()
-    user = sa.query(User).filter(User.username == USER).scalar()
+    user = User.get_by_username(USER)
     if user is None:
         raise Exception('user not found')
     repo = sa.query(Repository).filter(Repository.repo_name == HG_REPO).scalar()
     if repo is None:
         print '\trepo not found creating'
         form_data = {'repo_name':HG_REPO,
                      'repo_type':'hg',
                      'private':False,
                      'clone_uri':'' }
         rm = RepoModel(sa)
         rm.base_path = '/home/hg'
         rm.create(form_data, user)
 def set_anonymous_access(enable=True):
     sa = get_session()
     user = sa.query(User).filter(User.username == 'default').one()
     sa.expire(user)
     user = User.get_by_username('default')
     user.active = enable
     sa.add(user)
     sa.commit()
     sa.remove()
     import time;time.sleep(3)
     print '\tanonymous access is now:', enable
     if enable != User.get_by_username('default').active:
         raise Exception('Cannot set anonymous access')
 def get_anonymous_access():
     sa = get_session()
     obj1 = sa.query(User).filter(User.username == 'default').one()
     sa.expire(obj1)
     return obj1.active
     user = User.get_by_username('default')
     return user.active
 #==============================================================================
 # TESTS
 #==============================================================================
 @test_wrapp
 def test_clone_with_credentials(no_errors=False):
     cwd = path = jn(TESTS_TMP_PATH, HG_REPO)
     try:
         shutil.rmtree(path, ignore_errors=True)
         os.makedirs(path)
@@ @@ -369,44 +360,42 @@ def test_push_wrong_path(): @@
                   {'user':USER,
                    'pass':PASS,
                    'host':HOST,
                    'cloned_repo':HG_REPO + '_error',
                    'dest':jn(TESTS_TMP_PATH, HG_REPO)}
     stdout, stderr = Command(cwd).execute('hg push %s' % clone_url)
     if not """abort: HTTP Error 403: Forbidden"""  in stderr:
         raise Exception('Failure')
 @test_wrapp
 def get_logs():
     sa = get_session()
     return len(sa.query(UserLog).all())
     return UserLog.query().all()
 @test_wrapp
 def test_logs(initial):
     sa = get_session()
     logs = sa.query(UserLog).all()
     operations = 7
     if initial + operations != len(logs):
         raise Exception("missing number of logs %s vs %s" % (initial, len(logs)))
     logs = UserLog.query().all()
     operations = 4
     if len(initial) + operations != len(logs):
         raise Exception("missing number of logs initial:%s vs current:%s" % \
                             (len(initial), len(logs)))
 if __name__ == '__main__':
     create_test_user(force=False)
     create_test_repo()
     initial_logs = get_logs()
     print 'initial activity logs: %s' % len(initial_logs)
-#    test_push_modify_file()
+    #test_push_modify_file()
     test_clone_with_credentials()
     test_clone_wrong_credentials()
     test_push_new_file(commits=2, with_clone=True)
     test_clone_anonymous()
     test_push_wrong_path()
     test_push_wrong_credentials()
     test_logs(initial_logs)

0 comments (0 inline, 0 general)