(self)
| 3957 | self.check_frame_opcodes(pickled) |
| 3958 | |
| 3959 | def test_optional_frames(self): |
| 3960 | if self.py_version < (3, 4): |
| 3961 | self.skipTest('not supported in Python < 3.4') |
| 3962 | |
| 3963 | def remove_frames(pickled, keep_frame=None): |
| 3964 | """Remove frame opcodes from the given pickle.""" |
| 3965 | frame_starts = [] |
| 3966 | # 1 byte for the opcode and 8 for the argument |
| 3967 | frame_opcode_size = 9 |
| 3968 | for opcode, _, pos in pickletools.genops(pickled): |
| 3969 | if opcode.name == 'FRAME': |
| 3970 | frame_starts.append(pos) |
| 3971 | |
| 3972 | newpickle = bytearray() |
| 3973 | last_frame_end = 0 |
| 3974 | for i, pos in enumerate(frame_starts): |
| 3975 | if keep_frame and keep_frame(i): |
| 3976 | continue |
| 3977 | newpickle += pickled[last_frame_end:pos] |
| 3978 | last_frame_end = pos + frame_opcode_size |
| 3979 | newpickle += pickled[last_frame_end:] |
| 3980 | return newpickle |
| 3981 | |
| 3982 | frame_size = self.FRAME_SIZE_TARGET |
| 3983 | num_frames = 20 |
| 3984 | # Large byte objects (dict values) intermittent with small objects |
| 3985 | # (dict keys) |
| 3986 | for bytes_type in (bytes, bytearray): |
| 3987 | obj = {i: bytes_type([i]) * frame_size for i in range(num_frames)} |
| 3988 | |
| 3989 | for proto in range(4, pickle.HIGHEST_PROTOCOL + 1): |
| 3990 | pickled = self.dumps(obj, proto) |
| 3991 | |
| 3992 | frameless_pickle = remove_frames(pickled) |
| 3993 | self.assertEqual(count_opcode(pickle.FRAME, frameless_pickle), 0) |
| 3994 | self.assertEqual(obj, self.loads(frameless_pickle)) |
| 3995 | |
| 3996 | some_frames_pickle = remove_frames(pickled, lambda i: i % 2) |
| 3997 | self.assertLess(count_opcode(pickle.FRAME, some_frames_pickle), |
| 3998 | count_opcode(pickle.FRAME, pickled)) |
| 3999 | self.assertEqual(obj, self.loads(some_frames_pickle)) |
| 4000 | |
| 4001 | @support.skip_if_pgo_task |
| 4002 | def test_framed_write_sizes_with_delayed_writer(self): |
nothing calls this directly
no test coverage detected