(self)
| 4126 | |
| 4127 | @warnings_helper.ignore_fork_in_thread_deprecation_warnings() |
| 4128 | def test_dont_merge(self): |
| 4129 | a, b = self.Pipe() |
| 4130 | self.assertEqual(a.poll(0.0), False) |
| 4131 | self.assertEqual(a.poll(0.1), False) |
| 4132 | |
| 4133 | p = self.Process(target=self._child_dont_merge, args=(b,)) |
| 4134 | p.start() |
| 4135 | |
| 4136 | self.assertEqual(a.recv_bytes(), b'a') |
| 4137 | self.assertEqual(a.poll(1.0), True) |
| 4138 | self.assertEqual(a.poll(1.0), True) |
| 4139 | self.assertEqual(a.recv_bytes(), b'b') |
| 4140 | self.assertEqual(a.poll(1.0), True) |
| 4141 | self.assertEqual(a.poll(1.0), True) |
| 4142 | self.assertEqual(a.poll(0.0), True) |
| 4143 | self.assertEqual(a.recv_bytes(), b'cd') |
| 4144 | |
| 4145 | p.join() |
| 4146 | |
| 4147 | # |
| 4148 | # Test of sending connection and socket objects between processes |
nothing calls this directly
no test coverage detected