(self)
| 3655 | self.check_frame_opcodes(pickled) |
| 3656 | |
| 3657 | def test_optional_frames(self): |
| 3658 | if pickle.HIGHEST_PROTOCOL < 4: |
| 3659 | return |
| 3660 | |
| 3661 | def remove_frames(pickled, keep_frame=None): |
| 3662 | """Remove frame opcodes from the given pickle.""" |
| 3663 | frame_starts = [] |
| 3664 | # 1 byte for the opcode and 8 for the argument |
| 3665 | frame_opcode_size = 9 |
| 3666 | for opcode, _, pos in pickletools.genops(pickled): |
| 3667 | if opcode.name == 'FRAME': |
| 3668 | frame_starts.append(pos) |
| 3669 | |
| 3670 | newpickle = bytearray() |
| 3671 | last_frame_end = 0 |
| 3672 | for i, pos in enumerate(frame_starts): |
| 3673 | if keep_frame and keep_frame(i): |
| 3674 | continue |
| 3675 | newpickle += pickled[last_frame_end:pos] |
| 3676 | last_frame_end = pos + frame_opcode_size |
| 3677 | newpickle += pickled[last_frame_end:] |
| 3678 | return newpickle |
| 3679 | |
| 3680 | frame_size = self.FRAME_SIZE_TARGET |
| 3681 | num_frames = 20 |
| 3682 | # Large byte objects (dict values) intermittent with small objects |
| 3683 | # (dict keys) |
| 3684 | for bytes_type in (bytes, bytearray): |
| 3685 | obj = {i: bytes_type([i]) * frame_size for i in range(num_frames)} |
| 3686 | |
| 3687 | for proto in range(4, pickle.HIGHEST_PROTOCOL + 1): |
| 3688 | pickled = self.dumps(obj, proto) |
| 3689 | |
| 3690 | frameless_pickle = remove_frames(pickled) |
| 3691 | self.assertEqual(count_opcode(pickle.FRAME, frameless_pickle), 0) |
| 3692 | self.assertEqual(obj, self.loads(frameless_pickle)) |
| 3693 | |
| 3694 | some_frames_pickle = remove_frames(pickled, lambda i: i % 2) |
| 3695 | self.assertLess(count_opcode(pickle.FRAME, some_frames_pickle), |
| 3696 | count_opcode(pickle.FRAME, pickled)) |
| 3697 | self.assertEqual(obj, self.loads(some_frames_pickle)) |
| 3698 | |
| 3699 | @support.skip_if_pgo_task |
| 3700 | def test_framed_write_sizes_with_delayed_writer(self): |
nothing calls this directly
no test coverage detected