Merge pull request #169 from GooDer/master

Fixed maximum length of route metadata

Author: jell-o-fishi

rsocket/extensions/tagging.py

@@ -25,10 +25,10 @@ def _serialize_tags(self) -> bytes:
         serialized = b''
 
         for tag in list(map(ensure_bytes, self.tags)):
-            if len(tag) > 256:
-                raise RSocketError('Tag length longer than 256 characters: "%s"' % tag)
+            if len(tag) > 255:
+                raise RSocketError('Tag length longer than 255 characters: "%s"' % tag)
 
-            serialized += struct.pack('>b', len(tag))
+            serialized += struct.pack('>B', len(tag))
             serialized += tag
 
         return serialized
@@ -38,7 +38,7 @@ def parse(self, buffer: bytes):
         offset = 0
 
         while offset < len(buffer):
-            tag_length = struct.unpack('>b', buffer[offset:offset + 1])[0]
+            tag_length = struct.unpack('>B', buffer[offset:offset + 1])[0]
             offset += 1
             self.tags.append(buffer[offset:offset + tag_length])
             offset += tag_length

tests/rsocket/test_extentions.py

@@ -1,5 +1,11 @@
+import pytest
+import struct
+
+from rsocket.exceptions import RSocketError
 from rsocket.extensions.authentication import AuthenticationBearer, AuthenticationSimple
+from rsocket.extensions.mimetypes import WellKnownMimeTypes
 from rsocket.extensions.routing import RoutingMetadata
+from rsocket.extensions.tagging import TaggingMetadata
 
 
 def test_authentication_bearer():
@@ -39,3 +45,32 @@ def test_routing():
     parsed.parse(data)
 
     assert parsed == routing
+
+
+def test_tagging_metadata_serialize_max_length():
+    tag = 's' * 255
+    meta = TaggingMetadata(WellKnownMimeTypes.MESSAGE_RSOCKET_ROUTING, [tag])
+
+    serialized = meta.serialize()
+
+    length = struct.pack('>B', len(tag))
+    assert length + bytes(tag, 'utf-8') == serialized
+
+
+def test_tagging_metadata_serialize_exception_length():
+    tag = 's' * 256
+    meta = TaggingMetadata(WellKnownMimeTypes.MESSAGE_RSOCKET_ROUTING, [tag])
+
+    with pytest.raises(RSocketError) as e_info:
+        meta.serialize()
+
+    assert e_info.match(f'Tag length longer than 255 characters: "b\'{tag}\'"')
+
+
+def test_tagging_metadata_parse():
+    meta = TaggingMetadata(WellKnownMimeTypes.MESSAGE_RSOCKET_ROUTING)
+    tag = 's' * 255
+    length = struct.pack('>B', len(tag))
+
+    meta.parse(length + bytes(tag, 'utf-8'))
+    assert tag == meta.tags[0].decode()