Changeset - 21307b019ad4
[Not reviewed]
beta
0 1 0
Marcin Kuzminski - 13 years ago 2013-01-03 00:44:06
marcin@python-works.com
added integration tests for IP restriction option
1 file changed with 39 insertions and 0 deletions:
0 comments (0 inline, 0 general)
rhodecode/tests/scripts/test_vcs_operations.py
Show inline comments
 
@@ -18,48 +18,49 @@
 
# 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/>.
 

	
 
import os
 
import tempfile
 
import unittest
 
from os.path import join as jn
 
from os.path import dirname as dn
 

	
 
from tempfile import _RandomNameSequence
 
from subprocess import Popen, PIPE
 

	
 
from rhodecode.tests import *
 
from rhodecode.model.db import User, Repository, UserLog
 
from rhodecode.model.meta import Session
 
from rhodecode.model.repo import RepoModel
 
from rhodecode.model.user import UserModel
 

	
 
DEBUG = True
 
HOST = '127.0.0.1:5000'  # 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, stderr
 
        return stdout, stderr
 

	
 
@@ -399,24 +400,62 @@ class TestVCSOperations(unittest.TestCas
 

	
 
#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)
 
#
 
#        #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)
 
#        #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()
 
        new_ip = 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)
 
        assert 'abort: HTTP Error 403: Forbidden' in stderr
 

	
 
        #release IP restrictions
 
        clone_url = _construct_url(HG_REPO)
 
        user_model.delete_extra_ip(TEST_USER_ADMIN_LOGIN, new_ip.ip_id)
 
        Session().commit()
 
        stdout, stderr = Command('/tmp').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()
 
        new_ip = 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)
 
        assert 'error: The requested URL returned error: 403 Forbidden' in stderr
 

	
 
        #release IP restrictions
 
        clone_url = _construct_url(GIT_REPO)
 
        user_model.delete_extra_ip(TEST_USER_ADMIN_LOGIN, new_ip.ip_id)
 
        Session().commit()
 
        stdout, stderr = Command('/tmp').execute('git clone', clone_url)
 

	
 
        assert 'Cloning into' in stdout
 
        assert stderr == ''
0 comments (0 inline, 0 general)