kenneth.skovhede@gmail.com
20a6757d56
git-svn-id: https://duplicati.googlecode.com/svn/trunk@11 59da171f-624f-0410-aa54-27559c288bec
47 lines
1.6 KiB
Python
47 lines
1.6 KiB
Python
import unittest
|
|
|
|
|
|
import win32pipe
|
|
import win32file
|
|
import pywintypes
|
|
import winerror
|
|
import threading
|
|
|
|
class CurrentUserTestCase(unittest.TestCase):
|
|
pipename = "\\\\.\\pipe\\python_test_pipe"
|
|
def _workerThread(self, e):
|
|
data = win32pipe.CallNamedPipe(self.pipename,"foo\0bar", 1024, win32pipe.NMPWAIT_WAIT_FOREVER)
|
|
e.set()
|
|
self.failUnless(data == "bar\0foo")
|
|
|
|
def testCallNamedPipe(self):
|
|
openMode = win32pipe.PIPE_ACCESS_DUPLEX
|
|
pipeMode = win32pipe.PIPE_TYPE_MESSAGE | win32pipe.PIPE_WAIT
|
|
|
|
sa = pywintypes.SECURITY_ATTRIBUTES()
|
|
sa.SetSecurityDescriptorDacl ( 1, None, 0 )
|
|
|
|
pipeHandle = win32pipe.CreateNamedPipe(self.pipename,
|
|
openMode,
|
|
pipeMode,
|
|
win32pipe.PIPE_UNLIMITED_INSTANCES,
|
|
0,
|
|
0,
|
|
2000,
|
|
sa)
|
|
|
|
event = threading.Event()
|
|
threading.Thread(target=self._workerThread, args=(event,)).start()
|
|
|
|
hr = win32pipe.ConnectNamedPipe(pipeHandle)
|
|
self.failUnlessEqual(0, hr)
|
|
win32file.WriteFile(pipeHandle, "bar\0foo")
|
|
hr, got = win32file.ReadFile(pipeHandle, 100)
|
|
self.failUnless(got == "foo\0bar")
|
|
event.wait(5)
|
|
if not event.isSet():
|
|
self.fail("Failed to wait for event!")
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|