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