tests/bindings/python/bt2/test_stream_class.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # Copyright (C) 2019 EfficiOS Inc.
   4 #
   5
   6 import unittest
   7
   8 from bt2 import clock_class as bt2_clock_class
   9 from bt2 import event_class as bt2_event_class
  10 from bt2 import field_class as bt2_field_class
  11 from bt2 import trace_class as bt2_trace_class
  12 from bt2 import stream_class as bt2_stream_class
  13 from utils import run_in_component_init
  14
  15
  16 class StreamClassTestCase(unittest.TestCase):
  17     def setUp(self):
  18         def f(comp_self):
  19             tc = comp_self._create_trace_class(assigns_automatic_stream_class_id=True)
  20             cc = comp_self._create_clock_class()
  21             return tc, cc
  22
  23         self._tc, self._cc = run_in_component_init(f)
  24         self._trace = self._tc()
  25
  26     def test_create_default(self):
  27         sc = self._tc.create_stream_class()
  28
  29         self.assertIs(type(sc), bt2_stream_class._StreamClass)
  30         self.assertIsNone(sc.name)
  31         self.assertIsNone(sc.packet_context_field_class)
  32         self.assertIsNone(sc.event_common_context_field_class)
  33         self.assertIsNone(sc.default_clock_class)
  34         self.assertTrue(sc.assigns_automatic_event_class_id)
  35         self.assertTrue(sc.assigns_automatic_stream_id)
  36         self.assertFalse(sc.supports_packets)
  37         self.assertFalse(sc.packets_have_beginning_default_clock_snapshot)
  38         self.assertFalse(sc.packets_have_end_default_clock_snapshot)
  39         self.assertFalse(sc.supports_discarded_events)
  40         self.assertFalse(sc.discarded_events_have_default_clock_snapshots)
  41         self.assertFalse(sc.supports_discarded_packets)
  42         self.assertFalse(sc.discarded_packets_have_default_clock_snapshots)
  43         self.assertEqual(len(sc.user_attributes), 0)
  44
  45     def test_create_name(self):
  46         sc = self._tc.create_stream_class(name="bozo")
  47         self.assertEqual(sc.name, "bozo")
  48
  49     def test_create_invalid_name(self):
  50         with self.assertRaisesRegex(TypeError, "'int' is not a 'str' object"):
  51             self._tc.create_stream_class(name=17)
  52
  53         self.assertEqual(len(self._tc), 0)
  54
  55     def test_create_packet_context_field_class(self):
  56         fc = self._tc.create_structure_field_class()
  57         sc = self._tc.create_stream_class(
  58             packet_context_field_class=fc, supports_packets=True
  59         )
  60         self.assertEqual(sc.packet_context_field_class, fc)
  61         self.assertIs(
  62             type(sc.packet_context_field_class), bt2_field_class._StructureFieldClass
  63         )
  64
  65     def test_create_invalid_packet_context_field_class(self):
  66         with self.assertRaisesRegex(
  67             TypeError,
  68             "'int' is not a '<class 'bt2.field_class._StructureFieldClass'>' object",
  69         ):
  70             self._tc.create_stream_class(
  71                 packet_context_field_class=22, supports_packets=True
  72             )
  73
  74         self.assertEqual(len(self._tc), 0)
  75
  76     def test_create_invalid_packet_context_field_class_no_packets(self):
  77         fc = self._tc.create_structure_field_class()
  78
  79         with self.assertRaisesRegex(
  80             ValueError,
  81             "cannot have a packet context field class without supporting packets",
  82         ):
  83             self._tc.create_stream_class(packet_context_field_class=fc)
  84
  85         self.assertEqual(len(self._tc), 0)
  86
  87     def test_create_event_common_context_field_class(self):
  88         fc = self._tc.create_structure_field_class()
  89         sc = self._tc.create_stream_class(event_common_context_field_class=fc)
  90         self.assertEqual(sc.event_common_context_field_class, fc)
  91         self.assertIs(
  92             type(sc.event_common_context_field_class),
  93             bt2_field_class._StructureFieldClass,
  94         )
  95
  96     def test_create_invalid_event_common_context_field_class(self):
  97         with self.assertRaisesRegex(
  98             TypeError,
  99             "'int' is not a '<class 'bt2.field_class._StructureFieldClass'>' object",
 100         ):
 101             self._tc.create_stream_class(event_common_context_field_class=22)
 102
 103         self.assertEqual(len(self._tc), 0)
 104
 105     def test_create_default_clock_class(self):
 106         sc = self._tc.create_stream_class(default_clock_class=self._cc)
 107         self.assertEqual(sc.default_clock_class.addr, self._cc.addr)
 108         self.assertIs(type(sc.default_clock_class), bt2_clock_class._ClockClass)
 109
 110     def test_create_invalid_default_clock_class(self):
 111         with self.assertRaisesRegex(
 112             TypeError, "'int' is not a '<class 'bt2.clock_class._ClockClass'>' object"
 113         ):
 114             self._tc.create_stream_class(default_clock_class=12)
 115
 116         self.assertEqual(len(self._tc), 0)
 117
 118     def test_create_user_attributes(self):
 119         sc = self._tc.create_stream_class(user_attributes={"salut": 23})
 120         self.assertEqual(sc.user_attributes, {"salut": 23})
 121
 122     def test_create_invalid_user_attributes(self):
 123         with self.assertRaisesRegex(
 124             TypeError, "cannot create value object from 'object' object"
 125         ):
 126             self._tc.create_stream_class(user_attributes=object())
 127
 128         self.assertEqual(len(self._tc), 0)
 129
 130     def test_create_invalid_user_attributes_value_type(self):
 131         with self.assertRaisesRegex(
 132             TypeError,
 133             "'SignedIntegerValue' is not a '<class 'bt2.value.MapValue'>' object",
 134         ):
 135             self._tc.create_stream_class(user_attributes=23)
 136
 137         self.assertEqual(len(self._tc), 0)
 138
 139     def test_automatic_stream_ids(self):
 140         sc = self._tc.create_stream_class(assigns_automatic_stream_id=True)
 141         self.assertTrue(sc.assigns_automatic_stream_id)
 142
 143         stream = self._trace.create_stream(sc)
 144         self.assertIsNotNone(stream.id)
 145
 146     def test_automatic_stream_ids_raises(self):
 147         sc = self._tc.create_stream_class(assigns_automatic_stream_id=True)
 148         self.assertTrue(sc.assigns_automatic_stream_id)
 149
 150         with self.assertRaisesRegex(
 151             ValueError, "id provided, but stream class assigns automatic stream ids"
 152         ):
 153             self._trace.create_stream(sc, id=123)
 154
 155         self.assertEqual(len(self._trace), 0)
 156
 157     def test_automatic_stream_ids_wrong_type(self):
 158         with self.assertRaisesRegex(TypeError, "str' is not a 'bool' object"):
 159             self._tc.create_stream_class(assigns_automatic_stream_id="True")
 160
 161         self.assertEqual(len(self._tc), 0)
 162
 163     def test_no_automatic_stream_ids(self):
 164         sc = self._tc.create_stream_class(assigns_automatic_stream_id=False)
 165         self.assertFalse(sc.assigns_automatic_stream_id)
 166
 167         stream = self._trace.create_stream(sc, id=333)
 168         self.assertEqual(stream.id, 333)
 169
 170     def test_no_automatic_stream_ids_raises(self):
 171         sc = self._tc.create_stream_class(assigns_automatic_stream_id=False)
 172         self.assertFalse(sc.assigns_automatic_stream_id)
 173
 174         with self.assertRaisesRegex(
 175             ValueError,
 176             "id not provided, but stream class does not assign automatic stream ids",
 177         ):
 178             self._trace.create_stream(sc)
 179
 180         self.assertEqual(len(self._trace), 0)
 181
 182     def test_automatic_event_class_ids(self):
 183         sc = self._tc.create_stream_class(assigns_automatic_event_class_id=True)
 184         self.assertTrue(sc.assigns_automatic_event_class_id)
 185
 186         ec = sc.create_event_class()
 187         self.assertIsNotNone(ec.id)
 188
 189     def test_automatic_event_class_ids_raises(self):
 190         sc = self._tc.create_stream_class(assigns_automatic_event_class_id=True)
 191         self.assertTrue(sc.assigns_automatic_event_class_id)
 192
 193         with self.assertRaisesRegex(
 194             ValueError,
 195             "id provided, but stream class assigns automatic event class ids",
 196         ):
 197             sc.create_event_class(id=123)
 198
 199         self.assertEqual(len(sc), 0)
 200
 201     def test_automatic_event_class_ids_wrong_type(self):
 202         with self.assertRaisesRegex(TypeError, "'str' is not a 'bool' object"):
 203             self._tc.create_stream_class(assigns_automatic_event_class_id="True")
 204
 205         self.assertEqual(len(self._tc), 0)
 206
 207     def test_no_automatic_event_class_ids(self):
 208         sc = self._tc.create_stream_class(assigns_automatic_event_class_id=False)
 209         self.assertFalse(sc.assigns_automatic_event_class_id)
 210
 211         ec = sc.create_event_class(id=333)
 212         self.assertEqual(ec.id, 333)
 213
 214     def test_no_automatic_event_class_ids_raises(self):
 215         sc = self._tc.create_stream_class(assigns_automatic_event_class_id=False)
 216         self.assertFalse(sc.assigns_automatic_event_class_id)
 217
 218         with self.assertRaisesRegex(
 219             ValueError,
 220             "id not provided, but stream class does not assign automatic event class ids",
 221         ):
 222             sc.create_event_class()
 223
 224         self.assertEqual(len(sc), 0)
 225
 226     def test_supports_packets_without_cs(self):
 227         sc = self._tc.create_stream_class(
 228             default_clock_class=self._cc, supports_packets=True
 229         )
 230         self.assertTrue(sc.supports_packets)
 231         self.assertFalse(sc.packets_have_beginning_default_clock_snapshot)
 232         self.assertFalse(sc.packets_have_end_default_clock_snapshot)
 233
 234     def test_supports_packets_with_begin_cs(self):
 235         sc = self._tc.create_stream_class(
 236             default_clock_class=self._cc,
 237             supports_packets=True,
 238             packets_have_beginning_default_clock_snapshot=True,
 239         )
 240         self.assertTrue(sc.supports_packets)
 241         self.assertTrue(sc.packets_have_beginning_default_clock_snapshot)
 242         self.assertFalse(sc.packets_have_end_default_clock_snapshot)
 243
 244     def test_supports_packets_with_end_cs(self):
 245         sc = self._tc.create_stream_class(
 246             default_clock_class=self._cc,
 247             supports_packets=True,
 248             packets_have_end_default_clock_snapshot=True,
 249         )
 250         self.assertTrue(sc.supports_packets)
 251         self.assertFalse(sc.packets_have_beginning_default_clock_snapshot)
 252         self.assertTrue(sc.packets_have_end_default_clock_snapshot)
 253
 254     def test_supports_packets_raises_type_error(self):
 255         with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
 256             self._tc.create_stream_class(
 257                 default_clock_class=self._cc, supports_packets=23
 258             )
 259
 260         self.assertEqual(len(self._tc), 0)
 261
 262     def test_packets_have_begin_default_cs_raises_type_error(self):
 263         with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
 264             self._tc.create_stream_class(
 265                 default_clock_class=self._cc,
 266                 packets_have_beginning_default_clock_snapshot=23,
 267             )
 268
 269         self.assertEqual(len(self._tc), 0)
 270
 271     def test_packets_have_end_default_cs_raises_type_error(self):
 272         with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
 273             self._tc.create_stream_class(
 274                 default_clock_class=self._cc, packets_have_end_default_clock_snapshot=23
 275             )
 276
 277         self.assertEqual(len(self._tc), 0)
 278
 279     def test_does_not_support_packets_raises_with_begin_cs(self):
 280         with self.assertRaisesRegex(
 281             ValueError,
 282             "cannot not support packets, but have packet beginning default clock snapshot",
 283         ):
 284             self._tc.create_stream_class(
 285                 default_clock_class=self._cc,
 286                 packets_have_beginning_default_clock_snapshot=True,
 287             )
 288
 289         self.assertEqual(len(self._tc), 0)
 290
 291     def test_does_not_support_packets_raises_with_end_cs(self):
 292         with self.assertRaisesRegex(
 293             ValueError,
 294             "cannot not support packets, but have packet end default clock snapshots",
 295         ):
 296             self._tc.create_stream_class(
 297                 default_clock_class=self._cc,
 298                 packets_have_end_default_clock_snapshot=True,
 299             )
 300
 301         self.assertEqual(len(self._tc), 0)
 302
 303     def test_supports_discarded_events_without_cs(self):
 304         sc = self._tc.create_stream_class(
 305             default_clock_class=self._cc, supports_discarded_events=True
 306         )
 307         self.assertTrue(sc.supports_discarded_events)
 308         self.assertFalse(sc.discarded_events_have_default_clock_snapshots)
 309
 310     def test_supports_discarded_events_with_cs(self):
 311         sc = self._tc.create_stream_class(
 312             default_clock_class=self._cc,
 313             supports_discarded_events=True,
 314             discarded_events_have_default_clock_snapshots=True,
 315         )
 316         self.assertTrue(sc.supports_discarded_events)
 317         self.assertTrue(sc.discarded_events_have_default_clock_snapshots)
 318
 319     def test_supports_discarded_events_raises_type_error(self):
 320         with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
 321             self._tc.create_stream_class(
 322                 default_clock_class=self._cc, supports_discarded_events=23
 323             )
 324
 325         self.assertEqual(len(self._tc), 0)
 326
 327     def test_discarded_events_have_default_cs_raises_type_error(self):
 328         with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
 329             self._tc.create_stream_class(
 330                 default_clock_class=self._cc,
 331                 discarded_events_have_default_clock_snapshots=23,
 332             )
 333
 334         self.assertEqual(len(self._tc), 0)
 335
 336     def test_does_not_support_discarded_events_raises_with_cs(self):
 337         with self.assertRaisesRegex(
 338             ValueError,
 339             "cannot not support discarded events, but have default clock snapshots for discarded event messages",
 340         ):
 341             self._tc.create_stream_class(
 342                 default_clock_class=self._cc,
 343                 discarded_events_have_default_clock_snapshots=True,
 344             )
 345
 346         self.assertEqual(len(self._tc), 0)
 347
 348     def test_supports_discarded_events_with_clock_snapshots_without_default_clock_class_raises(
 349         self,
 350     ):
 351         with self.assertRaisesRegex(
 352             ValueError,
 353             "cannot have no default clock class, but have default clock snapshots for discarded event messages",
 354         ):
 355             self._tc.create_stream_class(
 356                 supports_discarded_events=True,
 357                 discarded_events_have_default_clock_snapshots=True,
 358             )
 359
 360         self.assertEqual(len(self._tc), 0)
 361
 362     def test_supports_discarded_packets_without_cs(self):
 363         sc = self._tc.create_stream_class(
 364             default_clock_class=self._cc,
 365             supports_discarded_packets=True,
 366             supports_packets=True,
 367         )
 368         self.assertTrue(sc.supports_discarded_packets)
 369         self.assertFalse(sc.discarded_packets_have_default_clock_snapshots)
 370
 371     def test_supports_discarded_packets_with_cs(self):
 372         sc = self._tc.create_stream_class(
 373             default_clock_class=self._cc,
 374             supports_discarded_packets=True,
 375             discarded_packets_have_default_clock_snapshots=True,
 376             supports_packets=True,
 377         )
 378         self.assertTrue(sc.supports_discarded_packets)
 379         self.assertTrue(sc.discarded_packets_have_default_clock_snapshots)
 380
 381     def test_supports_discarded_packets_raises_without_packet_support(self):
 382         with self.assertRaisesRegex(
 383             ValueError, "cannot support discarded packets, but not support packets"
 384         ):
 385             self._tc.create_stream_class(
 386                 default_clock_class=self._cc, supports_discarded_packets=True
 387             )
 388
 389         self.assertEqual(len(self._tc), 0)
 390
 391     def test_supports_discarded_packets_raises_type_error(self):
 392         with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
 393             self._tc.create_stream_class(
 394                 default_clock_class=self._cc,
 395                 supports_discarded_packets=23,
 396                 supports_packets=True,
 397             )
 398
 399         self.assertEqual(len(self._tc), 0)
 400
 401     def test_discarded_packets_have_default_cs_raises_type_error(self):
 402         with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
 403             self._tc.create_stream_class(
 404                 default_clock_class=self._cc,
 405                 discarded_packets_have_default_clock_snapshots=23,
 406                 supports_packets=True,
 407             )
 408
 409         self.assertEqual(len(self._tc), 0)
 410
 411     def test_does_not_support_discarded_packets_raises_with_cs(self):
 412         with self.assertRaisesRegex(
 413             ValueError,
 414             "cannot not support discarded packets, but have default clock snapshots for discarded packet messages",
 415         ):
 416             self._tc.create_stream_class(
 417                 default_clock_class=self._cc,
 418                 discarded_packets_have_default_clock_snapshots=True,
 419                 supports_packets=True,
 420             )
 421
 422         self.assertEqual(len(self._tc), 0)
 423
 424     def test_supports_discarded_packets_with_clock_snapshots_without_default_clock_class_raises(
 425         self,
 426     ):
 427         with self.assertRaisesRegex(
 428             ValueError,
 429             "cannot have no default clock class, but have default clock snapshots for discarded packet messages",
 430         ):
 431             self._tc.create_stream_class(
 432                 supports_packets=True,
 433                 supports_discarded_packets=True,
 434                 discarded_packets_have_default_clock_snapshots=True,
 435             )
 436
 437         self.assertEqual(len(self._tc), 0)
 438
 439     def test_trace_class(self):
 440         sc = self._tc.create_stream_class()
 441         self.assertEqual(sc.trace_class.addr, self._tc.addr)
 442         self.assertIs(type(sc.trace_class), bt2_trace_class._TraceClass)
 443
 444     def _create_stream_class_with_event_classes(self):
 445         sc = self._tc.create_stream_class(assigns_automatic_event_class_id=False)
 446         ec1 = sc.create_event_class(id=23)
 447         ec2 = sc.create_event_class(id=17)
 448         return sc, ec1, ec2
 449
 450     def test_getitem(self):
 451         sc, ec1, ec2 = self._create_stream_class_with_event_classes()
 452
 453         self.assertEqual(sc[23].addr, ec1.addr)
 454         self.assertEqual(type(sc[23]), bt2_event_class._EventClass)
 455         self.assertEqual(sc[17].addr, ec2.addr)
 456         self.assertEqual(type(sc[17]), bt2_event_class._EventClass)
 457
 458     def test_getitem_wrong_key_type(self):
 459         sc, _, _ = self._create_stream_class_with_event_classes()
 460
 461         with self.assertRaisesRegex(TypeError, "'str' is not an 'int' object"):
 462             sc["event23"]
 463
 464     def test_getitem_wrong_key(self):
 465         sc, _, _ = self._create_stream_class_with_event_classes()
 466
 467         with self.assertRaisesRegex(KeyError, "19"):
 468             sc[19]
 469
 470     def test_len(self):
 471         sc, _, _ = self._create_stream_class_with_event_classes()
 472
 473         self.assertEqual(len(sc), 2)
 474
 475     def test_iter(self):
 476         sc, _, _ = self._create_stream_class_with_event_classes()
 477
 478         ec_ids = sorted(sc)
 479         self.assertEqual(ec_ids, [17, 23])
 480
 481
 482 if __name__ == "__main__":
 483     unittest.main()