(total_size, chunksize, buffer_size=0)
| 577 | return cbuf, writer |
| 578 | |
| 579 | def test_writes(total_size, chunksize, buffer_size=0): |
| 580 | cbuf, writer = allocate(total_size) |
| 581 | arr, buf = make_random_buffer(size=total_size, target='host') |
| 582 | |
| 583 | if buffer_size > 0: |
| 584 | writer.buffer_size = buffer_size |
| 585 | |
| 586 | position = writer.tell() |
| 587 | assert position == 0 |
| 588 | writer.write(buf.slice(length=chunksize)) |
| 589 | assert writer.tell() == chunksize |
| 590 | writer.seek(0) |
| 591 | position = writer.tell() |
| 592 | assert position == 0 |
| 593 | |
| 594 | while position < total_size: |
| 595 | bytes_to_write = min(chunksize, total_size - position) |
| 596 | writer.write(buf.slice(offset=position, length=bytes_to_write)) |
| 597 | position += bytes_to_write |
| 598 | |
| 599 | writer.flush() |
| 600 | assert cbuf.size == total_size |
| 601 | cbuf.context.synchronize() |
| 602 | buf2 = cbuf.copy_to_host() |
| 603 | cbuf.context.synchronize() |
| 604 | assert buf2.size == total_size |
| 605 | arr2 = np.frombuffer(buf2, dtype=np.uint8) |
| 606 | np.testing.assert_equal(arr, arr2) |
| 607 | |
| 608 | total_size, chunk_size = 1 << 16, 1000 |
| 609 | test_writes(total_size, chunk_size) |
no test coverage detected