Changeset - 6304efbe37a3
Parent rev.
Child rev.
[Not reviewed]
default
0 10 0
Mads Kiilerich - 8 years ago 2017-05-30 02:59:45
mads@kiilerich.com
tests: put all temporary files in one top directory

Use KALLITHEA_TESTS_TMP_PATH or a create a temporary /tmp/kallithea-test-*
directory.

Some temporary files might still be placed in /tmp ... but nothing should be
left behind.
10 files changed with 65 insertions and 77 deletions:
kallithea/tests/base.py
3
4
kallithea/tests/functional/test_admin_settings.py
5
7
kallithea/tests/other/test_validators.py
1
2
kallithea/tests/other/test_vcs_operations.py
30
30
kallithea/tests/scripts/manual_test_concurrency.py
4
9
kallithea/tests/vcs/conf.py
8
8
kallithea/tests/vcs/test_archives.py
8
8
kallithea/tests/vcs/test_git.py
2
3
kallithea/tests/vcs/test_hg.py
2
3
kallithea/tests/vcs/test_utils.py
2
3
0 comments (0 inline, 0 general)
kallithea/tests/base.py
➞
Show inline comments
 
@@ -56,7 +56,9 @@ __all__ = [
 

			




	
	
	
		
 
#SOME GLOBALS FOR TESTS

			




	
	
	
		
 

			




	
	
	
		
 
TESTS_TMP_PATH = os.path.join(tempfile.gettempdir(), 'rc_test_%s' % tempfile._RandomNameSequence().next())

			




	
	
	
		
 
TESTS_TMP_PATH = os.environ.get('KALLITHEA_TESTS_TMP_PATH', tempfile.mkdtemp(prefix='kallithea-test-'))

			




	
	
	
		
 
os.environ['VCS_TEST_ROOT'] = TESTS_TMP_PATH

			




	
	
	
		
 

			




	
	
	
		
 
TEST_USER_ADMIN_LOGIN = 'test_admin'

			




	
	
	
		
 
TEST_USER_ADMIN_PASS = 'test12'

			




	
	
	
		
 
TEST_USER_ADMIN_EMAIL = 'test_admin@example.com'

			




	
	
		
 
@@ -99,9 +101,6 @@ TEST_HG_REPO = os.path.join(TESTS_TMP_PA

			




	
	
	
		
 
TEST_HG_REPO_CLONE = os.path.join(TESTS_TMP_PATH, 'vcshgclone%s' % uniq_suffix)

			




	
	
	
		
 
TEST_HG_REPO_PULL = os.path.join(TESTS_TMP_PATH, 'vcshgpull%s' % uniq_suffix)

			




	
	
	
		
 

			




	
	
	
		
 
TEST_DIR = tempfile.gettempdir()

			




	
	
	
		
 
TEST_REPO_PREFIX = 'vcs-test'

			




	
	
	
		
 

			




	
	
	
		
 
# cached repos if any !

			




	
	
	
		
 
# comment out to get some other repos from bb or github

			




	
	
	
		
 
GIT_REMOTE_REPO = os.path.join(TESTS_TMP_PATH, GIT_REPO)

			




        

    


    
    

    

        

                

                    kallithea/tests/functional/test_admin_settings.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
	
		
 
# -*- coding: utf-8 -*-

			




	
	
	
		
 

			




	
	
	
		
 
import tempfile

			




	
	
	
		
 

			




	
	
	
		
 
from kallithea.model.db import Setting, Ui

			




	
	
	
		
 
from kallithea.tests.base import *

			




	
	
	
		
 
from kallithea.tests.fixture import Fixture

			




	
	
		
 
@@ -39,23 +37,23 @@ class TestAdminSettingsController(TestCo

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        response = self.app.post(url('admin_settings_hooks'),

			




	
	
	
		
 
                                params=dict(new_hook_ui_key='test_hooks_1',

			




	
	
	
		
 
                                            new_hook_ui_value='cd %s' % tempfile.gettempdir(),

			




	
	
	
		
 
                                            new_hook_ui_value='cd %s' % TESTS_TMP_PATH,

			




	
	
	
		
 
                                            _authentication_token=self.authentication_token()))

			




	
	
	
		
 

			




	
	
	
		
 
        response = response.follow()

			




	
	
	
		
 
        response.mustcontain('test_hooks_1')

			




	
	
	
		
 
        response.mustcontain('cd %s' % tempfile.gettempdir())

			




	
	
	
		
 
        response.mustcontain('cd %s' % TESTS_TMP_PATH)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_create_custom_hook_delete(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        response = self.app.post(url('admin_settings_hooks'),

			




	
	
	
		
 
                                params=dict(new_hook_ui_key='test_hooks_2',

			




	
	
	
		
 
                                            new_hook_ui_value='cd %s2' % tempfile.gettempdir(),

			




	
	
	
		
 
                                            new_hook_ui_value='cd %s2' % TESTS_TMP_PATH,

			




	
	
	
		
 
                                            _authentication_token=self.authentication_token()))

			




	
	
	
		
 

			




	
	
	
		
 
        response = response.follow()

			




	
	
	
		
 
        response.mustcontain('test_hooks_2')

			




	
	
	
		
 
        response.mustcontain('cd %s2' % tempfile.gettempdir())

			




	
	
	
		
 
        response.mustcontain('cd %s2' % TESTS_TMP_PATH)

			




	
	
	
		
 

			




	
	
	
		
 
        hook_id = Ui.get_by_key('hooks', 'test_hooks_2').ui_id

			




	
	
	
		
 
        ## delete

			




	
	
		
 
@@ -63,7 +61,7 @@ class TestAdminSettingsController(TestCo

			




	
	
	
		
 
                        params=dict(hook_id=hook_id, _authentication_token=self.authentication_token()))

			




	
	
	
		
 
        response = self.app.get(url('admin_settings_hooks'))

			




	
	
	
		
 
        response.mustcontain(no=['test_hooks_2'])

			




	
	
	
		
 
        response.mustcontain(no=['cd %s2' % tempfile.gettempdir()])

			




	
	
	
		
 
        response.mustcontain(no=['cd %s2' % TESTS_TMP_PATH])

			




	
	
	
		
 

			




	
	
	
		
 
    def test_index_search(self):

			




	
	
	
		
 
        self.log_user()

			




        

    


    
    

    

        

                

                    kallithea/tests/other/test_validators.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
	
		
 
# -*- coding: utf-8 -*-

			




	
	
	
		
 
import formencode

			




	
	
	
		
 
import pytest

			




	
	
	
		
 
import tempfile

			




	
	
	
		
 

			




	
	
	
		
 
from kallithea.tests.base import *

			




	
	
	
		
 

			




	
	
		
 
@@ -205,7 +204,7 @@ class TestRepoGroups(TestController):

			




	
	
	
		
 

			




	
	
	
		
 
    def test_ValidPath(self):

			




	
	
	
		
 
            validator = v.ValidPath()

			




	
	
	
		
 
            assert tempfile.gettempdir() == validator.to_python(tempfile.gettempdir())

			




	
	
	
		
 
            assert TESTS_TMP_PATH == validator.to_python(TESTS_TMP_PATH)

			




	
	
	
		
 
            with pytest.raises(formencode.Invalid):

			




	
	
	
		
 
                validator.to_python('/no_such_dir')

			




	
	
	
		
 

			




        

    


    
    

    

        

                

                    kallithea/tests/other/test_vcs_operations.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -77,8 +77,8 @@ class Command(object):

			




	
	
	
		
 
        return stdout, stderr

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def _get_tmp_dir():

			




	
	
	
		
 
    return tempfile.mkdtemp(prefix='rc_integration_test')

			




	
	
	
		
 
def _get_tmp_dir(prefix='vcs_operations-', suffix=''):

			




	
	
	
		
 
    return tempfile.mkdtemp(dir=TESTS_TMP_PATH, prefix=prefix, suffix=suffix)

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def _add_files_and_push(webserver, vcs, DEST, ignoreReturnCode=False, files_no=3,

			




	
	
		
 
@@ -178,7 +178,7 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_hg_repo_by_admin(self, webserver):

			




	
	
	
		
 
        clone_url = webserver.repo_url(HG_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir())

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir())

			




	
	
	
		
 

			




	
	
	
		
 
        assert 'requesting all changes' in stdout

			




	
	
	
		
 
        assert 'adding changesets' in stdout

			




	
	
		
 
@@ -189,45 +189,45 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_git_repo_by_admin(self, webserver):

			




	
	
	
		
 
        clone_url = webserver.repo_url(GIT_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir())

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir())

			




	
	
	
		
 

			




	
	
	
		
 
        assert 'Cloning into' in stdout + stderr

			




	
	
	
		
 
        assert stderr == '' or stdout == ''

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_wrong_credentials_hg(self, webserver):

			




	
	
	
		
 
        clone_url = webserver.repo_url(HG_REPO, password='bad!')

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        assert 'abort: authorization failed' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_wrong_credentials_git(self, webserver):

			




	
	
	
		
 
        clone_url = webserver.repo_url(GIT_REPO, password='bad!')

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        assert 'fatal: Authentication failed' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_git_dir_as_hg(self, webserver):

			




	
	
	
		
 
        clone_url = webserver.repo_url(GIT_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        assert 'HTTP Error 404: Not Found' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_hg_repo_as_git(self, webserver):

			




	
	
	
		
 
        clone_url = webserver.repo_url(HG_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        assert 'not found' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_non_existing_path_hg(self, webserver):

			




	
	
	
		
 
        clone_url = webserver.repo_url('trololo')

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        assert 'HTTP Error 404: Not Found' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_non_existing_path_git(self, webserver):

			




	
	
	
		
 
        clone_url = webserver.repo_url('trololo')

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        assert 'not found' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_push_new_file_hg(self, webserver):

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(HG_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, DEST)

			




	
	
	
		
 

			




	
	
	
		
 
        fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())

			




	
	
	
		
 
        fixture.create_fork(HG_REPO, fork_name)

			




	
	
		
 
@@ -241,7 +241,7 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
    def test_push_new_file_git(self, webserver):

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(GIT_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, DEST)

			




	
	
	
		
 

			




	
	
	
		
 
        # commit some stuff into this repo

			




	
	
	
		
 
        fork_name = '%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())

			




	
	
		
 
@@ -263,7 +263,7 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(HG_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, DEST)

			




	
	
	
		
 

			




	
	
	
		
 
        fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())

			




	
	
	
		
 
        fixture.create_fork(HG_REPO, fork_name)

			




	
	
		
 
@@ -286,7 +286,7 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(GIT_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, DEST)

			




	
	
	
		
 

			




	
	
	
		
 
        # commit some stuff into this repo

			




	
	
	
		
 
        fork_name = '%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())

			




	
	
		
 
@@ -302,7 +302,7 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
    def test_push_wrong_credentials_hg(self, webserver):

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(HG_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, DEST)

			




	
	
	
		
 

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push(webserver, 'hg', DEST, username='bad',

			




	
	
	
		
 
                                             password='name', ignoreReturnCode=True)

			




	
	
		
 
@@ -312,7 +312,7 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
    def test_push_wrong_credentials_git(self, webserver):

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(GIT_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, DEST)

			




	
	
	
		
 

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push(webserver, 'git', DEST, username='bad',

			




	
	
	
		
 
                                             password='name', ignoreReturnCode=True)

			




	
	
		
 
@@ -322,7 +322,7 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
    def test_push_back_to_wrong_url_hg(self, webserver):

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(HG_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, DEST)

			




	
	
	
		
 

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push(webserver, 'hg', DEST,

			




	
	
	
		
 
                                    clone_url='http://%s:%s/tmp' % (webserver.server_address[0], webserver.server_address[1]),

			




	
	
		
 
@@ -333,7 +333,7 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
    def test_push_back_to_wrong_url_git(self, webserver):

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(GIT_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, DEST)

			




	
	
	
		
 

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push(webserver, 'git', DEST,

			




	
	
	
		
 
                                    clone_url='http://%s:%s/tmp' % (webserver.server_address[0], webserver.server_address[1]),

			




	
	
		
 
@@ -348,7 +348,7 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 
        # clone

			




	
	
	
		
 
        clone_url = webserver.repo_url(HG_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir())

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir())

			




	
	
	
		
 

			




	
	
	
		
 
        #check if lock was made

			




	
	
	
		
 
        r = Repository.get_by_repo_name(HG_REPO)

			




	
	
		
 
@@ -361,7 +361,7 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 
        # clone

			




	
	
	
		
 
        clone_url = webserver.repo_url(GIT_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir())

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir())

			




	
	
	
		
 

			




	
	
	
		
 
        #check if lock was made

			




	
	
	
		
 
        r = Repository.get_by_repo_name(GIT_REPO)

			




	
	
		
 
@@ -373,7 +373,7 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
        Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)

			




	
	
	
		
 
        #pull fails since repo is locked

			




	
	
	
		
 
        clone_url = webserver.repo_url(HG_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""

			




	
	
	
		
 
                % (HG_REPO, TEST_USER_ADMIN_LOGIN))

			




	
	
	
		
 
        assert msg in stderr

			




	
	
		
 
@@ -384,7 +384,7 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
        Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)

			




	
	
	
		
 
        #pull fails since repo is locked

			




	
	
	
		
 
        clone_url = webserver.repo_url(GIT_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        msg = ("""The requested URL returned error: 423""")

			




	
	
	
		
 
        assert msg in stderr

			




	
	
	
		
 

			




	
	
		
 
@@ -392,7 +392,7 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
        #clone some temp

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(HG_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, DEST)

			




	
	
	
		
 

			




	
	
	
		
 
        #lock repo

			




	
	
	
		
 
        r = Repository.get_by_repo_name(HG_REPO)

			




	
	
		
 
@@ -418,7 +418,7 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
        #clone some temp

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(GIT_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, DEST)

			




	
	
	
		
 

			




	
	
	
		
 
        #lock repo

			




	
	
	
		
 
        r = Repository.get_by_repo_name(GIT_REPO)

			




	
	
		
 
@@ -454,7 +454,7 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
        #clone some temp

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(fork_name)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, DEST)

			




	
	
	
		
 

			




	
	
	
		
 
        #check for lock repo after clone

			




	
	
	
		
 
        r = Repository.get_by_repo_name(fork_name)

			




	
	
		
 
@@ -480,7 +480,7 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
        #clone some temp

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(fork_name)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, DEST)

			




	
	
	
		
 

			




	
	
	
		
 
        #check for lock repo after clone

			




	
	
	
		
 
        r = Repository.get_by_repo_name(fork_name)

			




	
	
		
 
