(self)
| 5469 | conn.close() |
| 5470 | |
| 5471 | def test_timeout(self): |
| 5472 | old_timeout = socket.getdefaulttimeout() |
| 5473 | try: |
| 5474 | socket.setdefaulttimeout(0.1) |
| 5475 | parent, child = multiprocessing.Pipe(duplex=True) |
| 5476 | l = multiprocessing.connection.Listener(family='AF_INET') |
| 5477 | p = multiprocessing.Process(target=self._test_timeout, |
| 5478 | args=(child, l.address)) |
| 5479 | p.start() |
| 5480 | child.close() |
| 5481 | self.assertEqual(parent.recv(), 123) |
| 5482 | parent.close() |
| 5483 | conn = l.accept() |
| 5484 | self.assertEqual(conn.recv(), 456) |
| 5485 | conn.close() |
| 5486 | l.close() |
| 5487 | join_process(p) |
| 5488 | finally: |
| 5489 | socket.setdefaulttimeout(old_timeout) |
| 5490 | |
| 5491 | # |
| 5492 | # Test what happens with no "if __name__ == '__main__'" |
nothing calls this directly
no test coverage detected