(self)
| 3399 | conn.close() |
| 3400 | |
| 3401 | def test_connection(self): |
| 3402 | conn, child_conn = self.Pipe() |
| 3403 | |
| 3404 | p = self.Process(target=self._echo, args=(child_conn,)) |
| 3405 | p.daemon = True |
| 3406 | p.start() |
| 3407 | |
| 3408 | seq = [1, 2.25, None] |
| 3409 | msg = latin('hello world') |
| 3410 | longmsg = msg * 10 |
| 3411 | arr = array.array('i', list(range(4))) |
| 3412 | |
| 3413 | if self.TYPE == 'processes': |
| 3414 | self.assertEqual(type(conn.fileno()), int) |
| 3415 | |
| 3416 | self.assertEqual(conn.send(seq), None) |
| 3417 | self.assertEqual(conn.recv(), seq) |
| 3418 | |
| 3419 | self.assertEqual(conn.send_bytes(msg), None) |
| 3420 | self.assertEqual(conn.recv_bytes(), msg) |
| 3421 | |
| 3422 | if self.TYPE == 'processes': |
| 3423 | buffer = array.array('i', [0]*10) |
| 3424 | expected = list(arr) + [0] * (10 - len(arr)) |
| 3425 | self.assertEqual(conn.send_bytes(arr), None) |
| 3426 | self.assertEqual(conn.recv_bytes_into(buffer), |
| 3427 | len(arr) * buffer.itemsize) |
| 3428 | self.assertEqual(list(buffer), expected) |
| 3429 | |
| 3430 | buffer = array.array('i', [0]*10) |
| 3431 | expected = [0] * 3 + list(arr) + [0] * (10 - 3 - len(arr)) |
| 3432 | self.assertEqual(conn.send_bytes(arr), None) |
| 3433 | self.assertEqual(conn.recv_bytes_into(buffer, 3 * buffer.itemsize), |
| 3434 | len(arr) * buffer.itemsize) |
| 3435 | self.assertEqual(list(buffer), expected) |
| 3436 | |
| 3437 | buffer = bytearray(latin(' ' * 40)) |
| 3438 | self.assertEqual(conn.send_bytes(longmsg), None) |
| 3439 | try: |
| 3440 | res = conn.recv_bytes_into(buffer) |
| 3441 | except multiprocessing.BufferTooShort as e: |
| 3442 | self.assertEqual(e.args, (longmsg,)) |
| 3443 | else: |
| 3444 | self.fail('expected BufferTooShort, got %s' % res) |
| 3445 | |
| 3446 | poll = TimingWrapper(conn.poll) |
| 3447 | |
| 3448 | self.assertEqual(poll(), False) |
| 3449 | self.assertTimingAlmostEqual(poll.elapsed, 0) |
| 3450 | |
| 3451 | self.assertEqual(poll(-1), False) |
| 3452 | self.assertTimingAlmostEqual(poll.elapsed, 0) |
| 3453 | |
| 3454 | self.assertEqual(poll(TIMEOUT1), False) |
| 3455 | self.assertTimingAlmostEqual(poll.elapsed, TIMEOUT1) |
| 3456 | |
| 3457 | conn.send(None) |
| 3458 | time.sleep(.1) |
nothing calls this directly
no test coverage detected