--- 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()