SF.net SVN: fclient: [319] trunk/sandbox/fcp2/test_fcp

[<ju...@us...>]

Revision: 319
          http://fclient.svn.sourceforge.net/fclient/?rev=319&view=rev
Author:   jurner
Date:     2008-03-06 04:08:50 -0800 (Thu, 06 Mar 2008)

Log Message:
-----------
removed dummy socket. Use io now

Added Paths:
-----------
    trunk/sandbox/fcp2/test_fcp/dummy_io.py

Removed Paths:
-------------
    trunk/sandbox/fcp2/test_fcp/dummy_socket.py

Copied: trunk/sandbox/fcp2/test_fcp/dummy_io.py (from rev 318, trunk/sandbox/fcp2/test_fcp/dummy_socket.py)
===================================================================
--- trunk/sandbox/fcp2/test_fcp/dummy_io.py	                        (rev 0)
+++ trunk/sandbox/fcp2/test_fcp/dummy_io.py	2008-03-06 12:08:50 UTC (rev 319)
@@ -0,0 +1,108 @@
+"""Dummy socket object for testing"""
+
+import os, sys
+
+#--> rel import hack
+class _RelImportHack(object):
+	def __init__(self, n):
+		fpath = os.path.abspath(__file__)
+		for i in xrange(n): fpath = os.path.dirname(fpath)
+		sys.path.insert(0, fpath)
+	def __del__(self): sys.path.pop(0)
+hack = _RelImportHack(3)
+
+from fcp2 import client
+from fcp2.client import iohandler
+from fcp2 import consts
+
+del hack
+#<-- rel import hack
+#********************************************************************
+#
+#********************************************************************
+class TestIO(iohandler.IOObject):
+	
+	def __init__(self, ):
+		self.readBuffer = ''			# buffer client reads from
+		self.writeBuffer = ''			# buffer client writes to
+		
+		self._isOpen = False
+		self._isBroken = False
+		self._disallowConnect = False
+		self._reverseDirection = False
+		
+	def connect(self, **kwargs):
+		self._isOpen = True
+	
+	def read(self, n):
+		if self._isBroken:
+			raise iohandler.IOBroken('Broken')
+		if not self.isOpen():
+			raise iohandler.IOClosed('Closed')
+		
+		if self._reverseDirection:
+			if not self.writeBuffer:
+				raise iohandler.IOTimeout('Timeout')
+			bytes, self.writeBuffer = self.writeBuffer[ :n], self.writeBuffer[n: ]
+		else:
+			if not self.readBuffer:
+				raise iohandler.IOTimeout('Timeout')
+			bytes, self.readBuffer = self.readBuffer[ :n], self.readBuffer[n: ]
+		return bytes
+		
+	def write(self, bytes):
+		if self._isBroken:
+			raise iohandler.IOBroken('Broken')
+		if not self.isOpen():
+			raise iohandler.IOClosed('Closed')
+		self.writeBuffer += bytes
+			
+	def close(self):
+		self._isBroken = False
+		self._disallowConnect = False
+		self._reverseDirection = False
+		if self.isOpen():
+			self._isOpen = False
+			self.readBuffer = ''
+			self.writeBuffer = ''
+		else:
+			raise iohandler.IOClosed('Closed')
+		
+	def isOpen(self):
+		return self._isOpen
+	
+	def setTimeout(self, n):
+		pass
+	
+	
+	############################
+	## for testing...
+	
+	def setOpen(self, flag):
+		self._isOpen = flag
+
+	def setBroken(self, flag):
+		self._isBroken = flag
+
+	def setDissallowConnect(self, flag):
+		self.disallowConnect = flag
+
+	def setReverseDirection(self, flag):
+		self._reverseDirection = flag
+		
+	
+	def sendResponseMessage(self, name, data=None, **params):
+		buf = [name, ]
+		for name, value in params.items():
+			buf.append('%s=%s' % (name, value) )
+		if data is None:
+			buf.append('EndMessage\n')
+		else:
+			buf.append('Data\n')
+		self.readBuffer += '\n'.join(buf)
+		if data:
+			assert 'DataLength' in params
+			assert params['DataLength'] == len(data)
+			self.readBuffer += data
+		
+		

