fix writing unions of bytes or strings on py3
MODIFY
fastavro/six.py
fastavro/six.py
@@ -27,7 +27,7 @@ if sys.version_info >= (3, 0):
return obj.items()
def py3_is_str(obj):
- return isinstance(obj, (bytes, str,))
+ return isinstance(obj, str)
def py3_mk_bits(bits):
return bytes([bits & 0xff])

MODIFY
tests/test_str_py3.py
tests/test_str_py3.py
@@ -5,16 +5,16 @@ from string import ascii_uppercase, digits
import fastavro
+
+def test_str_py3():
letters = ascii_uppercase + digits
id_size = 100
seed('str_py3') # Repeatable results
-
def gen_id():
return ''.join(choice(letters) for _ in range(id_size))
-
keys = ['first', 'second', 'third', 'fourth']
testdata = [dict((key, gen_id()) for key in keys) for _ in range(50)]
@@ -26,8 +26,6 @@ schema = {
"type": "record"
}
-
-def test_str_py3():
buf = BytesIO()
fastavro.writer(buf, schema, testdata)
@@ -41,5 +39,18 @@ def test_str_py3():
assert rec == testdata[-1], 'bad last record'
-if __name__ == '__main__':
- test_str_py3()
+def test_py3_union_string_and_bytes():
+ schema = {
+ "fields": [{'name': 'field', 'type': ['string', 'bytes']}],
+ "namespace": "namespace",
+ "name": "union_string_bytes",
+ "type": "record"
+ }
+
+ records = [
+ {'field': u'string'},
+ {'field': b'bytes'}
+ ]
+
+ buf = BytesIO()
+ fastavro.writer(buf, schema, records)

fix writing unions of bytes or strings on py3