@@ -502,7 +502,7 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
            user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 
            clone_url = webserver.repo_url(HG_REPO)

			




	
	
	
		
 
            stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
            stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
            assert 'abort: HTTP Error 403: Forbidden' in stderr

			




	
	
	
		
 
        finally:

			




	
	
	
		
 
            #release IP restrictions

			




	
	
		
 
@@ -514,7 +514,7 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
        time.sleep(1.5)

			




	
	
	
		
 

			




	
	
	
		
 
        clone_url = webserver.repo_url(HG_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir())

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir())

			




	
	
	
		
 

			




	
	
	
		
 
        assert 'requesting all changes' in stdout

			




	
	
	
		
 
        assert 'adding changesets' in stdout

			




	
	
		
 
@@ -529,7 +529,7 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
            user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 
            clone_url = webserver.repo_url(GIT_REPO)

			




	
	
	
		
 
            stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
            stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
            # The message apparently changed in Git 1.8.3, so match it loosely.

			




	
	
	
		
 
            assert re.search(r'\b403\b', stderr)

			




	
	
	
		
 
        finally:

			




	
	
		
 
@@ -542,7 +542,7 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
        time.sleep(1.5)

			




	
	
	
		
 

			




	
	
	
		
 
        clone_url = webserver.repo_url(GIT_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir())

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir())

			




	
	
	
		
 

			




	
	
	
		
 
        assert 'Cloning into' in stdout + stderr

			




	
	
	
		
 
        assert stderr == '' or stdout == ''

			




        

    


    
    

    

        

                

                    kallithea/tests/scripts/manual_test_concurrency.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -159,12 +159,9 @@ def get_anonymous_access():

			




	
	
	
		
 