Deleted: trunk/sandbox/fcp2/test_fcp/dummy_socket.py
===================================================================
--- trunk/sandbox/fcp2/test_fcp/dummy_socket.py	2008-03-06 12:05:53 UTC (rev 318)
+++ trunk/sandbox/fcp2/test_fcp/dummy_socket.py	2008-03-06 12:08:50 UTC (rev 319)
@@ -1,108 +0,0 @@
-"""Dummy socket object for testing"""
-
-import os, sys
-
-#--> rel import hack
-class _RelImportHack(object):
-	def __init__(self, n):
-		fpath = os.path.abspath(__file__)
-		for i in xrange(n): fpath = os.path.dirname(fpath)
-		sys.path.insert(0, fpath)
-	def __del__(self): sys.path.pop(0)
-hack = _RelImportHack(3)
-
-from fcp2 import client
-from fcp2.client import iohandler
-from fcp2 import consts
-
-del hack
-#<-- rel import hack
-#********************************************************************
-#
-#********************************************************************
-class TestIO(iohandler.IOObject):
-	
-	def __init__(self, ):
-		self.readBuffer = ''			# buffer client reads from
-		self.writeBuffer = ''			# buffer client writes to
-		
-		self._isOpen = False
-		self._isBroken = False
-		self._disallowConnect = False
-		self._reverseDirection = False
-		
-	def connect(self, **kwargs):
-		self._isOpen = True
-	
-	def read(self, n):
-		if self._isBroken:
-			raise iohandler.IOBroken('Broken')
-		if not self.isOpen():
-			raise iohandler.IOClosed('Closed')
-		
-		if self._reverseDirection:
-			if not self.writeBuffer:
-				raise iohandler.IOTimeout('Timeout')
-			bytes, self.writeBuffer = self.writeBuffer[ :n], self.writeBuffer[n: ]
-		else:
-			if not self.readBuffer:
-				raise iohandler.IOTimeout('Timeout')
-			bytes, self.readBuffer = self.readBuffer[ :n], self.readBuffer[n: ]
-		return bytes
-		
-	def write(self, bytes):
-		if self._isBroken:
-			raise iohandler.IOBroken('Broken')
-		if not self.isOpen():
-			raise iohandler.IOClosed('Closed')
-		self.writeBuffer += bytes
-			
-	def close(self):
-		self._isBroken = False
-		self._disallowConnect = False
-		self._reverseDirection = False
-		if self.isOpen():
-			self._isOpen = False
-			self.readBuffer = ''
-			self.writeBuffer = ''
-		else:
-			raise iohandler.IOClosed('Closed')
-		
-	def isOpen(self):
-		return self._isOpen
-	
-	def setTimeout(self, n):
-		pass
-	
-	
-	############################
-	## for testing...
-	
-	def setOpen(self, flag):
-		self._isOpen = flag
-
-	def setBroken(self, flag):
-		self._isBroken = flag
-
-	def setDissallowConnect(self, flag):
-		self.disallowConnect = flag
-
-	def setReverseDirection(self, flag):
-		self._reverseDirection = flag
-		
-	
-	def sendResponseMessage(self, name, data=None, **params):
-		buf = [name, ]
-		for name, value in params.items():
-			buf.append('%s=%s' % (name, value) )
-		if data is None:
-			buf.append('EndMessage\n')
-		else:
-			buf.append('Data\n')
-		self.readBuffer += '\n'.join(buf)
-		if data:
-			assert 'DataLength' in params
-			assert params['DataLength'] == len(data)
-			self.readBuffer += data
-		
-		


This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.