diff --git a/Lib/test/test_multiprocessing.py b/Lib/test/test_multiprocessing.py --- a/Lib/test/test_multiprocessing.py +++ b/Lib/test/test_multiprocessing.py @@ -18,6 +18,7 @@ import random import logging import test.support +import struct # Skip tests if _multiprocessing wasn't built. @@ -1448,6 +1449,11 @@ conn.close() def test_connection(self): + def pack_list(fmt, lst): + """Pack values of lst into a bytearray, using fmt.""" + n = len(lst) + return bytearray(struct.pack(fmt * n, *lst)) + conn, child_conn = self.Pipe() p = self.Process(target=self._echo, args=(child_conn,)) @@ -1457,7 +1463,12 @@ seq = [1, 2.25, None] msg = latin('hello world') longmsg = msg * 10 - arr = array.array('i', list(range(4))) + + fmt = 'i' + nitems = 4 + arrlst = list(range(nitems)) + arr = pack_list(fmt, arrlst) + itemsize = struct.calcsize(fmt) if self.TYPE == 'processes': self.assertEqual(type(conn.fileno()), int) @@ -1469,19 +1480,20 @@ self.assertEqual(conn.recv_bytes(), msg) if self.TYPE == 'processes': - buffer = array.array('i', [0]*10) - expected = list(arr) + [0] * (10 - len(arr)) + nzeros = 10 + buffer = pack_list(fmt, [0]*nzeros) + expected = arrlst + [0] * (nzeros - nitems) self.assertEqual(conn.send_bytes(arr), None) - self.assertEqual(conn.recv_bytes_into(buffer), - len(arr) * buffer.itemsize) - self.assertEqual(list(buffer), expected) - - buffer = array.array('i', [0]*10) - expected = [0] * 3 + list(arr) + [0] * (10 - 3 - len(arr)) + self.assertEqual(conn.recv_bytes_into(buffer), len(arr)) + result = struct.unpack(fmt*nzeros, buffer) + self.assertEqual(list(result), expected) + + buffer = pack_list(fmt, [0]*nzeros) + expected = [0] * 3 + arrlst + [0] * (nzeros - 3 - nitems) self.assertEqual(conn.send_bytes(arr), None) - self.assertEqual(conn.recv_bytes_into(buffer, 3 * buffer.itemsize), - len(arr) * buffer.itemsize) - self.assertEqual(list(buffer), expected) + self.assertEqual(conn.recv_bytes_into(buffer, 3 * itemsize), len(arr)) + result = struct.unpack(fmt*nzeros, buffer) + self.assertEqual(list(result), expected) buffer = bytearray(latin(' ' * 40)) self.assertEqual(conn.send_bytes(longmsg), None)