tests/bindings/python/bt2/test_notification.py

   1 from bt2 import values
   2 import collections
   3 import unittest
   4 import copy
   5 import bt2
   6
   7
   8 class _NotificationTestCase(unittest.TestCase):
   9     def setUp(self):
  10         self._trace = bt2.Trace()
  11         self._sc = bt2.StreamClass()
  12         self._ec = bt2.EventClass('salut')
  13         self._my_int_ft = bt2.IntegerFieldType(32)
  14         self._ec.payload_field_type = bt2.StructureFieldType()
  15         self._ec.payload_field_type += collections.OrderedDict([
  16             ('my_int', self._my_int_ft),
  17         ])
  18         self._sc.add_event_class(self._ec)
  19         self._clock_class = bt2.ClockClass('allo', 1000)
  20         self._trace.add_clock_class(self._clock_class)
  21         self._trace.packet_header_field_type = bt2.StructureFieldType()
  22         self._trace.packet_header_field_type += collections.OrderedDict([
  23             ('hello', self._my_int_ft),
  24         ])
  25         self._trace.add_stream_class(self._sc)
  26         self._cc_prio_map = bt2.ClockClassPriorityMap()
  27         self._cc_prio_map[self._clock_class] = 231
  28         self._stream = self._sc()
  29         self._packet = self._stream.create_packet()
  30         self._packet.header_field['hello'] = 19487
  31         self._event = self._ec()
  32         self._event.clock_values.add(self._clock_class(1772))
  33         self._event.payload_field['my_int'] = 23
  34         self._event.packet = self._packet
  35
  36     def tearDown(self):
  37         del self._trace
  38         del self._sc
  39         del self._ec
  40         del self._my_int_ft
  41         del self._clock_class
  42         del self._cc_prio_map
  43         del self._stream
  44         del self._packet
  45         del self._event
  46
  47
  48 class EventNotificationTestCase(_NotificationTestCase):
  49     def test_create_no_cc_prio_map(self):
  50         notif = bt2.EventNotification(self._event)
  51         self.assertEqual(notif.event.addr, self._event.addr)
  52         self.assertEqual(len(notif.clock_class_priority_map), 0)
  53
  54     def test_create_with_cc_prio_map(self):
  55         notif = bt2.EventNotification(self._event, self._cc_prio_map)
  56         self.assertEqual(notif.event.addr, self._event.addr)
  57         self.assertEqual(len(notif.clock_class_priority_map), 1)
  58         self.assertEqual(notif.clock_class_priority_map.highest_priority_clock_class.addr,
  59                          self._clock_class.addr)
  60         self.assertEqual(notif.clock_class_priority_map[self._clock_class], 231)
  61
  62     def test_eq(self):
  63         notif = bt2.EventNotification(self._event, self._cc_prio_map)
  64         event_copy = copy.copy(self._event)
  65         event_copy.packet = self._packet
  66         cc_prio_map_copy = copy.copy(self._cc_prio_map)
  67         notif2 = bt2.EventNotification(event_copy, cc_prio_map_copy)
  68         self.assertEqual(notif, notif2)
  69
  70     def test_ne_event(self):
  71         notif = bt2.EventNotification(self._event, self._cc_prio_map)
  72         event_copy = copy.copy(self._event)
  73         event_copy.payload_field['my_int'] = 17
  74         event_copy.packet = self._packet
  75         cc_prio_map_copy = copy.copy(self._cc_prio_map)
  76         notif2 = bt2.EventNotification(event_copy, cc_prio_map_copy)
  77         self.assertNotEqual(notif, notif2)
  78
  79     def test_ne_cc_prio_map(self):
  80         notif = bt2.EventNotification(self._event)
  81         event_copy = copy.copy(self._event)
  82         event_copy.packet = self._packet
  83         cc_prio_map_copy = copy.copy(self._cc_prio_map)
  84         notif2 = bt2.EventNotification(event_copy, cc_prio_map_copy)
  85         self.assertNotEqual(notif, notif2)
  86
  87     def test_eq_invalid(self):
  88         notif = bt2.EventNotification(self._event)
  89         self.assertNotEqual(notif, 23)
  90
  91     def test_copy(self):
  92         notif = bt2.EventNotification(self._event, self._cc_prio_map)
  93         notif2 = copy.copy(notif)
  94         self.assertEqual(notif, notif2)
  95
  96     def test_deepcopy(self):
  97         notif = bt2.EventNotification(self._event, self._cc_prio_map)
  98         notif2 = copy.deepcopy(notif)
  99         self.assertEqual(notif, notif2)
 100
 101
 102 class PacketBeginningNotificationTestCase(_NotificationTestCase):
 103     def test_create(self):
 104         notif = bt2.PacketBeginningNotification(self._packet)
 105         self.assertEqual(notif.packet.addr, self._packet.addr)
 106
 107     def test_eq(self):
 108         notif = bt2.PacketBeginningNotification(self._packet)
 109         packet_copy = copy.copy(self._packet)
 110         notif2 = bt2.PacketBeginningNotification(packet_copy)
 111         self.assertEqual(notif, notif2)
 112
 113     def test_ne_packet(self):
 114         notif = bt2.PacketBeginningNotification(self._packet)
 115         packet_copy = copy.copy(self._packet)
 116         packet_copy.header_field['hello'] = 1847
 117         notif2 = bt2.PacketBeginningNotification(packet_copy)
 118         self.assertNotEqual(notif, notif2)
 119
 120     def test_eq_invalid(self):
 121         notif = bt2.PacketBeginningNotification(self._packet)
 122         self.assertNotEqual(notif, 23)
 123
 124     def test_copy(self):
 125         notif = bt2.PacketBeginningNotification(self._packet)
 126         notif2 = copy.copy(notif)
 127         self.assertEqual(notif, notif2)
 128
 129     def test_deepcopy(self):
 130         notif = bt2.PacketBeginningNotification(self._packet)
 131         notif2 = copy.deepcopy(notif)
 132         self.assertEqual(notif, notif2)
 133
 134
 135 class PacketEndNotificationTestCase(_NotificationTestCase):
 136     def test_create(self):
 137         notif = bt2.PacketEndNotification(self._packet)
 138         self.assertEqual(notif.packet.addr, self._packet.addr)
 139
 140     def test_eq(self):
 141         notif = bt2.PacketEndNotification(self._packet)
 142         packet_copy = copy.copy(self._packet)
 143         notif2 = bt2.PacketEndNotification(packet_copy)
 144         self.assertEqual(notif, notif2)
 145
 146     def test_ne_packet(self):
 147         notif = bt2.PacketEndNotification(self._packet)
 148         packet_copy = copy.copy(self._packet)
 149         packet_copy.header_field['hello'] = 1847
 150         notif2 = bt2.PacketEndNotification(packet_copy)
 151         self.assertNotEqual(notif, notif2)
 152
 153     def test_eq_invalid(self):
 154         notif = bt2.PacketEndNotification(self._packet)
 155         self.assertNotEqual(notif, 23)
 156
 157     def test_copy(self):
 158         notif = bt2.PacketEndNotification(self._packet)
 159         notif2 = copy.copy(notif)
 160         self.assertEqual(notif, notif2)
 161
 162     def test_deepcopy(self):
 163         notif = bt2.PacketEndNotification(self._packet)
 164         notif2 = copy.deepcopy(notif)
 165         self.assertEqual(notif, notif2)
 166
 167
 168 class StreamBeginningNotificationTestCase(_NotificationTestCase):
 169     def test_create(self):
 170         notif = bt2.StreamBeginningNotification(self._stream)
 171         self.assertEqual(notif.stream.addr, self._stream.addr)
 172
 173     def test_eq(self):
 174         notif = bt2.StreamBeginningNotification(self._stream)
 175         stream_copy = copy.copy(self._stream)
 176         notif2 = bt2.StreamBeginningNotification(stream_copy)
 177         self.assertEqual(notif, notif2)
 178
 179     def test_ne_stream(self):
 180         notif = bt2.StreamBeginningNotification(self._stream)
 181         stream_copy = self._sc(name='salut')
 182         notif2 = bt2.StreamBeginningNotification(stream_copy)
 183         self.assertNotEqual(notif, notif2)
 184
 185     def test_eq_invalid(self):
 186         notif = bt2.StreamBeginningNotification(self._stream)
 187         self.assertNotEqual(notif, 23)
 188
 189     def test_copy(self):
 190         notif = bt2.StreamBeginningNotification(self._stream)
 191         notif2 = copy.copy(notif)
 192         self.assertEqual(notif, notif2)
 193
 194     def test_deepcopy(self):
 195         notif = bt2.StreamBeginningNotification(self._stream)
 196         notif2 = copy.deepcopy(notif)
 197         self.assertEqual(notif, notif2)
 198
 199
 200 class StreamEndNotificationTestCase(_NotificationTestCase):
 201     def test_create(self):
 202         notif = bt2.StreamEndNotification(self._stream)
 203         self.assertEqual(notif.stream.addr, self._stream.addr)
 204
 205     def test_eq(self):
 206         notif = bt2.StreamEndNotification(self._stream)
 207         stream_copy = copy.copy(self._stream)
 208         notif2 = bt2.StreamEndNotification(stream_copy)
 209         self.assertEqual(notif, notif2)
 210
 211     def test_ne_stream(self):
 212         notif = bt2.StreamEndNotification(self._stream)
 213         stream_copy = self._sc(name='salut')
 214         notif2 = bt2.StreamEndNotification(stream_copy)
 215         self.assertNotEqual(notif, notif2)
 216
 217     def test_eq_invalid(self):
 218         notif = bt2.StreamEndNotification(self._stream)
 219         self.assertNotEqual(notif, 23)
 220
 221     def test_copy(self):
 222         notif = bt2.StreamEndNotification(self._stream)
 223         notif2 = copy.copy(notif)
 224         self.assertEqual(notif, notif2)
 225
 226     def test_deepcopy(self):
 227         notif = bt2.StreamEndNotification(self._stream)
 228         notif2 = copy.deepcopy(notif)
 229         self.assertEqual(notif, notif2)
 230
 231
 232 class InactivityNotificationTestCase(unittest.TestCase):
 233     def setUp(self):
 234         self._cc1 = bt2.ClockClass('cc1', 1000)
 235         self._cc2 = bt2.ClockClass('cc2', 2000)
 236         self._cc_prio_map = bt2.ClockClassPriorityMap()
 237         self._cc_prio_map[self._cc1] = 25
 238         self._cc_prio_map[self._cc2] = 50
 239
 240     def tearDown(self):
 241         del self._cc1
 242         del self._cc2
 243         del self._cc_prio_map
 244
 245     def test_create_no_cc_prio_map(self):
 246         notif = bt2.InactivityNotification()
 247         self.assertEqual(len(notif.clock_class_priority_map), 0)
 248
 249     def test_create_with_cc_prio_map(self):
 250         notif = bt2.InactivityNotification(self._cc_prio_map)
 251         notif.clock_values.add(self._cc1(123))
 252         notif.clock_values.add(self._cc2(19487))
 253         self.assertEqual(len(notif.clock_class_priority_map), 2)
 254         self.assertEqual(notif.clock_class_priority_map, self._cc_prio_map)
 255         self.assertEqual(notif.clock_values[self._cc1], 123)
 256         self.assertEqual(notif.clock_values[self._cc2], 19487)
 257
 258     def test_eq(self):
 259         notif = bt2.InactivityNotification(self._cc_prio_map)
 260         notif.clock_values.add(self._cc1(123))
 261         notif.clock_values.add(self._cc2(19487))
 262         cc_prio_map_copy = copy.copy(self._cc_prio_map)
 263         notif2 = bt2.InactivityNotification(cc_prio_map_copy)
 264         notif2.clock_values.add(self._cc1(123))
 265         notif2.clock_values.add(self._cc2(19487))
 266         self.assertEqual(notif, notif2)
 267
 268     def test_ne_cc_prio_map(self):
 269         notif = bt2.InactivityNotification(self._cc_prio_map)
 270         notif.clock_values.add(self._cc1(123))
 271         notif.clock_values.add(self._cc2(19487))
 272         cc_prio_map_copy = copy.copy(self._cc_prio_map)
 273         cc_prio_map_copy[self._cc2] = 23
 274         notif2 = bt2.InactivityNotification(cc_prio_map_copy)
 275         self.assertNotEqual(notif, notif2)
 276
 277     def test_ne_clock_value(self):
 278         notif = bt2.InactivityNotification(self._cc_prio_map)
 279         notif.clock_values.add(self._cc1(123))
 280         notif.clock_values.add(self._cc2(19487))
 281         notif2 = bt2.InactivityNotification(self._cc_prio_map)
 282         notif.clock_values.add(self._cc1(123))
 283         notif.clock_values.add(self._cc2(1847))
 284         self.assertNotEqual(notif, notif2)
 285
 286     def test_eq_invalid(self):
 287         notif = bt2.InactivityNotification(self._cc_prio_map)
 288         self.assertNotEqual(notif, 23)
 289
 290     def test_copy(self):
 291         notif = bt2.InactivityNotification(self._cc_prio_map)
 292         notif.clock_values.add(self._cc1(123))
 293         notif.clock_values.add(self._cc2(19487))
 294         notif_copy = copy.copy(notif)
 295         self.assertEqual(notif, notif_copy)
 296         self.assertNotEqual(notif.addr, notif_copy.addr)
 297         self.assertEqual(notif.clock_class_priority_map.addr,
 298                          notif_copy.clock_class_priority_map.addr)
 299         self.assertEqual(notif_copy.clock_values[self._cc1], 123)
 300         self.assertEqual(notif_copy.clock_values[self._cc2], 19487)
 301
 302     def test_deepcopy(self):
 303         notif = bt2.InactivityNotification(self._cc_prio_map)
 304         notif.clock_values.add(self._cc1(123))
 305         notif.clock_values.add(self._cc2(19487))
 306         notif_copy = copy.deepcopy(notif)
 307         self.assertEqual(notif, notif_copy)
 308         self.assertNotEqual(notif.addr, notif_copy.addr)
 309         self.assertNotEqual(notif.clock_class_priority_map.addr,
 310                             notif_copy.clock_class_priority_map.addr)
 311         self.assertEqual(notif.clock_class_priority_map,
 312                          notif_copy.clock_class_priority_map)
 313         self.assertNotEqual(list(notif.clock_class_priority_map)[0].addr,
 314                             list(notif_copy.clock_class_priority_map)[0].addr)
 315         self.assertIsNone(notif_copy.clock_values[self._cc1])
 316         self.assertIsNone(notif_copy.clock_values[self._cc2])
 317         self.assertEqual(notif_copy.clock_values[list(notif_copy.clock_class_priority_map)[0]], 123)
 318         self.assertEqual(notif_copy.clock_values[list(notif_copy.clock_class_priority_map)[1]], 19487)
 319
 320
 321 class DiscardedPacketsNotificationTestCase(unittest.TestCase):
 322     def setUp(self):
 323         self._trace = bt2.Trace()
 324         self._sc = bt2.StreamClass()
 325         self._ec = bt2.EventClass('salut')
 326         self._clock_class = bt2.ClockClass('yo', 1000)
 327         self._uint64_int_ft = bt2.IntegerFieldType(64, mapped_clock_class=self._clock_class)
 328         self._my_int_ft = bt2.IntegerFieldType(32)
 329         self._ec.payload_field_type = bt2.StructureFieldType()
 330         self._ec.payload_field_type += collections.OrderedDict([
 331             ('my_int', self._my_int_ft),
 332         ])
 333         self._sc.add_event_class(self._ec)
 334         self._sc.packet_context_field_type = bt2.StructureFieldType()
 335         self._sc.packet_context_field_type += collections.OrderedDict([
 336             ('packet_seq_num', self._my_int_ft),
 337             ('timestamp_begin', self._uint64_int_ft),
 338             ('timestamp_end', self._uint64_int_ft),
 339         ])
 340         self._trace.add_clock_class(self._clock_class)
 341         self._trace.add_stream_class(self._sc)
 342         self._stream = self._sc()
 343
 344     def tearDown(self):
 345         del self._trace
 346         del self._sc
 347         del self._ec
 348         del self._clock_class
 349         del self._uint64_int_ft
 350         del self._my_int_ft
 351         del self._stream
 352
 353     def _create_event(self, packet):
 354         event = self._ec()
 355         event.payload_field['my_int'] = 23
 356         event.packet = packet
 357         return event
 358
 359     def _get_notif(self):
 360         class MyIter(bt2._UserNotificationIterator):
 361             def __init__(iter_self):
 362                 packet1 = self._stream.create_packet()
 363                 packet1.context_field['packet_seq_num'] = 0
 364                 packet1.context_field['timestamp_begin'] = 3
 365                 packet1.context_field['timestamp_end'] = 6
 366                 packet2 = self._stream.create_packet()
 367                 packet2.context_field['packet_seq_num'] = 5
 368                 packet2.context_field['timestamp_begin'] = 7
 369                 packet2.context_field['timestamp_end'] = 10
 370                 iter_self._ev1 = self._create_event(packet1)
 371                 iter_self._ev2 = self._create_event(packet2)
 372                 iter_self._at = 0
 373
 374             def __next__(self):
 375                 if self._at == 0:
 376                     notif = bt2.EventNotification(self._ev1)
 377                 elif self._at == 1:
 378                     notif = bt2.EventNotification(self._ev2)
 379                 else:
 380                     raise bt2.Stop
 381
 382                 self._at += 1
 383                 return notif
 384
 385         class MySource(bt2._UserSourceComponent,
 386                        notification_iterator_class=MyIter):
 387             def __init__(self, params):
 388                 self._add_output_port('out')
 389
 390         class MySink(bt2._UserSinkComponent):
 391             def __init__(self, params):
 392                 self._add_input_port('in')
 393
 394             def _consume(comp_self):
 395                 nonlocal the_notif
 396                 notif = next(comp_self._notif_iter)
 397
 398                 if type(notif) is bt2._DiscardedPacketsNotification:
 399                     the_notif = notif
 400                     raise bt2.Stop
 401
 402             def _port_connected(self, port, other_port):
 403                 self._notif_iter = port.connection.create_notification_iterator()
 404
 405         the_notif = None
 406         graph = bt2.Graph()
 407         src = graph.add_component(MySource, 'src')
 408         sink = graph.add_component(MySink, 'sink')
 409         conn = graph.connect_ports(src.output_ports['out'],
 410                                    sink.input_ports['in'])
 411         graph.run()
 412         return the_notif
 413
 414     def test_create(self):
 415         self.assertIsInstance(self._get_notif(), bt2._DiscardedPacketsNotification)
 416
 417     def test_count(self):
 418         self.assertEqual(self._get_notif().count, 4)
 419
 420     def test_stream(self):
 421         self.assertEqual(self._get_notif().stream.addr, self._stream.addr)
 422
 423     def test_beginning_clock_value(self):
 424         notif = self._get_notif()
 425         beginning_clock_value = notif.beginning_clock_value
 426         self.assertEqual(beginning_clock_value.clock_class, self._clock_class)
 427         self.assertEqual(beginning_clock_value, 6)
 428
 429     def test_end_clock_value(self):
 430         notif = self._get_notif()
 431         end_clock_value = notif.end_clock_value
 432         self.assertEqual(end_clock_value.clock_class, self._clock_class)
 433         self.assertEqual(end_clock_value, 7)
 434
 435     def test_eq(self):
 436         notif1 = self._get_notif()
 437         notif2 = self._get_notif()
 438         self.assertEqual(notif1, notif2)
 439
 440     def test_eq_invalid(self):
 441         notif1 = self._get_notif()
 442         self.assertNotEqual(notif1, 23)
 443
 444
 445 class DiscardedEventsNotificationTestCase(unittest.TestCase):
 446     def setUp(self):
 447         self._trace = bt2.Trace()
 448         self._sc = bt2.StreamClass()
 449         self._ec = bt2.EventClass('salut')
 450         self._clock_class = bt2.ClockClass('yo', 1000)
 451         self._uint64_int_ft = bt2.IntegerFieldType(64, mapped_clock_class=self._clock_class)
 452         self._my_int_ft = bt2.IntegerFieldType(32)
 453         self._ec.payload_field_type = bt2.StructureFieldType()
 454         self._ec.payload_field_type += collections.OrderedDict([
 455             ('my_int', self._my_int_ft),
 456         ])
 457         self._sc.add_event_class(self._ec)
 458         self._sc.packet_context_field_type = bt2.StructureFieldType()
 459         self._sc.packet_context_field_type += collections.OrderedDict([
 460             ('events_discarded', self._my_int_ft),
 461             ('timestamp_begin', self._uint64_int_ft),
 462             ('timestamp_end', self._uint64_int_ft),
 463         ])
 464         self._trace.add_clock_class(self._clock_class)
 465         self._trace.add_stream_class(self._sc)
 466         self._stream = self._sc()
 467
 468     def tearDown(self):
 469         del self._trace
 470         del self._sc
 471         del self._ec
 472         del self._clock_class
 473         del self._uint64_int_ft
 474         del self._my_int_ft
 475         del self._stream
 476
 477     def _create_event(self, packet):
 478         event = self._ec()
 479         event.payload_field['my_int'] = 23
 480         event.packet = packet
 481         return event
 482
 483     def _get_notif(self):
 484         class MyIter(bt2._UserNotificationIterator):
 485             def __init__(iter_self):
 486                 packet1 = self._stream.create_packet()
 487                 packet1.context_field['events_discarded'] = 0
 488                 packet1.context_field['timestamp_begin'] = 3
 489                 packet1.context_field['timestamp_end'] = 6
 490                 packet2 = self._stream.create_packet()
 491                 packet2.context_field['events_discarded'] = 10
 492                 packet2.context_field['timestamp_begin'] = 7
 493                 packet2.context_field['timestamp_end'] = 10
 494                 iter_self._ev1 = self._create_event(packet1)
 495                 iter_self._ev2 = self._create_event(packet2)
 496                 iter_self._at = 0
 497
 498             def __next__(self):
 499                 if self._at == 0:
 500                     notif = bt2.EventNotification(self._ev1)
 501                 elif self._at == 1:
 502                     notif = bt2.EventNotification(self._ev2)
 503                 else:
 504                     raise bt2.Stop
 505
 506                 self._at += 1
 507                 return notif
 508
 509         class MySource(bt2._UserSourceComponent,
 510                        notification_iterator_class=MyIter):
 511             def __init__(self, params):
 512                 self._add_output_port('out')
 513
 514         class MySink(bt2._UserSinkComponent):
 515             def __init__(self, params):
 516                 self._add_input_port('in')
 517
 518             def _consume(comp_self):
 519                 nonlocal the_notif
 520                 notif = next(comp_self._notif_iter)
 521
 522                 if type(notif) is bt2._DiscardedEventsNotification:
 523                     the_notif = notif
 524                     raise bt2.Stop
 525
 526             def _port_connected(self, port, other_port):
 527                 self._notif_iter = port.connection.create_notification_iterator()
 528
 529         the_notif = None
 530         graph = bt2.Graph()
 531         src = graph.add_component(MySource, 'src')
 532         sink = graph.add_component(MySink, 'sink')
 533         conn = graph.connect_ports(src.output_ports['out'],
 534                                    sink.input_ports['in'])
 535         graph.run()
 536         return the_notif
 537
 538     def test_create(self):
 539         self.assertIsInstance(self._get_notif(), bt2._DiscardedEventsNotification)
 540
 541     def test_count(self):
 542         self.assertEqual(self._get_notif().count, 10)
 543
 544     def test_stream(self):
 545         self.assertEqual(self._get_notif().stream.addr, self._stream.addr)
 546
 547     def test_beginning_clock_value(self):
 548         notif = self._get_notif()
 549         beginning_clock_value = notif.beginning_clock_value
 550         self.assertEqual(beginning_clock_value.clock_class, self._clock_class)
 551         self.assertEqual(beginning_clock_value, 6)
 552
 553     def test_end_clock_value(self):
 554         notif = self._get_notif()
 555         end_clock_value = notif.end_clock_value
 556         self.assertEqual(end_clock_value.clock_class, self._clock_class)
 557         self.assertEqual(end_clock_value, 10)
 558
 559     def test_eq(self):
 560         notif1 = self._get_notif()
 561         notif2 = self._get_notif()
 562         self.assertEqual(notif1, notif2)
 563
 564     def test_eq_invalid(self):
 565         notif1 = self._get_notif()
 566         self.assertNotEqual(notif1, 23)