kallithea Changeset - 6304efbe37a3

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

kallithea/tests/functional/test_admin_settings.py

kallithea/tests/other/test_validators.py

kallithea/tests/other/test_vcs_operations.py

kallithea/tests/scripts/manual_test_concurrency.py

kallithea/tests/vcs/conf.py

kallithea/tests/vcs/test_archives.py

kallithea/tests/vcs/test_git.py

kallithea/tests/vcs/test_hg.py

kallithea/tests/vcs/test_utils.py

0 comments (0 inline, 0 general)

kallithea/tests/base.py

➞

Show inline comments

@@ @@ -35,94 +35,93 @@ skipif = pytest.mark.skipif @@
 parametrize = pytest.mark.parametrize
 # Hack: These module global values MUST be set to actual values before running any tests. This is currently done by conftest.py.
 url = None
 testapp = None
 __all__ = [
     'skipif', 'parametrize', 'url', 'TestController',
     'ldap_lib_installed', 'pam_lib_installed', 'invalidate_all_caches',
     'TESTS_TMP_PATH', 'HG_REPO', 'GIT_REPO', 'NEW_HG_REPO', 'NEW_GIT_REPO',
     'HG_FORK', 'GIT_FORK', 'TEST_USER_ADMIN_LOGIN', 'TEST_USER_ADMIN_PASS',
     'TEST_USER_ADMIN_EMAIL', 'TEST_USER_REGULAR_LOGIN', 'TEST_USER_REGULAR_PASS',
     'TEST_USER_REGULAR_EMAIL', 'TEST_USER_REGULAR2_LOGIN',
     'TEST_USER_REGULAR2_PASS', 'TEST_USER_REGULAR2_EMAIL', 'TEST_HG_REPO',
     'TEST_HG_REPO_CLONE', 'TEST_HG_REPO_PULL', 'TEST_GIT_REPO',
     'TEST_GIT_REPO_CLONE', 'TEST_GIT_REPO_PULL', 'HG_REMOTE_REPO',
     'GIT_REMOTE_REPO', 'SCM_TESTS', 'HG_TEST_REVISION', 'GIT_TEST_REVISION',
+]
 # Invoke websetup with the current config file
 # SetupCommand('setup-app').run([config_file])
 #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'
 TEST_USER_REGULAR_LOGIN = 'test_regular'
 TEST_USER_REGULAR_PASS = 'test12'
 TEST_USER_REGULAR_EMAIL = 'test_regular@example.com'
 TEST_USER_REGULAR2_LOGIN = 'test_regular2'
 TEST_USER_REGULAR2_PASS = 'test12'
 TEST_USER_REGULAR2_EMAIL = 'test_regular2@example.com'
 HG_REPO = u'vcs_test_hg'
 GIT_REPO = u'vcs_test_git'
 NEW_HG_REPO = u'vcs_test_hg_new'
 NEW_GIT_REPO = u'vcs_test_git_new'
 HG_FORK = u'vcs_test_hg_fork'
 GIT_FORK = u'vcs_test_git_fork'
 HG_TEST_REVISION = u"a53d9201d4bc278910d416d94941b7ea007ecd52"
 GIT_TEST_REVISION = u"7ab37bc680b4aa72c34d07b230c866c28e9fc204"
 ## VCS
 SCM_TESTS = ['hg', 'git']
 uniq_suffix = str(int(time.mktime(datetime.datetime.now().timetuple())))
 GIT_REMOTE_REPO = 'git://github.com/codeinn/vcs.git'
 TEST_GIT_REPO = os.path.join(TESTS_TMP_PATH, GIT_REPO)
 TEST_GIT_REPO_CLONE = os.path.join(TESTS_TMP_PATH, 'vcsgitclone%s' % uniq_suffix)
 TEST_GIT_REPO_PULL = os.path.join(TESTS_TMP_PATH, 'vcsgitpull%s' % uniq_suffix)
 HG_REMOTE_REPO = 'http://bitbucket.org/marcinkuzminski/vcs'
 TEST_HG_REPO = os.path.join(TESTS_TMP_PATH, HG_REPO)
 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)
 HG_REMOTE_REPO = os.path.join(TESTS_TMP_PATH, HG_REPO)
 #skip ldap tests if LDAP lib is not installed
 ldap_lib_installed = False
 try:
     import ldap
     ldap.API_VERSION
     ldap_lib_installed = True
 except ImportError:
     # means that python-ldap is not installed
     pass
 try:
     import pam
     pam.PAM_TEXT_INFO
     pam_lib_installed = True
 except ImportError:
     pam_lib_installed = False
 def invalidate_all_caches():
     """Invalidate all beaker caches currently configured.

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
 fixture = Fixture()
 class TestAdminSettingsController(TestController):
     def test_index_main(self):
         self.log_user()
         response = self.app.get(url('admin_settings'))
     def test_index_mapping(self):
         self.log_user()
         response = self.app.get(url('admin_settings_mapping'))
     def test_index_global(self):
         self.log_user()
         response = self.app.get(url('admin_settings_global'))
     def test_index_visual(self):
         self.log_user()
         response = self.app.get(url('admin_settings_visual'))
     def test_index_email(self):
         self.log_user()
         response = self.app.get(url('admin_settings_email'))
     def test_index_hooks(self):
         self.log_user()
         response = self.app.get(url('admin_settings_hooks'))
     def test_create_custom_hook(self):
         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
         self.app.post(url('admin_settings_hooks'),
                         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()
         response = self.app.get(url('admin_settings_search'))
     def test_index_system(self):
         self.log_user()
         response = self.app.get(url('admin_settings_system'))
     def test_ga_code_active(self):
         self.log_user()
         old_title = 'Kallithea'
         old_realm = 'Kallithea authentication'
         new_ga_code = 'ga-test-123456789'
         response = self.app.post(url('admin_settings_global'),
                         params=dict(title=old_title,
                                  realm=old_realm,
                                  ga_code=new_ga_code,
                                  captcha_private_key='',
                                  captcha_public_key='',
                                  _authentication_token=self.authentication_token(),
                                  ))
         self.checkSessionFlash(response, 'Updated application settings')

kallithea/tests/other/test_validators.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 import formencode
 import pytest
 import tempfile
 from kallithea.tests.base import *
 from kallithea.model import validators as v
 from kallithea.model.user_group import UserGroupModel
 from kallithea.model.meta import Session
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.tests.fixture import Fixture
 fixture = Fixture()
 @pytest.mark.usefixtures("test_context_fixture") # apply fixture for all test methods
 class TestRepoGroups(TestController):
     def teardown_method(self, method):
         Session.remove()
     def test_Message_extractor(self):
         validator = v.ValidUsername()
         with pytest.raises(formencode.Invalid):
             validator.to_python('default')
@@ @@ -184,49 +183,49 @@ class TestRepoGroups(TestController): @@
     def test_ValidCloneUri(self):
             #TODO: write this one
             pass
     def test_ValidForkType(self):
             validator = v.ValidForkType(old_data={'repo_type': 'hg'})
             assert 'hg' == validator.to_python('hg')
             with pytest.raises(formencode.Invalid):
                 validator.to_python('git')
     def test_ValidPerms(self):
             #TODO: write this one
             pass
     def test_ValidSettings(self):
         validator = v.ValidSettings()
         assert {'pass': 'pass'} == validator.to_python(value={'user': 'test',
                                                     'pass': 'pass'})
         assert {'user2': 'test', 'pass': 'pass'} == validator.to_python(value={'user2': 'test',
                                                     'pass': 'pass'})
     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')
     def test_UniqSystemEmail(self):
         validator = v.UniqSystemEmail(old_data={})
         assert 'mail@python.org' == validator.to_python('MaiL@Python.org')
         email = TEST_USER_REGULAR2_EMAIL
         with pytest.raises(formencode.Invalid):
             validator.to_python(email)
     def test_ValidSystemEmail(self):
         validator = v.ValidSystemEmail()
         email = TEST_USER_REGULAR2_EMAIL
         assert email == validator.to_python(email)
         with pytest.raises(formencode.Invalid):
             validator.to_python('err')
     def test_LdapLibValidator(self):
         if ldap_lib_installed:
             validator = v.LdapLibValidator()
             assert "DN" == validator.to_python('DN')

kallithea/tests/other/test_vcs_operations.py

➞

Show inline comments

@@ @@ -56,50 +56,50 @@ class Command(object): @@
         """
         Runs command on the system with given ``args``.
         """
         command = cmd + ' ' + ' '.join(args)
         ignoreReturnCode = environ.pop('ignoreReturnCode', False)
         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:
                 print 'stderr:', repr(stderr)
         if not ignoreReturnCode:
             assert p.returncode == 0
         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,
                         clone_url=None, username=TEST_USER_ADMIN_LOGIN, password=TEST_USER_ADMIN_PASS):
     """
     Generate some files, add it to DEST repo and push back
     vcs is git or hg and defines what VCS we want to make those files for
     :param vcs:
     :param DEST:
     """
     # commit some stuff into this repo
     cwd = os.path.join(DEST)
     #added_file = '%ssetupążźć.py' % _RandomNameSequence().next()
     added_file = '%ssetup.py' % _RandomNameSequence().next()
     Command(cwd).execute('touch %s' % added_file)
     Command(cwd).execute('%s add %s' % (vcs, added_file))
     email = 'me@example.com'
     if os.name == 'nt':
         author_str = 'User <%s>' % email
     else:
         author_str = 'User ǝɯɐᴎ <%s>' % email
     for i in xrange(files_no):
@@ @@ -157,392 +157,392 @@ def _check_proper_git_push(stdout, stder @@
     assert 'master -> master' in stderr
 @pytest.mark.usefixtures("test_context_fixture")
 class TestVCSOperations(TestController):
     @classmethod
     def setup_class(cls):
         #DISABLE ANONYMOUS ACCESS
         set_anonymous_access(False)
     def setup_method(self, method):
         r = Repository.get_by_repo_name(GIT_REPO)
         Repository.unlock(r)
         r.enable_locking = False
         Session().commit()
         r = Repository.get_by_repo_name(HG_REPO)
         Repository.unlock(r)
         r.enable_locking = False
         Session().commit()
     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
         assert 'adding manifests' in stdout
         assert 'adding file changes' in stdout
         assert stderr == ''
     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)
         clone_url = webserver.repo_url(fork_name)
         stdout, stderr = _add_files_and_push(webserver, '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, 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())
         fixture.create_fork(GIT_REPO, fork_name)
         clone_url = webserver.repo_url(fork_name)
         stdout, stderr = _add_files_and_push(webserver, 'git', DEST, clone_url=clone_url)
         print [(x.repo_full_path,x.repo_path) for x in Repository.query()] # TODO: what is this for
         _check_proper_git_push(stdout, stderr)
     def test_push_invalidates_cache_hg(self, webserver):
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                ==HG_REPO).scalar()
         if not key:
             key = CacheInvalidation(HG_REPO, HG_REPO)
             Session().add(key)
         key.cache_active = True
         Session().commit()
         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)
         clone_url = webserver.repo_url(fork_name)
         stdout, stderr = _add_files_and_push(webserver, 'hg', DEST, files_no=1, clone_url=clone_url)
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                ==fork_name).all()
         assert key == []
     def test_push_invalidates_cache_git(self, webserver):
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                ==GIT_REPO).scalar()
         if not key:
             key = CacheInvalidation(GIT_REPO, GIT_REPO)
             Session().add(key)
         key.cache_active = True
         Session().commit()
         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())
         fixture.create_fork(GIT_REPO, fork_name)
         clone_url = webserver.repo_url(fork_name)
         stdout, stderr = _add_files_and_push(webserver, 'git', DEST, files_no=1, clone_url=clone_url)
         _check_proper_git_push(stdout, stderr)
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                ==fork_name).all()
         assert key == []
     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)
         assert 'abort: authorization failed' in stderr
     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)
         assert 'fatal: Authentication failed' in stderr
     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]),
                                     ignoreReturnCode = True)
         assert 'HTTP Error 404: Not Found' in stderr
     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]),
                                     ignoreReturnCode = True)
         assert 'not found' in stderr
     def test_clone_and_create_lock_hg(self, webserver):
         # enable locking
         r = Repository.get_by_repo_name(HG_REPO)
         r.enable_locking = True
         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)
         assert r.locked[0] == User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id
     def test_clone_and_create_lock_git(self, webserver):
         # enable locking
         r = Repository.get_by_repo_name(GIT_REPO)
         r.enable_locking = True
         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)
         assert r.locked[0] == User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id
     def test_clone_after_repo_was_locked_hg(self, webserver):
         #lock repo
         r = Repository.get_by_repo_name(HG_REPO)
         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
     def test_clone_after_repo_was_locked_git(self, webserver):
         #lock repo
         r = Repository.get_by_repo_name(GIT_REPO)
         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
     def test_push_on_locked_repo_by_other_user_hg(self, webserver):
         #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)
         # 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(webserver, 'hg', DEST,
                                              username=TEST_USER_REGULAR_LOGIN,
                                              password=TEST_USER_REGULAR_PASS,
                                              ignoreReturnCode=True)
         msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""
                 % (HG_REPO, TEST_USER_ADMIN_LOGIN))
         assert msg in stderr
     def test_push_on_locked_repo_by_other_user_git(self, webserver):
         # Note: Git hooks must be executable on unix. This test will thus fail
         # for example on Linux if /tmp is mounted noexec.
         #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)
         # 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(webserver, 'git', DEST,
                                              username=TEST_USER_REGULAR_LOGIN,
                                              password=TEST_USER_REGULAR_PASS,
                                              ignoreReturnCode=True)
         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, webserver):
         # enable locking
         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().commit()
         #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)
         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(webserver, 'hg', DEST, clone_url=clone_url)
         assert str('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(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, webserver):
         # enable locking
         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().commit()
         #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)
         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(webserver, 'git', DEST, clone_url=clone_url)
         _check_proper_git_push(stdout, 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(fork_name)
         assert r.locked == [None, None]
     def test_ip_restriction_hg(self, webserver):
         user_model = UserModel()
         try:
             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
             for ip in UserIpMap.query():
                 UserIpMap.delete(ip.ip_id)
             Session().commit()
         # IP permissions are cached, need to wait for the cache in the server process to expire
         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
         assert 'adding manifests' in stdout
         assert 'adding file changes' in stdout
         assert stderr == ''
     def test_ip_restriction_git(self, webserver):
         user_model = UserModel()
         try:
             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:
             #release IP restrictions
             for ip in UserIpMap.query():
                 UserIpMap.delete(ip.ip_id)
             Session().commit()
         # IP permissions are cached, need to wait for the cache in the server process to expire
         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

@@ @@ -138,84 +138,79 @@ def create_test_repo(force=True): @@
         rm = RepoModel()
         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, 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)
         #print 'made dirs %s' % os.path.join(path)
     except OSError:
         raise
     clone_url = 'http://%(user)s:%(pass)s@%(host)s/%(cloned_repo)s' % \
                   {'user': USER,
                    'pass': PASS,
                    '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'
                 assert """abort""" not in stderr , 'got error from clone'
             elif backend == 'git':
                 assert """Cloning into""" in stdout, 'no messages about cloning'
 if __name__ == '__main__':
     try:
         create_test_user(force=False)
         seq = None
         import time
         try:
             METHOD = sys.argv[3]
         except IndexError:
             pass
         try:
             backend = sys.argv[4]
         except IndexError:
             backend = 'hg'
         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

 """
 Unit tests configuration module for vcs.
 """
 import os
 import time
 import hashlib
 import tempfile
 import datetime
 import shutil
 import uuid
 __all__ = (
     'TEST_HG_REPO', 'TEST_GIT_REPO', 'HG_REMOTE_REPO', 'GIT_REMOTE_REPO',
     'SCM_TESTS',
+)
 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:
     prefix-[title-]hexuuid
     Prefix is equal to value of variable TEST_REPO_PREFIX. The "hexuuid" is a
     hexadecimal value of a randomly generated UUID. Title will be added if
     specified.
     Args:
         title: Custom title to include as part of the resulting sub-directory
             name. Can be useful for debugging to identify destination. Defaults
             to None.
     Returns:
         Path to the new directory as a string.
     """
     if title:
         name = "%s-%s" % (TEST_REPO_PREFIX, title)
     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
     while os.path.exists("%s-%s" % (path, hex_uuid)):
         hex_uuid = uuid.uuid4().hex
     return "%s-%s" % (path, hex_uuid)
 PACKAGE_DIR = os.path.abspath(os.path.join(
     os.path.dirname(__file__), '..'))
 _dest = os.path.join(TEST_TMP_PATH, 'aconfig')
 shutil.copy(os.path.join(THIS, 'aconfig'), _dest)
 TEST_USER_CONFIG_FILE = _dest
 #overide default configurations with kallithea ones
 from kallithea.tests.base import *

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
 class ArchivesTestCaseMixin(_BackendTestMixin):
     @classmethod
     def _get_commits(cls):
         start_date = datetime.datetime(2010, 1, 1, 20)
         for x in xrange(5):
             yield {
                 'message': 'Commit %d' % x,
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': start_date + datetime.timedelta(hours=12 * x),
                 'added': [
                     FileNode('%d/file_%d.txt' % (x, x),
                         content='Foobar %d' % x),
                 ],
+            }
     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)
         for x in xrange(5):
             node_path = '%d/file_%d.txt' % (x, x)
             decompressed = StringIO.StringIO()
             decompressed.write(out.read('repo/' + node_path))
             self.assertEqual(
                 decompressed.getvalue(),
                 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)
         for x in xrange(5):
             node_path = '%d/file_%d.txt' % (x, x)
             self.assertEqual(
                 open(os.path.join(outdir, 'repo/' + node_path)).read(),
                 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)
         for x in xrange(5):
             node_path = '%d/file_%d.txt' % (x, x)
             self.assertEqual(
                 open(os.path.join(outdir, 'repo/' + node_path)).read(),
                 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()
         self.tip.fill_archive(stream=mystream)
         mystream.seek(0)
         with open(tmppath, 'rb') as f:
             file_content = f.read()
             stringio_content = mystream.read()
             # the gzip header contains a MTIME header
             # because is takes a little bit of time from one fill_archive call to the next
             # this part may differ so don't include that part in the comparison
             self.assertEqual(file_content[:4], stringio_content[:4])
             self.assertEqual(file_content[8:], stringio_content[8:])
     def test_archive_wrong_kind(self):
         with self.assertRaises(VCSError):
             self.tip.fill_archive(kind='wrong kind')
     def test_archive_empty_prefix(self):
         with self.assertRaises(VCSError):
             self.tip.fill_archive(prefix='')
     def test_archive_prefix_with_leading_slash(self):
         with self.assertRaises(VCSError):

kallithea/tests/vcs/test_git.py

➞

Show inline comments

 import os
 import sys
 import mock
 import datetime
 import urllib2
 import tempfile
 import pytest
 from kallithea.lib.vcs.backends.git import GitRepository, GitChangeset
 from kallithea.lib.vcs.exceptions import RepositoryError, VCSError, NodeDoesNotExistError
 from kallithea.lib.vcs.nodes import NodeKind, FileNode, DirNode, NodeState
 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):
     def __check_for_existing_repo(self):
         if os.path.exists(TEST_GIT_REPO_CLONE):
             pytest.fail('Cannot test git clone repo as location %s already '
                       'exists. You should manually remove it first.'
                       % TEST_GIT_REPO_CLONE)
     def setUp(self):
         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):
         repo_inject_path = TEST_GIT_REPO + '; echo "Cake";'
         with self.assertRaises(urllib2.URLError):
             # Should fail because URL will contain the parts after ; too
             urlerror_fail_repo = GitRepository(get_new_dir('injection-repo'), src_url=repo_inject_path, update_after_clone=True, create=True)
         with self.assertRaises(RepositoryError):
             # Should fail on direct clone call, which as of this writing does not happen outside of class
             clone_fail_repo = GitRepository(get_new_dir('injection-repo'), create=True)
             clone_fail_repo.clone(repo_inject_path, update_after_clone=True,)
         # Verify correct quoting of evil characters that should work on posix file systems
         if sys.platform == 'win32':
             # windows does not allow '"' in dir names
             # and some versions of the git client don't like ` and '
             tricky_path = get_new_dir("tricky-path-repo-$")
         else:
             tricky_path = get_new_dir("tricky-path-repo-$'\"`")
         successfully_cloned = GitRepository(tricky_path, src_url=TEST_GIT_REPO, update_after_clone=True, create=True)
         # Repo should have been created
         self.assertFalse(successfully_cloned._repo.bare)

