(tmp_dir, dvc, index, mocker)
| 71 | |
| 72 | |
| 73 | def test_partial_upload(tmp_dir, dvc, index, mocker): |
| 74 | from dvc_objects.fs import generic |
| 75 | |
| 76 | tmp_dir.dvc_gen({"foo": "foo content"}) |
| 77 | baz = tmp_dir.dvc_gen({"bar": {"baz": "baz content"}})[0].outs[0] |
| 78 | |
| 79 | original = generic.transfer |
| 80 | odb = dvc.cloud.get_remote_odb("upstream") |
| 81 | |
| 82 | def unreliable_upload(from_fs, from_info, to_fs, to_info, **kwargs): |
| 83 | on_error = kwargs["on_error"] |
| 84 | assert on_error |
| 85 | if isinstance(from_info, str): |
| 86 | from_info = [from_info] |
| 87 | else: |
| 88 | from_info = list(from_info) |
| 89 | if isinstance(to_info, str): |
| 90 | to_info = [to_info] |
| 91 | else: |
| 92 | to_info = list(to_info) |
| 93 | for i in range(len(from_info) - 1, -1, -1): |
| 94 | from_i = from_info[i] |
| 95 | to_i = to_info[i] |
| 96 | if os.path.abspath(to_i) == os.path.abspath( |
| 97 | odb.get(baz.hash_info.value).path |
| 98 | ): |
| 99 | if on_error: |
| 100 | on_error(from_i, to_i, Exception("stop baz")) |
| 101 | del from_info[i] |
| 102 | del to_info[i] |
| 103 | |
| 104 | return original(from_fs, from_info, to_fs, to_info, **kwargs) |
| 105 | |
| 106 | mocker.patch("dvc_objects.fs.generic.transfer", unreliable_upload) |
| 107 | with pytest.raises(UploadError): |
| 108 | dvc.push() |
| 109 | assert not list(index.hashes()) |
nothing calls this directly
no test coverage detected