kallithea Changeset - d7cdef68240b

Changeset - d7cdef68240b

Parent rev.

Child rev.

[Not reviewed]

default

0 2 0

domruf - 10 years ago 2016-05-02 22:41:51
dominikruf@gmail.com

tests: use webserver fixture for running vcs tests, removing need for serving manually

So far a bit hack-ish.

2 files changed with 32 insertions and 35 deletions:

kallithea/tests/conftest.py

kallithea/tests/other/manual_test_vcs_operations.py

0 comments (0 inline, 0 general)

kallithea/tests/conftest.py

➞

Show inline comments

@@ @@ -108,51 +108,55 @@ def auto_clear_ip_permissions(): @@
     user_model = UserModel()
     user_ids = []
     user_ids.append(User.get_default_user().user_id)
     user_ids.append(User.get_by_username(TEST_USER_REGULAR_LOGIN).user_id)
     for user_id in user_ids:
         for ip in UserIpMap.query().filter(UserIpMap.user_id == user_id):
             user_model.delete_extra_ip(user_id, ip.ip_id)
     # IP permissions are cached, need to invalidate this cache explicitly
     invalidate_all_caches()
 @pytest.fixture
 def test_context_fixture(app_fixture):
     """
     Encompass the entire test using this fixture in a test_context,
     making sure that certain functionality still works even if no call to
     self.app.get/post has been made.
     The typical error message indicating you need a test_context is:
         TypeError: No object (name: context) has been registered for this thread
     The standard way to fix this is simply using the test_context context
     manager directly inside your test:
         with test_context(self.app):
             <actions>
     but if test setup code (xUnit-style or pytest fixtures) also needs to be
     executed inside the test context, that method is not possible.
     Even if there is no such setup code, the fixture may reduce code complexity
     if the entire test needs to run inside a test context.
     To apply this fixture (like any other fixture) to all test methods of a
     class, use the following class decorator:
         @pytest.mark.usefixtures("test_context_fixture")
         class TestFoo(TestController):
             ...
     """
     with test_context(app_fixture):
         yield
 @pytest.yield_fixture(scope="session")
 def webserver():
     """Start web server while tests are running.
     Useful for debugging and necessary for vcs operation tests."""
     server = WSGIServer(application=kallithea.tests.base.testapp)
     server.start()
     # temporary hack for using this server for the vcs tests:
     from kallithea.tests.other import manual_test_vcs_operations
     manual_test_vcs_operations.HOST = '%s:%s' % server.server_address
     yield server
     server.stop()

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.
 Test suite for vcs push/pull operations.
 Run it in two terminals::
  gearbox serve -c kallithea/tests/test.ini
  KALLITHEA_WHOOSH_TEST_DISABLE=1 KALLITHEA_NO_TMP_PATH=1 py.test kallithea/tests/other/manual_test_vcs_operations.py
 You must have git > 1.8.1 for tests to work fine
 The tests need Git > 1.8.1.
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Dec 30, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import re
 import tempfile
 import time
 import pytest
 from tempfile import _RandomNameSequence
 from subprocess import Popen, PIPE
 from kallithea.tests.base import *
 from kallithea.tests.fixture import Fixture
 from kallithea.model.db import User, Repository, UserIpMap, CacheInvalidation
 from kallithea.model.meta import Session
 from kallithea.model.repo import RepoModel
 from kallithea.model.user import UserModel
 DEBUG = True
 HOST = '127.0.0.1:4999'  # test host
 fixture = Fixture()
 class Command(object):
     def __init__(self, cwd):
         self.cwd = cwd
     def execute(self, cmd, *args, **environ):
         """
         Runs command on the system with given ``args``.
         """
         command = cmd + ' ' + ' '.join(args)
         ignoreReturnCode = environ.pop('ignoreReturnCode', False)
         if DEBUG:
             print '*** CMD %s ***' % command
         testenv = dict(os.environ)
         testenv['LANG'] = 'en_US.UTF-8'
