PROTON-2531: Use Latin-1 when converting Delivery tag between bytes a…

…nd string

python/cproton.py

@@ -204,12 +204,12 @@ def bytes2string(b, encoding='utf8'):
     return ffi.unpack(b.start, b.size).decode(encoding)
 
 
-def py2bytes(py):
+def py2bytes(py, encoding='utf8'):
     if isinstance(py, (bytes, bytearray, memoryview)):
         s = ffi.from_buffer(py)
         return len(s), s
     elif isinstance(py, str):
-        s = ffi.from_buffer(py.encode('utf8'))
+        s = ffi.from_buffer(py.encode(encoding))
         return len(s), s
 
 
@@ -487,7 +487,7 @@ def pn_data_get_symbol(data):
 
 
 def pn_delivery_tag(delivery):
-    return bytes2string(lib.pn_delivery_tag(delivery))
+    return bytes2string(lib.pn_delivery_tag(delivery), 'latin-1')
 
 
 def pn_connection_get_container(connection):
@@ -543,7 +543,7 @@ def pn_receiver(session, name):
 
 
 def pn_delivery(link, tag):
-    return lib.pn_delivery(link, py2bytes(tag))
+    return lib.pn_delivery(link, py2bytes(tag, 'latin-1'))
 
 
 def pn_link_name(link):

python/tests/proton_tests/engine.py

@@ -978,6 +978,40 @@ def test_disposition(self):
         assert sd.local_state == rd.remote_state == Delivery.ACCEPTED
         sd.settle()
 
+    def test_delivery_tag_bytes(self):
+        test_tags = [
+            (bytes([0, 0, 0, 128]), '\x00\x00\x00\x80'),
+            ('tag', 'tag'),
+            (b'tag', 'tag'),
+            (bytearray([1, 2, 32, 254, 255]), '\x01\x02 \xfe\xff'),
+            (b'\xff'+(29*b' ')+b'\xff\x00', 'ÿ                             ÿ\x00')
+        ]
+
+        self.rcv.flow(len(test_tags))
+
+        self.pump(64 * len(test_tags))
+
+        for tt in test_tags:
+            self.snd.delivery(tt[0])
+            msg = b"this is a test"
+            self.snd.send(msg)
+            assert self.snd.advance()
+
+        self.pump(64 * len(test_tags))
+
+        for tt in test_tags:
+            rd = self.rcv.current
+            assert rd is not None
+            assert rd.tag == tt[1], (rd.tag, tt[1])
+            rd.update(Delivery.ACCEPTED)
+            rd.settle()
+
+        try:
+            self.snd.delivery(chr(256))
+            assert False, "Expected UnicodeEncodeError"
+        except UnicodeEncodeError:
+            pass
+
     def test_delivery_id_ordering(self):
         self.rcv.flow(1024)
         self.pump(buffer_size=64 * 1024)