1 from collections
import OrderedDict
8 @unittest.skip("this is broken")
9 class PacketTestCase(unittest
.TestCase
):
11 self
._packet
= self
._create
_packet
()
16 def _create_packet(self
, with_ph
=True, with_pc
=True):
18 eh
= bt2
.StructureFieldClass()
20 ('id', bt2
.IntegerFieldClass(8)),
21 ('ts', bt2
.IntegerFieldClass(32)),
24 # stream event context
25 sec
= bt2
.StructureFieldClass()
27 ('cpu_id', bt2
.IntegerFieldClass(8)),
28 ('stuff', bt2
.FloatingPointNumberFieldClass()),
33 pc
= bt2
.StructureFieldClass()
35 ('something', bt2
.IntegerFieldClass(8)),
36 ('something_else', bt2
.FloatingPointNumberFieldClass()),
42 sc
= bt2
.StreamClass()
43 sc
.packet_context_field_class
= pc
44 sc
.event_header_field_class
= eh
45 sc
.event_context_field_class
= sec
48 ec
= bt2
.StructureFieldClass()
50 ('ant', bt2
.IntegerFieldClass(16, is_signed
=True)),
51 ('msg', bt2
.StringFieldClass()),
55 ep
= bt2
.StructureFieldClass()
57 ('giraffe', bt2
.IntegerFieldClass(32)),
58 ('gnu', bt2
.IntegerFieldClass(8)),
59 ('mosquito', bt2
.IntegerFieldClass(8)),
63 event_class
= bt2
.EventClass('ec')
64 event_class
.context_field_class
= ec
65 event_class
.payload_field_class
= ep
66 sc
.add_event_class(event_class
)
70 ph
= bt2
.StructureFieldClass()
72 ('magic', bt2
.IntegerFieldClass(32)),
73 ('stream_id', bt2
.IntegerFieldClass(16)),
80 tc
.packet_header_field_class
= ph
81 tc
.add_stream_class(sc
)
87 return stream
.create_packet()
89 def test_attr_stream(self
):
90 self
.assertIsNotNone(self
._packet
.stream
)
92 def test_get_header_field(self
):
93 self
.assertIsNotNone(self
._packet
.header_field
)
95 def test_no_header_field(self
):
96 packet
= self
._create
_packet
(with_ph
=False)
97 self
.assertIsNone(packet
.header_field
)
99 def test_get_context_field(self
):
100 self
.assertIsNotNone(self
._packet
.context_field
)
102 def test_no_context_field(self
):
103 packet
= self
._create
_packet
(with_pc
=False)
104 self
.assertIsNone(packet
.context_field
)
106 def _fill_packet(self
, packet
):
107 packet
.header_field
['magic'] = 0xc1fc1fc1
108 packet
.header_field
['stream_id'] = 23
109 packet
.context_field
['something'] = 17
110 packet
.context_field
['something_else'] = 188.88
113 packet1
= self
._create
_packet
()
114 self
._fill
_packet
(packet1
)
115 packet2
= self
._create
_packet
()
116 self
._fill
_packet
(packet2
)
117 self
.assertEqual(packet1
, packet2
)
119 def test_ne_header_field(self
):
120 packet1
= self
._create
_packet
()
121 self
._fill
_packet
(packet1
)
122 packet2
= self
._create
_packet
()
123 self
._fill
_packet
(packet2
)
124 packet2
.header_field
['stream_id'] = 18
125 self
.assertNotEqual(packet1
, packet2
)
127 def test_ne_context_field(self
):
128 packet1
= self
._create
_packet
()
129 self
._fill
_packet
(packet1
)
130 packet2
= self
._create
_packet
()
131 self
._fill
_packet
(packet2
)
132 packet2
.context_field
['something_else'] = 1948.11
133 self
.assertNotEqual(packet1
, packet2
)
135 def test_eq_invalid(self
):
136 self
.assertFalse(self
._packet
== 23)
138 def _test_copy(self
, func
):
139 packet
= self
._create
_packet
()
140 self
._fill
_packet
(packet
)
142 self
.assertIsNot(packet
, cpy
)
143 self
.assertNotEqual(packet
.addr
, cpy
.addr
)
144 self
.assertEqual(packet
, cpy
)
147 self
._test
_copy
(copy
.copy
)
149 def test_deepcopy(self
):
150 self
._test
_copy
(copy
.deepcopy
)