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 @@
@@ -29,6 +29,8 @@
CreateRepositoryRequest,
CreateRepositoryRequest,
FindMergeBaseRequest,
FindMergeBaseRequest,
GetArchiveRequest,
GetArchiveRequest,
 
GetRawChangesRequest,
 
GetRawChangesResponse,
HasLocalBranchesRequest,
HasLocalBranchesRequest,
RepositoryExistsRequest,
RepositoryExistsRequest,
SetFullPathRequest,
SetFullPathRequest,
@@ -373,3 +375,103 @@
@@ -373,3 +375,103 @@
grpc_repo=Repository(
grpc_repo=Repository(
relative_path=grpc_repo.relative_path,
relative_path=grpc_repo.relative_path,
storage_name='unknown'))
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