(pd, seg, tmpdir)
| 33 | |
| 34 | |
| 35 | def test_atomic_download(pd, seg, tmpdir): |
| 36 | assert not pd.is_running(seg) |
| 37 | |
| 38 | pd.create(seg) |
| 39 | assert pd.is_running(seg) |
| 40 | |
| 41 | with pd.download(seg) as ad: |
| 42 | s = b'hello' |
| 43 | ad.tf.write(s) |
| 44 | ad.tf.flush() |
| 45 | assert pd.running_size(seg) == len(s) |
| 46 | |
| 47 | assert pd.contains(seg) |
| 48 | assert not pd.is_running(seg) |
| 49 | |
| 50 | promote_target = tmpdir.join('another-spot') |
| 51 | pd.promote(seg, str(promote_target)) |
| 52 | |
| 53 | pd.clear() |
| 54 | assert not pd.contains(seg) |
| 55 | |
| 56 | |
| 57 | def test_atomic_download_failure(pd, seg): |
nothing calls this directly
no test coverage detected