1 from collections
import OrderedDict
8 class PacketTestCase(unittest
.TestCase
):
10 self
._packet
= self
._create
_packet
()
15 def _create_packet(self
, with_ph
=True, with_pc
=True):
17 eh
= bt2
.StructureFieldType()
19 ('id', bt2
.IntegerFieldType(8)),
20 ('ts', bt2
.IntegerFieldType(32)),
23 # stream event context
24 sec
= bt2
.StructureFieldType()
26 ('cpu_id', bt2
.IntegerFieldType(8)),
27 ('stuff', bt2
.FloatingPointNumberFieldType()),
32 pc
= bt2
.StructureFieldType()
34 ('something', bt2
.IntegerFieldType(8)),
35 ('something_else', bt2
.FloatingPointNumberFieldType()),
41 sc
= bt2
.StreamClass()
42 sc
.packet_context_field_type
= pc
43 sc
.event_header_field_type
= eh
44 sc
.event_context_field_type
= sec
47 ec
= bt2
.StructureFieldType()
49 ('ant', bt2
.IntegerFieldType(16, is_signed
=True)),
50 ('msg', bt2
.StringFieldType()),
54 ep
= bt2
.StructureFieldType()
56 ('giraffe', bt2
.IntegerFieldType(32)),
57 ('gnu', bt2
.IntegerFieldType(8)),
58 ('mosquito', bt2
.IntegerFieldType(8)),
62 event_class
= bt2
.EventClass('ec')
63 event_class
.context_field_type
= ec
64 event_class
.payload_field_type
= ep
65 sc
.add_event_class(event_class
)
69 ph
= bt2
.StructureFieldType()
71 ('magic', bt2
.IntegerFieldType(32)),
72 ('stream_id', bt2
.IntegerFieldType(16)),
79 tc
.packet_header_field_type
= ph
80 tc
.add_stream_class(sc
)
86 return stream
.create_packet()
88 def test_attr_stream(self
):
89 self
.assertIsNotNone(self
._packet
.stream
)
91 def test_get_header_field(self
):
92 self
.assertIsNotNone(self
._packet
.header_field
)
94 def test_no_header_field(self
):
95 packet
= self
._create
_packet
(with_ph
=False)
96 self
.assertIsNone(packet
.header_field
)
98 def test_get_context_field(self
):
99 self
.assertIsNotNone(self
._packet
.context_field
)
101 def test_no_context_field(self
):
102 packet
= self
._create
_packet
(with_pc
=False)
103 self
.assertIsNone(packet
.context_field
)
105 def _fill_packet(self
, packet
):
106 packet
.header_field
['magic'] = 0xc1fc1fc1
107 packet
.header_field
['stream_id'] = 23
108 packet
.context_field
['something'] = 17
109 packet
.context_field
['something_else'] = 188.88
112 packet1
= self
._create
_packet
()
113 self
._fill
_packet
(packet1
)
114 packet2
= self
._create
_packet
()
115 self
._fill
_packet
(packet2
)
116 self
.assertEqual(packet1
, packet2
)
118 def test_ne_header_field(self
):
119 packet1
= self
._create
_packet
()
120 self
._fill
_packet
(packet1
)
121 packet2
= self
._create
_packet
()
122 self
._fill
_packet
(packet2
)
123 packet2
.header_field
['stream_id'] = 18
124 self
.assertNotEqual(packet1
, packet2
)
126 def test_ne_context_field(self
):
127 packet1
= self
._create
_packet
()
128 self
._fill
_packet
(packet1
)
129 packet2
= self
._create
_packet
()
130 self
._fill
_packet
(packet2
)
131 packet2
.context_field
['something_else'] = 1948.11
132 self
.assertNotEqual(packet1
, packet2
)
134 def test_eq_invalid(self
):
135 self
.assertFalse(self
._packet
== 23)
137 def _test_copy(self
, func
):
138 packet
= self
._create
_packet
()
139 self
._fill
_packet
(packet
)
141 self
.assertIsNot(packet
, cpy
)
142 self
.assertNotEqual(packet
.addr
, cpy
.addr
)
143 self
.assertEqual(packet
, cpy
)
146 self
._test
_copy
(copy
.copy
)
148 def test_deepcopy(self
):
149 self
._test
_copy
(copy
.deepcopy
)