Method test_write

tests/unit/ops/aio/test_gds.py:211–246

(self, tmpdir, async_queue)

    @pytest.mark.parametrize("async_queue", [2, 3])
    def test_write(self, tmpdir, async_queue):
        ref_files = []
        ref_buffers = []
        for i in range(async_queue):
            f, buf = _do_ref_write(tmpdir, i)
            ref_files.append(f)
            ref_buffers.append(buf)

        single_submit = True
        overlap_events = True
        h = GDSBuilder().load().gds_handle(BLOCK_SIZE, QUEUE_DEPTH, single_submit, overlap_events, IO_PARALLEL)

        gds_files = []
        gds_buffers = []
        for i in range(async_queue):
            f, buf = _get_test_write_file_and_device_buffer(tmpdir, ref_buffers[i], h, i)
            gds_files.append(f)
            gds_buffers.append(buf)

        _validate_handle_state(h, single_submit, overlap_events)

        for i in range(async_queue):
            read_status = h.async_pwrite(gds_buffers[i], gds_files[i], 0)
            assert read_status == 0

        wait_status = h.wait()
        assert wait_status == async_queue

        for t in gds_buffers:
            h.unpin_device_tensor(t)

        for i in range(async_queue):
            assert os.path.isfile(gds_files[i])

            filecmp.clear_cache()
            assert filecmp.cmp(ref_files[i], gds_files[i], shallow=False)


@pytest.mark.parametrize("use_new_api", [True, False])