test_executable.py 文件源码

def test_listen(self):
        logger = self.useFixture(FakeLogger(format="%(asctime)s %(message)s"))
        self.executable.listen(6666)
        self.executable.sleep(1)
        self.executable.spawn()

        # Try to connect to the socket
        sock = socket.socket()
        transient = (errno.ECONNREFUSED, errno.ECONNRESET)

        attempts = 10
        for i in range(1, attempts + 1):
            try:
                sock.connect(("127.0.0.1", self.executable.port))
            except socket.error as error:  # pragma: no cover
                if error.errno in transient and i != attempts:
                    time.sleep(0.05 * i)
                    continue
                logging.error("connection attempt %d failed", i)
                raise error
            break
        self.assertEqual("127.0.0.1", sock.getsockname()[0])