kallithea Changeset - 1a7611b9730e

Changeset - 1a7611b9730e

Parent rev.

Child rev.

[Not reviewed]

default

0 8 0

domruf - 10 years ago 2016-02-02 21:45:29
dominikruf@gmail.com

replace absolute /tmp paths to tempfile.gettempdir()

8 files changed with 45 insertions and 40 deletions:

kallithea/bin/kallithea_backup.py

kallithea/tests/__init__.py

kallithea/tests/functional/test_admin_settings.py

kallithea/tests/other/manual_test_vcs_operations.py

kallithea/tests/scripts/manual_test_crawler.py

kallithea/tests/vcs/conf.py

kallithea/tests/vcs/test_git.py

kallithea/tests/vcs/test_hg.py

0 comments (0 inline, 0 general)

kallithea/bin/kallithea_backup.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.bin.kallithea_backup
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Repositories backup manager, it allows to backups all
 repositories and send it to backup server using RSA key via ssh.
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Feb 28, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import sys
 import logging
 import tarfile
 import datetime
 import subprocess
 import tempfile
 logging.basicConfig(level=logging.DEBUG,
                     format="%(asctime)s %(levelname)-5.5s %(message)s")
 class BackupManager(object):
     def __init__(self, repos_location, rsa_key, backup_server):
         today = datetime.datetime.now().weekday() + 1
         self.backup_file_name = "repos.%s.tar.gz" % today
         self.id_rsa_path = self.get_id_rsa(rsa_key)
         self.repos_path = self.get_repos_path(repos_location)
         self.backup_server = backup_server
-        self.backup_file_path = '/tmp'
+        self.backup_file_path = tempfile.gettempdir()
         logging.info('starting backup for %s', self.repos_path)
         logging.info('backup target %s', self.backup_file_path)
     def get_id_rsa(self, rsa_key):
         if not os.path.isfile(rsa_key):
             logging.error('Could not load id_rsa key file in %s', rsa_key)
             sys.exit()
         return rsa_key
     def get_repos_path(self, path):
         if not os.path.isdir(path):
             logging.error('Wrong location for repositories in %s', path)
             sys.exit()
         return path
     def backup_repos(self):
         bckp_file = os.path.join(self.backup_file_path, self.backup_file_name)
         tar = tarfile.open(bckp_file, "w:gz")
         for dir_name in os.listdir(self.repos_path):
             logging.info('backing up %s', dir_name)
             tar.add(os.path.join(self.repos_path, dir_name), dir_name)
         tar.close()
         logging.info('finished backup of mercurial repositories')
     def transfer_files(self):
         params = {
                   'id_rsa_key': self.id_rsa_path,
                   'backup_file': os.path.join(self.backup_file_path,
                                              self.backup_file_name),
                   'backup_server': self.backup_server
+                  }
         cmd = ['scp', '-l', '40000', '-i', '%(id_rsa_key)s' % params,
                '%(backup_file)s' % params,
                '%(backup_server)s' % params]
         subprocess.call(cmd)
         logging.info('Transfered file %s to %s', self.backup_file_name, cmd[4])
     def rm_file(self):
         logging.info('Removing file %s', self.backup_file_name)
         os.remove(os.path.join(self.backup_file_path, self.backup_file_name))
 if __name__ == "__main__":
     repo_location = '/home/repo_path'
     backup_server = 'root@192.168.1.100:/backups/mercurial'
     rsa_key = '/home/id_rsa'
     B_MANAGER = BackupManager(repo_location, rsa_key, backup_server)
     B_MANAGER.backup_repos()
     B_MANAGER.transfer_files()
     B_MANAGER.rm_file()

