183 lines
6.6 KiB
Python
183 lines
6.6 KiB
Python
# Copyright (C) 2010-2024 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"], "branch": "feature"},
|
||
{"message": "a1", "branch": "feature"},
|
||
]
|
||
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_str_chars(directories)
|
||
assert_contains_only_str_chars(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_str_chars([d["name"] for d in directories])
|
||
assert_contains_only_str_chars([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_str_chars(structure):
|
||
assert structure
|
||
for value in structure:
|
||
assert isinstance(value, str)
|
||
|
||
|
||
@pytest.mark.backends("hg", "git")
|
||
def test_get_non_str_reference(backend):
|
||
model = scm.ScmModel()
|
||
special_name = "Adını"
|
||
non_str_list = [special_name]
|
||
|
||
def scm_instance():
|
||
return Mock(branches=non_str_list, bookmarks=non_str_list, tags=non_str_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", f"branch:{special_name}", f"book:{special_name}", f"tag:{special_name}"]
|
||
else:
|
||
valid_choices = ["rev:tip", f"branch:{special_name}", f"tag:{special_name}"]
|
||
|
||
assert choices == valid_choices
|