projects
/
babeltrace.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
bt2: add auto source discovery support to TraceCollectionMessageIterator
[babeltrace.git]
/
tests
/
bindings
/
python
/
bt2
/
test_connection.py
diff --git
a/tests/bindings/python/bt2/test_connection.py
b/tests/bindings/python/bt2/test_connection.py
index 7c1cbe5715de45e2e2d8e61dd064ab0930c6eba0..131e1d75f9618e0322be40ca448efc07082bedb9 100644
(file)
--- a/
tests/bindings/python/bt2/test_connection.py
+++ b/
tests/bindings/python/bt2/test_connection.py
@@
-26,53
+26,42
@@
class ConnectionTestCase(unittest.TestCase):
def __next__(self):
raise bt2.Stop
def __next__(self):
raise bt2.Stop
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
- def __init__(self, params):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ def __init__(self, params, obj):
self._add_output_port('out')
class MySink(bt2._UserSinkComponent):
self._add_output_port('out')
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params
, obj
):
self._add_input_port('in')
self._add_input_port('in')
- def _consume(self):
+ def _
user_
consume(self):
raise bt2.Stop
raise bt2.Stop
- def _graph_is_configured(self):
- pass
-
graph = bt2.Graph()
src = graph.add_component(MySource, 'src')
sink = graph.add_component(MySink, 'sink')
graph = bt2.Graph()
src = graph.add_component(MySource, 'src')
sink = graph.add_component(MySink, 'sink')
- conn = graph.connect_ports(src.output_ports['out'],
- sink.input_ports['in'])
- self.assertIsInstance(conn, bt2._Connection)
+ conn = graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
def test_downstream_port(self):
class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
def test_downstream_port(self):
class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
- def __init__(self, params):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ def __init__(self, params, obj):
self._add_output_port('out')
class MySink(bt2._UserSinkComponent):
self._add_output_port('out')
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params
, obj
):
self._add_input_port('in')
self._add_input_port('in')
- def _consume(self):
+ def _
user_
consume(self):
raise bt2.Stop
raise bt2.Stop
- def _graph_is_configured(self):
- pass
-
graph = bt2.Graph()
src = graph.add_component(MySource, 'src')
sink = graph.add_component(MySink, 'sink')
graph = bt2.Graph()
src = graph.add_component(MySource, 'src')
sink = graph.add_component(MySink, 'sink')
- conn = graph.connect_ports(src.output_ports['out'],
- sink.input_ports['in'])
+ conn = graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
self.assertEqual(conn.downstream_port.addr, sink.input_ports['in'].addr)
def test_upstream_port(self):
self.assertEqual(conn.downstream_port.addr, sink.input_ports['in'].addr)
def test_upstream_port(self):
@@
-80,24
+69,19
@@
class ConnectionTestCase(unittest.TestCase):
def __next__(self):
raise bt2.Stop
def __next__(self):
raise bt2.Stop
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
- def __init__(self, params):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ def __init__(self, params, obj):
self._add_output_port('out')
class MySink(bt2._UserSinkComponent):
self._add_output_port('out')
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params
, obj
):
self._add_input_port('in')
self._add_input_port('in')
- def _consume(self):
+ def _
user_
consume(self):
raise bt2.Stop
raise bt2.Stop
- def _graph_is_configured(self):
- pass
-
graph = bt2.Graph()
src = graph.add_component(MySource, 'src')
sink = graph.add_component(MySink, 'sink')
graph = bt2.Graph()
src = graph.add_component(MySource, 'src')
sink = graph.add_component(MySink, 'sink')
- conn = graph.connect_ports(src.output_ports['out'],
- sink.input_ports['in'])
+ conn = graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
self.assertEqual(conn.upstream_port.addr, src.output_ports['out'].addr)
self.assertEqual(conn.upstream_port.addr, src.output_ports['out'].addr)
This page took
0.024889 seconds
and
4
git commands to generate.