kallithea/tests/__init__.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 Pylons application test package
 This package assumes the Pylons environment is already loaded.
 This module initializes the application via ``websetup`` (`paster
 setup-app`) and provides the base testing objects.
 Refer to docs/contributing.rst for details on running the test suite.
 """
 import os
 import re
 import time
 import logging
 import datetime
 import hashlib
 import tempfile
 from os.path import join as jn
 from tempfile import _RandomNameSequence
 import pylons
 import pylons.test
 from pylons import config, url
 from pylons.i18n.translation import _get_translator
 from pylons.util import ContextObj
 from routes.util import URLGenerator
 from webtest import TestApp
 import pytest
 from kallithea.lib.compat import unittest
 from kallithea import is_windows
 from kallithea.model.db import Notification, User, UserNotification
 from kallithea.model.meta import Session
 from kallithea.tests.parameterized import parameterized
 from kallithea.lib.utils2 import safe_str
 os.environ['TZ'] = 'UTC'
 if not is_windows:
     time.tzset()
 log = logging.getLogger(__name__)
 skipif = pytest.mark.skipif
 parametrize = pytest.mark.parametrize
 __all__ = [
     'skipif', 'parametrize', 'parameterized', 'environ', 'url', 'TestController', 'TestControllerPytest',
     'ldap_lib_installed', 'pam_lib_installed', 'BaseTestCase', 'init_stack',
     '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', 'remove_all_notifications',
+]
 # Invoke websetup with the current config file
 # SetupCommand('setup-app').run([config_file])
 environ = {}
 #SOME GLOBALS FOR TESTS
-TESTS_TMP_PATH = jn('/', 'tmp', 'rc_test_%s' % _RandomNameSequence().next())
+TESTS_TMP_PATH = jn(tempfile.gettempdir(), 'rc_test_%s' % _RandomNameSequence().next())
 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'
 ## 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 = jn(TESTS_TMP_PATH, GIT_REPO)
 TEST_GIT_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcsgitclone%s' % uniq_suffix)
 TEST_GIT_REPO_PULL = jn(TESTS_TMP_PATH, 'vcsgitpull%s' % uniq_suffix)
 HG_REMOTE_REPO = 'http://bitbucket.org/marcinkuzminski/vcs'
 TEST_HG_REPO = jn(TESTS_TMP_PATH, HG_REPO)
 TEST_HG_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcshgclone%s' % uniq_suffix)
 TEST_HG_REPO_PULL = jn(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 = jn(TESTS_TMP_PATH, GIT_REPO)
 HG_REMOTE_REPO = jn(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
 class NullHandler(logging.Handler):
     def emit(self, record):
         pass
 def init_stack(config=None):
     if not config:
         config = pylons.test.pylonsapp.config
     url._push_object(URLGenerator(config['routes.map'], environ))
     pylons.app_globals._push_object(config['pylons.app_globals'])
     pylons.config._push_object(config)
     pylons.tmpl_context._push_object(ContextObj())
     # Initialize a translator for tests that utilize i18n
     translator = _get_translator(pylons.config.get('lang'))
     pylons.translator._push_object(translator)
     h = NullHandler()
     logging.getLogger("kallithea").addHandler(h)
 def remove_all_notifications():
     Notification.query().delete()
     # Because query().delete() does not (by default) trigger cascades.
     # http://docs.sqlalchemy.org/en/rel_0_7/orm/collections.html#passive-deletes
     UserNotification.query().delete()
     Session().commit()
 class BaseTestCase(unittest.TestCase):
     """Unittest-style base test case. Deprecated in favor of pytest style."""
     def __init__(self, *args, **kwargs):
         self.wsgiapp = pylons.test.pylonsapp
         init_stack(self.wsgiapp.config)
         unittest.TestCase.__init__(self, *args, **kwargs)
 class BaseTestController(object):
     """Base test controller used by pytest and unittest tests controllers."""
     # Note: pytest base classes cannot have an __init__ method
     def init(self):
         self.app = TestApp(self.wsgiapp)
         self.maxDiff = None
         self.index_location = config['app_conf']['index_dir']
     def log_user(self, username=TEST_USER_ADMIN_LOGIN,
                  password=TEST_USER_ADMIN_PASS):
         self._logged_username = username
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': username,
                                   'password': password})
         if 'Invalid username or password' in response.body:
             pytest.fail('could not login using %s %s' % (username, password))
         self.assertEqual(response.status, '302 Found')
         self.assert_authenticated_user(response, username)
         response = response.follow()
         return response.session['authuser']
     def _get_logged_user(self):
         return User.get_by_username(self._logged_username)
     def assert_authenticated_user(self, response, expected_username):
         cookie = response.session.get('authuser')
         user = cookie and cookie.get('user_id')
         user = user and User.get(user)
         user = user and user.username
         self.assertEqual(user, expected_username)
     def authentication_token(self):
         return self.app.get(url('authentication_token')).body
     def checkSessionFlash(self, response, msg=None, skip=0, _matcher=lambda msg, m: msg in m):
         if 'flash' not in response.session:
             pytest.fail(safe_str(u'msg `%s` not found - session has no flash:\n%s' % (msg, response)))
         try:
             level, m = response.session['flash'][-1 - skip]
             if _matcher(msg, m):
                 return
         except IndexError:
             pass
         pytest.fail(safe_str(u'msg `%s` not found in session flash (skipping %s): %s' %
                            (msg, skip,
                             ', '.join('`%s`' % m for level, m in response.session['flash']))))
     def checkSessionFlashRegex(self, response, regex, skip=0):
         self.checkSessionFlash(response, regex, skip=skip, _matcher=re.search)
 class TestController(BaseTestCase, BaseTestController):
     """Deprecated unittest-style test controller"""
     def __init__(self, *args, **kwargs):
         super(TestController, self).__init__(*args, **kwargs)
         self.init()
 class TestControllerPytest(BaseTestController):
     """Pytest-style test controller"""
     # Note: pytest base classes cannot have an __init__ method
     @pytest.fixture(autouse=True)
     def app_fixture(self):
         self.wsgiapp = pylons.test.pylonsapp
         init_stack(self.wsgiapp.config)
         self.init()
         return self.app
     # transitional implementations of unittest.TestCase asserts
     # Users of these should be converted to pytest's single 'assert' call
     def assertEqual(self, first, second, msg=None):
         assert first == second
     def assertNotEqual(self, first, second, msg=None):
         assert first != second
     def assertTrue(self, expr, msg=None):
         assert bool(expr) is True
     def assertFalse(self, expr, msg=None):
         assert bool(expr) is False
     def assertIn(self, first, second, msg=None):
         assert first in second
     def assertNotIn(self, first, second, msg=None):
         assert first not in second
     def assertSetEqual(self, first, second, msg=None):
         assert first == second
     def assertListEqual(self, first, second, msg=None):
         assert first == second
     def assertDictEqual(self, first, second, msg=None):
         assert first == second

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 import *
 from kallithea.tests.fixture import Fixture
 fixture = Fixture()
 class TestAdminSettingsController(TestControllerPytest):
     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 /tmp',
+                                            new_hook_ui_value='cd %s' % tempfile.gettempdir(),
                                             _authentication_token=self.authentication_token()))
         response = response.follow()
         response.mustcontain('test_hooks_1')
-        response.mustcontain('cd /tmp')
+        response.mustcontain('cd %s' % tempfile.gettempdir())
     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 /tmp2',
+                                            new_hook_ui_value='cd %s2' % tempfile.gettempdir(),
                                             _authentication_token=self.authentication_token()))
         response = response.follow()
         response.mustcontain('test_hooks_2')
-        response.mustcontain('cd /tmp2')
+        response.mustcontain('cd %s2' % tempfile.gettempdir())
         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 /tmp2'])
+        response.mustcontain(no=['cd %s2' % tempfile.gettempdir()])
     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')
         self.assertEqual(Setting
                          .get_app_settings()['ga_code'], new_ga_code)
         response = response.follow()
         response.mustcontain("""_gaq.push(['_setAccount', '%s']);""" % new_ga_code)
     def test_ga_code_inactive(self):
         self.log_user()
         old_title = 'Kallithea'
         old_realm = 'Kallithea authentication'
         new_ga_code = ''
         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')
         self.assertEqual(Setting
                         .get_app_settings()['ga_code'], new_ga_code)
         response = response.follow()
         response.mustcontain(no=["_gaq.push(['_setAccount', '%s']);" % new_ga_code])
     def test_captcha_activate(self):
         self.log_user()
         old_title = 'Kallithea'
         old_realm = 'Kallithea authentication'
         new_ga_code = ''
         response = self.app.post(url('admin_settings_global'),
                         params=dict(title=old_title,
                                  realm=old_realm,
                                  ga_code=new_ga_code,
                                  captcha_private_key='1234567890',
                                  captcha_public_key='1234567890',
                                  _authentication_token=self.authentication_token(),
                                  ))
         self.checkSessionFlash(response, 'Updated application settings')
         self.assertEqual(Setting
                         .get_app_settings()['captcha_private_key'], '1234567890')
         response = self.app.get(url('register'))
         response.mustcontain('captcha')
     def test_captcha_deactivate(self):
         self.log_user()
         old_title = 'Kallithea'
         old_realm = 'Kallithea authentication'
         new_ga_code = ''
         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='1234567890',
                                  _authentication_token=self.authentication_token(),
                                  ))
         self.checkSessionFlash(response, 'Updated application settings')
         self.assertEqual(Setting
                         .get_app_settings()['captcha_private_key'], '')
         response = self.app.get(url('register'))
         response.mustcontain(no=['captcha'])
     def test_title_change(self):
         self.log_user()
         old_title = 'Kallithea'
         new_title = old_title + '_changed'
         old_realm = 'Kallithea authentication'
         for new_title in ['Changed', 'Żółwik', old_title]:
             response = self.app.post(url('admin_settings_global'),
                         params=dict(title=new_title,
                                  realm=old_realm,
                                  ga_code='',
                                  captcha_private_key='',
                                  captcha_public_key='',
                                  _authentication_token=self.authentication_token(),
                                 ))
             self.checkSessionFlash(response, 'Updated application settings')
             self.assertEqual(Setting
                              .get_app_settings()['title'],
                              new_title.decode('utf-8'))
             response = response.follow()
             response.mustcontain("""<div class="branding">%s</div>""" % new_title)

kallithea/tests/other/manual_test_vcs_operations.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.tests.other.manual_test_vcs_operations
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Test suite for making push/pull operations.
 Run it in two terminals::
  paster serve kallithea/tests/test.ini
  KALLITHEA_WHOOSH_TEST_DISABLE=1 KALLITHEA_NO_TMP_PATH=1 nosetests kallithea/tests/other/manual_test_vcs_operations.py
 You must have git > 1.8.1 for tests to work fine
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Dec 30, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import re
 import tempfile
 import time
 from os.path import join as jn
 from tempfile import _RandomNameSequence
 from subprocess import Popen, PIPE
 from kallithea.tests import *
 from kallithea.model.db import User, Repository, UserIpMap, CacheInvalidation
 from kallithea.model.meta import Session
 from kallithea.model.repo import RepoModel
 from kallithea.model.user import UserModel
 DEBUG = True
 HOST = '127.0.0.1:4999'  # test host
 class Command(object):
     def __init__(self, cwd):
         self.cwd = cwd
     def execute(self, cmd, *args):
         """
         Runs command on the system with given ``args``.
         """
         command = cmd + ' ' + ' '.join(args)
         if DEBUG:
             print '*** CMD %s ***' % command
         p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE, cwd=self.cwd)
         stdout, stderr = p.communicate()
         if DEBUG:
             print 'stdout:', repr(stdout)
             print 'stderr:', repr(stderr)
         return stdout, stderr
 def _get_tmp_dir():
     return tempfile.mkdtemp(prefix='rc_integration_test')
 def _construct_url(repo, dest=None, **kwargs):
     if dest is None:
         #make temp clone
         dest = _get_tmp_dir()
     params = {
         'user': TEST_USER_ADMIN_LOGIN,
         'passwd': TEST_USER_ADMIN_PASS,
         'host': HOST,
         'cloned_repo': repo,
         'dest': dest
+    }
     params.update(**kwargs)
     if params['user'] and params['passwd']:
         _url = 'http://%(user)s:%(passwd)s@%(host)s/%(cloned_repo)s %(dest)s' % params
     else:
         _url = 'http://(host)s/%(cloned_repo)s %(dest)s' % params
     return _url
 def _add_files_and_push(vcs, DEST, **kwargs):
     """
     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 = path = jn(DEST)
     #added_file = jn(path, '%ssetupążźć.py' % _RandomNameSequence().next())
     added_file = jn(path, '%ssetup.py' % _RandomNameSequence().next())
     Command(cwd).execute('touch %s' % added_file)
     Command(cwd).execute('%s add %s' % (vcs, added_file))
     for i in xrange(kwargs.get('files_no', 3)):
         cmd = """echo 'added_line%s' >> %s""" % (i, added_file)
         Command(cwd).execute(cmd)
         author_str = 'User ǝɯɐᴎ <me@example.com>'
         if vcs == 'hg':
             cmd = """hg commit -m 'commited new %s' -u '%s' %s """ % (
                 i, author_str, added_file
+            )
         elif vcs == 'git':
             cmd = """EMAIL="me@example.com" git commit -m 'commited new %s' --author '%s' %s """ % (
                 i, author_str, added_file
+            )
         Command(cwd).execute(cmd)
     # PUSH it back
     _REPO = None
     if vcs == 'hg':
         _REPO = HG_REPO
     elif vcs == 'git':
         _REPO = GIT_REPO
     kwargs['dest'] = ''
     clone_url = _construct_url(_REPO, **kwargs)
     if 'clone_url' in kwargs:
         clone_url = kwargs['clone_url']
     stdout = stderr = None
     if vcs == 'hg':
         stdout, stderr = Command(cwd).execute('hg push --verbose', clone_url)
     elif vcs == 'git':
         stdout, stderr = Command(cwd).execute('git push --verbose', clone_url + " master")
     return stdout, stderr
 def set_anonymous_access(enable=True):
     user = User.get_by_username(User.DEFAULT_USER)
     user.active = enable
     Session().add(user)
     Session().commit()
     print '\tanonymous access is now:', enable
     if enable != User.get_by_username(User.DEFAULT_USER).active:
         raise Exception('Cannot set anonymous access')
 #==============================================================================
 # TESTS
 #==============================================================================
 def _check_proper_git_push(stdout, stderr):
     #WTF Git stderr is output ?!
     assert 'fatal' not in stderr
     assert 'rejected' not in stderr
     assert 'Pushing to' in stderr
     assert 'master -> master' in stderr
 class TestVCSOperations(BaseTestCase):
     @classmethod
     def setup_class(cls):
         #DISABLE ANONYMOUS ACCESS
         set_anonymous_access(False)
     def setUp(self):
         r = Repository.get_by_repo_name(GIT_REPO)
         Repository.unlock(r)
         r.enable_locking = False
         Session().add(r)
         Session().commit()
         r = Repository.get_by_repo_name(HG_REPO)
         Repository.unlock(r)
         r.enable_locking = False
         Session().add(r)
         Session().commit()
     def test_clone_hg_repo_by_admin(self):
         clone_url = _construct_url(HG_REPO)
-        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         assert 'requesting all changes' in stdout
         assert 'adding changesets' in stdout
         assert 'adding manifests' in stdout
         assert 'adding file changes' in stdout
         assert stderr == ''
     def test_clone_git_repo_by_admin(self):
         clone_url = _construct_url(GIT_REPO)
