Stanislav Hejny
124 lines
4.9 KiB
Python
124 lines
4.9 KiB
Python
from datetime import datetime, timezone
|
|
from unittest import TestCase
|
|
|
|
from amqp.adapter.data_message_factory import DataMessageFactory
|
|
|
|
|
|
class DataMessageFactoryTest(TestCase):
|
|
def test_from_bytes(self):
|
|
raw = "\002\027A64f3b5af4d00000a4000|m=POST|d=clever-pybook.localhost|p=/debug|h={Accept=[*/*],X-Forwarded-Host=[clever-pybook.localhost:8085],User-Agent=[curl/8.7.1],X-Forwarded-Proto=[http],X-Forwarded-For=[172.18.0.1],Host=[clever-pybook.localhost:8085],Accept-Encoding=[gzip],Content-Length=[1475],X-Forwarded-Port=[8085],X-Real-Ip=[172.18.0.1],X-Forwarded-Server=[c5945bef67d1],Content-Type=[multipart/form-data; boundary=------------------------j1RjIqLrFH07XMsyMvpS1v]}|t={traceparent=00-bed6eb7de0d8ccf4c80b145a4bfec357-8d1f3bafc471bf0f-01}|b=TEST_ME"
|
|
msg = DataMessageFactory.from_bytes(raw.encode("utf-8"))
|
|
self.assertIsNotNone(msg.id)
|
|
self.assertEqual(
|
|
msg.id().as_string(),
|
|
"64F3B5AF4D00000A4000".lower(),
|
|
"SnowflakeID incorrectly parsed.",
|
|
)
|
|
self.assertEqual(
|
|
msg.headers()["Content-Type"][0],
|
|
"multipart/form-data; boundary=------------------------j1RjIqLrFH07XMsyMvpS1v",
|
|
)
|
|
|
|
def test_generate_snowflake_id(self):
|
|
_id = DataMessageFactory.get_instance(1).generate_snowflake_id()
|
|
self.assertEqual("00001000", _id.as_string()[-8:], "Instance ID not set correctly")
|
|
|
|
_id = DataMessageFactory(2).generate_snowflake_id()
|
|
self.assertEqual("00002000", _id.as_string()[-8:], "Instance ID not set correctly")
|
|
|
|
_f = DataMessageFactory(127)
|
|
_i0 = _f.generate_snowflake_id()
|
|
_i1 = _f.generate_snowflake_id()
|
|
_i2 = _f.generate_snowflake_id()
|
|
_i3 = _f.generate_snowflake_id()
|
|
self.assertEqual(
|
|
"0007F000".lower(),
|
|
_i0.as_string()[-8:],
|
|
"Instance ID and sequence not set correctly",
|
|
)
|
|
self.assertEqual(
|
|
"0007F001".lower(),
|
|
_i1.as_string()[-8:],
|
|
"Instance ID and sequence not set correctly",
|
|
)
|
|
self.assertEqual(
|
|
"0007F002".lower(),
|
|
_i2.as_string()[-8:],
|
|
"Instance ID and sequence not set correctly",
|
|
)
|
|
self.assertEqual(
|
|
"0007F003".lower(),
|
|
_i3.as_string()[-8:],
|
|
"Instance ID and sequence not set correctly",
|
|
)
|
|
|
|
def test_create_request_message(self):
|
|
_f = DataMessageFactory(34)
|
|
_req = _f.create_request_message(
|
|
"PUT",
|
|
"localhost",
|
|
"/test/me",
|
|
{"k1": ["v1"], "k2": ["v2"]},
|
|
"Quick brown fox",
|
|
)
|
|
self.assertEqual("localhost", _req.domain(), "Domain not set correctly")
|
|
self.assertEqual("/test/me", _req.path(), "Context path not set correctly")
|
|
self.assertEqual("v1", _req.headers()["k1"][0], "Headers not set correctly")
|
|
self.assertEqual(b"Quick brown fox", _req.body(), "Body not correctly decoded")
|
|
|
|
def test_serialize(self):
|
|
_f = DataMessageFactory(34)
|
|
_req = _f.create_request_message(
|
|
"PUT",
|
|
"localhost",
|
|
"/test/me",
|
|
{"k1": ["v1"], "k2": ["v2"]},
|
|
"Quick brown fox",
|
|
)
|
|
_ser = _f.serialize(_req)
|
|
self.assertEqual(65, _ser[2])
|
|
_deser = _f.from_bytes(_ser)
|
|
self.assertEqual(_req.id().as_string(), _deser.id().as_string())
|
|
self.assertEqual("localhost", _deser.domain())
|
|
self.assertEqual("/test/me", _deser.path())
|
|
self.assertEqual(b"Quick brown fox", _deser.body())
|
|
self.assertEqual("v2", _deser.headers()["k2"][0])
|
|
|
|
def test_amq_message_routing_key(self):
|
|
_f = DataMessageFactory(35)
|
|
_req = _f.create_request_message(
|
|
"PUT",
|
|
"localhost",
|
|
"/test/me",
|
|
{"k1": ["v1"], "k2": ["v2"]},
|
|
"Quick brown fox",
|
|
)
|
|
_key = _f.amq_message_routing_key(_req)
|
|
self.assertEqual(
|
|
"localhost.test.me",
|
|
_key,
|
|
"Message routing key wrongly extracted from the message",
|
|
)
|
|
|
|
def test_get_timestamp_from_id(self):
|
|
_f = DataMessageFactory(35)
|
|
_timestamp = _f.get_timestamp_from_id(_f.generate_snowflake_id())
|
|
_x_timestamp = int(datetime.now(timezone.utc).timestamp() * 1000)
|
|
self.assertLessEqual(
|
|
_x_timestamp - _timestamp,
|
|
1,
|
|
"Timestamp in SnowflakeId differs too much from current_availability time",
|
|
)
|
|
|
|
def test_to_string(self):
|
|
_f = DataMessageFactory(35)
|
|
_req = _f.create_request_message(
|
|
"PUT",
|
|
"localhost",
|
|
"/test/me",
|
|
{"k1": ["v1"], "k2": ["v2"]},
|
|
"Quick brown fox",
|
|
)
|
|
_as_str = _f.to_string(_req)
|
|
self.assertGreater(_as_str.index("domain='localhost'"), 0, "Expected content not found")
|