tests/bindings/python/bt2/test_trace.py

   1 from bt2 import values
   2 import unittest
   3 import copy
   4 import uuid
   5 import bt2
   6
   7
   8 class TraceTestCase(unittest.TestCase):
   9     def setUp(self):
  10         self._sc = self._create_stream_class('sc1', 3)
  11         self._tc = bt2.Trace()
  12
  13     def tearDown(self):
  14         del self._sc
  15         del self._tc
  16
  17     def _create_stream_class(self, name, id):
  18         ec1, ec2 = self._create_event_classes()
  19         packet_context_ft = bt2.StructureFieldType()
  20         packet_context_ft.append_field('menu', bt2.FloatingPointNumberFieldType())
  21         packet_context_ft.append_field('sticker', bt2.StringFieldType())
  22         event_header_ft = bt2.StructureFieldType()
  23         event_header_ft.append_field('id', bt2.IntegerFieldType(19))
  24         event_context_ft = bt2.StructureFieldType()
  25         event_context_ft.append_field('msg', bt2.StringFieldType())
  26         return bt2.StreamClass(name=name, id=id,
  27                                packet_context_field_type=packet_context_ft,
  28                                event_header_field_type=event_header_ft,
  29                                event_context_field_type=event_context_ft,
  30                                event_classes=(ec1, ec2))
  31
  32     def _create_event_classes(self):
  33         context_ft = bt2.StructureFieldType()
  34         context_ft.append_field('allo', bt2.StringFieldType())
  35         context_ft.append_field('zola', bt2.IntegerFieldType(18))
  36         payload_ft = bt2.StructureFieldType()
  37         payload_ft.append_field('zoom', bt2.StringFieldType())
  38         ec1 = bt2.EventClass('event23', id=23, context_field_type=context_ft,
  39                              payload_field_type=payload_ft)
  40         ec2 = bt2.EventClass('event17', id=17, context_field_type=payload_ft,
  41                              payload_field_type=context_ft)
  42         return ec1, ec2
  43
  44     def test_create_default(self):
  45         self.assertEqual(len(self._tc), 0)
  46
  47     def _get_std_header(self):
  48         header_ft = bt2.StructureFieldType()
  49         header_ft.append_field('magic', bt2.IntegerFieldType(32))
  50         header_ft.append_field('stream_id', bt2.IntegerFieldType(32))
  51         return header_ft
  52
  53     def test_create_full(self):
  54         clock_classes = bt2.ClockClass('cc1', 1000), bt2.ClockClass('cc2', 30)
  55         sc = self._create_stream_class('sc1', 3)
  56         tc = bt2.Trace(name='my name',
  57                        native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
  58                        env={'the_string': 'value', 'the_int': 23},
  59                        packet_header_field_type=self._get_std_header(),
  60                        clock_classes=clock_classes,
  61                        stream_classes=(sc,))
  62         self.assertEqual(tc.name, 'my name')
  63         self.assertEqual(tc.native_byte_order, bt2.ByteOrder.LITTLE_ENDIAN)
  64         self.assertEqual(tc.env['the_string'], 'value')
  65         self.assertEqual(tc.env['the_int'], 23)
  66         self.assertEqual(tc.packet_header_field_type, self._get_std_header())
  67         self.assertEqual(tc.clock_classes['cc1'], clock_classes[0])
  68         self.assertEqual(tc.clock_classes['cc2'], clock_classes[1])
  69         self.assertEqual(tc[3], sc)
  70
  71     def test_assign_name(self):
  72         self._tc.name = 'lel'
  73         self.assertEqual(self._tc.name, 'lel')
  74
  75     def test_assign_invalid_name(self):
  76         with self.assertRaises(TypeError):
  77             self._tc.name = 17
  78
  79     def test_assign_static(self):
  80         self._tc.set_is_static()
  81         self.assertTrue(self._tc.is_static)
  82
  83     def test_assign_native_byte_order(self):
  84         self._tc.native_byte_order = bt2.ByteOrder.BIG_ENDIAN
  85         self.assertEqual(self._tc.native_byte_order, bt2.ByteOrder.BIG_ENDIAN)
  86
  87     def test_assign_invalid_native_byte_order(self):
  88         with self.assertRaises(TypeError):
  89             self._tc.native_byte_order = 'lel'
  90
  91     def test_assign_packet_header_field_type(self):
  92         header_ft = bt2.StructureFieldType()
  93         header_ft.append_field('magic', bt2.IntegerFieldType(32))
  94         self._tc.packet_header_field_type = header_ft
  95         self.assertEqual(self._tc.packet_header_field_type, header_ft)
  96
  97     def test_assign_no_packet_header_field_type(self):
  98         self._tc.packet_header_field_type = None
  99         self.assertIsNone(self._tc.packet_header_field_type)
 100
 101     def _test_copy(self, cpy):
 102         self.assertIsNot(cpy, self._tc)
 103         self.assertNotEqual(cpy.addr, self._tc.addr)
 104         self.assertEqual(cpy, self._tc)
 105         self.assertEqual(len(self._tc), len(cpy))
 106
 107     def _pre_copy(self):
 108         self._tc.packet_header_field_type = self._get_std_header()
 109         self._tc.name = 'the trace class'
 110         sc1 = self._create_stream_class('sc1', 3)
 111         sc2 = self._create_stream_class('sc2', 9)
 112         sc3 = self._create_stream_class('sc3', 17)
 113         self._tc.add_clock_class(bt2.ClockClass('cc1', 1000))
 114         self._tc.add_clock_class(bt2.ClockClass('cc2', 30))
 115         self._tc.env['allo'] = 'bateau'
 116         self._tc.env['bateau'] = 'cart'
 117         self._tc.add_stream_class(sc1)
 118         self._tc.add_stream_class(sc2)
 119         self._tc.add_stream_class(sc3)
 120
 121     def test_copy(self):
 122         self._pre_copy()
 123         cpy = copy.copy(self._tc)
 124         self._test_copy(cpy)
 125         self.assertEqual(self._tc.packet_header_field_type.addr, cpy.packet_header_field_type.addr)
 126         self.assertEqual(self._tc.clock_classes['cc1'].addr, cpy.clock_classes['cc1'].addr)
 127         self.assertEqual(self._tc.clock_classes['cc2'].addr, cpy.clock_classes['cc2'].addr)
 128         self.assertEqual(self._tc.env['allo'].addr, cpy.env['allo'].addr)
 129         self.assertEqual(self._tc.env['bateau'].addr, cpy.env['bateau'].addr)
 130
 131     def test_deepcopy(self):
 132         self._pre_copy()
 133         cpy = copy.deepcopy(self._tc)
 134         self._test_copy(cpy)
 135         self.assertNotEqual(self._tc.packet_header_field_type.addr, cpy.packet_header_field_type.addr)
 136         self.assertNotEqual(self._tc.clock_classes['cc1'].addr, cpy.clock_classes['cc1'].addr)
 137         self.assertNotEqual(self._tc.clock_classes['cc2'].addr, cpy.clock_classes['cc2'].addr)
 138         self.assertNotEqual(self._tc.env['allo'].addr, cpy.env['allo'].addr)
 139         self.assertNotEqual(self._tc.env['bateau'].addr, cpy.env['bateau'].addr)
 140
 141     def test_getitem(self):
 142         self._tc.add_stream_class(self._sc)
 143         self.assertEqual(self._tc[3].addr, self._sc.addr)
 144
 145     def test_getitem_wrong_key_type(self):
 146         self._tc.add_stream_class(self._sc)
 147         with self.assertRaises(TypeError):
 148             self._tc['hello']
 149
 150     def test_getitem_wrong_key(self):
 151         self._tc.add_stream_class(self._sc)
 152         with self.assertRaises(KeyError):
 153             self._tc[4]
 154
 155     def test_len(self):
 156         self.assertEqual(len(self._tc), 0)
 157         self._tc.add_stream_class(self._sc)
 158         self.assertEqual(len(self._tc), 1)
 159
 160     def test_iter(self):
 161         self._tc.packet_header_field_type = self._get_std_header()
 162         sc1 = self._create_stream_class('sc1', 3)
 163         sc2 = self._create_stream_class('sc2', 9)
 164         sc3 = self._create_stream_class('sc3', 17)
 165         self._tc.add_stream_class(sc1)
 166         self._tc.add_stream_class(sc2)
 167         self._tc.add_stream_class(sc3)
 168
 169         for sid, stream_class in self._tc.items():
 170             self.assertIsInstance(stream_class, bt2.StreamClass)
 171
 172             if sid == 3:
 173                 self.assertEqual(stream_class.addr, sc1.addr)
 174             elif sid == 9:
 175                 self.assertEqual(stream_class.addr, sc2.addr)
 176             elif sid == 17:
 177                 self.assertEqual(stream_class.addr, sc3.addr)
 178
 179     def test_env_getitem_wrong_key(self):
 180         with self.assertRaises(KeyError):
 181             self._tc.env['lel']
 182
 183     def test_clock_classes_getitem_wrong_key(self):
 184         with self.assertRaises(KeyError):
 185             self._tc.clock_classes['lel']
 186
 187     def test_streams_none(self):
 188         self.assertEqual(len(self._tc.streams), 0)
 189
 190     def test_streams_len(self):
 191         self._tc.add_stream_class(self._create_stream_class('sc1', 3))
 192         stream0 = self._tc[3]()
 193         stream1 = self._tc[3]()
 194         stream2 = self._tc[3]()
 195         self.assertEqual(len(self._tc.streams), 3)
 196
 197     def test_streams_iter(self):
 198         self._tc.add_stream_class(self._create_stream_class('sc1', 3))
 199         stream0 = self._tc[3](id=12)
 200         stream1 = self._tc[3](id=15)
 201         stream2 = self._tc[3](id=17)
 202         sids = set()
 203
 204         for stream in self._tc.streams:
 205             sids.add(stream.id)
 206
 207         self.assertEqual(len(sids), 3)
 208         self.assertTrue(12 in sids and 15 in sids and 17 in sids)
 209
 210     def _test_eq_create_objects(self):
 211         cc1_uuid = uuid.UUID('bc7f2f2d-2ee4-4e03-ab1f-2e0e1304e94f')
 212         cc1 = bt2.ClockClass('cc1', 1000, uuid=cc1_uuid)
 213         cc2_uuid = uuid.UUID('da7d6b6f-3108-4706-89bd-ab554732611b')
 214         cc2 = bt2.ClockClass('cc2', 30, uuid=cc2_uuid)
 215         sc1 = self._create_stream_class('sc1', 3)
 216         sc2 = self._create_stream_class('sc2', 9)
 217         return cc1, cc2, sc1, sc2, self._get_std_header()
 218
 219     def test_eq(self):
 220         cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects()
 221         tc1 = bt2.Trace(name='my name',
 222                         native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
 223                         env={'the_string': 'value', 'the_int': 23},
 224                         packet_header_field_type=header_ft,
 225                         clock_classes=(cc1, cc2),
 226                         stream_classes=(sc1, sc2))
 227         cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects()
 228         tc2 = bt2.Trace(name='my name',
 229                         native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
 230                         env={'the_string': 'value', 'the_int': 23},
 231                         packet_header_field_type=header_ft,
 232                         clock_classes=(cc1, cc2),
 233                         stream_classes=(sc1, sc2))
 234         self.assertEqual(tc1, tc2)
 235
 236     def test_ne_name(self):
 237         cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects()
 238         tc1 = bt2.Trace(name='my name2',
 239                         native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
 240                         env={'the_string': 'value', 'the_int': 23},
 241                         packet_header_field_type=header_ft,
 242                         clock_classes=(cc1, cc2),
 243                         stream_classes=(sc1, sc2))
 244         cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects()
 245         tc2 = bt2.Trace(name='my name',
 246                         native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
 247                         env={'the_string': 'value', 'the_int': 23},
 248                         packet_header_field_type=header_ft,
 249                         clock_classes=(cc1, cc2),
 250                         stream_classes=(sc1, sc2))
 251         self.assertNotEqual(tc1, tc2)
 252
 253     def test_ne_packet_header_field_type(self):
 254         cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects()
 255         tc1 = bt2.Trace(name='my name',
 256                         native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
 257                         env={'the_string': 'value', 'the_int': 23},
 258                         packet_header_field_type=header_ft,
 259                         clock_classes=(cc1, cc2),
 260                         stream_classes=(sc1, sc2))
 261         cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects()
 262         header_ft.append_field('yes', bt2.StringFieldType())
 263         tc2 = bt2.Trace(name='my name',
 264                         native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
 265                         env={'the_string': 'value', 'the_int': 23},
 266                         packet_header_field_type=header_ft,
 267                         clock_classes=(cc1, cc2),
 268                         stream_classes=(sc1, sc2))
 269         self.assertNotEqual(tc1, tc2)
 270
 271     def test_ne_native_byte_order(self):
 272         cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects()
 273         tc1 = bt2.Trace(name='my name',
 274                         native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
 275                         env={'the_string': 'value', 'the_int': 23},
 276                         packet_header_field_type=header_ft,
 277                         clock_classes=(cc1, cc2),
 278                         stream_classes=(sc1, sc2))
 279         cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects()
 280         tc2 = bt2.Trace(name='my name',
 281                         native_byte_order=bt2.ByteOrder.BIG_ENDIAN,
 282                         env={'the_string': 'value', 'the_int': 23},
 283                         packet_header_field_type=header_ft,
 284                         clock_classes=(cc1, cc2),
 285                         stream_classes=(sc1, sc2))
 286         self.assertNotEqual(tc1, tc2)
 287
 288     def test_ne_env(self):
 289         cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects()
 290         tc1 = bt2.Trace(name='my name',
 291                         native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
 292                         env={'the_string': 'value', 'the_int2': 23},
 293                         packet_header_field_type=header_ft,
 294                         clock_classes=(cc1, cc2),
 295                         stream_classes=(sc1, sc2))
 296         cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects()
 297         tc2 = bt2.Trace(name='my name',
 298                         native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
 299                         env={'the_string': 'value', 'the_int': 23},
 300                         packet_header_field_type=header_ft,
 301                         clock_classes=(cc1, cc2),
 302                         stream_classes=(sc1, sc2))
 303         self.assertNotEqual(tc1, tc2)
 304
 305     def test_ne_clock_classes(self):
 306         cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects()
 307         tc1 = bt2.Trace(name='my name',
 308                         native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
 309                         env={'the_string': 'value', 'the_int': 23},
 310                         packet_header_field_type=header_ft,
 311                         clock_classes=(cc1, cc2),
 312                         stream_classes=(sc1, sc2))
 313         cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects()
 314         cc2.frequency = 1234
 315         tc2 = bt2.Trace(name='my name',
 316                         native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
 317                         env={'the_string': 'value', 'the_int': 23},
 318                         packet_header_field_type=header_ft,
 319                         clock_classes=(cc1, cc2),
 320                         stream_classes=(sc1, sc2))
 321         self.assertNotEqual(tc1, tc2)
 322
 323     def test_ne_stream_classes(self):
 324         cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects()
 325         tc1 = bt2.Trace(name='my name',
 326                         native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
 327                         env={'the_string': 'value', 'the_int': 23},
 328                         packet_header_field_type=header_ft,
 329                         clock_classes=(cc1, cc2),
 330                         stream_classes=(sc1, sc2))
 331         cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects()
 332         sc2.id = 72632
 333         tc2 = bt2.Trace(name='my name',
 334                         native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
 335                         env={'the_string': 'value', 'the_int': 23},
 336                         packet_header_field_type=header_ft,
 337                         clock_classes=(cc1, cc2),
 338                         stream_classes=(sc1, sc2))
 339         self.assertNotEqual(tc1, tc2)
 340
 341     def test_eq_invalid(self):
 342         self.assertFalse(self._tc == 23)