-        stdout, stderr = Command('/tmp').execute('git clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         assert 'Cloning into' in stdout + stderr
         assert stderr == '' or stdout == ''
     def test_clone_wrong_credentials_hg(self):
         clone_url = _construct_url(HG_REPO, passwd='bad!')
-        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         assert 'abort: authorization failed' in stderr
     def test_clone_wrong_credentials_git(self):
         clone_url = _construct_url(GIT_REPO, passwd='bad!')
-        stdout, stderr = Command('/tmp').execute('git clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         assert 'fatal: Authentication failed' in stderr
     def test_clone_git_dir_as_hg(self):
         clone_url = _construct_url(GIT_REPO)
-        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         assert 'HTTP Error 404: Not Found' in stderr
     def test_clone_hg_repo_as_git(self):
         clone_url = _construct_url(HG_REPO)
-        stdout, stderr = Command('/tmp').execute('git clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         assert 'not found' in stderr
     def test_clone_non_existing_path_hg(self):
         clone_url = _construct_url('trololo')
-        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         assert 'HTTP Error 404: Not Found' in stderr
     def test_clone_non_existing_path_git(self):
         clone_url = _construct_url('trololo')
-        stdout, stderr = Command('/tmp').execute('git clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         assert 'not found' in stderr
     def test_push_new_file_hg(self):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(HG_REPO, dest=DEST)
-        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         stdout, stderr = _add_files_and_push('hg', DEST)
         assert 'pushing to' in stdout
         assert 'Repository size' in stdout
         assert 'Last revision is now' in stdout
     def test_push_new_file_git(self):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(GIT_REPO, dest=DEST)
-        stdout, stderr = Command('/tmp').execute('git clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         # commit some stuff into this repo
         stdout, stderr = _add_files_and_push('git', DEST)
         print [(x.repo_full_path,x.repo_path) for x in Repository.get_all()]
         _check_proper_git_push(stdout, stderr)
     def test_push_invalidates_cache_hg(self):
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                ==HG_REPO).scalar()
         if not key:
             key = CacheInvalidation(HG_REPO, HG_REPO)
         key.cache_active = True
         Session().add(key)
         Session().commit()
         DEST = _get_tmp_dir()
         clone_url = _construct_url(HG_REPO, dest=DEST)
-        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         stdout, stderr = _add_files_and_push('hg', DEST, files_no=1)
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                ==HG_REPO).all()
         self.assertEqual(key, [])
     def test_push_invalidates_cache_git(self):
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                ==GIT_REPO).scalar()
         if not key:
             key = CacheInvalidation(GIT_REPO, GIT_REPO)
         key.cache_active = True
         Session().add(key)
         Session().commit()
         DEST = _get_tmp_dir()
         clone_url = _construct_url(GIT_REPO, dest=DEST)
-        stdout, stderr = Command('/tmp').execute('git clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         # commit some stuff into this repo
         stdout, stderr = _add_files_and_push('git', DEST, files_no=1)
         _check_proper_git_push(stdout, stderr)
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                ==GIT_REPO).all()
         self.assertEqual(key, [])
     def test_push_wrong_credentials_hg(self):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(HG_REPO, dest=DEST)
-        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         stdout, stderr = _add_files_and_push('hg', DEST, user='bad',
                                              passwd='name')
         assert 'abort: authorization failed' in stderr
     def test_push_wrong_credentials_git(self):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(GIT_REPO, dest=DEST)
-        stdout, stderr = Command('/tmp').execute('git clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         stdout, stderr = _add_files_and_push('git', DEST, user='bad',
                                              passwd='name')
         assert 'fatal: Authentication failed' in stderr
     def test_push_back_to_wrong_url_hg(self):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(HG_REPO, dest=DEST)
-        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         stdout, stderr = _add_files_and_push('hg', DEST,
                                     clone_url='http://%s/tmp' % HOST)
         assert 'HTTP Error 404: Not Found' in stderr
     def test_push_back_to_wrong_url_git(self):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(GIT_REPO, dest=DEST)
-        stdout, stderr = Command('/tmp').execute('git clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         stdout, stderr = _add_files_and_push('git', DEST,
                                     clone_url='http://%s/tmp' % HOST)
         assert 'not found' in stderr
     def test_clone_and_create_lock_hg(self):
         # enable locking
         r = Repository.get_by_repo_name(HG_REPO)
         r.enable_locking = True
         Session().add(r)
         Session().commit()
         # clone
         clone_url = _construct_url(HG_REPO)
-        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         #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):
         # enable locking
         r = Repository.get_by_repo_name(GIT_REPO)
         r.enable_locking = True
         Session().add(r)
         Session().commit()
         # clone
         clone_url = _construct_url(GIT_REPO)
-        stdout, stderr = Command('/tmp').execute('git clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         #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):
         #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 = _construct_url(HG_REPO)
-        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         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):
         #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 = _construct_url(GIT_REPO)
-        stdout, stderr = Command('/tmp').execute('git clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         msg = ("""The requested URL returned error: 423""")
         assert msg in stderr
     def test_push_on_locked_repo_by_other_user_hg(self):
         #clone some temp
         DEST = _get_tmp_dir()
         clone_url = _construct_url(HG_REPO, dest=DEST)
-        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         #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('hg', DEST,
                                              user=TEST_USER_REGULAR_LOGIN,
                                              passwd=TEST_USER_REGULAR_PASS)
         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):
         #clone some temp
         DEST = _get_tmp_dir()
         clone_url = _construct_url(GIT_REPO, dest=DEST)
-        stdout, stderr = Command('/tmp').execute('git clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         #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('git', DEST,
                                              user=TEST_USER_REGULAR_LOGIN,
                                              passwd=TEST_USER_REGULAR_PASS)
         err = 'Repository `%s` locked by user `%s`' % (GIT_REPO, TEST_USER_ADMIN_LOGIN)
         assert err in stderr
         #TODO: fix this somehow later on Git, Git is stupid and even if we throw
         #back 423 to it, it makes ANOTHER request and we fail there with 405 :/
         msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""
                 % (GIT_REPO, TEST_USER_ADMIN_LOGIN))
         #msg = "405 Method Not Allowed"
         #assert msg in stderr
     def test_push_unlocks_repository_hg(self):
         # enable locking
         r = Repository.get_by_repo_name(HG_REPO)
         r.enable_locking = True
         Session().add(r)
         Session().commit()
         #clone some temp
         DEST = _get_tmp_dir()
         clone_url = _construct_url(HG_REPO, dest=DEST)
-        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         #check for lock repo after clone
         r = Repository.get_by_repo_name(HG_REPO)
         uid = User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id
         assert r.locked[0] == uid
         #push is ok and repo is now unlocked
         stdout, stderr = _add_files_and_push('hg', DEST)
         assert ('remote: Released lock on repo `%s`' % HG_REPO) in stdout
         #we need to cleanup the Session Here !
         Session.remove()
         r = Repository.get_by_repo_name(HG_REPO)
         assert r.locked == [None, None]
     #TODO: fix me ! somehow during tests hooks don't get called on Git
     def test_push_unlocks_repository_git(self):
         # enable locking
         r = Repository.get_by_repo_name(GIT_REPO)
         r.enable_locking = True
         Session().add(r)
         Session().commit()
         #clone some temp
         DEST = _get_tmp_dir()
         clone_url = _construct_url(GIT_REPO, dest=DEST)
-        stdout, stderr = Command('/tmp').execute('git clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         #check for lock repo after clone
         r = Repository.get_by_repo_name(GIT_REPO)
         assert r.locked[0] == User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id
         #push is ok and repo is now unlocked
         stdout, stderr = _add_files_and_push('git', DEST)
         _check_proper_git_push(stdout, stderr)
         #assert ('remote: Released lock on repo `%s`' % GIT_REPO) in stdout
         #we need to cleanup the Session Here !
         Session.remove()
         r = Repository.get_by_repo_name(GIT_REPO)
         assert r.locked == [None, None]
     def test_ip_restriction_hg(self):
         user_model = UserModel()
         try:
             user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')
             Session().commit()
             clone_url = _construct_url(HG_REPO)
-            stdout, stderr = Command('/tmp').execute('hg clone', clone_url)
+            stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
             assert 'abort: HTTP Error 403: Forbidden' in stderr
         finally:
             #release IP restrictions
             for ip in UserIpMap.getAll():
                 UserIpMap.delete(ip.ip_id)
             Session().commit()
         time.sleep(2)
         clone_url = _construct_url(HG_REPO)
-        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         assert 'requesting all changes' in stdout
         assert 'adding changesets' in stdout
         assert 'adding manifests' in stdout
         assert 'adding file changes' in stdout
         assert stderr == ''
     def test_ip_restriction_git(self):
         user_model = UserModel()
         try:
             user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')
             Session().commit()
             clone_url = _construct_url(GIT_REPO)
-            stdout, stderr = Command('/tmp').execute('git clone', clone_url)
+            stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
             # 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.getAll():
                 UserIpMap.delete(ip.ip_id)
             Session().commit()
         time.sleep(2)
         clone_url = _construct_url(GIT_REPO)
-        stdout, stderr = Command('/tmp').execute('git clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         assert 'Cloning into' in stdout + stderr
         assert stderr == '' or stdout == ''

kallithea/tests/scripts/manual_test_crawler.py

➞

Show inline comments

 #!/usr/bin/env python2
 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.tests.scripts.manual_test_crawler
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Test for crawling a project for memory usage
 This should be runned just as regular script together
 with a watch script that will show memory usage.
 watch -n1 ./kallithea/tests/mem_watch
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 21, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import cookielib
 import urllib
 import urllib2
 import time
 import os
 import sys
 import tempfile
 from os.path import join as jn
 from os.path import dirname as dn
 __here__ = os.path.abspath(__file__)
 __root__ = dn(dn(dn(__here__)))
 sys.path.append(__root__)
 from kallithea.lib import vcs
 from kallithea.lib.compat import OrderedSet
 from kallithea.lib.vcs.exceptions import RepositoryError
 PASES = 3
 HOST = 'http://127.0.0.1'
 PORT = 5000
 BASE_URI = '%s:%s/' % (HOST, PORT)
 if len(sys.argv) == 2:
     BASE_URI = sys.argv[1]
 if not BASE_URI.endswith('/'):
     BASE_URI += '/'
 print 'Crawling @ %s' % BASE_URI
 BASE_URI += '%s'
 PROJECT_PATH = jn('/', 'home', 'username', 'repos')
 PROJECTS = [
     #'linux-magx-pbranch',
     'CPython',
     'kallithea',
+]
-cj = cookielib.FileCookieJar('/tmp/rc_test_cookie.txt')
+cj = cookielib.FileCookieJar(jn(tempfile.gettempdir(), 'rc_test_cookie.txt'))
 o = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj))
 o.addheaders = [
     ('User-agent', 'kallithea-crawler'),
     ('Accept-Language', 'en - us, en;q = 0.5')
+]
 urllib2.install_opener(o)
 def _get_repo(proj):
     if isinstance(proj, basestring):
         repo = vcs.get_repo(jn(PROJECT_PATH, proj))
         proj = proj
     else:
         repo = proj
         proj = repo.name
     return repo, proj
 def test_changelog_walk(proj, pages=100):
     repo, proj = _get_repo(proj)
     total_time = 0
     for i in range(1, pages):
         page = '/'.join((proj, 'changelog',))
         full_uri = (BASE_URI % page) + '?' + urllib.urlencode({'page': i})
         s = time.time()
         f = o.open(full_uri)
         assert f.url == full_uri, 'URL:%s does not match %s' % (f.url, full_uri)
         size = len(f.read())
         e = time.time() - s
         total_time += e
         print 'visited %s size:%s req:%s ms' % (full_uri, size, e)
     print 'total_time', total_time
     print 'average on req', total_time / float(pages)
 def test_changeset_walk(proj, limit=None):
     repo, proj = _get_repo(proj)
     print 'processing', jn(PROJECT_PATH, proj)
     total_time = 0
     cnt = 0
     for i in repo:
         cnt += 1
         raw_cs = '/'.join((proj, 'changeset', i.raw_id))
         if limit and limit == cnt:
             break
         full_uri = (BASE_URI % raw_cs)
         print '%s visiting %s\%s' % (cnt, full_uri, i)
         s = time.time()
         f = o.open(full_uri)
         size = len(f.read())
         e = time.time() - s
         total_time += e
         print '%s visited %s\%s size:%s req:%s ms' % (cnt, full_uri, i, size, e)
     print 'total_time', total_time
     print 'average on req', total_time / float(cnt)
 def test_files_walk(proj, limit=100):
     repo, proj = _get_repo(proj)
     print 'processing', jn(PROJECT_PATH, proj)
     total_time = 0
     paths_ = OrderedSet([''])
     try:
         tip = repo.get_changeset('tip')
         for topnode, dirs, files in tip.walk('/'):
             for dir in dirs:
                 paths_.add(dir.path)
                 for f in dir:
                     paths_.add(f.path)
             for f in files:
                 paths_.add(f.path)
     except RepositoryError as e:
         pass
     cnt = 0
     for f in paths_:
         cnt += 1
         if limit and limit == cnt:
             break
         file_path = '/'.join((proj, 'files', 'tip', f))
         full_uri = (BASE_URI % file_path)
         print '%s visiting %s' % (cnt, full_uri)
         s = time.time()
         f = o.open(full_uri)
         size = len(f.read())
         e = time.time() - s
         total_time += e
         print '%s visited OK size:%s req:%s ms' % (cnt, size, e)
     print 'total_time', total_time
     print 'average on req', total_time / float(cnt)
 if __name__ == '__main__':
     for path in PROJECTS:
         repo = vcs.get_repo(jn(PROJECT_PATH, path))
         for i in range(PASES):
             print 'PASS %s/%s' % (i, PASES)
             test_changelog_walk(repo, pages=80)
             test_changeset_walk(repo, limit=100)
             test_files_walk(repo, limit=100)

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
 from os.path import join as jn
 __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'
