Skip to content
Snippets Groups Projects

repositoryService: implement GetRawChanges

Open Sushil Khanchi requested to merge topic/default/repository-GetRawChanges into branch/default
1 unresolved thread
Files
5
@@ -29,6 +29,8 @@
CreateRepositoryRequest,
FindMergeBaseRequest,
GetArchiveRequest,
GetRawChangesRequest,
GetRawChangesResponse,
HasLocalBranchesRequest,
RepositoryExistsRequest,
SetFullPathRequest,
@@ -373,3 +375,103 @@
grpc_repo=Repository(
relative_path=grpc_repo.relative_path,
storage_name='unknown'))
def test_get_raw_changes(grpc_channel, server_repos_root):
repo_stub = RepositoryServiceStub(grpc_channel)
wrapper, grpc_repo = make_empty_repo(server_repos_root)
# Repo structure:
#
# @ 4 Removed zar
# |
# o 3 Copy zar to zarcopy
# |
# o 2 Rename bar to zar
# |
# o 1 Modified bar
# |
# o 0 Add bar
#
cid0 = wrapper.commit_file('bar', content="Hello\n").hex()
cid1 = wrapper.commit_file('bar', content="Trello\n").hex()
wrapper.command(b'mv', wrapper.repo.root + b'/bar',
wrapper.repo.root + b'/zar')
cid2 = wrapper.commit([b'zar', b'bar'], message=b'Rename bar to zar').hex()
wrapper.command(b'cp', wrapper.repo.root + b'/zar',
wrapper.repo.root + b'/zarcopy')
cid3 = wrapper.commit(rel_paths=['zarcopy'], add_remove=True).hex()
wrapper.command(b'rm', wrapper.repo.root + b'/zar')
cid4 = wrapper.commit(rel_paths=['zar'], add_remove=True).hex()
def do_rpc(from_rev, to_rev):
request = GetRawChangesRequest(
repository=grpc_repo,
from_revision=from_rev,
to_revision=to_rev)
response = repo_stub.GetRawChanges(request)
changes = []
for chunked_resp in response:
changes.extend(chunked_resp.raw_changes)
return changes
def assert_do_rpc(from_rev, to_rev, **expected_kwargs):
resp = do_rpc(from_rev, to_rev)
resp = resp[0]
for attr in expected_kwargs:
assert getattr(resp, attr) == expected_kwargs[attr]
OPERATION_TYPES = dict(GetRawChangesResponse.RawChange.Operation.items())
# Add
assert_do_rpc(node_mod.nullhex, cid0,
size=len("Hello\n"),
old_path='',
new_path='bar',
old_mode=0,
new_mode=33188,
operation=OPERATION_TYPES['ADDED'])
# Modify
assert_do_rpc(cid0, cid1,
size=len("Trello\n"),
old_path='bar',
new_path='bar',
old_mode=33188,
new_mode=33188,
operation=OPERATION_TYPES['MODIFIED'])
# Rename
assert_do_rpc(cid1, cid2,
size=len("Trello\n"),
old_path='bar',
new_path='zar',
old_mode=33188,
new_mode=33188,
operation=OPERATION_TYPES['RENAMED'])
# Copy
assert_do_rpc(cid2, cid3,
size=len("Trello\n"),
old_path='zar',
new_path='zarcopy',
old_mode=33188,
new_mode=33188,
operation=OPERATION_TYPES['COPIED'])
# Delete
assert_do_rpc(cid3, cid4,
size=len("Trello\n"),
old_path='zar',
new_path='',
old_mode=33188,
new_mode=0,
operation=OPERATION_TYPES['DELETED'])
# test error case: when revision does not exist
non_existent_sha = b'boom' * 10
with pytest.raises(grpc.RpcError) as exc_info:
do_rpc(non_existent_sha, cid1)
assert exc_info.value.code() == grpc.StatusCode.INVALID_ARGUMENT
assert "invalid 'from' revision" in exc_info.value.details()
with pytest.raises(grpc.RpcError) as exc_info:
do_rpc(cid1, non_existent_sha)
assert exc_info.value.code() == grpc.StatusCode.INVALID_ARGUMENT
assert "invalid 'to' revision" in exc_info.value.details()
# test error case: when to_revision is NULLHEX
assert do_rpc(cid1, node_mod.nullhex) == []
Loading