@@ @@ -154,418 +147,418 @@ def _add_files_and_push(vcs, DEST, ignor @@
         stdout, stderr = Command(cwd).execute('hg push --verbose', clone_url, ignoreReturnCode=ignoreReturnCode)
     elif vcs == 'git':
         stdout, stderr = Command(cwd).execute('git push --verbose', clone_url, "master", ignoreReturnCode=ignoreReturnCode)
     return stdout, stderr
 def set_anonymous_access(enable=True):
     user = User.get_default_user()
     user.active = enable
     Session().commit()
     print '\tanonymous access is now:', enable
     if enable != User.get_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
 @pytest.mark.usefixtures("test_context_fixture")
 class TestVCSOperations(TestController):
     @classmethod
     def setup_class(cls):
         #DISABLE ANONYMOUS ACCESS
         set_anonymous_access(False)
     def setup_method(self, method):
         r = Repository.get_by_repo_name(GIT_REPO)
         Repository.unlock(r)
         r.enable_locking = False
         Session().commit()
         r = Repository.get_by_repo_name(HG_REPO)
         Repository.unlock(r)
         r.enable_locking = False
         Session().commit()
     def test_clone_hg_repo_by_admin(self):
+    def test_clone_hg_repo_by_admin(self, webserver):
         clone_url = _construct_url(HG_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir())
         assert 'requesting all changes' in stdout
         assert 'adding changesets' in stdout
         assert 'adding manifests' in stdout
         assert 'adding file changes' in stdout
         assert stderr == ''
     def test_clone_git_repo_by_admin(self):
+    def test_clone_git_repo_by_admin(self, webserver):
         clone_url = _construct_url(GIT_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir())
         assert 'Cloning into' in stdout + stderr
         assert stderr == '' or stdout == ''
     def test_clone_wrong_credentials_hg(self):
+    def test_clone_wrong_credentials_hg(self, webserver):
         clone_url = _construct_url(HG_REPO, passwd='bad!')
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
         assert 'abort: authorization failed' in stderr
     def test_clone_wrong_credentials_git(self):
+    def test_clone_wrong_credentials_git(self, webserver):
         clone_url = _construct_url(GIT_REPO, passwd='bad!')
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
         assert 'fatal: Authentication failed' in stderr
     def test_clone_git_dir_as_hg(self):
+    def test_clone_git_dir_as_hg(self, webserver):
         clone_url = _construct_url(GIT_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
         assert 'HTTP Error 404: Not Found' in stderr
     def test_clone_hg_repo_as_git(self):
+    def test_clone_hg_repo_as_git(self, webserver):
         clone_url = _construct_url(HG_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
         assert 'not found' in stderr
     def test_clone_non_existing_path_hg(self):
+    def test_clone_non_existing_path_hg(self, webserver):
         clone_url = _construct_url('trololo')
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
         assert 'HTTP Error 404: Not Found' in stderr
     def test_clone_non_existing_path_git(self):
+    def test_clone_non_existing_path_git(self, webserver):
         clone_url = _construct_url('trololo')
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
         assert 'not found' in stderr
     def test_push_new_file_hg(self):
+    def test_push_new_file_hg(self, webserver):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(HG_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)
         fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())
         fixture.create_fork(HG_REPO, fork_name)
         clone_url = _construct_url(fork_name).split()[0]
         stdout, stderr = _add_files_and_push('hg', DEST, clone_url=clone_url)
         assert 'pushing to' in stdout
         assert 'Repository size' in stdout
         assert 'Last revision is now' in stdout
     def test_push_new_file_git(self):
+    def test_push_new_file_git(self, webserver):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(GIT_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)
         # commit some stuff into this repo
         fork_name = '%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())
         fixture.create_fork(GIT_REPO, fork_name)
         clone_url = _construct_url(fork_name).split()[0]
         stdout, stderr = _add_files_and_push('git', DEST, clone_url=clone_url)
         print [(x.repo_full_path,x.repo_path) for x in Repository.query()] # TODO: what is this for
         _check_proper_git_push(stdout, stderr)
     def test_push_invalidates_cache_hg(self):
+    def test_push_invalidates_cache_hg(self, webserver):
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                ==HG_REPO).scalar()
         if not key:
             key = CacheInvalidation(HG_REPO, HG_REPO)
             Session().add(key)
         key.cache_active = True
         Session().commit()
         DEST = _get_tmp_dir()
         clone_url = _construct_url(HG_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)
         fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())
         fixture.create_fork(HG_REPO, fork_name)
         clone_url = _construct_url(fork_name).split()[0]
         stdout, stderr = _add_files_and_push('hg', DEST, files_no=1, clone_url=clone_url)
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                ==fork_name).all()
         assert key == []
     def test_push_invalidates_cache_git(self):
+    def test_push_invalidates_cache_git(self, webserver):
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                ==GIT_REPO).scalar()
         if not key:
             key = CacheInvalidation(GIT_REPO, GIT_REPO)
             Session().add(key)
         key.cache_active = True
         Session().commit()
         DEST = _get_tmp_dir()
         clone_url = _construct_url(GIT_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)
         # commit some stuff into this repo
         fork_name = '%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())
         fixture.create_fork(GIT_REPO, fork_name)
         clone_url = _construct_url(fork_name).split()[0]
         stdout, stderr = _add_files_and_push('git', DEST, files_no=1, clone_url=clone_url)
         _check_proper_git_push(stdout, stderr)
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                ==fork_name).all()
         assert key == []
     def test_push_wrong_credentials_hg(self):
+    def test_push_wrong_credentials_hg(self, webserver):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(HG_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)
         stdout, stderr = _add_files_and_push('hg', DEST, user='bad',
                                              passwd='name', ignoreReturnCode=True)
         assert 'abort: authorization failed' in stderr
     def test_push_wrong_credentials_git(self):
+    def test_push_wrong_credentials_git(self, webserver):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(GIT_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)
         stdout, stderr = _add_files_and_push('git', DEST, user='bad',
                                              passwd='name', ignoreReturnCode=True)
         assert 'fatal: Authentication failed' in stderr
     def test_push_back_to_wrong_url_hg(self):
+    def test_push_back_to_wrong_url_hg(self, webserver):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(HG_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)
         stdout, stderr = _add_files_and_push('hg', DEST,
                                     clone_url='http://%s/tmp' % HOST,
                                     ignoreReturnCode = True)
         assert 'HTTP Error 404: Not Found' in stderr
     def test_push_back_to_wrong_url_git(self):
+    def test_push_back_to_wrong_url_git(self, webserver):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(GIT_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)
         stdout, stderr = _add_files_and_push('git', DEST,
                                     clone_url='http://%s/tmp' % HOST,
                                     ignoreReturnCode = True)
         assert 'not found' in stderr
     def test_clone_and_create_lock_hg(self):
+    def test_clone_and_create_lock_hg(self, webserver):
         # enable locking
         r = Repository.get_by_repo_name(HG_REPO)
         r.enable_locking = True
         Session().commit()
         # clone
         clone_url = _construct_url(HG_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir())
         #check if lock was made
         r = Repository.get_by_repo_name(HG_REPO)
         assert r.locked[0] == User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id
     def test_clone_and_create_lock_git(self):
+    def test_clone_and_create_lock_git(self, webserver):
         # enable locking
         r = Repository.get_by_repo_name(GIT_REPO)
         r.enable_locking = True
         Session().commit()
         # clone
         clone_url = _construct_url(GIT_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir())
         #check if lock was made
         r = Repository.get_by_repo_name(GIT_REPO)
         assert r.locked[0] == User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id
     def test_clone_after_repo_was_locked_hg(self):
+    def test_clone_after_repo_was_locked_hg(self, webserver):
         #lock repo
         r = Repository.get_by_repo_name(HG_REPO)
         Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)
         #pull fails since repo is locked
         clone_url = _construct_url(HG_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
         msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""
                 % (HG_REPO, TEST_USER_ADMIN_LOGIN))
         assert msg in stderr
     def test_clone_after_repo_was_locked_git(self):
+    def test_clone_after_repo_was_locked_git(self, webserver):
         #lock repo
         r = Repository.get_by_repo_name(GIT_REPO)
         Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)
         #pull fails since repo is locked
         clone_url = _construct_url(GIT_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
         msg = ("""The requested URL returned error: 423""")
         assert msg in stderr
     def test_push_on_locked_repo_by_other_user_hg(self):
+    def test_push_on_locked_repo_by_other_user_hg(self, webserver):
         #clone some temp
         DEST = _get_tmp_dir()
         clone_url = _construct_url(HG_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)
         #lock repo
         r = Repository.get_by_repo_name(HG_REPO)
         # let this user actually push !
         RepoModel().grant_user_permission(repo=r, user=TEST_USER_REGULAR_LOGIN,
                                           perm='repository.write')
         Session().commit()
         Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)
         #push fails repo is locked by other user !
         stdout, stderr = _add_files_and_push('hg', DEST,
                                              user=TEST_USER_REGULAR_LOGIN,
                                              passwd=TEST_USER_REGULAR_PASS,
                                              ignoreReturnCode=True)
         msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""
                 % (HG_REPO, TEST_USER_ADMIN_LOGIN))
         assert msg in stderr
     def test_push_on_locked_repo_by_other_user_git(self):
