Add configuration for python code formatter black
[babeltrace.git] / tests / bindings / python / bt2 / test_packet.py
CommitLineData
9cf643d1 1from collections import OrderedDict
c4239792 2from bt2 import value
9cf643d1
PP
3import unittest
4import copy
5import bt2
6
7
976c241d 8@unittest.skip("this is broken")
9cf643d1
PP
9class PacketTestCase(unittest.TestCase):
10 def setUp(self):
11 self._packet = self._create_packet()
12
811644b8
PP
13 def tearDown(self):
14 del self._packet
15
9cf643d1
PP
16 def _create_packet(self, with_ph=True, with_pc=True):
17 # event header
b4f45851 18 eh = bt2.StructureFieldClass()
9cf643d1 19 eh += OrderedDict((
b4f45851
SM
20 ('id', bt2.IntegerFieldClass(8)),
21 ('ts', bt2.IntegerFieldClass(32)),
9cf643d1
PP
22 ))
23
24 # stream event context
b4f45851 25 sec = bt2.StructureFieldClass()
9cf643d1 26 sec += OrderedDict((
b4f45851
SM
27 ('cpu_id', bt2.IntegerFieldClass(8)),
28 ('stuff', bt2.FloatingPointNumberFieldClass()),
9cf643d1
PP
29 ))
30
31 # packet context
32 if with_pc:
b4f45851 33 pc = bt2.StructureFieldClass()
9cf643d1 34 pc += OrderedDict((
b4f45851
SM
35 ('something', bt2.IntegerFieldClass(8)),
36 ('something_else', bt2.FloatingPointNumberFieldClass()),
9cf643d1
PP
37 ))
38 else:
39 pc = None
40
41 # stream class
42 sc = bt2.StreamClass()
b4f45851
SM
43 sc.packet_context_field_class = pc
44 sc.event_header_field_class = eh
45 sc.event_context_field_class = sec
9cf643d1
PP
46
47 # event context
b4f45851 48 ec = bt2.StructureFieldClass()
9cf643d1 49 ec += OrderedDict((
b4f45851
SM
50 ('ant', bt2.IntegerFieldClass(16, is_signed=True)),
51 ('msg', bt2.StringFieldClass()),
9cf643d1
PP
52 ))
53
54 # event payload
b4f45851 55 ep = bt2.StructureFieldClass()
9cf643d1 56 ep += OrderedDict((
b4f45851
SM
57 ('giraffe', bt2.IntegerFieldClass(32)),
58 ('gnu', bt2.IntegerFieldClass(8)),
59 ('mosquito', bt2.IntegerFieldClass(8)),
9cf643d1
PP
60 ))
61
62 # event class
63 event_class = bt2.EventClass('ec')
b4f45851
SM
64 event_class.context_field_class = ec
65 event_class.payload_field_class = ep
9cf643d1
PP
66 sc.add_event_class(event_class)
67
68 # packet header
69 if with_ph:
b4f45851 70 ph = bt2.StructureFieldClass()
9cf643d1 71 ph += OrderedDict((
b4f45851
SM
72 ('magic', bt2.IntegerFieldClass(32)),
73 ('stream_id', bt2.IntegerFieldClass(16)),
9cf643d1
PP
74 ))
75 else:
76 ph = None
77
78 # trace c;ass
79 tc = bt2.Trace()
b4f45851 80 tc.packet_header_field_class = ph
9cf643d1
PP
81 tc.add_stream_class(sc)
82
83 # stream
84 stream = sc()
85
86 # packet
87 return stream.create_packet()
88
89 def test_attr_stream(self):
90 self.assertIsNotNone(self._packet.stream)
91
92 def test_get_header_field(self):
93 self.assertIsNotNone(self._packet.header_field)
94
95 def test_no_header_field(self):
96 packet = self._create_packet(with_ph=False)
97 self.assertIsNone(packet.header_field)
98
99 def test_get_context_field(self):
100 self.assertIsNotNone(self._packet.context_field)
101
102 def test_no_context_field(self):
103 packet = self._create_packet(with_pc=False)
104 self.assertIsNone(packet.context_field)
105
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
111
112 def test_eq(self):
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)
118
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)
126
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)
134
135 def test_eq_invalid(self):
136 self.assertFalse(self._packet == 23)
137
138 def _test_copy(self, func):
139 packet = self._create_packet()
140 self._fill_packet(packet)
141 cpy = func(packet)
142 self.assertIsNot(packet, cpy)
143 self.assertNotEqual(packet.addr, cpy.addr)
144 self.assertEqual(packet, cpy)
145
146 def test_copy(self):
147 self._test_copy(copy.copy)
148
149 def test_deepcopy(self):
150 self._test_copy(copy.deepcopy)
This page took 0.045074 seconds and 4 git commands to generate.