test_win32pipe.py 文件源码

def _serverThread(self, pipe_handle, event, wait_time):
        # just do one connection and terminate.
        hr = win32pipe.ConnectNamedPipe(pipe_handle)
        self.failUnless(hr in (0, winerror.ERROR_PIPE_CONNECTED), "Got error code 0x%x" % (hr,))
        hr, got = win32file.ReadFile(pipe_handle, 100)
        self.failUnlessEqual(got, str2bytes("foo\0bar"))
        time.sleep(wait_time)
        win32file.WriteFile(pipe_handle, str2bytes("bar\0foo"))
        pipe_handle.Close()
        event.set()