Stanislav Hejny
57 lines
2.4 KiB
Python
57 lines
2.4 KiB
Python
from unittest import TestCase
|
|
|
|
from amqp.adapter.service_message_factory import (
|
|
INVALID_MESSAGE,
|
|
RS,
|
|
ServiceMessageFactory,
|
|
)
|
|
from amqp.model.model import ServiceMessage
|
|
from amqp.model.service_message_type import ServiceMessageType
|
|
|
|
|
|
class ServiceMessageTest(TestCase):
|
|
|
|
def test_to_string(self):
|
|
_f: ServiceMessageFactory = ServiceMessageFactory("bumble-bee")
|
|
message: ServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA, "Duck", "Donald")
|
|
message_str = _f.to_string(message)
|
|
self.assertTrue("|type=01|" in message_str)
|
|
self.assertTrue("|body=Duck" in message_str)
|
|
message: ServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA_REPLACE, "Duck", "Donald")
|
|
message_str = _f.to_string(message)
|
|
self.assertTrue("|type=03|" in message_str)
|
|
self.assertTrue("|body=Duck" in message_str)
|
|
self.assertTrue("|replyTo=bumble-bee" in message_str)
|
|
nullMsg = _f.to_string(None)
|
|
self.assertEqual("null", nullMsg)
|
|
|
|
def test_from(self):
|
|
_f: ServiceMessageFactory = ServiceMessageFactory("amqp-router")
|
|
message = _f.of(ServiceMessageType.ROUTING_DATA, "Duck", "Donald")
|
|
serialized = _f.serialize(message)
|
|
deserialized = _f.from_bytes(serialized)
|
|
assert deserialized.is_valid()
|
|
assert deserialized.message_type == ServiceMessageType.ROUTING_DATA
|
|
assert deserialized.recipient_name == "Donald"
|
|
assert deserialized.message == "Duck"
|
|
assert deserialized.reply_to == "amqp-router"
|
|
invalidMsg = _f.from_bytes(b"wrong")
|
|
self.assertEqual(invalidMsg, INVALID_MESSAGE)
|
|
|
|
def test_base64(self):
|
|
_f: ServiceMessageFactory = ServiceMessageFactory("amqp-router")
|
|
message = _f.as_base64(ServiceMessageType.ROUTING_DATA, "Duck", "Donald")
|
|
serialized = _f.serialize(message)
|
|
assert b"|81|" in serialized
|
|
deserialized = _f.from_bytes(serialized)
|
|
assert deserialized.is_valid()
|
|
assert deserialized.message_type == ServiceMessageType.ROUTING_DATA
|
|
assert deserialized.message == "Duck"
|
|
|
|
def test_serialize(self):
|
|
_f: ServiceMessageFactory = ServiceMessageFactory("amqp-router")
|
|
message = _f.of(ServiceMessageType.ROUTING_DATA, "Duck", "Donald")
|
|
serialized = _f.serialize(message)
|
|
assert RS + "0" + str(ServiceMessageType.ROUTING_DATA.value)
|
|
assert len(serialized) > 16
|