-TEST_TMP_PATH = os.environ.get('VCS_TEST_ROOT', '/tmp')
+TEST_TMP_PATH = os.environ.get('VCS_TEST_ROOT', tempfile.gettempdir())
 TEST_GIT_REPO = os.environ.get('VCS_TEST_GIT_REPO',
                               jn(TEST_TMP_PATH, 'vcs-git'))
 TEST_GIT_REPO_CLONE = os.environ.get('VCS_TEST_GIT_REPO_CLONE',
                             jn(TEST_TMP_PATH, 'vcsgitclone%s' % uniq_suffix))
 TEST_GIT_REPO_PULL = os.environ.get('VCS_TEST_GIT_REPO_PULL',
                             jn(TEST_TMP_PATH, 'vcsgitpull%s' % uniq_suffix))
 HG_REMOTE_REPO = 'http://bitbucket.org/marcinkuzminski/vcs'
 TEST_HG_REPO = os.environ.get('VCS_TEST_HG_REPO',
                               jn(TEST_TMP_PATH, 'vcs-hg'))
 TEST_HG_REPO_CLONE = os.environ.get('VCS_TEST_HG_REPO_CLONE',
                               jn(TEST_TMP_PATH, 'vcshgclone%s' % uniq_suffix))
 TEST_HG_REPO_PULL = os.environ.get('VCS_TEST_HG_REPO_PULL',
                               jn(TEST_TMP_PATH, 'vcshgpull%s' % uniq_suffix))
 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.
     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)
     # 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 = jn(TEST_TMP_PATH, 'aconfig')
 shutil.copy(jn(THIS, 'aconfig'), _dest)
 TEST_USER_CONFIG_FILE = _dest
 #overide default configurations with kallithea ones
 from kallithea.tests import *

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
 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 = '/tmp/errorrepo'
