# HG changeset patch # User RĂ©mi Cardona <remi.cardona@free.fr> # Date 1442314287 -7200 # Node ID f454404733c1e6cc0d542d45e2021e0e50be6fef # Parent 6c497fe389d26ecd3a87d985e47eb12e1076fb41 Port cw.Binary to io.BytesIO And add plenty of tests too! diff -r 6c497fe389d2 -r f454404733c1 __init__.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 diff -r 6c497fe389d2 -r f454404733c1 test/unittest_binary.py --- /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()