kallithea Changeset - a69bcf8b28b3

Changeset - a69bcf8b28b3

Parent rev.

Child rev.

[Not reviewed]

default

0 1 0

domruf - 9 years ago 2016-06-15 18:33:27
dominikruf@gmail.com

tests: use forks when modifying repositories in vcs tests

Otherwise other tests (primarily checking the size of the repo) would fail
because they expect the repos to be unmodified.

1 file changed with 38 insertions and 20 deletions:

kallithea/tests/other/manual_test_vcs_operations.py

0 comments (0 inline, 0 general)

kallithea/tests/other/manual_test_vcs_operations.py

➞

Show inline comments

@@ @@ -21,56 +21,59 @@ Run it in two terminals:: @@
  paster serve kallithea/tests/test.ini
  KALLITHEA_WHOOSH_TEST_DISABLE=1 KALLITHEA_NO_TMP_PATH=1 py.test kallithea/tests/other/manual_test_vcs_operations.py
 You must have git > 1.8.1 for tests to work fine
 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: Dec 30, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import re
 import tempfile
 import time
 from os.path import join as jn
 from tempfile import _RandomNameSequence
 from subprocess import Popen, PIPE
 from kallithea.tests import *
 from kallithea.tests.fixture import Fixture
 from kallithea.model.db import User, Repository, UserIpMap, CacheInvalidation
 from kallithea.model.meta import Session
 from kallithea.model.repo import RepoModel
 from kallithea.model.user import UserModel
 DEBUG = True
 HOST = '127.0.0.1:4999'  # test host
 fixture = Fixture()
 class Command(object):
     def __init__(self, cwd):
         self.cwd = cwd
     def execute(self, cmd, *args, **environ):
         """
         Runs command on the system with given ``args``.
         """
         command = cmd + ' ' + ' '.join(args)
         if DEBUG:
             print '*** CMD %s ***' % command
         testenv = dict(os.environ)
         testenv['LANG'] = 'en_US.UTF-8'
         testenv['LANGUAGE'] = 'en_US:en'
         testenv.update(environ)
         p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE, cwd=self.cwd, env=testenv)
         stdout, stderr = p.communicate()
         if DEBUG:
             if stdout:
                 print 'stdout:', repr(stdout)
             if stderr:
@@ @@ -230,105 +233,116 @@ class TestVCSOperations(TestController): @@
         clone_url = _construct_url(GIT_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         assert 'HTTP Error 404: Not Found' in stderr
     def test_clone_hg_repo_as_git(self):
         clone_url = _construct_url(HG_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         assert 'not found' in stderr
     def test_clone_non_existing_path_hg(self):
         clone_url = _construct_url('trololo')
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         assert 'HTTP Error 404: Not Found' in stderr
     def test_clone_non_existing_path_git(self):
         clone_url = _construct_url('trololo')
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         assert 'not found' in stderr
     def test_push_new_file_hg(self):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(HG_REPO, dest=DEST)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         stdout, stderr = _add_files_and_push('hg', DEST)
         fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())
         fixture.create_fork(HG_REPO, fork_name)
         clone_url = _construct_url(fork_name).split()[0]
         stdout, stderr = _add_files_and_push('hg', DEST, clone_url=clone_url)
         assert 'pushing to' in stdout
         assert 'Repository size' in stdout
         assert 'Last revision is now' in stdout
     def test_push_new_file_git(self):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(GIT_REPO, dest=DEST)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         # commit some stuff into this repo
         stdout, stderr = _add_files_and_push('git', DEST)
         print [(x.repo_full_path,x.repo_path) for x in Repository.get_all()]
         fork_name = '%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())
         fixture.create_fork(GIT_REPO, fork_name)
         clone_url = _construct_url(fork_name).split()[0]
         stdout, stderr = _add_files_and_push('git', DEST, clone_url=clone_url)
         print [(x.repo_full_path,x.repo_path) for x in Repository.get_all()] # TODO: what is this for
         _check_proper_git_push(stdout, stderr)
     def test_push_invalidates_cache_hg(self):
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                ==HG_REPO).scalar()
         if not key:
             key = CacheInvalidation(HG_REPO, HG_REPO)
         key.cache_active = True
         Session().add(key)
         Session().commit()
         DEST = _get_tmp_dir()
         clone_url = _construct_url(HG_REPO, dest=DEST)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         stdout, stderr = _add_files_and_push('hg', DEST, files_no=1)
         fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())
         fixture.create_fork(HG_REPO, fork_name)
         clone_url = _construct_url(fork_name).split()[0]
         stdout, stderr = _add_files_and_push('hg', DEST, files_no=1, clone_url=clone_url)
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
-                                               ==HG_REPO).all()
+                                               ==fork_name).all()
         assert key == []
     def test_push_invalidates_cache_git(self):
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                ==GIT_REPO).scalar()
         if not key:
             key = CacheInvalidation(GIT_REPO, GIT_REPO)
         key.cache_active = True
         Session().add(key)
         Session().commit()
         DEST = _get_tmp_dir()
         clone_url = _construct_url(GIT_REPO, dest=DEST)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         # commit some stuff into this repo
         stdout, stderr = _add_files_and_push('git', DEST, files_no=1)
         fork_name = '%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())
         fixture.create_fork(GIT_REPO, fork_name)
         clone_url = _construct_url(fork_name).split()[0]
         stdout, stderr = _add_files_and_push('git', DEST, files_no=1, clone_url=clone_url)
         _check_proper_git_push(stdout, stderr)
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
-                                               ==GIT_REPO).all()
+                                               ==fork_name).all()
         assert key == []
     def test_push_wrong_credentials_hg(self):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(HG_REPO, dest=DEST)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         stdout, stderr = _add_files_and_push('hg', DEST, user='bad',
                                              passwd='name')
         assert 'abort: authorization failed' in stderr
     def test_push_wrong_credentials_git(self):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(GIT_REPO, dest=DEST)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         stdout, stderr = _add_files_and_push('git', DEST, user='bad',
                                              passwd='name')
         assert 'fatal: Authentication failed' in stderr
     def test_push_back_to_wrong_url_hg(self):
         DEST = _get_tmp_dir()
