test_win32file.py 文件源码

def acceptWorker(self, port, running_event, stopped_event):
        listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        listener.bind(('', port))
        listener.listen(200)

        # create accept socket
        accepter = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        # An overlapped
        overlapped = pywintypes.OVERLAPPED()
        overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
        # accept the connection.
        # We used to allow strings etc to be passed here, and they would be
        # modified!  Obviously this is evil :)
        buffer = " " * 1024 # EVIL - SHOULD NOT BE ALLOWED.
        self.assertRaises(TypeError, win32file.AcceptEx, listener, accepter, buffer, overlapped)

        # This is the correct way to allocate the buffer...
        buffer = win32file.AllocateReadBuffer(1024)
        rc = win32file.AcceptEx(listener, accepter, buffer, overlapped)
        self.failUnlessEqual(rc, winerror.ERROR_IO_PENDING)
        # Set the event to say we are all ready
        running_event.set()
        # and wait for the connection.
        rc = win32event.WaitForSingleObject(overlapped.hEvent, 2000)
        if rc == win32event.WAIT_TIMEOUT:
            self.fail("timed out waiting for a connection")
        nbytes = win32file.GetOverlappedResult(listener.fileno(), overlapped, False)
        #fam, loc, rem = win32file.GetAcceptExSockaddrs(accepter, buffer)
        accepter.send(buffer[:nbytes])
        # NOT set in a finally - this means *successfully* stopped!
        stopped_event.set()