kallithea Changeset - a404060706c0

Changeset - a404060706c0

Parent rev.

Child rev.

[Not reviewed]

beta

0 1 0

Marcin Kuzminski - 14 years ago 2011-11-11 17:46:39
marcin@python-works.com

test concurency script updates
can test clone or pull now

1 file changed with 24 insertions and 7 deletions:

rhodecode/tests/_test_concurency.py

0 comments (0 inline, 0 general)

rhodecode/tests/_test_concurency.py

➞

Show inline comments

@@ @@ -33,48 +33,49 @@ from tempfile import _RandomNameSequence @@
 from subprocess import Popen, PIPE
 from paste.deploy import appconfig
 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
 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)
 load_environment(conf.global_conf, conf.local_conf)
 add_cache(conf)
 USER = 'test_admin'
 PASS = 'test12'
 HOST = '127.0.0.1:5000'
 METHOD = 'pull'
 DEBUG = True
 log = logging.getLogger(__name__)
 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)
         if DEBUG:
             print command
         p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE, cwd=self.cwd)
         stdout, stderr = p.communicate()
         if DEBUG:
             print stdout, stderr
         return stdout, stderr
 def get_session():
@@ @@ -132,58 +133,74 @@ def create_test_repo(force=True): @@
                      'repo_type':'hg',
                      'private':False,
                      'clone_uri':'' }
         rm = RepoModel(sa)
         rm.base_path = '/home/hg'
         rm.create(form_data, user)
     print 'done'
 def set_anonymous_access(enable=True):
     sa = get_session()
     user = sa.query(User).filter(User.username == 'default').one()
     user.active = enable
     sa.add(user)
     sa.commit()
 def get_anonymous_access():
     sa = get_session()
     return sa.query(User).filter(User.username == 'default').one().active
 #==============================================================================
 # TESTS
 #==============================================================================
 def test_clone_with_credentials(no_errors=False, repo=HG_REPO):
 def test_clone_with_credentials(no_errors=False, repo=HG_REPO, method=METHOD,
                                 seq=None):
     cwd = path = jn(TESTS_TMP_PATH, repo)
     if seq == None:
         seq = _RandomNameSequence().next()
     try:
         shutil.rmtree(path, ignore_errors=True)
         os.makedirs(path)
         #print 'made dirs %s' % jn(path)
     except OSError:
         raise
-    clone_url = 'http://%(user)s:%(pass)s@%(host)s/%(cloned_repo)s %(dest)s' % \
     clone_url = 'http://%(user)s:%(pass)s@%(host)s/%(cloned_repo)s' % \
                   {'user':USER,
                    'pass':PASS,
                    'host':HOST,
                    'cloned_repo':repo,
                    'dest':path + _RandomNameSequence().next()}
                    'cloned_repo':repo, }
     stdout, stderr = Command(cwd).execute('hg clone', clone_url)
     dest = path + seq
     if method == 'pull':
         stdout, stderr = Command(cwd).execute('hg', method, '--cwd', dest, clone_url)
     else:
         stdout, stderr = Command(cwd).execute('hg', method, clone_url, dest)
     if no_errors is False:
         assert """adding file changes""" in stdout, 'no messages about cloning'
         assert """abort""" not in stderr , 'got error from clone'
 if __name__ == '__main__':
     try:
         create_test_user(force=False)
         seq = None
         import time
         if METHOD == 'pull':
             seq = _RandomNameSequence().next()
             test_clone_with_credentials(repo=sys.argv[1], method='clone',
                                         seq=seq)
         s = time.time()
         for i in range(int(sys.argv[2])):
             test_clone_with_credentials(repo=sys.argv[1])
             print 'take', i
             test_clone_with_credentials(repo=sys.argv[1], method=METHOD,
                                         seq=seq)
         print 'time taken %.3f' % (time.time() - s)
     except Exception, e:
         raise
         sys.exit('stop on %s' % e)

0 comments (0 inline, 0 general)