(self)
| 3684 | |
| 3685 | @warnings_helper.ignore_fork_in_thread_deprecation_warnings() |
| 3686 | def test_connection(self): |
| 3687 | conn, child_conn = self.Pipe() |
| 3688 | |
| 3689 | p = self.Process(target=self._echo, args=(child_conn,)) |
| 3690 | p.daemon = True |
| 3691 | p.start() |
| 3692 | |
| 3693 | seq = [1, 2.25, None] |
| 3694 | msg = latin('hello world') |
| 3695 | longmsg = msg * 10 |
| 3696 | arr = array.array('i', list(range(4))) |
| 3697 | |
| 3698 | if self.TYPE == 'processes': |
| 3699 | self.assertEqual(type(conn.fileno()), int) |
| 3700 | |
| 3701 | self.assertEqual(conn.send(seq), None) |
| 3702 | self.assertEqual(conn.recv(), seq) |
| 3703 | |
| 3704 | self.assertEqual(conn.send_bytes(msg), None) |
| 3705 | self.assertEqual(conn.recv_bytes(), msg) |
| 3706 | |
| 3707 | if self.TYPE == 'processes': |
| 3708 | buffer = array.array('i', [0]*10) |
| 3709 | expected = list(arr) + [0] * (10 - len(arr)) |
| 3710 | self.assertEqual(conn.send_bytes(arr), None) |
| 3711 | self.assertEqual(conn.recv_bytes_into(buffer), |
| 3712 | len(arr) * buffer.itemsize) |
| 3713 | self.assertEqual(list(buffer), expected) |
| 3714 | |
| 3715 | buffer = array.array('i', [0]*10) |
| 3716 | expected = [0] * 3 + list(arr) + [0] * (10 - 3 - len(arr)) |
| 3717 | self.assertEqual(conn.send_bytes(arr), None) |
| 3718 | self.assertEqual(conn.recv_bytes_into(buffer, 3 * buffer.itemsize), |
| 3719 | len(arr) * buffer.itemsize) |
| 3720 | self.assertEqual(list(buffer), expected) |
| 3721 | |
| 3722 | buffer = bytearray(latin(' ' * 40)) |
| 3723 | self.assertEqual(conn.send_bytes(longmsg), None) |
| 3724 | try: |
| 3725 | res = conn.recv_bytes_into(buffer) |
| 3726 | except multiprocessing.BufferTooShort as e: |
| 3727 | self.assertEqual(e.args, (longmsg,)) |
| 3728 | else: |
| 3729 | self.fail('expected BufferTooShort, got %s' % res) |
| 3730 | |
| 3731 | poll = TimingWrapper(conn.poll) |
| 3732 | |
| 3733 | self.assertEqual(poll(), False) |
| 3734 | self.assertTimingAlmostEqual(poll.elapsed, 0) |
| 3735 | |
| 3736 | self.assertEqual(poll(-1), False) |
| 3737 | self.assertTimingAlmostEqual(poll.elapsed, 0) |
| 3738 | |
| 3739 | self.assertEqual(poll(TIMEOUT1), False) |
| 3740 | self.assertTimingAlmostEqual(poll.elapsed, TIMEOUT1) |
| 3741 | |
| 3742 | conn.send(None) |
| 3743 | time.sleep(.1) |
nothing calls this directly
no test coverage detected