+    def test_push_on_locked_repo_by_other_user_git(self, webserver):
         # Note: Git hooks must be executable on unix. This test will thus fail
         # for example on Linux if /tmp is mounted noexec.
         #clone some temp
         DEST = _get_tmp_dir()
         clone_url = _construct_url(GIT_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)
         #lock repo
         r = Repository.get_by_repo_name(GIT_REPO)
         # let this user actually push !
         RepoModel().grant_user_permission(repo=r, user=TEST_USER_REGULAR_LOGIN,
                                           perm='repository.write')
         Session().commit()
         Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)
         #push fails repo is locked by other user !
         stdout, stderr = _add_files_and_push('git', DEST,
                                              user=TEST_USER_REGULAR_LOGIN,
                                              passwd=TEST_USER_REGULAR_PASS,
                                              ignoreReturnCode=True)
         err = 'Repository `%s` locked by user `%s`' % (GIT_REPO, TEST_USER_ADMIN_LOGIN)
         assert err in stderr
         #TODO: fix this somehow later on Git, Git is stupid and even if we throw
         #back 423 to it, it makes ANOTHER request and we fail there with 405 :/
         msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""
                 % (GIT_REPO, TEST_USER_ADMIN_LOGIN))
         #msg = "405 Method Not Allowed"
         #assert msg in stderr
     def test_push_unlocks_repository_hg(self):
+    def test_push_unlocks_repository_hg(self, webserver):
         # enable locking
         fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())
         fixture.create_fork(HG_REPO, fork_name)
         r = Repository.get_by_repo_name(fork_name)
         r.enable_locking = True
         Session().commit()
         #clone some temp
         DEST = _get_tmp_dir()
         clone_url = _construct_url(fork_name)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)
         #check for lock repo after clone
         r = Repository.get_by_repo_name(fork_name)
         uid = User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id
         assert r.locked[0] == uid
         #push is ok and repo is now unlocked
         stdout, stderr = _add_files_and_push('hg', DEST, clone_url=clone_url.split()[0])
         assert str('remote: Released lock on repo `%s`' % fork_name) in stdout
         #we need to cleanup the Session Here !
         Session.remove()
         r = Repository.get_by_repo_name(fork_name)
         assert r.locked == [None, None]
     #TODO: fix me ! somehow during tests hooks don't get called on Git
     def test_push_unlocks_repository_git(self):
+    def test_push_unlocks_repository_git(self, webserver):
         # enable locking
         fork_name = '%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())
         fixture.create_fork(GIT_REPO, fork_name)
         r = Repository.get_by_repo_name(fork_name)
         r.enable_locking = True
         Session().commit()
         #clone some temp
         DEST = _get_tmp_dir()
         clone_url = _construct_url(fork_name)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)
         #check for lock repo after clone
         r = Repository.get_by_repo_name(fork_name)
         assert r.locked[0] == User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id
         #push is ok and repo is now unlocked
         stdout, stderr = _add_files_and_push('git', DEST, clone_url=clone_url.split()[0])
         _check_proper_git_push(stdout, stderr)
         assert ('remote: Released lock on repo `%s`' % fork_name) in stderr
         #we need to cleanup the Session Here !
         Session.remove()
         r = Repository.get_by_repo_name(fork_name)
         assert r.locked == [None, None]
     def test_ip_restriction_hg(self):
+    def test_ip_restriction_hg(self, webserver):
         user_model = UserModel()
         try:
             user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')
             Session().commit()
             clone_url = _construct_url(HG_REPO)
             stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
             assert 'abort: HTTP Error 403: Forbidden' in stderr
         finally:
             #release IP restrictions
             for ip in UserIpMap.query():
                 UserIpMap.delete(ip.ip_id)
             Session().commit()
         # IP permissions are cached, need to wait for the cache in the server process to expire
         time.sleep(1.5)
         clone_url = _construct_url(HG_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir())
         assert 'requesting all changes' in stdout
         assert 'adding changesets' in stdout
         assert 'adding manifests' in stdout
         assert 'adding file changes' in stdout
         assert stderr == ''
     def test_ip_restriction_git(self):
+    def test_ip_restriction_git(self, webserver):
         user_model = UserModel()
         try:
             user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')
             Session().commit()
             clone_url = _construct_url(GIT_REPO)
             stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
             # The message apparently changed in Git 1.8.3, so match it loosely.
             assert re.search(r'\b403\b', stderr)
         finally:
             #release IP restrictions
             for ip in UserIpMap.query():
                 UserIpMap.delete(ip.ip_id)
             Session().commit()
         # IP permissions are cached, need to wait for the cache in the server process to expire
         time.sleep(1.5)
         clone_url = _construct_url(GIT_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir())
         assert 'Cloning into' in stdout + stderr
         assert stderr == '' or stdout == ''

0 comments (0 inline, 0 general)