##// END OF EJS Templates
dependencies: bumped psutil==5.7.0
dependencies: bumped psutil==5.7.0

File last commit:

r3951:041016db default
r4261:10fa903a default
Show More
test_scm.py
189 lines | 6.7 KiB | text/x-python | PythonLexer
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2019 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 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)
repo.scm_instance(vcs_full_cache=False)
mocks['_get_instance'].assert_called_with(
config=None, cache=False)
repo.scm_instance(vcs_full_cache=True)
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, 'repo_state_uid': None})
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, 'repo_state_uid': None})
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_commits()]
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_commits()]
assert len(rest_commit_ids) == 2
assert commit_ids['b'] not in rest_commit_ids
def test_get_nodes_returns_unicode_flat(backend):
repo = backend.repo
commit_id = repo.get_commit(commit_idx=0).raw_id
directories, files = scm.ScmModel().get_nodes(repo.repo_name, commit_id, flat=True)
assert_contains_only_unicode(directories)
assert_contains_only_unicode(files)
def test_get_nodes_returns_unicode_non_flat(backend):
repo = backend.repo
commit_id = repo.get_commit(commit_idx=0).raw_id
directories, files = scm.ScmModel().get_nodes(repo.repo_name, commit_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(translator=lambda s: s, 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