@@ @@ -431,94 +445,98 @@ class TestVCSOperations(TestController): @@
         r = Repository.get_by_repo_name(GIT_REPO)
         # let this user actually push !
         RepoModel().grant_user_permission(repo=r, user=TEST_USER_REGULAR_LOGIN,
                                           perm='repository.write')
         Session().commit()
         Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)
         #push fails repo is locked by other user !
         stdout, stderr = _add_files_and_push('git', DEST,
                                              user=TEST_USER_REGULAR_LOGIN,
                                              passwd=TEST_USER_REGULAR_PASS)
         err = 'Repository `%s` locked by user `%s`' % (GIT_REPO, TEST_USER_ADMIN_LOGIN)
         assert err in stderr
         #TODO: fix this somehow later on Git, Git is stupid and even if we throw
         #back 423 to it, it makes ANOTHER request and we fail there with 405 :/
         msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""
                 % (GIT_REPO, TEST_USER_ADMIN_LOGIN))
         #msg = "405 Method Not Allowed"
         #assert msg in stderr
     def test_push_unlocks_repository_hg(self):
         # enable locking
         r = Repository.get_by_repo_name(HG_REPO)
         fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())
         fixture.create_fork(HG_REPO, fork_name)
         r = Repository.get_by_repo_name(fork_name)
         r.enable_locking = True
         Session().add(r)
         Session().commit()
         #clone some temp
         DEST = _get_tmp_dir()
-        clone_url = _construct_url(HG_REPO, dest=DEST)
+        clone_url = _construct_url(fork_name, dest=DEST)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         #check for lock repo after clone
-        r = Repository.get_by_repo_name(HG_REPO)
+        r = Repository.get_by_repo_name(fork_name)
         uid = User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id
         assert r.locked[0] == uid
         #push is ok and repo is now unlocked
         stdout, stderr = _add_files_and_push('hg', DEST)
         assert ('remote: Released lock on repo `%s`' % HG_REPO) in stdout
         stdout, stderr = _add_files_and_push('hg', DEST, clone_url=clone_url.split()[0])
         assert ('remote: Released lock on repo `%s`' % fork_name) in stdout
         #we need to cleanup the Session Here !
         Session.remove()
-        r = Repository.get_by_repo_name(HG_REPO)
+        r = Repository.get_by_repo_name(fork_name)
         assert r.locked == [None, None]
     #TODO: fix me ! somehow during tests hooks don't get called on Git
     def test_push_unlocks_repository_git(self):
         # enable locking
         r = Repository.get_by_repo_name(GIT_REPO)
         fork_name = '%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())
         fixture.create_fork(GIT_REPO, fork_name)
         r = Repository.get_by_repo_name(fork_name)
         r.enable_locking = True
         Session().add(r)
         Session().commit()
         #clone some temp
         DEST = _get_tmp_dir()
-        clone_url = _construct_url(GIT_REPO, dest=DEST)
+        clone_url = _construct_url(fork_name, dest=DEST)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         #check for lock repo after clone
-        r = Repository.get_by_repo_name(GIT_REPO)
+        r = Repository.get_by_repo_name(fork_name)
         assert r.locked[0] == User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id
         #push is ok and repo is now unlocked
         stdout, stderr = _add_files_and_push('git', DEST)
+        stdout, stderr = _add_files_and_push('git', DEST, clone_url=clone_url.split()[0])
         _check_proper_git_push(stdout, stderr)
-        assert ('remote: Released lock on repo `%s`' % GIT_REPO) in stderr
+        assert ('remote: Released lock on repo `%s`' % fork_name) in stderr
         #we need to cleanup the Session Here !
         Session.remove()
-        r = Repository.get_by_repo_name(GIT_REPO)
+        r = Repository.get_by_repo_name(fork_name)
         assert r.locked == [None, None]
     def test_ip_restriction_hg(self):
         user_model = UserModel()
         try:
             user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')
             Session().commit()
             clone_url = _construct_url(HG_REPO)
             stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
             assert 'abort: HTTP Error 403: Forbidden' in stderr
         finally:
             #release IP restrictions
             for ip in UserIpMap.getAll():
                 UserIpMap.delete(ip.ip_id)
             Session().commit()
         time.sleep(2)
         clone_url = _construct_url(HG_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         assert 'requesting all changes' in stdout
         assert 'adding changesets' in stdout
         assert 'adding manifests' in stdout
         assert 'adding file changes' in stdout

0 comments (0 inline, 0 general)