kallithea/tests/vcs/test_hg.py

➞

Show inline comments

 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
 # Use only clean mercurial's ui
 from kallithea.lib.vcs.utils.hgcompat import mercurial
 mercurial.scmutil.rcpath()
 if mercurial.scmutil._rcpath:
     mercurial.scmutil._rcpath = mercurial.scmutil._rcpath[:1]
 class MercurialRepositoryTest(unittest.TestCase):
     def __check_for_existing_repo(self):
         if os.path.exists(TEST_HG_REPO_CLONE):
             pytest.fail('Cannot test mercurial clone repo as location %s already '
                       'exists. You should manually remove it first.'
                       % TEST_HG_REPO_CLONE)
     def setUp(self):
         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):
         self.assertRaises(VCSError,lambda:MercurialRepository(u'iShouldFail'))
     def test_repo_clone(self):
         self.__check_for_existing_repo()
         repo = MercurialRepository(safe_str(TEST_HG_REPO))
         repo_clone = MercurialRepository(TEST_HG_REPO_CLONE,
             src_url=TEST_HG_REPO, update_after_clone=True)
         self.assertEqual(len(repo.revisions), len(repo_clone.revisions))
         # Checking hashes of changesets should be enough
         for changeset in repo.get_changesets():
             raw_id = changeset.raw_id
             self.assertEqual(raw_id, repo_clone.get_changeset(raw_id).raw_id)
     def test_repo_clone_with_update(self):
         repo = MercurialRepository(safe_str(TEST_HG_REPO))
         repo_clone = MercurialRepository(TEST_HG_REPO_CLONE + '_w_update',
             src_url=TEST_HG_REPO, update_after_clone=True)
         self.assertEqual(len(repo.revisions), len(repo_clone.revisions))
         #check if current workdir was updated
         self.assertEqual(os.path.isfile(os.path.join(TEST_HG_REPO_CLONE \

kallithea/tests/vcs/test_utils.py

➞

Show inline comments

 # -*- coding: utf-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
 from kallithea.lib.vcs.utils.helpers import get_scm
 from kallithea.lib.vcs.utils.helpers import get_scms_for_path
 from kallithea.lib.vcs.utils.helpers import get_total_seconds
 from kallithea.lib.vcs.utils.helpers import parse_changesets
 from kallithea.lib.vcs.utils.helpers import parse_datetime
 from kallithea.lib.vcs.utils import author_email, author_name
 from kallithea.lib.vcs.utils.paths import get_user_home
 from kallithea.lib.vcs.exceptions import VCSError
 from kallithea.tests.vcs.conf import TEST_HG_REPO, TEST_GIT_REPO, TEST_TMP_PATH
 class PathsTest(unittest.TestCase):
     def _test_get_dirs_for_path(self, path, expected):
         """
         Tests if get_dirs_for_path returns same as expected.
         """
         expected = sorted(expected)
         result = sorted(get_dirs_for_path(path))
         self.assertEqual(result, expected,
@@ @@ -40,50 +40,49 @@ class PathsTest(unittest.TestCase): @@
             ('foo/bar/', ['foo', 'foo/bar']),
             ('foo/bar', ['foo']),
+        )
         for path, expected in paths_and_results:
             self._test_get_dirs_for_path(path, expected)
     def test_get_scm(self):
         self.assertEqual(('hg', TEST_HG_REPO), get_scm(TEST_HG_REPO))
         self.assertEqual(('git', TEST_GIT_REPO), get_scm(TEST_GIT_REPO))
     def test_get_two_scms_for_path(self):
         multialias_repo_path = os.path.join(TEST_TMP_PATH, 'hg-git-repo-2')
         if os.path.isdir(multialias_repo_path):
             shutil.rmtree(multialias_repo_path)
         os.mkdir(multialias_repo_path)
         self.assertRaises(VCSError, get_scm, multialias_repo_path)
     def test_get_scm_error_path(self):
         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), [])
         os.mkdir(os.path.join(new, '.tux'))
         self.assertEqual(get_scms_for_path(new), [])
         os.mkdir(os.path.join(new, '.git'))
         self.assertEqual(set(get_scms_for_path(new)), set(['git']))
         os.mkdir(os.path.join(new, '.hg'))
         self.assertEqual(set(get_scms_for_path(new)), set(['git', 'hg']))
 class TestParseChangesets(unittest.TestCase):
     def test_main_is_returned_correctly(self):
         self.assertEqual(parse_changesets('123456'), {
             'start': None,
             'main': '123456',
             'end': None,
         })
     def test_start_is_returned_correctly(self):
         self.assertEqual(parse_changesets('aaabbb..'), {

0 comments (0 inline, 0 general)