+        wrong_repo_path = os.path.join(tempfile.gettempdir(), '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
             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)
         if sys.platform == 'win32':
             # windows does not allow '"' in dir names
             tricky_path_2 = get_new_dir("tricky-path-2-repo-$'`")
         else:
             tricky_path_2 = get_new_dir("tricky-path-2-repo-$'\"`")
         successfully_cloned2 = GitRepository(tricky_path_2, src_url=tricky_path, bare=True, create=True)
         # Repo should have been created and thus used correct quoting for clone
         self.assertTrue(successfully_cloned2._repo.bare)
         # Should pass because URL has been properly quoted
         successfully_cloned.pull(tricky_path_2)
         successfully_cloned2.fetch(tricky_path)
     def test_repo_create_with_spaces_in_path(self):
         repo_path = get_new_dir("path with spaces")
         repo = GitRepository(repo_path, src_url=None, bare=True, create=True)
         # Repo should have been created
         self.assertTrue(repo._repo.bare)
     def test_repo_clone(self):
         self.__check_for_existing_repo()
         repo = GitRepository(TEST_GIT_REPO)
         repo_clone = GitRepository(TEST_GIT_REPO_CLONE,
             src_url=TEST_GIT_REPO, create=True, 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_spaces_in_path(self):
         repo_path = get_new_dir("path with spaces")
         successfully_cloned = GitRepository(repo_path, src_url=TEST_GIT_REPO, update_after_clone=True, create=True)
         # Repo should have been created
         self.assertFalse(successfully_cloned._repo.bare)
         successfully_cloned.pull(TEST_GIT_REPO)
         self.repo.fetch(repo_path)
     def test_repo_clone_without_create(self):
         self.assertRaises(RepositoryError, GitRepository,
             TEST_GIT_REPO_CLONE + '_wo_create', src_url=TEST_GIT_REPO)
     def test_repo_clone_with_update(self):
         repo = GitRepository(TEST_GIT_REPO)
         clone_path = TEST_GIT_REPO_CLONE + '_with_update'
         repo_clone = GitRepository(clone_path,
             create=True, src_url=TEST_GIT_REPO, update_after_clone=True)
         self.assertEqual(len(repo.revisions), len(repo_clone.revisions))
         #check if current workdir was updated
         fpath = os.path.join(clone_path, 'MANIFEST.in')
         self.assertEqual(True, os.path.isfile(fpath),
             'Repo was cloned and updated but file %s could not be found'
             % fpath)
     def test_repo_clone_without_update(self):
         repo = GitRepository(TEST_GIT_REPO)
         clone_path = TEST_GIT_REPO_CLONE + '_without_update'
         repo_clone = GitRepository(clone_path,
             create=True, src_url=TEST_GIT_REPO, update_after_clone=False)
         self.assertEqual(len(repo.revisions), len(repo_clone.revisions))
         #check if current workdir was *NOT* updated
         fpath = os.path.join(clone_path, 'MANIFEST.in')
         # Make sure it's not bare repo
         self.assertFalse(repo_clone._repo.bare)
         self.assertEqual(False, os.path.isfile(fpath),
             'Repo was cloned and updated but file %s was found'
             % fpath)
     def test_repo_clone_into_bare_repo(self):
         repo = GitRepository(TEST_GIT_REPO)
         clone_path = TEST_GIT_REPO_CLONE + '_bare.git'
         repo_clone = GitRepository(clone_path, create=True,
             src_url=repo.path, bare=True)
         self.assertTrue(repo_clone._repo.bare)
     def test_create_repo_is_not_bare_by_default(self):
         repo = GitRepository(get_new_dir('not-bare-by-default'), create=True)
         self.assertFalse(repo._repo.bare)
     def test_create_bare_repo(self):
         repo = GitRepository(get_new_dir('bare-repo'), create=True, bare=True)
         self.assertTrue(repo._repo.bare)
     def test_revisions(self):
         # there are 112 revisions (by now)
         # so we can assume they would be available from now on
         subset = set([
             'c1214f7e79e02fc37156ff215cd71275450cffc3',
             '38b5fe81f109cb111f549bfe9bb6b267e10bc557',
             'fa6600f6848800641328adbf7811fd2372c02ab2',
             '102607b09cdd60e2793929c4f90478be29f85a17',
             '49d3fd156b6f7db46313fac355dca1a0b94a0017',
             '2d1028c054665b962fa3d307adfc923ddd528038',
             'd7e0d30fbcae12c90680eb095a4f5f02505ce501',
             'ff7ca51e58c505fec0dd2491de52c622bb7a806b',
             'dd80b0f6cf5052f17cc738c2951c4f2070200d7f',
             '8430a588b43b5d6da365400117c89400326e7992',
             'd955cd312c17b02143c04fa1099a352b04368118',
             'f67b87e5c629c2ee0ba58f85197e423ff28d735b',
             'add63e382e4aabc9e1afdc4bdc24506c269b7618',
             'f298fe1189f1b69779a4423f40b48edf92a703fc',
             'bd9b619eb41994cac43d67cf4ccc8399c1125808',
             '6e125e7c890379446e98980d8ed60fba87d0f6d1',
             'd4a54db9f745dfeba6933bf5b1e79e15d0af20bd',
             '0b05e4ed56c802098dfc813cbe779b2f49e92500',
             '191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',
             '45223f8f114c64bf4d6f853e3c35a369a6305520',
             'ca1eb7957a54bce53b12d1a51b13452f95bc7c7e',
             'f5ea29fc42ef67a2a5a7aecff10e1566699acd68',
             '27d48942240f5b91dfda77accd2caac94708cc7d',
             '622f0eb0bafd619d2560c26f80f09e3b0b0d78af',
             'e686b958768ee96af8029fe19c6050b1a8dd3b2b'])
         self.assertTrue(subset.issubset(set(self.repo.revisions)))
     def test_slicing(self):
         #4 1 5 10 95
         for sfrom, sto, size in [(0, 4, 4), (1, 2, 1), (10, 15, 5),
                                  (10, 20, 10), (5, 100, 95)]:
             revs = list(self.repo[sfrom:sto])
             self.assertEqual(len(revs), size)
             self.assertEqual(revs[0], self.repo.get_changeset(sfrom))
             self.assertEqual(revs[-1], self.repo.get_changeset(sto - 1))
     def test_branches(self):
         # TODO: Need more tests here
         # Removed (those are 'remotes' branches for cloned repo)
         #self.assertTrue('master' in self.repo.branches)
         #self.assertTrue('gittree' in self.repo.branches)
         #self.assertTrue('web-branch' in self.repo.branches)
         for name, id in self.repo.branches.items():
             self.assertTrue(isinstance(
                 self.repo.get_changeset(id), GitChangeset))
     def test_tags(self):
         # TODO: Need more tests here
         self.assertTrue('v0.1.1' in self.repo.tags)
         self.assertTrue('v0.1.2' in self.repo.tags)
         for name, id in self.repo.tags.items():
             self.assertTrue(isinstance(
                 self.repo.get_changeset(id), GitChangeset))
     def _test_single_changeset_cache(self, revision):
         chset = self.repo.get_changeset(revision)
         self.assertTrue(revision in self.repo.changesets)
         self.assertTrue(chset is self.repo.changesets[revision])
     def test_initial_changeset(self):
         id = self.repo.revisions[0]
         init_chset = self.repo.get_changeset(id)
         self.assertEqual(init_chset.message, 'initial import\n')
         self.assertEqual(init_chset.author,
             'Marcin Kuzminski <marcin@python-blog.com>')
         for path in ('vcs/__init__.py',
                      'vcs/backends/BaseRepository.py',
                      'vcs/backends/__init__.py'):
             self.assertTrue(isinstance(init_chset.get_node(path), FileNode))
         for path in ('', 'vcs', 'vcs/backends'):
             self.assertTrue(isinstance(init_chset.get_node(path), DirNode))
         self.assertRaises(NodeDoesNotExistError, init_chset.get_node, path='foobar')
         node = init_chset.get_node('vcs/')
         self.assertTrue(hasattr(node, 'kind'))
         self.assertEqual(node.kind, NodeKind.DIR)
         node = init_chset.get_node('vcs')
         self.assertTrue(hasattr(node, 'kind'))
         self.assertEqual(node.kind, NodeKind.DIR)
         node = init_chset.get_node('vcs/__init__.py')
         self.assertTrue(hasattr(node, 'kind'))
         self.assertEqual(node.kind, NodeKind.FILE)
     def test_not_existing_changeset(self):
         self.assertRaises(RepositoryError, self.repo.get_changeset,
             'f' * 40)
     def test_changeset10(self):
         chset10 = self.repo.get_changeset(self.repo.revisions[9])
         README = """===
 VCS
 ===
 Various Version Control System management abstraction layer for Python.
 Introduction
 ------------
 TODO: To be written...
 """
         node = chset10.get_node('README.rst')
         self.assertEqual(node.kind, NodeKind.FILE)
         self.assertEqual(node.content, README)
 class GitChangesetTest(unittest.TestCase):
     def setUp(self):
         self.repo = GitRepository(TEST_GIT_REPO)
     def test_default_changeset(self):
         tip = self.repo.get_changeset()
         self.assertEqual(tip, self.repo.get_changeset(None))
         self.assertEqual(tip, self.repo.get_changeset('tip'))
     def test_root_node(self):
         tip = self.repo.get_changeset()
         self.assertTrue(tip.root is tip.get_node(''))
     def test_lazy_fetch(self):
         """
         Test if changeset's nodes expands and are cached as we walk through
         the revision. This test is somewhat hard to write as order of tests
         is a key here. Written by running command after command in a shell.
         """
         hex = '2a13f185e4525f9d4b59882791a2d397b90d5ddc'
         self.assertTrue(hex in self.repo.revisions)
         chset = self.repo.get_changeset(hex)
         self.assertTrue(len(chset.nodes) == 0)
         root = chset.root
         self.assertTrue(len(chset.nodes) == 1)
         self.assertTrue(len(root.nodes) == 8)
         # accessing root.nodes updates chset.nodes
         self.assertTrue(len(chset.nodes) == 9)
         docs = root.get_node('docs')
         # we haven't yet accessed anything new as docs dir was already cached
         self.assertTrue(len(chset.nodes) == 9)
         self.assertTrue(len(docs.nodes) == 8)
         # accessing docs.nodes updates chset.nodes
         self.assertTrue(len(chset.nodes) == 17)
         self.assertTrue(docs is chset.get_node('docs'))
         self.assertTrue(docs is root.nodes[0])
         self.assertTrue(docs is root.dirs[0])
         self.assertTrue(docs is chset.get_node('docs'))
     def test_nodes_with_changeset(self):
         hex = '2a13f185e4525f9d4b59882791a2d397b90d5ddc'
         chset = self.repo.get_changeset(hex)
         root = chset.root
         docs = root.get_node('docs')
         self.assertTrue(docs is chset.get_node('docs'))
         api = docs.get_node('api')
         self.assertTrue(api is chset.get_node('docs/api'))
         index = api.get_node('index.rst')
         self.assertTrue(index is chset.get_node('docs/api/index.rst'))
         self.assertTrue(index is chset.get_node('docs') \
             .get_node('api') \
             .get_node('index.rst'))
     def test_branch_and_tags(self):
         """
         rev0 = self.repo.revisions[0]
         chset0 = self.repo.get_changeset(rev0)
         self.assertEqual(chset0.branch, 'master')
         self.assertEqual(chset0.tags, [])
         rev10 = self.repo.revisions[10]
         chset10 = self.repo.get_changeset(rev10)
         self.assertEqual(chset10.branch, 'master')
         self.assertEqual(chset10.tags, [])
         rev44 = self.repo.revisions[44]
         chset44 = self.repo.get_changeset(rev44)
         self.assertEqual(chset44.branch, 'web-branch')
         tip = self.repo.get_changeset('tip')
         self.assertTrue('tip' in tip.tags)
         """
         # Those tests would fail - branches are now going
         # to be changed at main API in order to support git backend
         pass
     def _test_slices(self, limit, offset):
         count = self.repo.count()
         changesets = self.repo.get_changesets(limit=limit, offset=offset)
         idx = 0
         for changeset in changesets:
             rev = offset + idx
             idx += 1
             rev_id = self.repo.revisions[rev]
             if idx > limit:
                 pytest.fail("Exceeded limit already (getting revision %s, "
                     "there are %s total revisions, offset=%s, limit=%s)"
                     % (rev_id, count, offset, limit))
             self.assertEqual(changeset, self.repo.get_changeset(rev_id))
         result = list(self.repo.get_changesets(limit=limit, offset=offset))
         start = offset
         end = limit and offset + limit or None
         sliced = list(self.repo[start:end])
         pytest.failUnlessEqual(result, sliced,
             msg="Comparison failed for limit=%s, offset=%s"
             "(get_changeset returned: %s and sliced: %s"
             % (limit, offset, result, sliced))
     def _test_file_size(self, revision, path, size):
         node = self.repo.get_changeset(revision).get_node(path)
         self.assertTrue(node.is_file())
         self.assertEqual(node.size, size)
     def test_file_size(self):
         to_check = (
             ('c1214f7e79e02fc37156ff215cd71275450cffc3',
                 'vcs/backends/BaseRepository.py', 502),
             ('d7e0d30fbcae12c90680eb095a4f5f02505ce501',
                 'vcs/backends/hg.py', 854),
             ('6e125e7c890379446e98980d8ed60fba87d0f6d1',
                 'setup.py', 1068),
             ('d955cd312c17b02143c04fa1099a352b04368118',
                 'vcs/backends/base.py', 2921),
             ('ca1eb7957a54bce53b12d1a51b13452f95bc7c7e',
                 'vcs/backends/base.py', 3936),
             ('f50f42baeed5af6518ef4b0cb2f1423f3851a941',
                 'vcs/backends/base.py', 6189),
+        )
         for revision, path, size in to_check:
             self._test_file_size(revision, path, size)
     def test_file_history(self):
         # we can only check if those revisions are present in the history
         # as we cannot update this test every time file is changed
         files = {
             'setup.py': [
                 '54386793436c938cff89326944d4c2702340037d',
                 '51d254f0ecf5df2ce50c0b115741f4cf13985dab',
                 '998ed409c795fec2012b1c0ca054d99888b22090',
                 '5e0eb4c47f56564395f76333f319d26c79e2fb09',
                 '0115510b70c7229dbc5dc49036b32e7d91d23acd',
                 '7cb3fd1b6d8c20ba89e2264f1c8baebc8a52d36e',
                 '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                 '191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',
                 'ff7ca51e58c505fec0dd2491de52c622bb7a806b',
             ],
             'vcs/nodes.py': [
                 '33fa3223355104431402a888fa77a4e9956feb3e',
                 'fa014c12c26d10ba682fadb78f2a11c24c8118e1',
                 'e686b958768ee96af8029fe19c6050b1a8dd3b2b',
                 'ab5721ca0a081f26bf43d9051e615af2cc99952f',
                 'c877b68d18e792a66b7f4c529ea02c8f80801542',
                 '4313566d2e417cb382948f8d9d7c765330356054',
                 '6c2303a793671e807d1cfc70134c9ca0767d98c2',
                 '54386793436c938cff89326944d4c2702340037d',
                 '54000345d2e78b03a99d561399e8e548de3f3203',
                 '1c6b3677b37ea064cb4b51714d8f7498f93f4b2b',
                 '2d03ca750a44440fb5ea8b751176d1f36f8e8f46',
                 '2a08b128c206db48c2f0b8f70df060e6db0ae4f8',
                 '30c26513ff1eb8e5ce0e1c6b477ee5dc50e2f34b',
                 'ac71e9503c2ca95542839af0ce7b64011b72ea7c',
                 '12669288fd13adba2a9b7dd5b870cc23ffab92d2',
                 '5a0c84f3e6fe3473e4c8427199d5a6fc71a9b382',
                 '12f2f5e2b38e6ff3fbdb5d722efed9aa72ecb0d5',
                 '5eab1222a7cd4bfcbabc218ca6d04276d4e27378',
                 'f50f42baeed5af6518ef4b0cb2f1423f3851a941',
                 'd7e390a45f6aa96f04f5e7f583ad4f867431aa25',
                 'f15c21f97864b4f071cddfbf2750ec2e23859414',
                 'e906ef056cf539a4e4e5fc8003eaf7cf14dd8ade',
                 'ea2b108b48aa8f8c9c4a941f66c1a03315ca1c3b',
                 '84dec09632a4458f79f50ddbbd155506c460b4f9',
                 '0115510b70c7229dbc5dc49036b32e7d91d23acd',
                 '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                 '3bf1c5868e570e39569d094f922d33ced2fa3b2b',
                 'b8d04012574729d2c29886e53b1a43ef16dd00a1',
                 '6970b057cffe4aab0a792aa634c89f4bebf01441',
                 'dd80b0f6cf5052f17cc738c2951c4f2070200d7f',
                 'ff7ca51e58c505fec0dd2491de52c622bb7a806b',
             ],
             'vcs/backends/git.py': [
                 '4cf116ad5a457530381135e2f4c453e68a1b0105',
                 '9a751d84d8e9408e736329767387f41b36935153',
                 'cb681fb539c3faaedbcdf5ca71ca413425c18f01',
                 '428f81bb652bcba8d631bce926e8834ff49bdcc6',
                 '180ab15aebf26f98f714d8c68715e0f05fa6e1c7',
                 '2b8e07312a2e89e92b90426ab97f349f4bce2a3a',
                 '50e08c506174d8645a4bb517dd122ac946a0f3bf',
                 '54000345d2e78b03a99d561399e8e548de3f3203',
             ],
+        }
         for path, revs in files.items():
             node = self.repo.get_changeset(revs[0]).get_node(path)
             node_revs = [chset.raw_id for chset in node.history]
             self.assertTrue(set(revs).issubset(set(node_revs)),
                 "We assumed that %s is subset of revisions for which file %s "
                 "has been changed, and history of that node returned: %s"
                 % (revs, path, node_revs))
     def test_file_annotate(self):
         files = {
             'vcs/backends/__init__.py': {
                 'c1214f7e79e02fc37156ff215cd71275450cffc3': {
                     'lines_no': 1,
                     'changesets': [
                         'c1214f7e79e02fc37156ff215cd71275450cffc3',
                     ],
                 },
                 '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647': {
                     'lines_no': 21,
                     'changesets': [
                         '49d3fd156b6f7db46313fac355dca1a0b94a0017',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                     ],
                 },
                 'e29b67bd158580fc90fc5e9111240b90e6e86064': {
                     'lines_no': 32,
                     'changesets': [
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '5eab1222a7cd4bfcbabc218ca6d04276d4e27378',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '54000345d2e78b03a99d561399e8e548de3f3203',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '78c3f0c23b7ee935ec276acb8b8212444c33c396',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '78c3f0c23b7ee935ec276acb8b8212444c33c396',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                     ],
                 },
             },
+        }
         for fname, revision_dict in files.items():
             for rev, data in revision_dict.items():
                 cs = self.repo.get_changeset(rev)
                 l1_1 = [x[1] for x in cs.get_file_annotate(fname)]
                 l1_2 = [x[2]().raw_id for x in cs.get_file_annotate(fname)]
                 self.assertEqual(l1_1, l1_2)
                 l1 = l1_1
                 l2 = files[fname][rev]['changesets']
                 self.assertTrue(l1 == l2 , "The lists of revision for %s@rev %s"
                                 "from annotation list should match each other, "
                                 "got \n%s \nvs \n%s " % (fname, rev, l1, l2))
     def test_files_state(self):
         """
         Tests state of FileNodes.
         """
         node = self.repo \
             .get_changeset('e6ea6d16e2f26250124a1f4b4fe37a912f9d86a0') \
             .get_node('vcs/utils/diffs.py')
         self.assertTrue(node.state, NodeState.ADDED)
         self.assertTrue(node.added)
         self.assertFalse(node.changed)
         self.assertFalse(node.not_changed)
         self.assertFalse(node.removed)
         node = self.repo \
             .get_changeset('33fa3223355104431402a888fa77a4e9956feb3e') \
             .get_node('.hgignore')
         self.assertTrue(node.state, NodeState.CHANGED)
         self.assertFalse(node.added)
         self.assertTrue(node.changed)
         self.assertFalse(node.not_changed)
         self.assertFalse(node.removed)
         node = self.repo \
             .get_changeset('e29b67bd158580fc90fc5e9111240b90e6e86064') \
             .get_node('setup.py')
         self.assertTrue(node.state, NodeState.NOT_CHANGED)
         self.assertFalse(node.added)
         self.assertFalse(node.changed)
         self.assertTrue(node.not_changed)
         self.assertFalse(node.removed)
         # If node has REMOVED state then trying to fetch it would raise
         # ChangesetError exception
         chset = self.repo.get_changeset(
             'fa6600f6848800641328adbf7811fd2372c02ab2')
         path = 'vcs/backends/BaseRepository.py'
         self.assertRaises(NodeDoesNotExistError, chset.get_node, path)
         # but it would be one of ``removed`` (changeset's attribute)
         self.assertTrue(path in [rf.path for rf in chset.removed])
         chset = self.repo.get_changeset(
             '54386793436c938cff89326944d4c2702340037d')
         changed = ['setup.py', 'tests/test_nodes.py', 'vcs/backends/hg.py',
             'vcs/nodes.py']
         self.assertEqual(set(changed), set([f.path for f in chset.changed]))
     def test_commit_message_is_unicode(self):
         for cs in self.repo:
             self.assertEqual(type(cs.message), unicode)
     def test_changeset_author_is_unicode(self):
         for cs in self.repo:
             self.assertEqual(type(cs.author), unicode)
     def test_repo_files_content_is_unicode(self):
         changeset = self.repo.get_changeset()
         for node in changeset.get_node('/'):
             if node.is_file():
                 self.assertEqual(type(node.content), unicode)
     def test_wrong_path(self):
         # There is 'setup.py' in the root dir but not there:
         path = 'foo/bar/setup.py'
         tip = self.repo.get_changeset()
         self.assertRaises(VCSError, tip.get_node, path)
     def test_author_email(self):
         self.assertEqual('marcin@python-blog.com',
           self.repo.get_changeset('c1214f7e79e02fc37156ff215cd71275450cffc3') \
           .author_email)
         self.assertEqual('lukasz.balcerzak@python-center.pl',
           self.repo.get_changeset('ff7ca51e58c505fec0dd2491de52c622bb7a806b') \
           .author_email)
         self.assertEqual('',
           self.repo.get_changeset('8430a588b43b5d6da365400117c89400326e7992') \
           .author_email)
     def test_author_username(self):
         self.assertEqual('Marcin Kuzminski',
           self.repo.get_changeset('c1214f7e79e02fc37156ff215cd71275450cffc3') \
           .author_name)
         self.assertEqual('Lukasz Balcerzak',
           self.repo.get_changeset('ff7ca51e58c505fec0dd2491de52c622bb7a806b') \
           .author_name)
         self.assertEqual('marcink none@none',
           self.repo.get_changeset('8430a588b43b5d6da365400117c89400326e7992') \
           .author_name)
 class GitSpecificTest(unittest.TestCase):
     def test_error_is_raised_for_added_if_diff_name_status_is_wrong(self):
         repo = mock.MagicMock()
         changeset = GitChangeset(repo, 'foobar')
         changeset._diff_name_status = 'foobar'
         with self.assertRaises(VCSError):
             changeset.added
     def test_error_is_raised_for_changed_if_diff_name_status_is_wrong(self):
         repo = mock.MagicMock()
         changeset = GitChangeset(repo, 'foobar')
         changeset._diff_name_status = 'foobar'
         with self.assertRaises(VCSError):
             changeset.added
     def test_error_is_raised_for_removed_if_diff_name_status_is_wrong(self):
         repo = mock.MagicMock()
         changeset = GitChangeset(repo, 'foobar')
         changeset._diff_name_status = 'foobar'
         with self.assertRaises(VCSError):
             changeset.added
 class GitSpecificWithRepoTest(_BackendTestMixin, unittest.TestCase):
     backend_alias = 'git'
     @classmethod
     def _get_commits(cls):
         return [
+            {
                 'message': 'Initial',
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode('foobar/static/js/admin/base.js', content='base'),
                     FileNode('foobar/static/admin', content='admin',
                         mode=0120000), # this is a link
                     FileNode('foo', content='foo'),
                 ],
             },
+            {
                 'message': 'Second',
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 22),
                 'added': [
                     FileNode('foo2', content='foo2'),
                 ],
             },
