(self, tmpdir, single_submit, overlap_events)
| 144 | assert filecmp.cmp(ref_file, gds_file, shallow=False) |
| 145 | |
| 146 | def test_async_write(self, tmpdir, single_submit, overlap_events): |
| 147 | ref_file, ref_buffer = _do_ref_write(tmpdir) |
| 148 | |
| 149 | h = GDSBuilder().load().gds_handle(BLOCK_SIZE, QUEUE_DEPTH, single_submit, overlap_events, IO_PARALLEL) |
| 150 | gds_file, gds_buffer = _get_test_write_file_and_device_buffer(tmpdir, ref_buffer, h) |
| 151 | |
| 152 | _validate_handle_state(h, single_submit, overlap_events) |
| 153 | |
| 154 | write_status = h.async_pwrite(gds_buffer, gds_file, 0) |
| 155 | assert write_status == 0 |
| 156 | |
| 157 | wait_status = h.wait() |
| 158 | assert wait_status == 1 |
| 159 | |
| 160 | h.unpin_device_tensor(gds_buffer) |
| 161 | |
| 162 | assert os.path.isfile(gds_file) |
| 163 | |
| 164 | filecmp.clear_cache() |
| 165 | assert filecmp.cmp(ref_file, gds_file, shallow=False) |
| 166 | |
| 167 | |
| 168 | @pytest.mark.sequential |
nothing calls this directly
no test coverage detected