tests/bindings/python/bt2/test_connection.py

   1 from bt2 import values
   2 import unittest
   3 import copy
   4 import bt2
   5
   6
   7 class ConnectionTestCase(unittest.TestCase):
   8     def test_create(self):
   9         class MyIter(bt2._UserNotificationIterator):
  10             def __next__(self):
  11                 raise bt2.Stop
  12
  13         class MySource(bt2._UserSourceComponent,
  14                        notification_iterator_class=MyIter):
  15             def __init__(self, params):
  16                 self._add_output_port('out')
  17
  18         class MySink(bt2._UserSinkComponent):
  19             def __init__(self, params):
  20                 self._add_input_port('in')
  21
  22             def _consume(self):
  23                 raise bt2.Stop
  24
  25         graph = bt2.Graph()
  26         src = graph.add_component(MySource, 'src')
  27         sink = graph.add_component(MySink, 'sink')
  28         conn = graph.connect_ports(src.output_ports['out'],
  29                                    sink.input_ports['in'])
  30         self.assertIsInstance(conn, bt2._Connection)
  31         self.assertNotIsInstance(conn, bt2._PrivateConnection)
  32
  33     def test_is_ended_false(self):
  34         class MyIter(bt2._UserNotificationIterator):
  35             def __next__(self):
  36                 raise bt2.Stop
  37
  38         class MySource(bt2._UserSourceComponent,
  39                        notification_iterator_class=MyIter):
  40             def __init__(self, params):
  41                 self._add_output_port('out')
  42
  43         class MySink(bt2._UserSinkComponent):
  44             def __init__(self, params):
  45                 self._add_input_port('in')
  46
  47             def _consume(self):
  48                 raise bt2.Stop
  49
  50         graph = bt2.Graph()
  51         src = graph.add_component(MySource, 'src')
  52         sink = graph.add_component(MySink, 'sink')
  53         conn = graph.connect_ports(src.output_ports['out'],
  54                                    sink.input_ports['in'])
  55         self.assertFalse(conn.is_ended)
  56
  57     def test_is_ended_true(self):
  58         class MyIter(bt2._UserNotificationIterator):
  59             def __next__(self):
  60                 raise bt2.Stop
  61
  62         class MySource(bt2._UserSourceComponent,
  63                        notification_iterator_class=MyIter):
  64             def __init__(self, params):
  65                 self._add_output_port('out')
  66
  67         class MySink(bt2._UserSinkComponent):
  68             def __init__(self, params):
  69                 self._add_input_port('in')
  70
  71             def _consume(self):
  72                 raise bt2.Stop
  73
  74         graph = bt2.Graph()
  75         src = graph.add_component(MySource, 'src')
  76         sink = graph.add_component(MySink, 'sink')
  77         conn = graph.connect_ports(src.output_ports['out'],
  78                                    sink.input_ports['in'])
  79         src.output_ports['out'].disconnect()
  80         self.assertTrue(conn.is_ended)
  81
  82     def test_downstream_port(self):
  83         class MyIter(bt2._UserNotificationIterator):
  84             def __next__(self):
  85                 raise bt2.Stop
  86
  87         class MySource(bt2._UserSourceComponent,
  88                        notification_iterator_class=MyIter):
  89             def __init__(self, params):
  90                 self._add_output_port('out')
  91
  92         class MySink(bt2._UserSinkComponent):
  93             def __init__(self, params):
  94                 self._add_input_port('in')
  95
  96             def _consume(self):
  97                 raise bt2.Stop
  98
  99         graph = bt2.Graph()
 100         src = graph.add_component(MySource, 'src')
 101         sink = graph.add_component(MySink, 'sink')
 102         conn = graph.connect_ports(src.output_ports['out'],
 103                                    sink.input_ports['in'])
 104         self.assertEqual(conn.downstream_port.addr, sink.input_ports['in'].addr)
 105
 106     def test_upstream_port(self):
 107         class MyIter(bt2._UserNotificationIterator):
 108             def __next__(self):
 109                 raise bt2.Stop
 110
 111         class MySource(bt2._UserSourceComponent,
 112                        notification_iterator_class=MyIter):
 113             def __init__(self, params):
 114                 self._add_output_port('out')
 115
 116         class MySink(bt2._UserSinkComponent):
 117             def __init__(self, params):
 118                 self._add_input_port('in')
 119
 120             def _consume(self):
 121                 raise bt2.Stop
 122
 123         graph = bt2.Graph()
 124         src = graph.add_component(MySource, 'src')
 125         sink = graph.add_component(MySink, 'sink')
 126         conn = graph.connect_ports(src.output_ports['out'],
 127                                    sink.input_ports['in'])
 128         self.assertEqual(conn.upstream_port.addr, src.output_ports['out'].addr)
 129
 130     def test_eq(self):
 131         class MyIter(bt2._UserNotificationIterator):
 132             def __next__(self):
 133                 raise bt2.Stop
 134
 135         class MySource(bt2._UserSourceComponent,
 136                        notification_iterator_class=MyIter):
 137             def __init__(self, params):
 138                 self._add_output_port('out')
 139
 140         class MySink(bt2._UserSinkComponent):
 141             def __init__(self, params):
 142                 self._add_input_port('in')
 143
 144             def _consume(self):
 145                 raise bt2.Stop
 146
 147         graph = bt2.Graph()
 148         src = graph.add_component(MySource, 'src')
 149         sink = graph.add_component(MySink, 'sink')
 150         conn = graph.connect_ports(src.output_ports['out'],
 151                                    sink.input_ports['in'])
 152         self.assertEqual(conn, conn)
 153
 154     def test_eq_invalid(self):
 155         class MyIter(bt2._UserNotificationIterator):
 156             def __next__(self):
 157                 raise bt2.Stop
 158
 159         class MySource(bt2._UserSourceComponent,
 160                        notification_iterator_class=MyIter):
 161             def __init__(self, params):
 162                 self._add_output_port('out')
 163
 164         class MySink(bt2._UserSinkComponent):
 165             def __init__(self, params):
 166                 self._add_input_port('in')
 167
 168             def _consume(self):
 169                 raise bt2.Stop
 170
 171         graph = bt2.Graph()
 172         src = graph.add_component(MySource, 'src')
 173         sink = graph.add_component(MySink, 'sink')
 174         conn = graph.connect_ports(src.output_ports['out'],
 175                                    sink.input_ports['in'])
 176         self.assertNotEqual(conn, 23)
 177
 178
 179 class PrivateConnectionTestCase(unittest.TestCase):
 180     def test_create(self):
 181         class MyIter(bt2._UserNotificationIterator):
 182             def __next__(self):
 183                 raise bt2.Stop
 184
 185         class MySource(bt2._UserSourceComponent,
 186                        notification_iterator_class=MyIter):
 187             def __init__(self, params):
 188                 self._add_output_port('out')
 189
 190         class MySink(bt2._UserSinkComponent):
 191             def __init__(self, params):
 192                 self._add_input_port('in')
 193
 194             def _consume(self):
 195                 raise bt2.Stop
 196
 197             def _port_connected(self, port, other_port):
 198                 nonlocal priv_conn
 199                 priv_conn = port.connection
 200
 201         priv_conn = None
 202         graph = bt2.Graph()
 203         src = graph.add_component(MySource, 'src')
 204         sink = graph.add_component(MySink, 'sink')
 205         conn = graph.connect_ports(src.output_ports['out'],
 206                                    sink.input_ports['in'])
 207         self.assertIsInstance(priv_conn, bt2._PrivateConnection)
 208         self.assertEqual(conn, priv_conn)
 209         del priv_conn
 210
 211     def test_is_ended_false(self):
 212         class MyIter(bt2._UserNotificationIterator):
 213             def __next__(self):
 214                 raise bt2.Stop
 215
 216         class MySource(bt2._UserSourceComponent,
 217                        notification_iterator_class=MyIter):
 218             def __init__(self, params):
 219                 self._add_output_port('out')
 220
 221         class MySink(bt2._UserSinkComponent):
 222             def __init__(self, params):
 223                 self._add_input_port('in')
 224
 225             def _consume(self):
 226                 raise bt2.Stop
 227
 228             def _port_connected(self, port, other_port):
 229                 nonlocal priv_conn
 230                 priv_conn = port.connection
 231
 232         priv_conn = None
 233         graph = bt2.Graph()
 234         src = graph.add_component(MySource, 'src')
 235         sink = graph.add_component(MySink, 'sink')
 236         conn = graph.connect_ports(src.output_ports['out'],
 237                                    sink.input_ports['in'])
 238         self.assertFalse(priv_conn.is_ended)
 239         del priv_conn
 240
 241     def test_is_ended_true(self):
 242         class MyIter(bt2._UserNotificationIterator):
 243             def __next__(self):
 244                 raise bt2.Stop
 245
 246         class MySource(bt2._UserSourceComponent,
 247                        notification_iterator_class=MyIter):
 248             def __init__(self, params):
 249                 self._add_output_port('out')
 250
 251         class MySink(bt2._UserSinkComponent):
 252             def __init__(self, params):
 253                 self._add_input_port('in')
 254
 255             def _consume(self):
 256                 raise bt2.Stop
 257
 258             def _port_connected(self, port, other_port):
 259                 nonlocal priv_conn
 260                 priv_conn = port.connection
 261
 262         priv_conn = None
 263         graph = bt2.Graph()
 264         src = graph.add_component(MySource, 'src')
 265         sink = graph.add_component(MySink, 'sink')
 266         conn = graph.connect_ports(src.output_ports['out'],
 267                                    sink.input_ports['in'])
 268         sink.input_ports['in'].disconnect()
 269         self.assertTrue(priv_conn.is_ended)
 270         del priv_conn
 271
 272     def test_downstream_port(self):
 273         class MyIter(bt2._UserNotificationIterator):
 274             def __next__(self):
 275                 raise bt2.Stop
 276
 277         class MySource(bt2._UserSourceComponent,
 278                        notification_iterator_class=MyIter):
 279             def __init__(self, params):
 280                 self._add_output_port('out')
 281
 282         class MySink(bt2._UserSinkComponent):
 283             def __init__(self, params):
 284                 self._add_input_port('in')
 285
 286             def _consume(self):
 287                 raise bt2.Stop
 288
 289             def _port_connected(self, port, other_port):
 290                 nonlocal priv_port
 291                 priv_conn = port.connection
 292                 priv_port = priv_conn.downstream_port
 293
 294         priv_port = None
 295         graph = bt2.Graph()
 296         src = graph.add_component(MySource, 'src')
 297         sink = graph.add_component(MySink, 'sink')
 298         conn = graph.connect_ports(src.output_ports['out'],
 299                                    sink.input_ports['in'])
 300         self.assertEqual(priv_port.addr, sink.input_ports['in'].addr)
 301         del priv_port
 302
 303     def test_upstream_port(self):
 304         class MyIter(bt2._UserNotificationIterator):
 305             def __next__(self):
 306                 raise bt2.Stop
 307
 308         class MySource(bt2._UserSourceComponent,
 309                        notification_iterator_class=MyIter):
 310             def __init__(self, params):
 311                 self._add_output_port('out')
 312
 313         class MySink(bt2._UserSinkComponent):
 314             def __init__(self, params):
 315                 self._add_input_port('in')
 316
 317             def _consume(self):
 318                 raise bt2.Stop
 319
 320             def _port_connected(self, port, other_port):
 321                 nonlocal priv_port
 322                 priv_conn = port.connection
 323                 priv_port = priv_conn.upstream_port
 324
 325         priv_port = None
 326         graph = bt2.Graph()
 327         src = graph.add_component(MySource, 'src')
 328         sink = graph.add_component(MySink, 'sink')
 329         conn = graph.connect_ports(src.output_ports['out'],
 330                                    sink.input_ports['in'])
 331         self.assertEqual(priv_port.addr, src.output_ports['out'].addr)
 332         del priv_port
 333
 334     def test_eq(self):
 335         class MyIter(bt2._UserNotificationIterator):
 336             def __next__(self):
 337                 raise bt2.Stop
 338
 339         class MySource(bt2._UserSourceComponent,
 340                        notification_iterator_class=MyIter):
 341             def __init__(self, params):
 342                 self._add_output_port('out')
 343
 344         class MySink(bt2._UserSinkComponent):
 345             def __init__(self, params):
 346                 self._add_input_port('in')
 347
 348             def _consume(self):
 349                 raise bt2.Stop
 350
 351             def _port_connected(self, port, other_port):
 352                 nonlocal priv_conn
 353                 priv_conn = port.connection
 354
 355         priv_conn = None
 356         graph = bt2.Graph()
 357         src = graph.add_component(MySource, 'src')
 358         sink = graph.add_component(MySink, 'sink')
 359         conn = graph.connect_ports(src.output_ports['out'],
 360                                    sink.input_ports['in'])
 361         self.assertEqual(priv_conn, conn)
 362         del priv_conn
 363
 364     def test_eq_invalid(self):
 365         class MyIter(bt2._UserNotificationIterator):
 366             def __next__(self):
 367                 raise bt2.Stop
 368
 369         class MySource(bt2._UserSourceComponent,
 370                        notification_iterator_class=MyIter):
 371             def __init__(self, params):
 372                 self._add_output_port('out')
 373
 374         class MySink(bt2._UserSinkComponent):
 375             def __init__(self, params):
 376                 self._add_input_port('in')
 377
 378             def _consume(self):
 379                 raise bt2.Stop
 380
 381             def _port_connected(self, port, other_port):
 382                 nonlocal priv_conn
 383                 priv_conn = port.connection
 384
 385         priv_conn = None
 386         graph = bt2.Graph()
 387         src = graph.add_component(MySource, 'src')
 388         sink = graph.add_component(MySink, 'sink')
 389         conn = graph.connect_ports(src.output_ports['out'],
 390                                    sink.input_ports['in'])
 391         self.assertNotEqual(priv_conn, 23)
 392         del priv_conn