[Inteproxy-commits] r333 - in branches/streaming: . test

[scm-commit@wald.intevation.org]

Author: teichmann
Date: 2012-01-05 13:04:46 +0100 (Thu, 05 Jan 2012)
New Revision: 333

Added:
   branches/streaming/test/test_chunkedwriter.py
Modified:
   branches/streaming/ChangeLog
Log:
Added unit tests for module inteproxy.chunkedwriter.

Modified: branches/streaming/ChangeLog
===================================================================
--- branches/streaming/ChangeLog	2012-01-03 17:31:29 UTC (rev 332)
+++ branches/streaming/ChangeLog	2012-01-05 12:04:46 UTC (rev 333)
@@ -1,3 +1,8 @@
+2012-01-05	Sascha L. Teichmann	<sascha.teichmann at intevation.de>
+
+	* test/test_chunkedwriter.py: New. Unit tests for the chunkedwriter
+	  module.
+
 2012-01-03	Sascha L. Teichmann	<sascha.teichmann at intevation.de>
 
 	* inteproxy/proxycore.py: Fixed: If a response is rewritten in

Added: branches/streaming/test/test_chunkedwriter.py
===================================================================
--- branches/streaming/test/test_chunkedwriter.py	2012-01-03 17:31:29 UTC (rev 332)
+++ branches/streaming/test/test_chunkedwriter.py	2012-01-05 12:04:46 UTC (rev 333)
@@ -0,0 +1,50 @@
+# Copyright (C) 2012 by Intevation GmbH
+# Authors:
+# Sascha L. Teichmann <sascha.teichmann at intevation.de>
+#
+# This program is free software under the GPL (>=v2)
+# Read the file COPYING coming with the software for details
+
+"""Tests for the inteproxy.chunkedwriter module"""
+
+import unittest
+import StringIO
+
+from random import Random
+
+from inteproxy.chunkedwriter import ChunkedTransferEncodingWriter
+
+class ChunkedTransferEncodingWriterTest(unittest.TestCase):
+
+    def test_chunked_transfer_encoding_writer(self):
+        """Test for the ChunkedTransferEncodingWriter"""
+
+        SEED  = 339394899170L
+        CHUNK = 4096
+        SMALL = 100
+
+        raw = [chr(c)*CHUNK for c in range(ord('A'), ord('Z')+1)]
+        source = ''.join(raw)
+
+        output = StringIO.StringIO()
+        write = output.write
+
+        writer = ChunkedTransferEncodingWriter(write, CHUNK)
+
+        r = Random(SEED)
+        randint, rand = r.randint, r.random
+        rest = len(source)
+        while rest:
+            to_write = (randint(1, SMALL) 
+                        if rand() < 0.7
+                        else randint(CHUNK-SMALL, CHUNK+SMALL))
+            to_write = min(rest, to_write)
+            p = len(source)-rest
+            writer.append(source[p:p+to_write])
+            rest -= to_write
+
+        writer.finish()
+        result = output.getvalue()
+        expected = '%x\r\n%s\r\n0\r\n\r\n' % (CHUNK, (
+                                              '\r\n%x\r\n' % CHUNK).join(raw))
+        self.assertEqual(result, expected)