tests/bindings/python/bt2/test_connection.py

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