(self)
| 594 | @unittest.skipUnless(hasattr(os, 'splice'), 'test needs os.splice()') |
| 595 | @requires_splice_pipe |
| 596 | def test_splice_offset_out(self): |
| 597 | TESTFN4 = os_helper.TESTFN + ".4" |
| 598 | data = b'0123456789' |
| 599 | bytes_to_copy = 6 |
| 600 | out_seek = 3 |
| 601 | |
| 602 | create_file(os_helper.TESTFN, data) |
| 603 | self.addCleanup(os_helper.unlink, os_helper.TESTFN) |
| 604 | |
| 605 | read_fd, write_fd = os.pipe() |
| 606 | self.addCleanup(lambda: os.close(read_fd)) |
| 607 | self.addCleanup(lambda: os.close(write_fd)) |
| 608 | os.write(write_fd, data) |
| 609 | |
| 610 | out_file = open(TESTFN4, 'w+b') |
| 611 | self.addCleanup(os_helper.unlink, TESTFN4) |
| 612 | self.addCleanup(out_file.close) |
| 613 | out_fd = out_file.fileno() |
| 614 | |
| 615 | try: |
| 616 | i = os.splice(read_fd, out_fd, bytes_to_copy, offset_dst=out_seek) |
| 617 | except OSError as e: |
| 618 | # Handle the case in which Python was compiled |
| 619 | # in a system with the syscall but without support |
| 620 | # in the kernel. |
| 621 | if e.errno != errno.ENOSYS: |
| 622 | raise |
| 623 | self.skipTest(e) |
| 624 | else: |
| 625 | # The number of copied bytes can be less than |
| 626 | # the number of bytes originally requested. |
| 627 | self.assertIn(i, range(0, bytes_to_copy+1)); |
| 628 | |
| 629 | with open(TESTFN4, 'rb') as in_file: |
| 630 | read = in_file.read() |
| 631 | # seeked bytes (5) are zero'ed |
| 632 | self.assertEqual(read[:out_seek], b'\x00'*out_seek) |
| 633 | # 012 are skipped (in_skip) |
| 634 | # 345678 are copied in the file (in_skip + bytes_to_copy) |
| 635 | self.assertEqual(read[out_seek:], data[:i]) |
| 636 | |
| 637 | |
| 638 | # Test attributes on return values from os.*stat* family. |
nothing calls this directly
no test coverage detected