# TESTS

			




	
	
	
		
 
#==============================================================================

			




	
	
	
		
 
def test_clone_with_credentials(no_errors=False, repo=HG_REPO, method=METHOD,

			




	
	
	
		
 
                                seq=None, backend='hg'):

			




	
	
	
		
 
                                backend='hg'):

			




	
	
	
		
 
    cwd = path = os.path.join(Ui.get_by_key('paths', '/').ui_value, repo)

			




	
	
	
		
 

			




	
	
	
		
 
    if seq is None:

			




	
	
	
		
 
        seq = _RandomNameSequence().next()

			




	
	
	
		
 

			




	
	
	
		
 
    try:

			




	
	
	
		
 
        shutil.rmtree(path, ignore_errors=True)

			




	
	
	
		
 
        os.makedirs(path)

			




	
	
		
 
@@ -178,12 +175,11 @@ def test_clone_with_credentials(no_error

			




	
	
	
		
 
                   'host': HOST,

			




	
	
	
		
 
                   'cloned_repo': repo, }

			




	
	
	
		
 

			




	
	
	
		
 
    dest = path + seq

			




	
	
	
		
 
    dest = tempfile.mktemp(dir=path, prefix='dest-')

			




	
	
	
		
 
    if method == 'pull':

			




	
	
	
		
 
        stdout, stderr = Command(cwd).execute(backend, method, '--cwd', dest, clone_url)

			




	
	
	
		
 
    else:

			




	
	
	
		
 
        stdout, stderr = Command(cwd).execute(backend, method, clone_url, dest)

			




	
	
	
		
 
        print stdout,'sdasdsadsa'

			




	
	
	
		
 
        if not no_errors:

			




	
	
	
		
 
            if backend == 'hg':

			




	
	
	
		
 
                assert """adding file changes""" in stdout, 'no messages about cloning'

			




	
	
		
 
