Mercurial > kallithea

changeset 909:1f0e37c0854d beta

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
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()
+
+