##// END OF EJS Templates
dependencies: Adding Bower out of nixpkgs...
dependencies: Adding Bower out of nixpkgs Usually bower would be installed globally, so it seems to be appropriate to use the one out of the nixpkgs so that we don't have to pin a specific version ourselves.

File last commit:

r502:8bb872ec default
r709:a198b78f default
Show More
test_scm.py
335 lines | 11.9 KiB | text/x-python | PythonLexer
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2016 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# 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 Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import os
import stat
import sys
import pytest
from mock import Mock, patch, DEFAULT
import rhodecode
from rhodecode.model import db, scm
def test_scm_instance_config(backend):
repo = backend.create_repo()
with patch.multiple('rhodecode.model.db.Repository',
_get_instance=DEFAULT,
_get_instance_cached=DEFAULT) as mocks:
repo.scm_instance()
mocks['_get_instance'].assert_called_with(
config=None, cache=False)
config = {'some': 'value'}
repo.scm_instance(config=config)
mocks['_get_instance'].assert_called_with(
config=config, cache=False)
with patch.dict(rhodecode.CONFIG, {'vcs_full_cache': 'true'}):
repo.scm_instance(config=config)
mocks['_get_instance_cached'].assert_called()
def test__get_instance_config(backend):
repo = backend.create_repo()
vcs_class = Mock()
with patch.multiple('rhodecode.lib.vcs.backends',
get_scm=DEFAULT,
get_backend=DEFAULT) as mocks:
mocks['get_scm'].return_value = backend.alias
mocks['get_backend'].return_value = vcs_class
with patch('rhodecode.model.db.Repository._config') as config_mock:
repo._get_instance()
vcs_class.assert_called_with(
repo_path=repo.repo_full_path, config=config_mock,
create=False, with_wire={'cache': True})
new_config = {'override': 'old_config'}
repo._get_instance(config=new_config)
vcs_class.assert_called_with(
repo_path=repo.repo_full_path, config=new_config, create=False,
with_wire={'cache': True})
def test_mark_for_invalidation_config(backend):
repo = backend.create_repo()
with patch('rhodecode.model.db.Repository.update_commit_cache') as _mock:
scm.ScmModel().mark_for_invalidation(repo.repo_name)
_, kwargs = _mock.call_args
assert kwargs['config'].__dict__ == repo._config.__dict__
def test_mark_for_invalidation_with_delete_updates_last_commit(backend):
commits = [{'message': 'A'}, {'message': 'B'}]
repo = backend.create_repo(commits=commits)
scm.ScmModel().mark_for_invalidation(repo.repo_name, delete=True)
assert repo.changeset_cache['revision'] == 1
def test_mark_for_invalidation_with_delete_updates_last_commit_empty(backend):
repo = backend.create_repo()
scm.ScmModel().mark_for_invalidation(repo.repo_name, delete=True)
assert repo.changeset_cache['revision'] == -1
def test_strip_with_multiple_heads(backend_hg):
commits = [
{'message': 'A'},
{'message': 'a'},
{'message': 'b'},
{'message': 'B', 'parents': ['A']},
{'message': 'a1'},
]
repo = backend_hg.create_repo(commits=commits)
commit_ids = backend_hg.commit_ids
model = scm.ScmModel()
model.strip(repo, commit_ids['b'], branch=None)
vcs_repo = repo.scm_instance()
rest_commit_ids = [c.raw_id for c in vcs_repo.get_changesets()]
assert len(rest_commit_ids) == 4
assert commit_ids['b'] not in rest_commit_ids
def test_strip_with_single_heads(backend_hg):
commits = [
{'message': 'A'},
{'message': 'a'},
{'message': 'b'},
]
repo = backend_hg.create_repo(commits=commits)
commit_ids = backend_hg.commit_ids
model = scm.ScmModel()
model.strip(repo, commit_ids['b'], branch=None)
vcs_repo = repo.scm_instance()
rest_commit_ids = [c.raw_id for c in vcs_repo.get_changesets()]
assert len(rest_commit_ids) == 2
assert commit_ids['b'] not in rest_commit_ids
def test_get_nodes_returns_unicode_flat(backend_random):
repo = backend_random.repo
directories, files = scm.ScmModel().get_nodes(
repo.repo_name, repo.get_commit(commit_idx=0).raw_id,
flat=True)
assert_contains_only_unicode(directories)
assert_contains_only_unicode(files)
def test_get_nodes_returns_unicode_non_flat(backend_random):
repo = backend_random.repo
directories, files = scm.ScmModel().get_nodes(
repo.repo_name, repo.get_commit(commit_idx=0).raw_id,
flat=False)
# johbo: Checking only the names for now, since that is the critical
# part.
assert_contains_only_unicode([d['name'] for d in directories])
assert_contains_only_unicode([f['name'] for f in files])
def test_get_nodes_max_file_bytes(backend_random):
repo = backend_random.repo
max_file_bytes = 10
directories, files = scm.ScmModel().get_nodes(
repo.repo_name, repo.get_commit(commit_idx=0).raw_id, content=True,
extended_info=True, flat=False)
assert any(file['content'] and len(file['content']) > max_file_bytes
for file in files)
directories, files = scm.ScmModel().get_nodes(
repo.repo_name, repo.get_commit(commit_idx=0).raw_id, content=True,
extended_info=True, flat=False, max_file_bytes=max_file_bytes)
assert all(
file['content'] is None if file['size'] > max_file_bytes else True
for file in files)
def assert_contains_only_unicode(structure):
assert structure
for value in structure:
assert isinstance(value, unicode)
@pytest.mark.backends("hg", "git")
def test_get_non_unicode_reference(backend):
model = scm.ScmModel()
non_unicode_list = ["Adını".decode("cp1254")]
def scm_instance():
return Mock(
branches=non_unicode_list, bookmarks=non_unicode_list,
tags=non_unicode_list, alias=backend.alias)
repo = Mock(__class__=db.Repository, scm_instance=scm_instance)
choices, __ = model.get_repo_landing_revs(repo=repo)
if backend.alias == 'hg':
valid_choices = [
'rev:tip', u'branch:Ad\xc4\xb1n\xc4\xb1',
u'book:Ad\xc4\xb1n\xc4\xb1', u'tag:Ad\xc4\xb1n\xc4\xb1']
else:
valid_choices = [
'rev:tip', u'branch:Ad\xc4\xb1n\xc4\xb1',
u'tag:Ad\xc4\xb1n\xc4\xb1']
assert choices == valid_choices
class TestInstallSvnHooks(object):
HOOK_FILES = ('pre-commit', 'post-commit')
def test_new_hooks_are_created(self, backend_svn):
model = scm.ScmModel()
repo = backend_svn.create_repo()
vcs_repo = repo.scm_instance()
model.install_svn_hooks(vcs_repo)
hooks_path = os.path.join(vcs_repo.path, 'hooks')
assert os.path.isdir(hooks_path)
for file_name in self.HOOK_FILES:
file_path = os.path.join(hooks_path, file_name)
self._check_hook_file_mode(file_path)
self._check_hook_file_content(file_path)
def test_rc_hooks_are_replaced(self, backend_svn):
model = scm.ScmModel()
repo = backend_svn.create_repo()
vcs_repo = repo.scm_instance()
hooks_path = os.path.join(vcs_repo.path, 'hooks')
file_paths = [os.path.join(hooks_path, f) for f in self.HOOK_FILES]
for file_path in file_paths:
self._create_fake_hook(
file_path, content="RC_HOOK_VER = 'abcde'\n")
model.install_svn_hooks(vcs_repo)
for file_path in file_paths:
self._check_hook_file_content(file_path)
def test_non_rc_hooks_are_not_replaced_without_force_create(
self, backend_svn):
model = scm.ScmModel()
repo = backend_svn.create_repo()
vcs_repo = repo.scm_instance()
hooks_path = os.path.join(vcs_repo.path, 'hooks')
file_paths = [os.path.join(hooks_path, f) for f in self.HOOK_FILES]
non_rc_content = "exit 0\n"
for file_path in file_paths:
self._create_fake_hook(file_path, content=non_rc_content)
model.install_svn_hooks(vcs_repo)
for file_path in file_paths:
with open(file_path, 'rt') as hook_file:
content = hook_file.read()
assert content == non_rc_content
def test_non_rc_hooks_are_replaced_with_force_create(self, backend_svn):
model = scm.ScmModel()
repo = backend_svn.create_repo()
vcs_repo = repo.scm_instance()
hooks_path = os.path.join(vcs_repo.path, 'hooks')
file_paths = [os.path.join(hooks_path, f) for f in self.HOOK_FILES]
non_rc_content = "exit 0\n"
for file_path in file_paths:
self._create_fake_hook(file_path, content=non_rc_content)
model.install_svn_hooks(vcs_repo, force_create=True)
for file_path in file_paths:
self._check_hook_file_content(file_path)
def _check_hook_file_mode(self, file_path):
assert os.path.exists(file_path)
stat_info = os.stat(file_path)
file_mode = stat.S_IMODE(stat_info.st_mode)
expected_mode = int('755', 8)
assert expected_mode == file_mode
def _check_hook_file_content(self, file_path):
with open(file_path, 'rt') as hook_file:
content = hook_file.read()
expected_env = '#!{}'.format(sys.executable)
expected_rc_version = "\nRC_HOOK_VER = '{}'\n".format(
rhodecode.__version__)
assert content.strip().startswith(expected_env)
assert expected_rc_version in content
def _create_fake_hook(self, file_path, content):
with open(file_path, 'w') as hook_file:
hook_file.write(content)
class TestCheckRhodecodeHook(object):
@patch('os.path.exists', Mock(return_value=False))
def test_returns_true_when_no_hook_found(self):
result = scm._check_rhodecode_hook('/tmp/fake_hook_file.py')
assert result
@pytest.mark.parametrize("file_content, expected_result", [
("RC_HOOK_VER = '3.3.3'\n", True),
("RC_HOOK = '3.3.3'\n", False),
])
@patch('os.path.exists', Mock(return_value=True))
def test_signatures(self, file_content, expected_result):
hook_content_patcher = patch.object(
scm, '_read_hook', return_value=file_content)
with hook_content_patcher:
result = scm._check_rhodecode_hook('/tmp/fake_hook_file.py')
assert result is expected_result
class TestInstallHooks(object):
def test_hooks_are_installed_for_git_repo(self, backend_git):
repo = backend_git.create_repo()
model = scm.ScmModel()
scm_repo = repo.scm_instance()
with patch.object(model, 'install_git_hook') as hooks_mock:
model.install_hooks(scm_repo, repo_type='git')
hooks_mock.assert_called_once_with(scm_repo)
def test_hooks_are_installed_for_svn_repo(self, backend_svn):
repo = backend_svn.create_repo()
scm_repo = repo.scm_instance()
model = scm.ScmModel()
with patch.object(scm.ScmModel, 'install_svn_hooks') as hooks_mock:
model.install_hooks(scm_repo, repo_type='svn')
hooks_mock.assert_called_once_with(scm_repo)
@pytest.mark.parametrize('hook_method', [
'install_svn_hooks',
'install_git_hook'])
def test_mercurial_doesnt_trigger_hooks(self, backend_hg, hook_method):
repo = backend_hg.create_repo()
scm_repo = repo.scm_instance()
model = scm.ScmModel()
with patch.object(scm.ScmModel, hook_method) as hooks_mock:
model.install_hooks(scm_repo, repo_type='hg')
assert hooks_mock.call_count == 0