tests/bindings/python/bt2/test_notification_iterator.py

   1 from bt2 import values
   2 import collections
   3 import unittest
   4 import copy
   5 import bt2
   6
   7
   8 class UserNotificationIteratorTestCase(unittest.TestCase):
   9     @staticmethod
  10     def _create_graph(src_comp_cls):
  11         class MySink(bt2._UserSinkComponent):
  12             def __init__(self, params):
  13                 self._add_input_port('in')
  14
  15             def _consume(self):
  16                 next(self._notif_iter)
  17
  18             def _port_connected(self, port, other_port):
  19                 self._notif_iter = port.connection.create_notification_iterator()
  20
  21         graph = bt2.Graph()
  22         src_comp = graph.add_component(src_comp_cls, 'src')
  23         sink_comp = graph.add_component(MySink, 'sink')
  24         graph.connect_ports(src_comp.output_ports['out'],
  25                             sink_comp.input_ports['in'])
  26         return graph
  27
  28     def test_init(self):
  29         class MyIter(bt2._UserNotificationIterator):
  30             def __init__(self):
  31                 nonlocal initialized
  32                 initialized = True
  33
  34         class MySource(bt2._UserSourceComponent,
  35                        notification_iterator_class=MyIter):
  36             def __init__(self, params):
  37                 self._add_output_port('out')
  38
  39         initialized = False
  40         graph = self._create_graph(MySource)
  41         self.assertTrue(initialized)
  42
  43     def test_finalize(self):
  44         class MyIter(bt2._UserNotificationIterator):
  45             def _finalize(self):
  46                 nonlocal finalized
  47                 finalized = True
  48
  49         class MySource(bt2._UserSourceComponent,
  50                        notification_iterator_class=MyIter):
  51             def __init__(self, params):
  52                 self._add_output_port('out')
  53
  54         finalized = False
  55         graph = self._create_graph(MySource)
  56         del graph
  57         self.assertTrue(finalized)
  58
  59     def test_component(self):
  60         class MyIter(bt2._UserNotificationIterator):
  61             def __init__(self):
  62                 nonlocal salut
  63                 salut = self._component._salut
  64
  65         class MySource(bt2._UserSourceComponent,
  66                        notification_iterator_class=MyIter):
  67             def __init__(self, params):
  68                 self._add_output_port('out')
  69                 self._salut = 23
  70
  71         salut = None
  72         graph = self._create_graph(MySource)
  73         self.assertEqual(salut, 23)
  74
  75     def test_addr(self):
  76         class MyIter(bt2._UserNotificationIterator):
  77             def __init__(self):
  78                 nonlocal addr
  79                 addr = self.addr
  80
  81         class MySource(bt2._UserSourceComponent,
  82                        notification_iterator_class=MyIter):
  83             def __init__(self, params):
  84                 self._add_output_port('out')
  85
  86         addr = None
  87         graph = self._create_graph(MySource)
  88         self.assertIsNotNone(addr)
  89         self.assertNotEqual(addr, 0)
  90
  91
  92 class PrivateConnectionNotificationIteratorTestCase(unittest.TestCase):
  93     def test_component(self):
  94         class MyIter(bt2._UserNotificationIterator):
  95             pass
  96
  97         class MySource(bt2._UserSourceComponent,
  98                        notification_iterator_class=MyIter):
  99             def __init__(self, params):
 100                 self._add_output_port('out')
 101
 102         class MySink(bt2._UserSinkComponent):
 103             def __init__(self, params):
 104                 self._add_input_port('in')
 105
 106             def _consume(self):
 107                 next(self._notif_iter)
 108
 109             def _port_connected(self, port, other_port):
 110                 nonlocal upstream_comp
 111                 self._notif_iter = port.connection.create_notification_iterator()
 112                 upstream_comp = self._notif_iter.component
 113
 114         upstream_comp = None
 115         graph = bt2.Graph()
 116         src_comp = graph.add_component(MySource, 'src')
 117         sink_comp = graph.add_component(MySink, 'sink')
 118         graph.connect_ports(src_comp.output_ports['out'],
 119                             sink_comp.input_ports['in'])
 120         self.assertEqual(src_comp, upstream_comp)
 121         del upstream_comp
 122
 123
 124 class OutputPortNotificationIteratorTestCase(unittest.TestCase):
 125     def test_component(self):
 126         class MyIter(bt2._UserNotificationIterator):
 127             def __init__(self):
 128                 self._build_meta()
 129                 self._at = 0
 130
 131             def _build_meta(self):
 132                 self._trace = bt2.Trace()
 133                 self._sc = bt2.StreamClass()
 134                 self._ec = bt2.EventClass('salut')
 135                 self._my_int_ft = bt2.IntegerFieldType(32)
 136                 self._ec.payload_field_type = bt2.StructureFieldType()
 137                 self._ec.payload_field_type += collections.OrderedDict([
 138                     ('my_int', self._my_int_ft),
 139                 ])
 140                 self._sc.add_event_class(self._ec)
 141                 self._trace.add_stream_class(self._sc)
 142                 self._stream = self._sc()
 143                 self._packet = self._stream.create_packet()
 144
 145             def _create_event(self, value):
 146                 ev = self._ec()
 147                 ev.payload_field['my_int'] = value
 148                 ev.packet = self._packet
 149                 return ev
 150
 151             def __next__(self):
 152                 if self._at == 5:
 153                     raise bt2.Stop
 154
 155                 notif = bt2.EventNotification(self._create_event(self._at * 3))
 156                 self._at += 1
 157                 return notif
 158
 159         class MySource(bt2._UserSourceComponent,
 160                        notification_iterator_class=MyIter):
 161             def __init__(self, params):
 162                 self._add_output_port('out')
 163
 164         graph = bt2.Graph()
 165         src = graph.add_component(MySource, 'src')
 166         types = [bt2.EventNotification]
 167         notif_iter = src.output_ports['out'].create_notification_iterator(types)
 168
 169         for at, notif in enumerate(notif_iter):
 170             self.assertIsInstance(notif, bt2.EventNotification)
 171             self.assertEqual(notif.event.event_class.name, 'salut')
 172             field = notif.event.payload_field['my_int']
 173             self.assertEqual(field, at * 3)