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

[Chris McDonough]

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

Modified Files:
      Tag: Zope-2_7-branch
	test_responses.py 
Log Message:
Merge chrism-publishfile-branch.  See http://dev.zope.org/Wikis/DevSite/Proposals/FasterStaticContentServing for more information.


=== Zope/lib/python/ZServer/tests/test_responses.py 1.1.2.1 => 1.1.2.2 ===
--- 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 28 06:11:49 2004
@@ -18,8 +18,9 @@
 from ZServer.FTPResponse import FTPResponse
 from ZServer.PCGIServer import PCGIResponse
 from ZServer.FCGIServer import FCGIResponse
+from ZPublisher.Iterators import IStreamIterator
 import unittest
-
+from cStringIO import StringIO
 
 class ZServerResponseTestCase(unittest.TestCase):
     """Test ZServer response objects."""
@@ -40,7 +41,56 @@
         response = FCGIResponse()
         self.assertRaises(TypeError, response.write, u'bad')
 
-
+    def test_setBodyIterator(self):
+        channel = DummyChannel()
+        one = ZServerHTTPResponse(stdout=channel)
+        one.setHeader('content-length', 5)
+        one.setBody(test_streamiterator())
+        one.outputBody()
+        all = channel.all()
+        lines = all.split('\r\n')
+        self.assertEqual(lines[-2], '')    # end of headers
+        self.assertEqual(lines[-1], 'hello') # payload
+
+    def test_setBodyIteratorFailsWithoutContentLength(self):
+        one = ZServerHTTPResponse(stdout=DummyChannel())
+        self.assertRaises(AssertionError,
+                          one.setBody, test_streamiterator())
+
+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)
+
+class test_streamiterator:
+    __implements__ = IStreamIterator
+    data = "hello"
+    done = 0
+
+    def next(self):
+        if not self.done:
+            self.done = 1
+            return self.data
+        raise StopIteration
 
 def test_suite():
     return unittest.makeSuite(ZServerResponseTestCase)