test_socket.py 文件源码

def test_stopOnlyCloses(self):
        """
        When the L{IListeningPort} returned by
        L{IReactorSocket.adoptDatagramPort} is stopped using
        C{stopListening}, the underlying socket is closed but not
        shutdown.  This allows another process which still has a
        reference to it to continue reading and writing to it.
        """
        reactor = self.buildReactor()

        portSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.addCleanup(portSocket.close)

        portSocket.bind(("127.0.0.1", 0))
        portSocket.setblocking(False)

        # The file descriptor is duplicated by adoptDatagramPort
        port = reactor.adoptDatagramPort(
            portSocket.fileno(), portSocket.family, DatagramProtocol())
        d = port.stopListening()
        def stopped(ignored):
            # Should still be possible to recv on portSocket.  If
            # it was shutdown, the exception would be EINVAL instead.
            exc = self.assertRaises(socket.error, portSocket.recvfrom, 1)
            if platform.isWindows() and _PY3:
                self.assertEqual(exc.args[0], errno.WSAEWOULDBLOCK)
            else:
                self.assertEqual(exc.args[0], errno.EAGAIN)
        d.addCallback(stopped)
        d.addErrback(err, "Failed to read on original port.")

        needsRunningReactor(
            reactor,
            lambda: d.addCallback(lambda ignored: reactor.stop()))

        reactor.run()