Return to test_protocol.py CVS log

Up to [ELWIX - Embedded LightWeight unIX -] / embedaddon / strongswan / src / libcharon / plugins / vici / python / test

Annotation of embedaddon/strongswan/src/libcharon/plugins/vici/python/test/test_protocol.py, revision 1.1

1.1     ! misho       1: import pytest
        !             2: 
        !             3: from vici.protocol import Packet, Message, FiniteStream
        !             4: from vici.exception import DeserializationException
        !             5: 
        !             6: 
        !             7: class TestPacket(object):
        !             8:     # test data definitions for outgoing packet types
        !             9:     cmd_request = b"\x00\x0c" b"command_type"
        !            10:     cmd_request_msg = b"\x00\x07" b"command" b"payload"
        !            11:     event_register = b"\x03\x0a" b"event_type"
        !            12:     event_unregister = b"\x04\x0a" b"event_type"
        !            13: 
        !            14:     # test data definitions for incoming packet types
        !            15:     cmd_response = b"\x01" b"reply"
        !            16:     cmd_unknown = b"\x02"
        !            17:     event_confirm = b"\x05"
        !            18:     event_unknown = b"\x06"
        !            19:     event = b"\x07\x03" b"log" b"message"
        !            20: 
        !            21:     def test_request(self):
        !            22:         assert Packet.request("command_type") == self.cmd_request
        !            23:         assert Packet.request("command", b"payload") == self.cmd_request_msg
        !            24: 
        !            25:     def test_register_event(self):
        !            26:         assert Packet.register_event("event_type") == self.event_register
        !            27: 
        !            28:     def test_unregister_event(self):
        !            29:         assert Packet.unregister_event("event_type") == self.event_unregister
        !            30: 
        !            31:     def test_parse(self):
        !            32:         parsed_cmd_response = Packet.parse(self.cmd_response)
        !            33:         assert parsed_cmd_response.response_type == Packet.CMD_RESPONSE
        !            34:         assert parsed_cmd_response.payload.getvalue() == self.cmd_response
        !            35: 
        !            36:         parsed_cmd_unknown = Packet.parse(self.cmd_unknown)
        !            37:         assert parsed_cmd_unknown.response_type == Packet.CMD_UNKNOWN
        !            38:         assert parsed_cmd_unknown.payload.getvalue() == self.cmd_unknown
        !            39: 
        !            40:         parsed_event_confirm = Packet.parse(self.event_confirm)
        !            41:         assert parsed_event_confirm.response_type == Packet.EVENT_CONFIRM
        !            42:         assert parsed_event_confirm.payload.getvalue() == self.event_confirm
        !            43: 
        !            44:         parsed_event_unknown = Packet.parse(self.event_unknown)
        !            45:         assert parsed_event_unknown.response_type == Packet.EVENT_UNKNOWN
        !            46:         assert parsed_event_unknown.payload.getvalue() == self.event_unknown
        !            47: 
        !            48:         parsed_event = Packet.parse(self.event)
        !            49:         assert parsed_event.response_type == Packet.EVENT
        !            50:         assert parsed_event.payload.getvalue() == self.event
        !            51: 
        !            52: 
        !            53: class TestMessage(object):
        !            54:     """Message (de)serialization test."""
        !            55: 
        !            56:     # data definitions for test of de(serialization)
        !            57:     # serialized messages holding a section
        !            58:     ser_sec_unclosed = b"\x01\x08unclosed"
        !            59:     ser_sec_single = b"\x01\x07section\x02"
        !            60:     ser_sec_nested = b"\x01\x05outer\x01\x0asubsection\x02\x02"
        !            61: 
        !            62:     # serialized messages holding a list
        !            63:     ser_list_invalid = b"\x04\x07invalid\x05\x00\x02e1\x02\x03sec\x06"
        !            64:     ser_list_0_item = b"\x04\x05empty\x06"
        !            65:     ser_list_1_item = b"\x04\x01l\x05\x00\x02e1\x06"
        !            66:     ser_list_2_item = b"\x04\x01l\x05\x00\x02e1\x05\x00\x02e2\x06"
        !            67: 
        !            68:     # serialized messages with key value pairs
        !            69:     ser_kv_pair = b"\x03\x03key\x00\x05value"
        !            70:     ser_kv_zero = b"\x03\x0azerolength\x00\x00"
        !            71: 
        !            72:     # deserialized messages holding a section
        !            73:     des_sec_single = {"section": {}}
        !            74:     des_sec_nested = {"outer": {"subsection": {}}}
        !            75: 
        !            76:     # deserialized messages holding a list
        !            77:     des_list_0_item = {"empty": []}
        !            78:     des_list_1_item = {"l": [b"e1"]}
        !            79:     des_list_2_item = {"l": [b"e1", b"e2"]}
        !            80: 
        !            81:     # deserialized messages with key value pairs
        !            82:     des_kv_pair = {"key": b"value"}
        !            83:     des_kv_zero = {"zerolength": b""}
        !            84: 
        !            85:     def test_section_serialization(self):
        !            86:         assert Message.serialize(self.des_sec_single) == self.ser_sec_single
        !            87:         assert Message.serialize(self.des_sec_nested) == self.ser_sec_nested
        !            88: 
        !            89:     def test_list_serialization(self):
        !            90:         assert Message.serialize(self.des_list_0_item) == self.ser_list_0_item
        !            91:         assert Message.serialize(self.des_list_1_item) == self.ser_list_1_item
        !            92:         assert Message.serialize(self.des_list_2_item) == self.ser_list_2_item
        !            93: 
        !            94:     def test_key_serialization(self):
        !            95:         assert Message.serialize(self.des_kv_pair) == self.ser_kv_pair
        !            96:         assert Message.serialize(self.des_kv_zero) == self.ser_kv_zero
        !            97: 
        !            98:     def test_section_deserialization(self):
        !            99:         single = Message.deserialize(FiniteStream(self.ser_sec_single))
        !           100:         nested = Message.deserialize(FiniteStream(self.ser_sec_nested))
        !           101: 
        !           102:         assert single == self.des_sec_single
        !           103:         assert nested == self.des_sec_nested
        !           104: 
        !           105:         with pytest.raises(DeserializationException):
        !           106:             Message.deserialize(FiniteStream(self.ser_sec_unclosed))
        !           107: 
        !           108:     def test_list_deserialization(self):
        !           109:         l0 = Message.deserialize(FiniteStream(self.ser_list_0_item))
        !           110:         l1 = Message.deserialize(FiniteStream(self.ser_list_1_item))
        !           111:         l2 = Message.deserialize(FiniteStream(self.ser_list_2_item))
        !           112: 
        !           113:         assert l0 == self.des_list_0_item
        !           114:         assert l1 == self.des_list_1_item
        !           115:         assert l2 == self.des_list_2_item
        !           116: 
        !           117:         with pytest.raises(DeserializationException):
        !           118:             Message.deserialize(FiniteStream(self.ser_list_invalid))
        !           119: 
        !           120:     def test_key_deserialization(self):
        !           121:         pair = Message.deserialize(FiniteStream(self.ser_kv_pair))
        !           122:         zerolength = Message.deserialize(FiniteStream(self.ser_kv_zero))
        !           123: 
        !           124:         assert pair == self.des_kv_pair
        !           125:         assert zerolength == self.des_kv_zero
        !           126: 
        !           127:     def test_roundtrip(self):
        !           128:         message = {
        !           129:             "key1": "value1",
        !           130:             "section1": {
        !           131:                 "sub-section": {
        !           132:                     "key2": b"value2",
        !           133:                 },
        !           134:                 "list1": ["item1", "item2"],
        !           135:             },
        !           136:         }
        !           137:         serialized_message = FiniteStream(Message.serialize(message))
        !           138:         deserialized_message = Message.deserialize(serialized_message)
        !           139: 
        !           140:         # ensure that list items and key values remain as undecoded bytes
        !           141:         deserialized_section = deserialized_message["section1"]
        !           142:         assert deserialized_message["key1"] == b"value1"
        !           143:         assert deserialized_section["sub-section"]["key2"] == b"value2"
        !           144:         assert deserialized_section["list1"] == [b"item1", b"item2"]