+        ]
     def test_paths_slow_traversing(self):
         cs = self.repo.get_changeset()
         self.assertEqual(cs.get_node('foobar').get_node('static').get_node('js')
             .get_node('admin').get_node('base.js').content, 'base')
     def test_paths_fast_traversing(self):
         cs = self.repo.get_changeset()
         self.assertEqual(cs.get_node('foobar/static/js/admin/base.js').content,
             'base')
     def test_workdir_get_branch(self):
         self.repo.run_git_command(['checkout', '-b', 'production'])
         # Regression test: one of following would fail if we don't check
         # .git/HEAD file
         self.repo.run_git_command(['checkout', 'production'])
         self.assertEqual(self.repo.workdir.get_branch(), 'production')
         self.repo.run_git_command(['checkout', 'master'])
         self.assertEqual(self.repo.workdir.get_branch(), 'master')
     def test_get_diff_runs_git_command_with_hashes(self):
         self.repo.run_git_command = mock.Mock(return_value=['', ''])
         self.repo.get_diff(0, 1)
         self.repo.run_git_command.assert_called_once_with(
             ['diff', '-U3', '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(0), self.repo._get_revision(1)])
     def test_get_diff_runs_git_command_with_str_hashes(self):
         self.repo.run_git_command = mock.Mock(return_value=['', ''])
         self.repo.get_diff(self.repo.EMPTY_CHANGESET, 1)
         self.repo.run_git_command.assert_called_once_with(
             ['show', '-U3', '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(1)])
     def test_get_diff_runs_git_command_with_path_if_its_given(self):
         self.repo.run_git_command = mock.Mock(return_value=['', ''])
         self.repo.get_diff(0, 1, 'foo')
         self.repo.run_git_command.assert_called_once_with(
             ['diff', '-U3', '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'])
 class GitRegressionTest(_BackendTestMixin, unittest.TestCase):
     backend_alias = 'git'
     @classmethod
     def _get_commits(cls):
         return [
+            {
                 'message': 'Initial',
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode('bot/__init__.py', content='base'),
                     FileNode('bot/templates/404.html', content='base'),
                     FileNode('bot/templates/500.html', content='base'),
                 ],
             },
+            {
                 'message': 'Second',
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 22),
                 'added': [
                     FileNode('bot/build/migrations/1.py', content='foo2'),
                     FileNode('bot/build/migrations/2.py', content='foo2'),
                     FileNode('bot/build/static/templates/f.html', content='foo2'),
                     FileNode('bot/build/static/templates/f1.html', content='foo2'),
                     FileNode('bot/build/templates/err.html', content='foo2'),
                     FileNode('bot/build/templates/err2.html', content='foo2'),
                 ],
             },
+        ]
     def test_similar_paths(self):
         cs = self.repo.get_changeset()
         paths = lambda *n:[x.path for x in n]
         self.assertEqual(paths(*cs.get_nodes('bot')), ['bot/build', 'bot/templates', 'bot/__init__.py'])
         self.assertEqual(paths(*cs.get_nodes('bot/build')), ['bot/build/migrations', 'bot/build/static', 'bot/build/templates'])
         self.assertEqual(paths(*cs.get_nodes('bot/build/static')), ['bot/build/static/templates'])
         # this get_nodes below causes troubles !
         self.assertEqual(paths(*cs.get_nodes('bot/build/static/templates')), ['bot/build/static/templates/f.html', 'bot/build/static/templates/f1.html'])
         self.assertEqual(paths(*cs.get_nodes('bot/build/templates')), ['bot/build/templates/err.html', 'bot/build/templates/err2.html'])
         self.assertEqual(paths(*cs.get_nodes('bot/templates/')), ['bot/templates/404.html', 'bot/templates/500.html'])
 class GitHooksTest(unittest.TestCase):
     """
     Tests related to hook functionality of Git repositories.
     """
     def setUp(self):
         # For each run we want a fresh repo.
         self.repo_directory = get_new_dir("githookrepo")
         self.repo = GitRepository(self.repo_directory, create=True)
         # Create a dictionary where keys are hook names, and values are paths to
         # them. Deduplicates code in tests a bit.
         self.hook_directory = self.repo.get_hook_location()
         self.kallithea_hooks = {h: os.path.join(self.hook_directory, h) for h in ("pre-receive", "post-receive")}
     def test_hooks_created_if_missing(self):
         """
         Tests if hooks are installed in repository if they are missing.
         """
         for hook, hook_path in self.kallithea_hooks.iteritems():
             if os.path.exists(hook_path):
                 os.remove(hook_path)
         ScmModel().install_git_hooks(repo=self.repo)
         for hook, hook_path in self.kallithea_hooks.iteritems():
             self.assertTrue(os.path.exists(hook_path))
     def test_kallithea_hooks_updated(self):
         """
         Tests if hooks are updated if they are Kallithea hooks already.
         """
         for hook, hook_path in self.kallithea_hooks.iteritems():
             with open(hook_path, "w") as f:
                 f.write("KALLITHEA_HOOK_VER=0.0.0\nJUST_BOGUS")
         ScmModel().install_git_hooks(repo=self.repo)
         for hook, hook_path in self.kallithea_hooks.iteritems():
             with open(hook_path) as f:
                 self.assertNotIn("JUST_BOGUS", f.read())
     def test_custom_hooks_untouched(self):
         """
         Tests if hooks are left untouched if they are not Kallithea hooks.
         """
         for hook, hook_path in self.kallithea_hooks.iteritems():
             with open(hook_path, "w") as f:
                 f.write("#!/bin/bash\n#CUSTOM_HOOK")
         ScmModel().install_git_hooks(repo=self.repo)
         for hook, hook_path in self.kallithea_hooks.iteritems():
             with open(hook_path) as f:
                 self.assertIn("CUSTOM_HOOK", f.read())
     def test_custom_hooks_forced_update(self):
         """
         Tests if hooks are forcefully updated even though they are custom hooks.
         """
         for hook, hook_path in self.kallithea_hooks.iteritems():
             with open(hook_path, "w") as f:
                 f.write("#!/bin/bash\n#CUSTOM_HOOK")
         ScmModel().install_git_hooks(repo=self.repo, force_create=True)
         for hook, hook_path in self.kallithea_hooks.iteritems():
             with open(hook_path) as f:
                 self.assertIn("KALLITHEA_HOOK_VER", f.read())
 if __name__ == '__main__':
     unittest.main()

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
 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 = '/tmp/errorrepo'
+        wrong_repo_path = os.path.join(tempfile.gettempdir(), '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 \
                                                     + '_w_update',
                                                     'MANIFEST.in')), True,)
     def test_repo_clone_without_update(self):
         repo = MercurialRepository(safe_str(TEST_HG_REPO))
         repo_clone = MercurialRepository(TEST_HG_REPO_CLONE + '_wo_update',
             src_url=TEST_HG_REPO, update_after_clone=False)
         self.assertEqual(len(repo.revisions), len(repo_clone.revisions))
         self.assertEqual(os.path.isfile(os.path.join(TEST_HG_REPO_CLONE \
                                                     + '_wo_update',
                                                     'MANIFEST.in')), False,)
     def test_pull(self):
         if os.path.exists(TEST_HG_REPO_PULL):
             pytest.fail('Cannot test mercurial pull command as location %s '
                       'already exists. You should manually remove it first'
                       % TEST_HG_REPO_PULL)
         repo_new = MercurialRepository(TEST_HG_REPO_PULL, create=True)
         self.assertTrue(len(self.repo.revisions) > len(repo_new.revisions))
         repo_new.pull(self.repo.path)
         repo_new = MercurialRepository(TEST_HG_REPO_PULL)
         self.assertTrue(len(self.repo.revisions) == len(repo_new.revisions))
     def test_revisions(self):
         # there are 21 revisions at bitbucket now
         # so we can assume they would be available from now on
         subset = set(['b986218ba1c9b0d6a259fac9b050b1724ed8e545',
                  '3d8f361e72ab303da48d799ff1ac40d5ac37c67e',
                  '6cba7170863a2411822803fa77a0a264f1310b35',
                  '56349e29c2af3ac913b28bde9a2c6154436e615b',
                  '2dda4e345facb0ccff1a191052dd1606dba6781d',
                  '6fff84722075f1607a30f436523403845f84cd9e',
                  '7d4bc8ec6be56c0f10425afb40b6fc315a4c25e7',
                  '3803844fdbd3b711175fc3da9bdacfcd6d29a6fb',
                  'dc5d2c0661b61928834a785d3e64a3f80d3aad9c',
                  'be90031137367893f1c406e0a8683010fd115b79',
                  'db8e58be770518cbb2b1cdfa69146e47cd481481',
                  '84478366594b424af694a6c784cb991a16b87c21',
                  '17f8e105dddb9f339600389c6dc7175d395a535c',
                  '20a662e756499bde3095ffc9bc0643d1def2d0eb',
                  '2e319b85e70a707bba0beff866d9f9de032aa4f9',
                  '786facd2c61deb9cf91e9534735124fb8fc11842',
                  '94593d2128d38210a2fcd1aabff6dda0d6d9edf8',
                  'aa6a0de05b7612707db567078e130a6cd114a9a7',
                  'eada5a770da98ab0dd7325e29d00e0714f228d09'
                 ])
         self.assertTrue(subset.issubset(set(self.repo.revisions)))
         # check if we have the proper order of revisions
         org = ['b986218ba1c9b0d6a259fac9b050b1724ed8e545',
                 '3d8f361e72ab303da48d799ff1ac40d5ac37c67e',
                 '6cba7170863a2411822803fa77a0a264f1310b35',
                 '56349e29c2af3ac913b28bde9a2c6154436e615b',
                 '2dda4e345facb0ccff1a191052dd1606dba6781d',
                 '6fff84722075f1607a30f436523403845f84cd9e',
                 '7d4bc8ec6be56c0f10425afb40b6fc315a4c25e7',
                 '3803844fdbd3b711175fc3da9bdacfcd6d29a6fb',
                 'dc5d2c0661b61928834a785d3e64a3f80d3aad9c',
                 'be90031137367893f1c406e0a8683010fd115b79',
                 'db8e58be770518cbb2b1cdfa69146e47cd481481',
                 '84478366594b424af694a6c784cb991a16b87c21',
                 '17f8e105dddb9f339600389c6dc7175d395a535c',
                 '20a662e756499bde3095ffc9bc0643d1def2d0eb',
                 '2e319b85e70a707bba0beff866d9f9de032aa4f9',
                 '786facd2c61deb9cf91e9534735124fb8fc11842',
                 '94593d2128d38210a2fcd1aabff6dda0d6d9edf8',
                 'aa6a0de05b7612707db567078e130a6cd114a9a7',
                 'eada5a770da98ab0dd7325e29d00e0714f228d09',
                 '2c1885c735575ca478bf9e17b0029dca68824458',
                 'd9bcd465040bf869799b09ad732c04e0eea99fe9',
                 '469e9c847fe1f6f7a697b8b25b4bc5b48780c1a7',
                 '4fb8326d78e5120da2c7468dcf7098997be385da',
                 '62b4a097164940bd66030c4db51687f3ec035eed',
                 '536c1a19428381cfea92ac44985304f6a8049569',
                 '965e8ab3c44b070cdaa5bf727ddef0ada980ecc4',
                 '9bb326a04ae5d98d437dece54be04f830cf1edd9',
                 'f8940bcb890a98c4702319fbe36db75ea309b475',
                 'ff5ab059786ebc7411e559a2cc309dfae3625a3b',
                 '6b6ad5f82ad5bb6190037671bd254bd4e1f4bf08',
                 'ee87846a61c12153b51543bf860e1026c6d3dcba', ]
         self.assertEqual(org, self.repo.revisions[:31])
     def test_iter_slice(self):
         sliced = list(self.repo[:10])
         itered = list(self.repo)[:10]
         self.assertEqual(sliced, itered)
     def test_slicing(self):
         #4 1 5 10 95
         for sfrom, sto, size in [(0, 4, 4), (1, 2, 1), (10, 15, 5),
                                  (10, 20, 10), (5, 100, 95)]:
             revs = list(self.repo[sfrom:sto])
             self.assertEqual(len(revs), size)
             self.assertEqual(revs[0], self.repo.get_changeset(sfrom))
             self.assertEqual(revs[-1], self.repo.get_changeset(sto - 1))
     def test_branches(self):
         # TODO: Need more tests here
         #active branches
         self.assertTrue('default' in self.repo.branches)
         self.assertTrue('stable' in self.repo.branches)
         # closed
         self.assertTrue('git' in self.repo._get_branches(closed=True))
         self.assertTrue('web' in self.repo._get_branches(closed=True))
         for name, id in self.repo.branches.items():
             self.assertTrue(isinstance(
                 self.repo.get_changeset(id), MercurialChangeset))
     def test_tip_in_tags(self):
         # tip is always a tag
         self.assertIn('tip', self.repo.tags)
     def test_tip_changeset_in_tags(self):
         tip = self.repo.get_changeset()
         self.assertEqual(self.repo.tags['tip'], tip.raw_id)
     def test_initial_changeset(self):
         init_chset = self.repo.get_changeset(0)
         self.assertEqual(init_chset.message, 'initial import')
         self.assertEqual(init_chset.author,
             'Marcin Kuzminski <marcin@python-blog.com>')
         self.assertEqual(sorted(init_chset._file_paths),
             sorted([
                 'vcs/__init__.py',
                 'vcs/backends/BaseRepository.py',
                 'vcs/backends/__init__.py',
             ])
+        )
         self.assertEqual(sorted(init_chset._dir_paths),
             sorted(['', 'vcs', 'vcs/backends']))
         self.assertRaises(NodeDoesNotExistError, init_chset.get_node, path='foobar')
         node = init_chset.get_node('vcs/')
         self.assertTrue(hasattr(node, 'kind'))
         self.assertEqual(node.kind, NodeKind.DIR)
         node = init_chset.get_node('vcs')
         self.assertTrue(hasattr(node, 'kind'))
         self.assertEqual(node.kind, NodeKind.DIR)
         node = init_chset.get_node('vcs/__init__.py')
         self.assertTrue(hasattr(node, 'kind'))
         self.assertEqual(node.kind, NodeKind.FILE)
     def test_not_existing_changeset(self):
         #rawid
         self.assertRaises(RepositoryError, self.repo.get_changeset,
             'abcd' * 10)
         #shortid
         self.assertRaises(RepositoryError, self.repo.get_changeset,
             'erro' * 4)
         #numeric
         self.assertRaises(RepositoryError, self.repo.get_changeset,
             self.repo.count() + 1)
         # Small chance we ever get to this one
         revision = pow(2, 30)
         self.assertRaises(RepositoryError, self.repo.get_changeset, revision)
     def test_changeset10(self):
         chset10 = self.repo.get_changeset(10)
         README = """===
 VCS
 ===
 Various Version Control System management abstraction layer for Python.
 Introduction
 ------------
 TODO: To be written...
 """
         node = chset10.get_node('README.rst')
         self.assertEqual(node.kind, NodeKind.FILE)
         self.assertEqual(node.content, README)
 class MercurialChangesetTest(unittest.TestCase):
     def setUp(self):
         self.repo = MercurialRepository(safe_str(TEST_HG_REPO))
     def _test_equality(self, changeset):
         revision = changeset.revision
         self.assertEqual(changeset, self.repo.get_changeset(revision))
     def test_equality(self):
         self.setUp()
         revs = [0, 10, 20]
         changesets = [self.repo.get_changeset(rev) for rev in revs]
         for changeset in changesets:
             self._test_equality(changeset)
     def test_default_changeset(self):
         tip = self.repo.get_changeset('tip')
         self.assertEqual(tip, self.repo.get_changeset())
         self.assertEqual(tip, self.repo.get_changeset(revision=None))
         self.assertEqual(tip, list(self.repo[-1:])[0])
     def test_root_node(self):
         tip = self.repo.get_changeset('tip')
         self.assertTrue(tip.root is tip.get_node(''))
     def test_lazy_fetch(self):
         """
         Test if changeset's nodes expands and are cached as we walk through
         the revision. This test is somewhat hard to write as order of tests
         is a key here. Written by running command after command in a shell.
         """
         self.setUp()
         chset = self.repo.get_changeset(45)
         self.assertTrue(len(chset.nodes) == 0)
         root = chset.root
         self.assertTrue(len(chset.nodes) == 1)
         self.assertTrue(len(root.nodes) == 8)
         # accessing root.nodes updates chset.nodes
         self.assertTrue(len(chset.nodes) == 9)
         docs = root.get_node('docs')
         # we haven't yet accessed anything new as docs dir was already cached
         self.assertTrue(len(chset.nodes) == 9)
         self.assertTrue(len(docs.nodes) == 8)
         # accessing docs.nodes updates chset.nodes
         self.assertTrue(len(chset.nodes) == 17)
         self.assertTrue(docs is chset.get_node('docs'))
         self.assertTrue(docs is root.nodes[0])
         self.assertTrue(docs is root.dirs[0])
         self.assertTrue(docs is chset.get_node('docs'))
     def test_nodes_with_changeset(self):
         self.setUp()
         chset = self.repo.get_changeset(45)
         root = chset.root
         docs = root.get_node('docs')
         self.assertTrue(docs is chset.get_node('docs'))
         api = docs.get_node('api')
         self.assertTrue(api is chset.get_node('docs/api'))
         index = api.get_node('index.rst')
         self.assertTrue(index is chset.get_node('docs/api/index.rst'))
         self.assertTrue(index is chset.get_node('docs') \
             .get_node('api') \
             .get_node('index.rst'))
     def test_branch_and_tags(self):
         chset0 = self.repo.get_changeset(0)
         self.assertEqual(chset0.branch, 'default')
         self.assertEqual(chset0.tags, [])
         chset10 = self.repo.get_changeset(10)
         self.assertEqual(chset10.branch, 'default')
         self.assertEqual(chset10.tags, [])
         chset44 = self.repo.get_changeset(44)
         self.assertEqual(chset44.branch, 'web')
         tip = self.repo.get_changeset('tip')
         self.assertTrue('tip' in tip.tags)
     def _test_file_size(self, revision, path, size):
         node = self.repo.get_changeset(revision).get_node(path)
         self.assertTrue(node.is_file())
         self.assertEqual(node.size, size)
     def test_file_size(self):
         to_check = (
             (10, 'setup.py', 1068),
             (20, 'setup.py', 1106),
             (60, 'setup.py', 1074),
             (10, 'vcs/backends/base.py', 2921),
             (20, 'vcs/backends/base.py', 3936),
             (60, 'vcs/backends/base.py', 6189),
+        )
         for revision, path, size in to_check:
             self._test_file_size(revision, path, size)
     def test_file_history(self):
         # we can only check if those revisions are present in the history
         # as we cannot update this test every time file is changed
         files = {
             'setup.py': [7, 18, 45, 46, 47, 69, 77],
             'vcs/nodes.py': [7, 8, 24, 26, 30, 45, 47, 49, 56, 57, 58, 59, 60,
 , 73, 76],
             'vcs/backends/hg.py': [4, 5, 6, 11, 12, 13, 14, 15, 16, 21, 22, 23,
 , 27, 28, 30, 31, 33, 35, 36, 37, 38, 39, 40, 41, 44, 45, 47,
 , 49, 53, 54, 55, 58, 60, 61, 67, 68, 69, 70, 73, 77, 78, 79,
 ],
+        }
         for path, revs in files.items():
             tip = self.repo.get_changeset(revs[-1])
             node = tip.get_node(path)
             node_revs = [chset.revision for chset in node.history]
             self.assertTrue(set(revs).issubset(set(node_revs)),
                 "We assumed that %s is subset of revisions for which file %s "
                 "has been changed, and history of that node returned: %s"
                 % (revs, path, node_revs))
     def test_file_annotate(self):
         files = {
                  'vcs/backends/__init__.py':
                   {89: {'lines_no': 31,
                         'changesets': [32, 32, 61, 32, 32, 37, 32, 32, 32, 44,
 , 37, 37, 37, 45, 37, 44, 37, 37, 37,
 , 32, 32, 32, 37, 32, 37, 37, 32,
 , 32]},
 : {'lines_no': 1,
                         'changesets': [4]},
 : {'lines_no': 31,
                         'changesets': [32, 32, 45, 32, 32, 37, 32, 32, 32, 44,
 , 37, 37, 37, 45, 37, 44, 37, 37, 37,
 , 32, 32, 32, 37, 32, 37, 37, 32,
 , 32]}},
                  'vcs/exceptions.py':
                  {89: {'lines_no': 18,
                        'changesets': [16, 16, 16, 16, 16, 16, 16, 16, 16, 16,
 , 16, 17, 16, 16, 18, 18, 18]},
 : {'lines_no': 18,
                        'changesets': [16, 16, 16, 16, 16, 16, 16, 16, 16, 16,
 , 16, 17, 16, 16, 18, 18, 18]},
 : {'lines_no': 18, 'changesets': [16, 16, 16, 16, 16, 16,
 , 16, 16, 16, 16, 16,
 , 16, 16, 18, 18, 18]}},
                  'MANIFEST.in': {89: {'lines_no': 5,
                                       'changesets': [7, 7, 7, 71, 71]},
 : {'lines_no': 3,
                                       'changesets': [7, 7, 7]},
 : {'lines_no': 3,
                                      'changesets': [7, 7, 7]}}}
         for fname, revision_dict in files.items():
             for rev, data in revision_dict.items():
                 cs = self.repo.get_changeset(rev)
                 l1_1 = [x[1] for x in cs.get_file_annotate(fname)]
                 l1_2 = [x[2]().raw_id for x in cs.get_file_annotate(fname)]
                 self.assertEqual(l1_1, l1_2)
                 l1 = l1_2 = [x[2]().revision for x in cs.get_file_annotate(fname)]
                 l2 = files[fname][rev]['changesets']
                 self.assertTrue(l1 == l2 , "The lists of revision for %s@rev%s"
                                 "from annotation list should match each other,"
                                 "got \n%s \nvs \n%s " % (fname, rev, l1, l2))
     def test_changeset_state(self):
         """
         Tests which files have been added/changed/removed at particular revision
         """
         # rev 46ad32a4f974:
         # hg st --rev 46ad32a4f974
         #    changed: 13
         #    added:   20
         #    removed: 1
         changed = set(['.hgignore'
             , 'README.rst' , 'docs/conf.py' , 'docs/index.rst' , 'setup.py'
             , 'tests/test_hg.py' , 'tests/test_nodes.py' , 'vcs/__init__.py'
             , 'vcs/backends/__init__.py' , 'vcs/backends/base.py'
             , 'vcs/backends/hg.py' , 'vcs/nodes.py' , 'vcs/utils/__init__.py'])
         added = set(['docs/api/backends/hg.rst'
             , 'docs/api/backends/index.rst' , 'docs/api/index.rst'
             , 'docs/api/nodes.rst' , 'docs/api/web/index.rst'
             , 'docs/api/web/simplevcs.rst' , 'docs/installation.rst'
             , 'docs/quickstart.rst' , 'setup.cfg' , 'vcs/utils/baseui_config.py'
             , 'vcs/utils/web.py' , 'vcs/web/__init__.py' , 'vcs/web/exceptions.py'
             , 'vcs/web/simplevcs/__init__.py' , 'vcs/web/simplevcs/exceptions.py'
             , 'vcs/web/simplevcs/middleware.py' , 'vcs/web/simplevcs/models.py'
             , 'vcs/web/simplevcs/settings.py' , 'vcs/web/simplevcs/utils.py'
             , 'vcs/web/simplevcs/views.py'])
         removed = set(['docs/api.rst'])
         chset64 = self.repo.get_changeset('46ad32a4f974')
         self.assertEqual(set((node.path for node in chset64.added)), added)
         self.assertEqual(set((node.path for node in chset64.changed)), changed)
         self.assertEqual(set((node.path for node in chset64.removed)), removed)
         # rev b090f22d27d6:
         # hg st --rev b090f22d27d6
         #    changed: 13
         #    added:   20
         #    removed: 1
         chset88 = self.repo.get_changeset('b090f22d27d6')
         self.assertEqual(set((node.path for node in chset88.added)), set())
         self.assertEqual(set((node.path for node in chset88.changed)),
             set(['.hgignore']))
         self.assertEqual(set((node.path for node in chset88.removed)), set())
