Changeset - 335241410ee6
Parent rev.
Child rev.
[Not reviewed]
default
0 1 0
domruf - 8 years ago 2018-03-05 21:13:20
dominikruf@gmail.com
Grafted from: ed6bf72cc873
tests: change vcs repo name strings to unicode

This fixes sqlalchemy warnings like
SAWarning: Unicode type received non-unicode bind param value 'vcs_test_git_forkSeWYRg'.
1 file changed with 8 insertions and 8 deletions:
kallithea/tests/other/test_vcs_operations.py
8
8
0 comments (0 inline, 0 general)
kallithea/tests/other/test_vcs_operations.py
➞
Show inline comments
 
@@ -190,27 +190,27 @@ class TestVCSOperations(TestController):
 
    def testhook_cleanup(self):
 
        yield
 
        # remove hook
 
        for hook in ['prechangegroup', 'pretxnchangegroup', 'preoutgoing', 'changegroup', 'outgoing', 'incoming']:
 
            entry = Ui.get_by_key('hooks', '%s.testhook' % hook)
 
            if entry:
 
                Session().delete(entry)
 
        Session().commit()
 

			




	
	
	
		
 
    @pytest.fixture(scope="module")

			




	
	
	
		
 
    def testfork(self):

			




	
	
	
		
 
        # create fork so the repo stays untouched

			




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

			




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

			




	
	
	
		
 
        fixture.create_fork(GIT_REPO, git_fork_name)

			




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

			




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

			




	
	
	
		
 
        fixture.create_fork(HG_REPO, hg_fork_name)

			




	
	
	
		
 
        return {'git': git_fork_name, 'hg': hg_fork_name}

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_hg_repo_by_admin(self, webserver):

			




	
	
	
		
 
        clone_url = webserver.repo_url(HG_REPO)

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        assert 'requesting all changes' in stdout

			




	
	
	
		
 
        assert 'adding changesets' in stdout

			




	
	
	
		
 
        assert 'adding manifests' in stdout

			




	
	
	
		
 
        assert 'adding file changes' in stdout

			




	
	
	
		
 

			




	
	
		
 