@@ -194,7 +190,6 @@ def test_clone_with_credentials(no_error

			




	
	
	
		
 
if __name__ == '__main__':

			




	
	
	
		
 
    try:

			




	
	
	
		
 
        create_test_user(force=False)

			




	
	
	
		
 
        seq = None

			




	
	
	
		
 
        import time

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
		
 
@@ -210,12 +205,12 @@ if __name__ == '__main__':

			




	
	
	
		
 
        if METHOD == 'pull':

			




	
	
	
		
 
            seq = _RandomNameSequence().next()

			




	
	
	
		
 
            test_clone_with_credentials(repo=sys.argv[1], method='clone',

			




	
	
	
		
 
                                        seq=seq, backend=backend)

			




	
	
	
		
 
                                        backend=backend)

			




	
	
	
		
 
        s = time.time()

			




	
	
	
		
 
        for i in range(1, int(sys.argv[2]) + 1):

			




	
	
	
		
 
            print 'take', i

			




	
	
	
		
 
            test_clone_with_credentials(repo=sys.argv[1], method=METHOD,

			




	
	
	
		
 
                                        seq=seq, backend=backend)

			




	
	
	
		
 
                                        backend=backend)

			




	
	
	
		
 
        print 'time taken %.3f' % (time.time() - s)

			




	
	
	
		
 
    except Exception as e:

			




	
	
	
		
 
        sys.exit('stop on %s' % e)

			




        

    


    
    

    

        

                

                    kallithea/tests/vcs/conf.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -15,35 +15,35 @@ __all__ = (

			




	
	
	
		
 
)

			




	
	
	
		
 

			




	
	
	
		
 
SCM_TESTS = ['hg', 'git']

			




	
	
	
		
 
uniq_suffix = str(int(time.mktime(datetime.datetime.now().timetuple())))

			




	
	
	
		
 

			




	
	
	
		
 
THIS = os.path.abspath(os.path.dirname(__file__))

			




	
	
	
		
 

			




	
	
	
		
 
GIT_REMOTE_REPO = 'git://github.com/codeinn/vcs.git'

			




	
	
	
		
 

			




	
	
	
		
 
# Note: TEST_TMP_PATH, not TESTS_TMP_PATH

			




	
	
	
		
 
TEST_TMP_PATH = os.environ.get('VCS_TEST_ROOT', tempfile.gettempdir())

			




	
	
	
		
 

			




	
	
	
		
 
TEST_GIT_REPO = os.environ.get('VCS_TEST_GIT_REPO',

			




	
	
	
		
 
                               os.path.join(TEST_TMP_PATH, 'vcs-git'))

			




	
	
	
		
 
