Annotation of embedaddon/strongswan/src/libcharon/plugins/vici/python/test/test_protocol.py, revision 1.1.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>