(self)
| 556 | @unittest.skipUnless(hasattr(os, 'splice'), 'test needs os.splice()') |
| 557 | @requires_splice_pipe |
| 558 | def test_splice_offset_in(self): |
| 559 | TESTFN4 = os_helper.TESTFN + ".4" |
| 560 | data = b'0123456789' |
| 561 | bytes_to_copy = 6 |
| 562 | in_skip = 3 |
| 563 | |
| 564 | create_file(os_helper.TESTFN, data) |
| 565 | self.addCleanup(os_helper.unlink, os_helper.TESTFN) |
| 566 | |
| 567 | in_file = open(os_helper.TESTFN, 'rb') |
| 568 | self.addCleanup(in_file.close) |
| 569 | in_fd = in_file.fileno() |
| 570 | |
| 571 | read_fd, write_fd = os.pipe() |
| 572 | self.addCleanup(lambda: os.close(read_fd)) |
| 573 | self.addCleanup(lambda: os.close(write_fd)) |
| 574 | |
| 575 | try: |
| 576 | i = os.splice(in_fd, write_fd, bytes_to_copy, offset_src=in_skip) |
| 577 | except OSError as e: |
| 578 | # Handle the case in which Python was compiled |
| 579 | # in a system with the syscall but without support |
| 580 | # in the kernel. |
| 581 | if e.errno != errno.ENOSYS: |
| 582 | raise |
| 583 | self.skipTest(e) |
| 584 | else: |
| 585 | # The number of copied bytes can be less than |
| 586 | # the number of bytes originally requested. |
| 587 | self.assertIn(i, range(0, bytes_to_copy+1)); |
| 588 | |
| 589 | read = os.read(read_fd, 100) |
| 590 | # 012 are skipped (in_skip) |
| 591 | # 345678 are copied in the file (in_skip + bytes_to_copy) |
| 592 | self.assertEqual(read, data[in_skip:in_skip+i]) |
| 593 | |
| 594 | @unittest.skipUnless(hasattr(os, 'splice'), 'test needs os.splice()') |
| 595 | @requires_splice_pipe |
nothing calls this directly
no test coverage detected