TEST_GIT_REPO_CLONE = os.environ.get('VCS_TEST_GIT_REPO_CLONE',

			




	
	
	
		
 
                                     os.path.join(TEST_TMP_PATH, 'vcsgitclone%s' % uniq_suffix))

			




	
	
	
		
 
                                     os.path.join(TEST_TMP_PATH, 'vcs-git-clone'))

			




	
	
	
		
 
TEST_GIT_REPO_PULL = os.environ.get('VCS_TEST_GIT_REPO_PULL',

			




	
	
	
		
 
                                    os.path.join(TEST_TMP_PATH, 'vcsgitpull%s' % uniq_suffix))

			




	
	
	
		
 
                                    os.path.join(TEST_TMP_PATH, 'vcs-git-pull'))

			




	
	
	
		
 

			




	
	
	
		
 
HG_REMOTE_REPO = 'http://bitbucket.org/marcinkuzminski/vcs'

			




	
	
	
		
 
TEST_HG_REPO = os.environ.get('VCS_TEST_HG_REPO',

			




	
	
	
		
 
                              os.path.join(TEST_TMP_PATH, 'vcs-hg'))

			




	
	
	
		
 
TEST_HG_REPO_CLONE = os.environ.get('VCS_TEST_HG_REPO_CLONE',

			




	
	
	
		
 
                                    os.path.join(TEST_TMP_PATH, 'vcshgclone%s' % uniq_suffix))

			




	
	
	
		
 
                                    os.path.join(TEST_TMP_PATH, 'vcs-hg-clone'))

			




	
	
	
		
 
