Mercurial > kallithea
changeset 909:1f0e37c0854d beta
fixed hg operations test script
author | Marcin Kuzminski <marcin@python-works.com> |
---|---|
date | Mon, 03 Jan 2011 00:30:05 +0100 |
parents | de560c47dd03 |
children | 811fa5d45de8 |
files | rhodecode/tests/test_hg_operations.py |
diffstat | 1 files changed, 127 insertions(+), 40 deletions(-) [+] |
line wrap: on
line diff
--- a/rhodecode/tests/test_hg_operations.py Sun Jan 02 20:52:24 2011 +0100 +++ b/rhodecode/tests/test_hg_operations.py Mon Jan 03 00:30:05 2011 +0100 @@ -26,38 +26,98 @@ log = logging.getLogger(__name__) -def __execute_cmd(cmd, *args): - """Runs command on the system with given ``args``. - """ + +class Command(object): + + def __init__(self, cwd): + self.cwd = cwd - command = cmd + ' ' + ' '.join(args) - log.debug('Executing %s' % command) - print command - p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE) - stdout, stderr = p.communicate() - print stdout, stderr - return stdout, stderr + def execute(self, cmd, *args): + """Runs command on the system with given ``args``. + """ + + command = cmd + ' ' + ' '.join(args) + log.debug('Executing %s' % command) + print command + p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE, cwd=self.cwd) + stdout, stderr = p.communicate() + print stdout, stderr + return stdout, stderr #=============================================================================== # TESTS #=============================================================================== def test_clone(): - #rm leftovers + cwd = path = jn(TESTS_TMP_PATH, HG_REPO) + try: - log.debug('removing old directory') - shutil.rmtree(jn(TESTS_TMP_PATH, HG_REPO)) + shutil.rmtree(path, ignore_errors=True) + os.makedirs(path) + #print 'made dirs %s' % jn(path) except OSError: - pass + raise + clone_url = 'http://%(user)s:%(pass)s@%(host)s/%(cloned_repo)s %(dest)s' % \ {'user':USER, 'pass':PASS, 'host':HOST, 'cloned_repo':HG_REPO, - 'dest':jn(TESTS_TMP_PATH, HG_REPO)} + 'dest':path} + + stdout, stderr = Command(cwd).execute('hg clone', clone_url) + + assert """adding file changes""" in stdout, 'no messages about cloning' + assert """abort""" not in stderr , 'got error from clone' + + + +def test_clone_anonymous_ok(): + cwd = path = jn(TESTS_TMP_PATH, HG_REPO) + + try: + shutil.rmtree(path, ignore_errors=True) + os.makedirs(path) + #print 'made dirs %s' % jn(path) + except OSError: + raise + + + clone_url = 'http://%(host)s/%(cloned_repo)s %(dest)s' % \ + {'user':USER, + 'pass':PASS, + 'host':HOST, + 'cloned_repo':HG_REPO, + 'dest':path} - stdout, stderr = __execute_cmd('hg clone', clone_url) + stdout, stderr = Command(cwd).execute('hg clone', clone_url) + print stdout, stderr + assert """adding file changes""" in stdout, 'no messages about cloning' + assert """abort""" not in stderr , 'got error from clone' + +def test_clone_wrong_credentials(): + cwd = path = jn(TESTS_TMP_PATH, HG_REPO) + + 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' % \ + {'user':USER + 'error', + 'pass':PASS, + 'host':HOST, + 'cloned_repo':HG_REPO, + 'dest':path} + + stdout, stderr = Command(cwd).execute('hg clone', clone_url) + + assert """abort: authorization failed""" in stderr , 'no error from wrong credentials' + def test_pull(): pass @@ -67,28 +127,39 @@ modified_file = jn(TESTS_TMP_PATH, HG_REPO, 'setup.py') for i in xrange(5): cmd = """echo 'added_line%s' >> %s""" % (i, modified_file) - __execute_cmd(cmd) + Command(cwd).execute(cmd) cmd = """hg ci -m 'changed file %s' %s """ % (i, modified_file) - __execute_cmd(cmd) + Command(cwd).execute(cmd) - __execute_cmd('hg push %s' % jn(TESTS_TMP_PATH, HG_REPO)) + Command(cwd).execute('hg push %s' % jn(TESTS_TMP_PATH, HG_REPO)) def test_push_new_file(): - added_file = jn(TESTS_TMP_PATH, HG_REPO, 'setup.py') + + test_clone() - __execute_cmd('touch %s' % added_file) + cwd = path = jn(TESTS_TMP_PATH, HG_REPO) + added_file = jn(path, 'setup.py') - __execute_cmd('hg add %s' % added_file) + Command(cwd).execute('touch %s' % added_file) + + Command(cwd).execute('hg add %s' % added_file) for i in xrange(15): cmd = """echo 'added_line%s' >> %s""" % (i, added_file) - __execute_cmd(cmd) + Command(cwd).execute(cmd) cmd = """hg ci -m 'commited new %s' %s """ % (i, added_file) - __execute_cmd(cmd) + Command(cwd).execute(cmd) - __execute_cmd('hg push %s' % jn(TESTS_TMP_PATH, HG_REPO)) + push_url = 'http://%(user)s:%(pass)s@%(host)s/%(cloned_repo)s' % \ + {'user':USER, + 'pass':PASS, + 'host':HOST, + 'cloned_repo':HG_REPO, + 'dest':jn(TESTS_TMP_PATH, HG_REPO)} + + Command(cwd).execute('hg push %s' % push_url) def test_push_wrong_credentials(): @@ -102,37 +173,53 @@ modified_file = jn(TESTS_TMP_PATH, HG_REPO, 'setup.py') for i in xrange(5): cmd = """echo 'added_line%s' >> %s""" % (i, modified_file) - __execute_cmd(cmd) + Command(cwd).execute(cmd) cmd = """hg ci -m 'commited %s' %s """ % (i, modified_file) - __execute_cmd(cmd) + Command(cwd).execute(cmd) - __execute_cmd('hg push %s' % clone_url) + Command(cwd).execute('hg push %s' % clone_url) def test_push_wrong_path(): - added_file = jn(TESTS_TMP_PATH, HG_REPO, 'somefile.py') + cwd = path = jn(TESTS_TMP_PATH, HG_REPO) + added_file = jn(path, 'somefile.py') try: - os.makedirs(jn(TESTS_TMP_PATH, HG_REPO)) + shutil.rmtree(path, ignore_errors=True) + os.makedirs(path) + print 'made dirs %s' % jn(path) except OSError: - pass + raise - __execute_cmd("""echo '' > %s""" % added_file) - - __execute_cmd("""hg add %s""" % added_file) + Command(cwd).execute("""echo '' > %s""" % added_file) + Command(cwd).execute("""hg init %s""" % path) + Command(cwd).execute("""hg add %s""" % added_file) for i in xrange(2): cmd = """echo 'added_line%s' >> %s""" % (i, added_file) - __execute_cmd(cmd) + Command(cwd).execute(cmd) cmd = """hg ci -m 'commited new %s' %s """ % (i, added_file) - __execute_cmd(cmd) + Command(cwd).execute(cmd) - __execute_cmd('hg push %s' % jn(TESTS_TMP_PATH, HG_REPO + '_error')) + clone_url = 'http://%(user)s:%(pass)s@%(host)s/%(cloned_repo)s' % \ + {'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) + assert """abort: HTTP Error 403: Forbidden""" in stderr + if __name__ == '__main__': test_clone() - test_push_wrong_path() - test_push_wrong_credentials() - test_push_new_file() + test_clone_wrong_credentials() + ##test_clone_anonymous_ok() + #test_push_new_file() + #test_push_wrong_path() + #test_push_wrong_credentials() + +