mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-04 17:08:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 13

bpo-32034: Make IncompleteReadError & LimitOverrunError pickleable #4409

Browse source
This commit is contained in:
Yury Selivanov 2017-11-15 17:14:28 -05:00 • committed by GitHub
parent 4bd41c9b52
commit 43605e6bfa
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 25 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

18
Lib/test/test_asyncio/test_streams.py
View file

@ -3,6 +3,7 @@
import gc
import os
import queue
import pickle
import socket
import sys
import threading
@ -845,6 +846,23 @@ os.close(fd)
stream._transport.__repr__.return_value = "<Transport>"
self.assertEqual("<StreamReader t=<Transport>>", repr(stream))
def test_IncompleteReadError_pickleable(self):
e = asyncio.IncompleteReadError(b'abc', 10)
for proto in range(pickle.HIGHEST_PROTOCOL + 1):
with self.subTest(pickle_protocol=proto):
e2 = pickle.loads(pickle.dumps(e, protocol=proto))
self.assertEqual(str(e), str(e2))
self.assertEqual(e.partial, e2.partial)
self.assertEqual(e.expected, e2.expected)
def test_LimitOverrunError_pickleable(self):
e = asyncio.LimitOverrunError('message', 10)
for proto in range(pickle.HIGHEST_PROTOCOL + 1):
with self.subTest(pickle_protocol=proto):
e2 = pickle.loads(pickle.dumps(e, protocol=proto))
self.assertEqual(str(e), str(e2))
self.assertEqual(e.consumed, e2.consumed)
if __name__ == '__main__':
unittest.main()