Commit 0d3b273b authored by Sushil Khanchi's avatar Sushil Khanchi 🐨
Browse files

mercurial-repository: implement BackupAdditional and RestoreAdditional

Also added the tests referenced/copied from py-heptapod side testing
of `hpd-backup-additional` and `hpd-restore-additional` commands.
parent e2fa26418ffc
Pipeline #26160 passed with stages
in 2 minutes and 2 seconds
......@@ -7,12 +7,18 @@
# SPDX-License-Identifier: GPL-2.0-or-later
import logging
from hgext3rd.heptapod import (
from mercurial import (
from ..errors import (
from ..stub.mercurial_repository_pb2 import (
......@@ -56,12 +62,19 @@ class MercurialRepositoryServicer(MercurialRepositoryServiceServicer,
def BackupAdditional(self, request: BackupAdditionalRequest,
context) -> BackupAdditionalResponse:
return not_implemented( # pragma no cover
context, BackupAdditionalResponse,
repo = self.load_repo(request.repository, context)
save_path = pycompat.sysbytes(request.save_path)
backup_additional(repo.ui, repo, save_path)
return BackupAdditionalResponse()
def RestoreAdditional(self, request: RestoreAdditionalRequest,
context) -> RestoreAdditionalResponse:
return not_implemented( # pragma no cover
context, RestoreAdditionalResponse,
context) -> RestoreAdditionalResponse:
repo = self.load_repo(request.repository, context)
bundle_path = pycompat.sysbytes(request.bundle_path)
restore_additional(repo.ui, repo, bundle_path)
except error.Abort as exc_info:
return failed_precondition(
context, RestoreAdditionalResponse, exc_info.message
return RestoreAdditionalResponse()
......@@ -4,15 +4,26 @@
# GNU General Public License version 2 or any later version.
# SPDX-License-Identifier: GPL-2.0-or-later
from hgitaly.tests.common import make_empty_repo
import pytest
import tarfile
import grpc
from io import BytesIO
from hgitaly.tests.common import make_empty_repo
from hgext3rd.heptapod.special_ref import (
from hgitaly.stub.mercurial_repository_pb2 import (
from hgitaly.stub.mercurial_repository_pb2_grpc import (
from mercurial_testhelpers import as_bytes
def test_config_item(grpc_channel, server_repos_root):
......@@ -31,3 +42,60 @@ def test_config_item(grpc_channel, server_repos_root):
# some other ones if they change.
assert rpc_config_bool('format', 'usestore') is True
assert rpc_config_bool('commands', 'status.verbose') is False
def test_additional_backup_restore(grpc_channel, server_repos_root, tmpdir):
hg_repo_stub = MercurialRepositoryServiceStub(grpc_channel)
wrapper, grpc_repo = make_empty_repo(server_repos_root)
repo = wrapper.repo
def rpc_backup_additional(repo, save_path):
return hg_repo_stub.BackupAdditional(
BackupAdditionalRequest(repository=repo, save_path=save_path)
def rpc_restore_additional(repo, bundle_path):
return hg_repo_stub.RestoreAdditional(
RestoreAdditionalRequest(repository=repo, bundle_path=bundle_path)
main_hgrc = b'[foo]\nsome=bar\n'
managed_hgrc = b'[heptapod]\nallow-bookmars=true\n'
repo.vfs.write(b'hgrc', main_hgrc)
repo.vfs.write(b'hgrc.managed', managed_hgrc)
# some repo content and GitLab refs
sha = wrapper.write_commit('foo', branch='surprise').hex()
write_gitlab_special_ref(repo, b'merge_requests/5/head', sha)
# Backup additional
tar_path = tmpdir / 'additional_empty_repo.tar'
rpc_backup_additional(grpc_repo, as_bytes(tar_path))
# Restore additional
dest_wrapper, dest_grpc_repo = make_empty_repo(
server_repos_root, relative_path='dest_repo'
rpc_restore_additional(dest_grpc_repo, as_bytes(tar_path))
# Assertions for configs and Gitlab refs
dst_repo = dest_wrapper.repo
assert dst_repo.vfs(b'hgrc').read() == main_hgrc
assert dst_repo.vfs(b'hgrc.managed').read() == managed_hgrc
assert special_refs(dst_repo) == {b'merge_requests/5/head': sha}
# errors: test bad tar restoration
tar_path = tmpdir / 'bad.tar'
dest_wrapper, dest_grpc_repo = make_empty_repo(
server_repos_root, relative_path='badtar_dest_repo'
# str() because we support Python 3.7 (for CI)
# and path-like objects are accepted in tarfile from Python 3.8 onwards
with, 'w') as tarf:
tinfo = tarfile.TarInfo(name='/etc/shadow')
tarf.addfile(tinfo, BytesIO(b'evil input'))
with pytest.raises(grpc.RpcError) as exc_info:
rpc_restore_additional(dest_grpc_repo, as_bytes(tar_path))
assert exc_info.value.code() == grpc.StatusCode.FAILED_PRECONDITION
assert "Prohibited member" in exc_info.value.details()
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment