- self.assertEqual(calls[2][1].name, 'in')
- self.assertIs(calls[3][0], port_removed_listener)
- self.assertEqual(calls[3][1].name, 'zero')
- self.assertIs(calls[4][0], port_added_listener)
- self.assertEqual(calls[4][1].name, 'taste')
- self.assertIs(calls[5][0], ports_connected_listener)
- self.assertEqual(calls[5][1].name, 'out')
- self.assertEqual(calls[5][2].name, 'in')
- self.assertIs(calls[6][0], port_removed_listener)
- self.assertEqual(calls[6][1].name, 'in')
- self.assertIs(calls[7][0], ports_disconnected_listener)
- self.assertEqual(calls[7][1].name, 'src')
- self.assertEqual(calls[7][2].name, 'sink')
- self.assertEqual(calls[7][3].name, 'out')
- self.assertEqual(calls[7][4].name, 'in')
- del calls
+ self.assertEqual(calls[2][1].name, 'sink')
+ self.assertEqual(calls[2][2].name, 'in')
+
+ self.assertIs(calls[3][0], port_added_listener)
+ self.assertEqual(calls[3][1].name, 'sink')
+ self.assertEqual(calls[3][2].name, 'taste')
+
+ def test_invalid_listeners(self):
+ class MySource(
+ bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
+ def __init__(self, config, params, obj):
+ self._add_output_port('out')
+ self._add_output_port('zero')
+
+ class MySink(bt2._UserSinkComponent):
+ def __init__(self, config, params, obj):
+ self._add_input_port('in')
+
+ def _user_consume(self):
+ raise bt2.Stop
+
+ def _user_port_connected(self, port, other_port):
+ self._add_input_port('taste')
+
+ with self.assertRaises(TypeError):
+ self._graph.add_port_added_listener(1234)
+
+ def test_raise_in_component_init(self):
+ class MySink(bt2._UserSinkComponent):
+ def __init__(self, config, params, obj):
+ raise ValueError('oops!')
+
+ def _user_consume(self):
+ raise bt2.Stop
+
+ graph = bt2.Graph()
+
+ with self.assertRaises(bt2._Error):
+ graph.add_component(MySink, 'comp')
+
+ def test_raise_in_port_added_listener(self):
+ class MySink(bt2._UserSinkComponent):
+ def __init__(self, config, params, obj):
+ self._add_input_port('in')
+
+ def _user_consume(self):
+ raise bt2.Stop
+
+ def port_added_listener(component, port):
+ raise ValueError('oh noes!')
+
+ graph = bt2.Graph()
+ graph.add_port_added_listener(port_added_listener)
+
+ with self.assertRaises(bt2._Error):
+ graph.add_component(MySink, 'comp')
+
+
+if __name__ == '__main__':
+ unittest.main()