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"]
FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>