Port cw.Binary to io.BytesIO
authorRémi Cardona <remi.cardona@free.fr>
Tue, 15 Sep 2015 12:51:27 +0200
changeset 10616 f454404733c1
parent 10615 6c497fe389d2
child 10617 c36cda9074c5
Port cw.Binary to io.BytesIO And add plenty of tests too!
__init__.py
test/unittest_binary.py
--- a/__init__.py	Fri Sep 18 18:26:07 2015 +0200
+++ b/__init__.py	Tue Sep 15 12:51:27 2015 +0200
@@ -29,6 +29,7 @@
 
 
 import __builtin__
+from six import PY2, binary_type
 # '_' is available in builtins to mark internationalized string but should
 # not be used to do the actual translation
 if not hasattr(__builtin__, '_'):
@@ -37,7 +38,7 @@
 CW_SOFTWARE_ROOT = __path__[0]
 
 import sys, os, logging
-from StringIO import StringIO
+from io import BytesIO
 
 from six.moves import cPickle as pickle
 
@@ -67,17 +68,19 @@
 #import threading
 #threading.settrace(log_thread)
 
-class Binary(StringIO):
-    """customize StringIO to make sure we don't use unicode"""
-    def __init__(self, buf=''):
-        assert isinstance(buf, (str, buffer, bytearray)), \
-               "Binary objects must use raw strings, not %s" % buf.__class__
-        StringIO.__init__(self, buf)
+class Binary(BytesIO):
+    """class to hold binary data. Use BytesIO to prevent use of unicode data"""
+    _allowed_types = (binary_type, bytearray, buffer if PY2 else memoryview)
+
+    def __init__(self, buf=b''):
+        assert isinstance(buf, self._allowed_types), \
+               "Binary objects must use bytes/buffer objects, not %s" % buf.__class__
+        super(Binary, self).__init__(buf)
 
     def write(self, data):
-        assert isinstance(data, (str, buffer, bytearray)), \
-               "Binary objects must use raw strings, not %s" % data.__class__
-        StringIO.write(self, data)
+        assert isinstance(data, self._allowed_types), \
+               "Binary objects must use bytes/buffer objects, not %s" % data.__class__
+        super(Binary, self).write(data)
 
     def to_file(self, fobj):
         """write a binary to disk
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/test/unittest_binary.py	Tue Sep 15 12:51:27 2015 +0200
@@ -0,0 +1,54 @@
+from six import PY2
+
+from unittest import TestCase
+from tempfile import NamedTemporaryFile
+import os.path as osp
+
+from logilab.common.shellutils import tempdir
+from cubicweb import Binary
+
+
+class BinaryTC(TestCase):
+    def test_init(self):
+        Binary()
+        Binary(b'toto')
+        Binary(bytearray(b'toto'))
+        if PY2:
+            Binary(buffer('toto'))
+        else:
+            Binary(memoryview(b'toto'))
+        with self.assertRaises((AssertionError, TypeError)):
+            # TypeError is raised by BytesIO if python runs with -O
+            Binary(u'toto')
+
+    def test_write(self):
+        b = Binary()
+        b.write(b'toto')
+        b.write(bytearray(b'toto'))
+        if PY2:
+            b.write(buffer('toto'))
+        else:
+            b.write(memoryview(b'toto'))
+        with self.assertRaises((AssertionError, TypeError)):
+            # TypeError is raised by BytesIO if python runs with -O
+            b.write(u'toto')
+
+    def test_gzpickle_roundtrip(self):
+        old = (u'foo', b'bar', 42, {})
+        new = Binary.zpickle(old).unzpickle()
+        self.assertEqual(old, new)
+        self.assertIsNot(old, new)
+
+    def test_from_file_to_file(self):
+        with tempdir() as dpath:
+            fpath = osp.join(dpath, 'binary.bin')
+            with open(fpath, 'wb') as fobj:
+                Binary(b'binaryblob').to_file(fobj)
+
+            bobj = Binary.from_file(fpath)
+            self.assertEqual(bobj.getvalue(), b'binaryblob')
+
+
+if __name__ == '__main__':
+    from unittest import main
+    main()