1 from collections
import OrderedDict
8 class PacketTestCase(unittest
.TestCase
):
10 self
._packet
= self
._create
_packet
()
12 def _create_packet(self
, with_ph
=True, with_pc
=True):
14 eh
= bt2
.StructureFieldType()
16 ('id', bt2
.IntegerFieldType(8)),
17 ('ts', bt2
.IntegerFieldType(32)),
20 # stream event context
21 sec
= bt2
.StructureFieldType()
23 ('cpu_id', bt2
.IntegerFieldType(8)),
24 ('stuff', bt2
.FloatingPointNumberFieldType()),
29 pc
= bt2
.StructureFieldType()
31 ('something', bt2
.IntegerFieldType(8)),
32 ('something_else', bt2
.FloatingPointNumberFieldType()),
38 sc
= bt2
.StreamClass()
39 sc
.packet_context_field_type
= pc
40 sc
.event_header_field_type
= eh
41 sc
.event_context_field_type
= sec
44 ec
= bt2
.StructureFieldType()
46 ('ant', bt2
.IntegerFieldType(16, is_signed
=True)),
47 ('msg', bt2
.StringFieldType()),
51 ep
= bt2
.StructureFieldType()
53 ('giraffe', bt2
.IntegerFieldType(32)),
54 ('gnu', bt2
.IntegerFieldType(8)),
55 ('mosquito', bt2
.IntegerFieldType(8)),
59 event_class
= bt2
.EventClass('ec')
60 event_class
.context_field_type
= ec
61 event_class
.payload_field_type
= ep
62 sc
.add_event_class(event_class
)
66 ph
= bt2
.StructureFieldType()
68 ('magic', bt2
.IntegerFieldType(32)),
69 ('stream_id', bt2
.IntegerFieldType(16)),
76 tc
.packet_header_field_type
= ph
77 tc
.add_stream_class(sc
)
83 return stream
.create_packet()
85 def test_attr_stream(self
):
86 self
.assertIsNotNone(self
._packet
.stream
)
88 def test_get_header_field(self
):
89 self
.assertIsNotNone(self
._packet
.header_field
)
91 def test_no_header_field(self
):
92 packet
= self
._create
_packet
(with_ph
=False)
93 self
.assertIsNone(packet
.header_field
)
95 def test_get_context_field(self
):
96 self
.assertIsNotNone(self
._packet
.context_field
)
98 def test_no_context_field(self
):
99 packet
= self
._create
_packet
(with_pc
=False)
100 self
.assertIsNone(packet
.context_field
)
102 def _fill_packet(self
, packet
):
103 packet
.header_field
['magic'] = 0xc1fc1fc1
104 packet
.header_field
['stream_id'] = 23
105 packet
.context_field
['something'] = 17
106 packet
.context_field
['something_else'] = 188.88
109 packet1
= self
._create
_packet
()
110 self
._fill
_packet
(packet1
)
111 packet2
= self
._create
_packet
()
112 self
._fill
_packet
(packet2
)
113 self
.assertEqual(packet1
, packet2
)
115 def test_ne_header_field(self
):
116 packet1
= self
._create
_packet
()
117 self
._fill
_packet
(packet1
)
118 packet2
= self
._create
_packet
()
119 self
._fill
_packet
(packet2
)
120 packet2
.header_field
['stream_id'] = 18
121 self
.assertNotEqual(packet1
, packet2
)
123 def test_ne_context_field(self
):
124 packet1
= self
._create
_packet
()
125 self
._fill
_packet
(packet1
)
126 packet2
= self
._create
_packet
()
127 self
._fill
_packet
(packet2
)
128 packet2
.context_field
['something_else'] = 1948.11
129 self
.assertNotEqual(packet1
, packet2
)
131 def test_eq_invalid(self
):
132 self
.assertFalse(self
._packet
== 23)
134 def _test_copy(self
, func
):
135 packet
= self
._create
_packet
()
136 self
._fill
_packet
(packet
)
138 self
.assertIsNot(packet
, cpy
)
139 self
.assertNotEqual(packet
.addr
, cpy
.addr
)
140 self
.assertEqual(packet
, cpy
)
143 self
._test
_copy
(copy
.copy
)
145 def test_deepcopy(self
):
146 self
._test
_copy
(copy
.deepcopy
)