+#
         # 85:
         #    added:   2 ['vcs/utils/diffs.py', 'vcs/web/simplevcs/views/diffs.py']
         #    changed: 4 ['vcs/web/simplevcs/models.py', ...]
         #    removed: 1 ['vcs/utils/web.py']
         chset85 = self.repo.get_changeset(85)
         self.assertEqual(set((node.path for node in chset85.added)), set([
             'vcs/utils/diffs.py',
             'vcs/web/simplevcs/views/diffs.py']))
         self.assertEqual(set((node.path for node in chset85.changed)), set([
             'vcs/web/simplevcs/models.py',
             'vcs/web/simplevcs/utils.py',
             'vcs/web/simplevcs/views/__init__.py',
             'vcs/web/simplevcs/views/repository.py',
             ]))
         self.assertEqual(set((node.path for node in chset85.removed)),
             set(['vcs/utils/web.py']))
     def test_files_state(self):
         """
         Tests state of FileNodes.
         """
         chset = self.repo.get_changeset(85)
         node = chset.get_node('vcs/utils/diffs.py')
         self.assertTrue(node.state, NodeState.ADDED)
         self.assertTrue(node.added)
         self.assertFalse(node.changed)
         self.assertFalse(node.not_changed)
         self.assertFalse(node.removed)
         chset = self.repo.get_changeset(88)
         node = chset.get_node('.hgignore')
         self.assertTrue(node.state, NodeState.CHANGED)
         self.assertFalse(node.added)
         self.assertTrue(node.changed)
         self.assertFalse(node.not_changed)
         self.assertFalse(node.removed)
         chset = self.repo.get_changeset(85)
         node = chset.get_node('setup.py')
         self.assertTrue(node.state, NodeState.NOT_CHANGED)
         self.assertFalse(node.added)
         self.assertFalse(node.changed)
         self.assertTrue(node.not_changed)
         self.assertFalse(node.removed)
         # If node has REMOVED state then trying to fetch it would raise
         # ChangesetError exception
         chset = self.repo.get_changeset(2)
         path = 'vcs/backends/BaseRepository.py'
         self.assertRaises(NodeDoesNotExistError, chset.get_node, path)
         # but it would be one of ``removed`` (changeset's attribute)
         self.assertTrue(path in [rf.path for rf in chset.removed])
     def test_commit_message_is_unicode(self):
         for cm in self.repo:
             self.assertEqual(type(cm.message), unicode)
     def test_changeset_author_is_unicode(self):
         for cm in self.repo:
             self.assertEqual(type(cm.author), unicode)
     def test_repo_files_content_is_unicode(self):
         test_changeset = self.repo.get_changeset(100)
         for node in test_changeset.get_node('/'):
             if node.is_file():
                 self.assertEqual(type(node.content), unicode)
     def test_wrong_path(self):
         # There is 'setup.py' in the root dir but not there:
         path = 'foo/bar/setup.py'
         self.assertRaises(VCSError, self.repo.get_changeset().get_node, path)
     def test_archival_file(self):
         #TODO:
         pass
     def test_archival_as_generator(self):
         #TODO:
         pass
     def test_archival_wrong_kind(self):
         tip = self.repo.get_changeset()
         self.assertRaises(VCSError, tip.fill_archive, kind='error')
     def test_archival_empty_prefix(self):
         #TODO:
         pass
     def test_author_email(self):
         self.assertEqual('marcin@python-blog.com',
                          self.repo.get_changeset('b986218ba1c9').author_email)
         self.assertEqual('lukasz.balcerzak@python-center.pl',
                          self.repo.get_changeset('3803844fdbd3').author_email)
         self.assertEqual('',
                          self.repo.get_changeset('84478366594b').author_email)
     def test_author_username(self):
         self.assertEqual('Marcin Kuzminski',
                          self.repo.get_changeset('b986218ba1c9').author_name)
         self.assertEqual('Lukasz Balcerzak',
                          self.repo.get_changeset('3803844fdbd3').author_name)
         self.assertEqual('marcink',
                          self.repo.get_changeset('84478366594b').author_name)

0 comments (0 inline, 0 general)