Update of /cvsroot/pywin32/pywin32/win32/test
In directory sc8-pr-cvs1.sourceforge.net:/tmp/cvs-serv32483/test
Modified Files:
test_win32file.py
Log Message:
As noticed by Roger, AcceptEx and WSARecv used the buffer interface
incorrectly, so allowed you to pass strings as a buffer that would be
modified!
Index: test_win32file.py
===================================================================
RCS file: /cvsroot/pywin32/pywin32/win32/test/test_win32file.py,v
retrieving revision 1.12
retrieving revision 1.13
diff -C2 -d -r1.12 -r1.13
*** test_win32file.py 20 Nov 2006 22:21:48 -0000 1.12
--- test_win32file.py 28 May 2008 04:32:22 -0000 1.13
***************
*** 9,12 ****
--- 9,13 ----
import time
import shutil
+ import socket
class TestSimpleOps(unittest.TestCase):
***************
*** 155,160 ****
# Mainly checking that we can "associate" an existing handle. This
# failed in build 203.
- import socket
-
ioport = win32file.CreateIoCompletionPort(win32file.INVALID_HANDLE_VALUE,
0, 0, 0)
--- 156,159 ----
***************
*** 253,256 ****
--- 252,315 ----
d[overlapped] = "hello"
+ class TestSocketExtensions(unittest.TestCase):
+ 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()
+
+ def testAcceptEx(self):
+ port = 4680
+ running = threading.Event()
+ stopped = threading.Event()
+ t = threading.Thread(target=self.acceptWorker, args=(port, running,stopped))
+ t.start()
+ running.wait(2)
+ if not running.isSet():
+ self.fail("AcceptEx Worker thread failed to start")
+ s = socket.create_connection(('127.0.0.1', port), 10)
+ win32file.WSASend(s, "hello", None)
+ overlapped = pywintypes.OVERLAPPED()
+ overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
+ # Like above - WSARecv used to allow strings as the receive buffer!!
+ buffer = " " * 10
+ self.assertRaises(TypeError, win32file.WSARecv, s, buffer, overlapped)
+ # This one should work :)
+ buffer = win32file.AllocateReadBuffer(10)
+ win32file.WSARecv(s, buffer, overlapped)
+ nbytes = win32file.GetOverlappedResult(s.fileno(), overlapped, True)
+ got = buffer[:nbytes]
+ self.failUnlessEqual(got, "hello")
+ # thread should have stopped
+ stopped.wait(2)
+ if not stopped.isSet():
+ self.fail("AcceptEx Worker thread failed to successfully stop")
+
class TestFindFiles(unittest.TestCase):
def testIter(self):
|
