[Zope-Checkins] CVS: Zope/lib/python/ZServer/tests - test_responses.py:1.1.2.1.4.1

[Chris McDonough]

Update of /cvs-repository/Zope/lib/python/ZServer/tests
In directory cvs.zope.org:/tmp/cvs-serv11574/tests

Modified Files:
      Tag: chrism-publishfile-branch
	test_responses.py 
Log Message:
Initial stab at being able to return producer objects from Zope code instead of strings (to allow the serving of file-like objects to be faster).


=== Zope/lib/python/ZServer/tests/test_responses.py 1.1.2.1 => 1.1.2.1.4.1 ===
--- Zope/lib/python/ZServer/tests/test_responses.py:1.1.2.1	Thu Jan  8 18:34:03 2004
+++ Zope/lib/python/ZServer/tests/test_responses.py	Sun Mar 21 14:38:52 2004
@@ -19,7 +19,7 @@
 from ZServer.PCGIServer import PCGIResponse
 from ZServer.FCGIServer import FCGIResponse
 import unittest
-
+from cStringIO import StringIO
 
 class ZServerResponseTestCase(unittest.TestCase):
     """Test ZServer response objects."""
@@ -40,7 +40,60 @@
         response = FCGIResponse()
         self.assertRaises(TypeError, response.write, u'bad')
 
-
+    def test_setBodyProducer(self):
+        channel = DummyChannel()
+        one = ZServerHTTPResponse(stdout=channel)
+        one.setHeader('content-length', 5)
+        one.setBodyProducer(dummy_producer("hello"))
+        expected = 'HTTP/1.0 200 OK\r\nServer: Zope/2.0 ZServer/2.0\r\nDate: Sun, 21 Mar 2004 19:28:47 GMT\r\nContent-Length: 5\r\nEtag: \r\nConnection: close\r\n\r\nhello'
+        one.outputBody()
+        all = channel.all()
+        lines = all.split('\r\n')
+        self.assertEqual(lines[7], 'hello')
+
+    def test_setBodyProducerFailsWithoutContentLength(self):
+        one = ZServerHTTPResponse(stdout=DummyChannel())
+        self.assertRaises(AssertionError,
+                          one.setBodyProducer, dummy_producer("hello"))
+
+    def test_setBadBodyProducer(self):
+        one = ZServerHTTPResponse(stdout=DummyChannel())
+        one.setHeader('content-length', 5)
+        self.assertRaises(AssertionError, one.setBodyProducer, None)
+
+class dummy_producer:
+    def __init__(self, data):
+        self.data = data
+        self._done = 0
+        
+    def more(self):
+        if not self._done:
+            self._done = 1
+            return self.data
+
+class DummyChannel:
+    def __init__(self):
+        self.out = StringIO()
+
+    def all(self):
+        self.out.seek(0)
+        return self.out.read()
+
+    def read(self):
+        pass
+
+    def write(self, data, len=None):
+        try:
+            if isinstance(data, str):
+                self.out.write(data)
+                return
+        except TypeError:
+            pass
+        while 1:
+            s = data.more()
+            if not s:
+                break
+            self.out.write(s)
 
 def test_suite():
     return unittest.makeSuite(ZServerResponseTestCase)