diff --git a/amqp/adapter/serializer.py b/amqp/adapter/serializer.py index a112507..ce70d2e 100644 --- a/amqp/adapter/serializer.py +++ b/amqp/adapter/serializer.py @@ -18,7 +18,7 @@ def bytes_to_long(b: bytes) -> int: def read_long(bis: BytesIO) -> int: try: return struct.unpack("q", bis.read(8))[0] - except IOError: + except struct.error: return 0 @@ -51,7 +51,10 @@ def csv_as_list(input_str: str) -> List[str]: def add_to_map(map_: Dict[str, List[str]], token: str) -> None: - eq_idx = token.index("=") + try: + eq_idx = token.index("=") + except ValueError: + eq_idx = -1 # set to -1 when not found if eq_idx > 0: key = token[:eq_idx] map_[key] = csv_as_list(token[eq_idx + 1 :]) diff --git a/tests/adapter/test_serializer.py b/tests/adapter/test_serializer.py index d15a91f..3c3bc03 100644 --- a/tests/adapter/test_serializer.py +++ b/tests/adapter/test_serializer.py @@ -1,42 +1,114 @@ -def test_long_to_bytes(): - assert False +import unittest +from io import BytesIO + +from amqp.adapter.serializer import ( + bytes_to_long, + csv_as_list, + long_to_bytes, + map_as_string, + map_with_list_as_string, + object_or_list_as_string, + parse_map_list_string, + parse_map_string, + read_long, + str_to_bool, +) -def test_bytes_to_long(): - assert False +class TestSerializer(unittest.TestCase): + def test_long_to_bytes_and_bytes_to_long(self): + # Test conversion in both directions + test_numbers = [ + -9223372036854775808, + -1, + 0, + 1, + 9223372036854775807, + ] # Min and max long values + for num in test_numbers: + bytes_val = long_to_bytes(num) + self.assertEqual(bytes_to_long(bytes_val), num) + def test_read_long(self): + # Test reading valid long + test_num = 12345 + bio = BytesIO(long_to_bytes(test_num)) + self.assertEqual(read_long(bio), test_num) -def test_read_long(): - assert False + # Test reading from empty BytesIO + empty_bio = BytesIO() + self.assertEqual(read_long(empty_bio), 0) + def test_object_or_list_as_string(self): + # Test with string + self.assertEqual(object_or_list_as_string("test"), "test") -def test_object_or_list_as_string(): - assert False + # Test with list + self.assertEqual(object_or_list_as_string(["a", "b", "c"]), "[a,b,c]") + # Test with empty list + self.assertEqual(object_or_list_as_string([]), "[]") -def test_map_as_string(): - assert False + def test_map_as_string(self): + # Test with simple key-value pairs + test_map = {"key1": "value1", "key2": "value2"} + self.assertEqual(map_as_string(test_map), "key1=value1,key2=value2") + # Test with empty map + self.assertEqual(map_as_string({}), "") -def test_map_with_list_as_string(): - assert False + def test_map_with_list_as_string(self): + # Test with lists as values + test_map = {"key1": ["a", "b"], "key2": ["c", "d"]} + self.assertEqual(map_with_list_as_string(test_map), "key1=[a,b],key2=[c,d]") + # Test with empty map + self.assertEqual(map_with_list_as_string({}), "") -def test_parse_map_string(): - assert False + def test_parse_map_string(self): + # Test valid map string + test_str = "{key1=value1,key2=value2}" + expected = {"key1": "value1", "key2": "value2"} + self.assertEqual(parse_map_string(test_str), expected) + # Test empty map string + self.assertEqual(parse_map_string("{}"), {}) -def test_csv_as_list(): - assert False + # Test invalid format + self.assertEqual(parse_map_string("invalid"), {}) + def test_csv_as_list(self): + # Test normal CSV string + self.assertEqual(csv_as_list("a,b,c"), ["a", "b", "c"]) -def test_add_to_map(): - assert False + # Test with spaces + self.assertEqual(csv_as_list(" a , b , c "), ["a", "b", "c"]) + # Test with brackets + self.assertEqual(csv_as_list("[a,b,c]"), ["a", "b", "c"]) -def test_parse_map_list_string(): - assert False + # Test empty list + self.assertEqual(csv_as_list("[]"), [""]) + def test_parse_map_list_string(self): + # Test valid map list string + test_str = "{key1=[a,b],key2=[c,d]}" + expected = {"key1": ["a", "b"], "key2": ["c", "d"]} + self.assertEqual(parse_map_list_string(test_str), expected) -def test_str_to_bool(): - assert False + # Test empty map + self.assertEqual(parse_map_list_string("{}"), {}) + + # Test single entry + self.assertEqual(parse_map_list_string("{key1=[a,b]}"), {"key1": ["a", "b"]}) + + def test_str_to_bool(self): + # Test true values + true_values = ["true", "yes", "1", "t", "y", "on", "TRUE", "YES", "T", "Y", "ON"] + for value in true_values: + self.assertTrue(str_to_bool(value)) + + # Test false values + false_values = ["false", "no", "0", "f", "n", "off", "FALSE", "NO", "F", "N", "OFF"] + for value in false_values: + self.assertFalse(str_to_bool(value))