TEST_HG_REPO_PULL = os.environ.get('VCS_TEST_HG_REPO_PULL',

			




	
	
	
		
 
                                   os.path.join(TEST_TMP_PATH, 'vcshgpull%s' % uniq_suffix))

			




	
	
	
		
 
                                   os.path.join(TEST_TMP_PATH, 'vcs-hg-pull'))

			




	
	
	
		
 

			




	
	
	
		
 
TEST_DIR = os.environ.get('VCS_TEST_ROOT', tempfile.gettempdir())

			




	
	
	
		
 
TEST_REPO_PREFIX = 'vcs-test'

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def get_new_dir(title=None):

			




	
	
	
		
 
    """

			




	
	
	
		
 
    Calculates a path for a new, non-existant, unique sub-directory in TEST_DIR.

			




	
	
	
		
 
    Calculates a path for a new, non-existant, unique sub-directory in TEST_TMP_PATH.

			




	
	
	
		
 

			




	
	
	
		
 
    Resulting directory name will have format:

			




	
	
	
		
 

			




	
	
		
 
@@ -67,7 +67,7 @@ def get_new_dir(title=None):

			




	
	
	
		
 
    else:

			




	
	
	
		
 
        name = TEST_REPO_PREFIX

			




	
	
	
		
 

			




	
	
	
		
 
    path = os.path.join(TEST_DIR, name)

			




	
	
	
		
 
    path = os.path.join(TEST_TMP_PATH, name)

			




	
	
	
		
 

			




	
	
	
		
 
    # Generate new hexes until we get a unique name (just in case).

			




	
	
	
		
 
    hex_uuid = uuid.uuid4().hex

			




        

    


    
    

    

        

                

                    kallithea/tests/vcs/test_archives.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
	
		
 

			




	
	
	
		
 
import os

			




	
	
	
		
 
import tarfile

			




	
	
	
		
 
import zipfile

			




	
	
	
		
 
import datetime

			




	
	
	
		
 
import tempfile

			




	
	
	
		
 
import StringIO

			




	
	
	
		
 

			




	
	
	
		
 
from kallithea.tests.vcs.base import _BackendTestMixin

			




	
	
	
		
 
from kallithea.tests.vcs.conf import SCM_TESTS

			




	
	
	
		
 
from kallithea.tests.vcs.conf import SCM_TESTS, TEST_TMP_PATH

			




	
	
	
		
 
from kallithea.lib.vcs.exceptions import VCSError

			




	
	
	
		
 
from kallithea.lib.vcs.nodes import FileNode

			




	
	
	
		
 
from kallithea.lib.vcs.utils.compat import unittest

			




	
	
		
 
@@ -29,7 +29,7 @@ class ArchivesTestCaseMixin(_BackendTest

			




	
	
	
		
 
            }

			




	
	
	
		
 

			




	
	
	
		
 
    def test_archive_zip(self):

			




	
	
	
		
 
        path = tempfile.mkstemp()[1]

			




	
	
	
		
 
        path = tempfile.mkstemp(dir=TEST_TMP_PATH, prefix='test_archive_zip-')[1]

			




	
	
	
		
 
        with open(path, 'wb') as f:

			




	
	
	
		
 
            self.tip.fill_archive(stream=f, kind='zip', prefix='repo')

			




	
	
	
		
 
        out = zipfile.ZipFile(path)

			




	
	
		
 
@@ -43,10 +43,10 @@ class ArchivesTestCaseMixin(_BackendTest

			




	
	
	
		
 
                self.tip.get_node(node_path).content)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_archive_tgz(self):

			




	
	
	
		
 
        path = tempfile.mkstemp()[1]

			




	
	
	
		
 
        path = tempfile.mkstemp(dir=TEST_TMP_PATH, prefix='test_archive_tgz-')[1]

			




	
	
	
		
 
        with open(path, 'wb') as f:

			




	
	
	
		
 
            self.tip.fill_archive(stream=f, kind='tgz', prefix='repo')

			




	
	
	
		
 
        outdir = tempfile.mkdtemp()

			




	
	
	
		
 
        outdir = tempfile.mkdtemp(dir=TEST_TMP_PATH, prefix='test_archive_tgz-', suffix='-outdir')

			




	
	
	
		
 

			




	
	
	
		
 
        outfile = tarfile.open(path, 'r|gz')

			




	
	
	
		
 
        outfile.extractall(outdir)

			




	
	
		
 
@@ -58,10 +58,10 @@ class ArchivesTestCaseMixin(_BackendTest

			




	
	
	
		
 
                self.tip.get_node(node_path).content)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_archive_tbz2(self):

			




	
	
	
		
 
        path = tempfile.mkstemp()[1]

			




	
	
	
		
 
        path = tempfile.mkstemp(dir=TEST_TMP_PATH, prefix='test_archive_tbz2-')[1]

			




	
	
	
		
 
        with open(path, 'w+b') as f:

			




	
	
	
		
 
            self.tip.fill_archive(stream=f, kind='tbz2', prefix='repo')

			




	
	
	
		
 
        outdir = tempfile.mkdtemp()

			




	
	
	
		
 
        outdir = tempfile.mkdtemp(dir=TEST_TMP_PATH, prefix='test_archive_tbz2-', suffix='-outdir')

			




	
	
	
		
 

			




	
	
	
		
 
        outfile = tarfile.open(path, 'r|bz2')

			




	
	
	
		
 
        outfile.extractall(outdir)

			




	
	
		
 
@@ -73,7 +73,7 @@ class ArchivesTestCaseMixin(_BackendTest

			




	
	
	
		
 
                self.tip.get_node(node_path).content)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_archive_default_stream(self):

			




	
	
	
		
 
        tmppath = tempfile.mkstemp()[1]

			




	
	
	
		
 
        tmppath = tempfile.mkstemp(dir=TEST_TMP_PATH, prefix='test_archive_default_stream-')[1]

			




	
	
	
		
 
        with open(tmppath, 'wb') as stream:

			




	
	
	
		
 
            self.tip.fill_archive(stream=stream)

			




	
	
	
		
 
        mystream = StringIO.StringIO()

			




        

    


    
    

    

        

                

                    kallithea/tests/vcs/test_git.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -4,7 +4,6 @@ import sys

			




	
	
	
		
 
import mock

			




	
	
	
		
 
import datetime

			




	
	
	
		
 
import urllib2

			




	
	
	
		
 
import tempfile

			




	
	
	
		
 

			




	
	
	
		
 
import pytest

			




	
	
	
		
 

			




	
	
		
 
@@ -14,7 +13,7 @@ from kallithea.lib.vcs.nodes import Node

			




	
	
	
		
 
from kallithea.lib.vcs.utils.compat import unittest

			




	
	
	
		
 
from kallithea.model.scm import ScmModel

			




	
	
	
		
 
from kallithea.tests.vcs.base import _BackendTestMixin

			




	
	
	
		
 
from kallithea.tests.vcs.conf import TEST_GIT_REPO, TEST_GIT_REPO_CLONE, get_new_dir

			




	
	
	
		
 
from kallithea.tests.vcs.conf import TEST_GIT_REPO, TEST_GIT_REPO_CLONE, TEST_TMP_PATH, get_new_dir

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class GitRepositoryTest(unittest.TestCase):

			




	
	
		
 
@@ -29,7 +28,7 @@ class GitRepositoryTest(unittest.TestCas

			




	
	
	
		
 
        self.repo = GitRepository(TEST_GIT_REPO)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_wrong_repo_path(self):

			




	
	
	
		
 
        wrong_repo_path = os.path.join(tempfile.gettempdir(), 'errorrepo')

			




	
	
	
		
 
        wrong_repo_path = os.path.join(TEST_TMP_PATH, 'errorrepo')

			




	
	
	
		
 
        self.assertRaises(RepositoryError, GitRepository, wrong_repo_path)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_git_cmd_injection(self):

			




        

    


    
    

    

        

                

                    kallithea/tests/vcs/test_hg.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -3,13 +3,12 @@ import os

			




	
	
	
		
 

			




	
	
	
		
 
import pytest

			




	
	
	
		
 

			




	
	
	
		
 
import tempfile

			




	
	
	
		
 
from kallithea.lib.utils2 import safe_str

			




	
	
	
		
 
from kallithea.lib.vcs.backends.hg import MercurialRepository, MercurialChangeset

			




	
	
	
		
 
from kallithea.lib.vcs.exceptions import RepositoryError, VCSError, NodeDoesNotExistError

			




	
	
	
		
 
from kallithea.lib.vcs.nodes import NodeKind, NodeState

			




	
	
	
		
 
from kallithea.tests.vcs.conf import TEST_HG_REPO, TEST_HG_REPO_CLONE, \

			




	
	
	
		
 
    TEST_HG_REPO_PULL

			




	
	
	
		
 
    TEST_HG_REPO_PULL, TEST_TMP_PATH

			




	
	
	
		
 
from kallithea.lib.vcs.utils.compat import unittest

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
		
 
@@ -32,7 +31,7 @@ class MercurialRepositoryTest(unittest.T

			




	
	
	
		
 
        self.repo = MercurialRepository(safe_str(TEST_HG_REPO))

			




	
	
	
		
 

			




	
	
	
		
 
    def test_wrong_repo_path(self):

			




	
	
	
		
 
        wrong_repo_path = os.path.join(tempfile.gettempdir(), 'errorrepo')

			




	
	
	
		
 
        wrong_repo_path = os.path.join(TEST_TMP_PATH, 'errorrepo')

			




	
	
	
		
 
        self.assertRaises(RepositoryError, MercurialRepository, wrong_repo_path)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_unicode_path_repo(self):

			




        

    


    
    

    

        

                

                    kallithea/tests/vcs/test_utils.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -4,8 +4,8 @@ import os

			




	
	
	
		
 
import mock

			




	
	
	
		
 
import time

			




	
	
	
		
 
import shutil

			




	
	
	
		
 
import tempfile

			




	
	
	
		
 
import datetime

			




	
	
	
		
 

			




	
	
	
		
 
from kallithea.lib.vcs.utils.compat import unittest

			




	
	
	
		
 
from kallithea.lib.vcs.utils.paths import get_dirs_for_path

			




	
	
	
		
 
from kallithea.lib.vcs.utils.helpers import get_dict_for_attrs

			




	
	
		
 
@@ -61,8 +61,7 @@ class PathsTest(unittest.TestCase):

			




	
	
	
		
 
        self.assertRaises(VCSError, get_scm, 'err')

			




	
	
	
		
 

			




	
	
	
		
 
    def test_get_scms_for_path(self):

			




	
	
	
		
 
        dirpath = tempfile.gettempdir()

			




	
	
	
		
 
        new = os.path.join(dirpath, 'vcs-scms-for-path-%s' % time.time())

			




	
	
	
		
 
        new = os.path.join(TEST_TMP_PATH, 'vcs-scms-for-path-%s' % time.time())

			




	
	
	
		
 
        os.mkdir(new)

			




	
	
	
		
 
        self.assertEqual(get_scms_for_path(new), [])

			




	
	
	
		
 

			




        

    



    


    



    



      

      





    
    0 comments (0 inline, 0 general)
    





    


  

  







  


    




    








    
        
    
    
        This site is powered by
            Kallithea 0.7.0,
        which is
        © 2010–2025 by various authors & licensed under GPLv3.