@@ -249,85 +249,85 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
        assert 'HTTP Error 404: Not Found' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_non_existing_path_git(self, webserver):

			




	
	
	
		
 
        clone_url = webserver.repo_url('trololo')

			




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

			




	
	
	
		
 
        assert 'not found' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_push_new_file_hg(self, webserver):

			




	
	
	
		
 
        dest_dir = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(HG_REPO)

			




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

			




	
	
	
		
 

			




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

			




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

			




	
	
	
		
 
        fixture.create_fork(HG_REPO, fork_name)

			




	
	
	
		
 
        clone_url = webserver.repo_url(fork_name)

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push(webserver, 'hg', dest_dir, clone_url=clone_url)

			




	
	
	
		
 

			




	
	
	
		
 
        assert 'pushing to' in stdout

			




	
	
	
		
 
        assert 'Repository size' in stdout

			




	
	
	
		
 
        assert 'Last revision is now' in stdout

			




	
	
	
		
 

			




	
	
	
		
 
    def test_push_new_file_git(self, webserver):

			




	
	
	
		
 
        dest_dir = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(GIT_REPO)

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        # commit some stuff into this repo

			




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

			




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

			




	
	
	
		
 
        fixture.create_fork(GIT_REPO, fork_name)

			




	
	
	
		
 
        clone_url = webserver.repo_url(fork_name)

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push(webserver, 'git', dest_dir, clone_url=clone_url)

			




	
	
	
		
 
        print [(x.repo_full_path,x.repo_path) for x in Repository.query()] # TODO: what is this for

			




	
	
	
		
 
        _check_proper_git_push(stdout, stderr)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_push_invalidates_cache_hg(self, webserver):

			




	
	
	
		
 
        key = CacheInvalidation.query().filter(CacheInvalidation.cache_key

			




	
	
	
		
 
                                               == HG_REPO).scalar()

			




	
	
	
		
 
        if not key:

			




	
	
	
		
 
            key = CacheInvalidation(HG_REPO, HG_REPO)

			




	
	
	
		
 
            Session().add(key)

			




	
	
	
		
 

			




	
	
	
		
 
        key.cache_active = True

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
        dest_dir = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(HG_REPO)

			




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

			




	
	
	
		
 

			




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

			




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

			




	
	
	
		
 
        fixture.create_fork(HG_REPO, fork_name)

			




	
	
	
		
 
        clone_url = webserver.repo_url(fork_name)

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push(webserver, 'hg', dest_dir, files_no=1, clone_url=clone_url)

			




	
	
	
		
 

			




	
	
	
		
 
        key = CacheInvalidation.query().filter(CacheInvalidation.cache_key

			




	
	
	
		
 
                                               == fork_name).all()

			




	
	
	
		
 
        assert key == []

			




	
	
	
		
 

			




	
	
	
		
 
    def test_push_invalidates_cache_git(self, webserver):

			




	
	
	
		
 
        key = CacheInvalidation.query().filter(CacheInvalidation.cache_key

			




	
	
	
		
 
                                               == GIT_REPO).scalar()

			




	
	
	
		
 
        if not key:

			




	
	
	
		
 
            key = CacheInvalidation(GIT_REPO, GIT_REPO)

			




	
	
	
		
 
            Session().add(key)

			




	
	
	
		
 

			




	
	
	
		
 
        key.cache_active = True

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
        dest_dir = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(GIT_REPO)

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        # commit some stuff into this repo

			




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

			




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

			




	
	
	
		
 
        fixture.create_fork(GIT_REPO, fork_name)

			




	
	
	
		
 
        clone_url = webserver.repo_url(fork_name)

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push(webserver, 'git', dest_dir, files_no=1, clone_url=clone_url)

			




	
	
	
		
 
        _check_proper_git_push(stdout, stderr)

			




	
	
	
		
 

			




	
	
	
		
 
        key = CacheInvalidation.query().filter(CacheInvalidation.cache_key

			




	
	
	
		
 
                                               == fork_name).all()

			




	
	
	
		
 
        assert key == []

			




	
	
	
		
 

			




	
	
	
		
 
    def test_push_wrong_credentials_hg(self, webserver):

			




	
	
	
		
 
        dest_dir = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(HG_REPO)

			




	
	
		
 
@@ -488,25 +488,25 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
        assert err in stderr

			




	
	
	
		
 

			




	
	
	
		
 
        # TODO: fix this somehow later on Git, Git is stupid and even if we throw

			




	
	
	
		
 
        # back 423 to it, it makes ANOTHER request and we fail there with 405 :/

			




	
	
	
		
 

			




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

			




	
	
	
		
 
                % (GIT_REPO, TEST_USER_ADMIN_LOGIN))

			




	
	
	
		
 
        #msg = "405 Method Not Allowed"

			




	
	
	
		
 
        #assert msg in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_push_unlocks_repository_hg(self, webserver):

			




	
	
	
		
 
        # enable locking

			




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

			




	
	
	
		
 
        fork_name = u'%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_dir = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(fork_name)

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        # 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

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        # push is ok and repo is now unlocked

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push(webserver, 'hg', dest_dir, clone_url=clone_url)

			




	
	
	
		
 
        assert str('remote: Released lock on repo `%s`' % fork_name) in stdout

			




	
	
	
		
 
        # we need to cleanup the Session Here !

			




	
	
	
		
 
        Session.remove()

			




	
	
	
		
 
        r = Repository.get_by_repo_name(fork_name)

			




	
	
	
		
 
        assert r.locked == [None, None]

			




	
	
	
		
 

			




	
	
	
		
 
    # TODO: fix me ! somehow during tests hooks don't get called on Git

			




	
	
	
		
 
    def test_push_unlocks_repository_git(self, webserver):

			




	
	
	
		
 
        # enable locking

			




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

			




	
	
	
		
 
        fork_name = u'%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_dir = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(fork_name)

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        # 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

			




        

    



    


    



    



      

      





    
    0 comments (0 inline, 0 general)
    





    


  

  







  


    




    








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