[pywin32-checkins] pywin32/win32/test test_win32file.py,1.12,1.13

[Mark H. <mha...@us...>]

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):