This patch does the essential to make test_port.py pass.
Everything related to equality between ports was removed from
test_port.py, as we decided that it was not particularly useful.
Everything related to disconnection of ports and removal of ports from
components was also removed, since these concepts don't exist in
Babeltrace anymore.
I have also removed the possibility of getting a component from a port.
This was the only place where an explicit downcast from bt_component to
the specific component type would have been necessary. Instead, we have
concluded that every time you have access to the port, you already have
access to its component, already with the right type.
The possibility to create an output message iterator from an output port
is also removed, that operation will be done from a graph instead.
The other notable changes in the bindings are:
- Graph: Change add_sink_component to add_component, and handle the
various component types.
- Port: Simple adaptations to the current API, using the now famous pattern
of _as_*_ptr static methods.
- Component: Adjust how ports are listed and iterated on to account for
changes in the API.
Change-Id: Iee42bb4f8508e08d7f2b2cca451cae05d52a0ed0
Signed-off-by: Simon Marchi <simon.marchi@efficios.com>
Signed-off-by: Francis Deslauriers <francis.deslauriers@efficios.com>
Reviewed-on: https://review.lttng.org/c/babeltrace/+/1279
Reviewed-by: Philippe Proulx <eeppeliteloop@gmail.com>
Tested-by: jenkins
from bt2.packet import _Packet
from bt2.plugin import *
from bt2.port import *
-from bt2.port import _InputPort
-from bt2.port import _OutputPort
-from bt2.port import _Port
-from bt2.port import _PrivateInputPort
-from bt2.port import _PrivateOutputPort
-from bt2.port import _PrivatePort
from bt2.py_plugin import *
from bt2.query_executor import *
from bt2.stream import _Stream
def _handle_component_status(status, gen_error_msg):
- if status == native_bt.COMPONENT_STATUS_END:
+ if status == native_bt.SELF_COMPONENT_STATUS_END:
raise bt2.Stop
- elif status == native_bt.COMPONENT_STATUS_AGAIN:
+ elif status == native_bt.SELF_COMPONENT_STATUS_AGAIN:
raise bt2.TryAgain
- elif status == native_bt.COMPONENT_STATUS_UNSUPPORTED:
- raise bt2.UnsupportedFeature
- elif status == native_bt.COMPONENT_STATUS_REFUSE_PORT_CONNECTION:
+ elif status == native_bt.SELF_COMPONENT_STATUS_REFUSE_PORT_CONNECTION:
raise bt2.PortConnectionRefused
- elif status == native_bt.COMPONENT_STATUS_GRAPH_IS_CANCELED:
- raise bt2.GraphCanceled
elif status < 0:
raise bt2.Error(gen_error_msg)
raise StopIteration
comp_ports = self._comp_ports
- comp_ptr = comp_ports._component._ptr
- port_ptr = comp_ports._get_port_at_index_fn(comp_ptr, self._at)
- assert(port_ptr)
-
- if comp_ports._is_private:
- port_pub_ptr = native_bt.port_from_private(port_ptr)
- name = native_bt.port_get_name(port_pub_ptr)
- native_bt.put(port_pub_ptr)
- else:
- name = native_bt.port_get_name(port_ptr)
+ comp_ptr = comp_ports._component_ptr
+
+ port_ptr = comp_ports._borrow_port_ptr_at_index(comp_ptr, self._at)
+ assert port_ptr is not None
+
+ name = native_bt.port_get_name(comp_ports._port_pycls._as_port_ptr(port_ptr))
+ assert name is not None
- assert(name is not None)
- native_bt.put(port_ptr)
self._at += 1
return name
class _ComponentPorts(collections.abc.Mapping):
- def __init__(self, is_private, component,
- get_port_by_name_fn, get_port_at_index_fn,
- get_port_count_fn):
- self._is_private = is_private
- self._component = component
- self._get_port_by_name_fn = get_port_by_name_fn
- self._get_port_at_index_fn = get_port_at_index_fn
- self._get_port_count_fn = get_port_count_fn
+
+ # component_ptr is a bt_component_source *, bt_component_filter * or
+ # bt_component_sink *. Its type must match the type expected by the
+ # functions passed as arguments.
+
+ def __init__(self, component_ptr,
+ borrow_port_ptr_by_name,
+ borrow_port_ptr_at_index,
+ get_port_count,
+ port_pycls):
+ self._component_ptr = component_ptr
+ self._borrow_port_ptr_by_name = borrow_port_ptr_by_name
+ self._borrow_port_ptr_at_index = borrow_port_ptr_at_index
+ self._get_port_count = get_port_count
+ self._port_pycls = port_pycls
def __getitem__(self, key):
utils._check_str(key)
- port_ptr = self._get_port_by_name_fn(self._component._ptr, key)
+ port_ptr = self._borrow_port_ptr_by_name(self._component_ptr, key)
if port_ptr is None:
raise KeyError(key)
- if self._is_private:
- return bt2.port._create_private_from_ptr(port_ptr)
- else:
- return bt2.port._create_from_ptr(port_ptr)
+ return self._port_pycls._create_from_ptr_and_get_ref(port_ptr)
def __len__(self):
- if self._is_private:
- pub_ptr = native_bt.component_from_private(self._component._ptr)
- count = self._get_port_count_fn(pub_ptr)
- native_bt.put(pub_ptr)
- else:
- count = self._get_port_count_fn(self._component._ptr)
-
- assert(count >= 0)
+ count = self._get_port_count(self._component_ptr)
+ assert count >= 0
return count
def __iter__(self):
# This is analogous to _GenericSourceComponentClass, but for source
# component objects.
class _GenericSourceComponent(object._SharedObject, _SourceComponent):
+ _get_ref = staticmethod(native_bt.component_source_get_ref)
+ _put_ref = staticmethod(native_bt.component_source_put_ref)
+
@property
def output_ports(self):
- return _ComponentPorts(False, self,
- native_bt.component_source_get_output_port_by_name,
- native_bt.component_source_get_output_port_by_index,
- native_bt.component_source_get_output_port_count)
+ return _ComponentPorts(self._ptr,
+ native_bt.component_source_borrow_output_port_by_name_const,
+ native_bt.component_source_borrow_output_port_by_index_const,
+ native_bt.component_source_get_output_port_count,
+ bt2.port._OutputPort)
# This is analogous to _GenericFilterComponentClass, but for filter
# component objects.
class _GenericFilterComponent(object._SharedObject, _FilterComponent):
+ _get_ref = staticmethod(native_bt.component_filter_get_ref)
+ _put_ref = staticmethod(native_bt.component_filter_put_ref)
+
@property
def output_ports(self):
- return _ComponentPorts(False, self,
- native_bt.component_filter_get_output_port_by_name,
- native_bt.component_filter_get_output_port_by_index,
- native_bt.component_filter_get_output_port_count)
+ return _ComponentPorts(self._ptr,
+ native_bt.component_filter_borrow_output_port_by_name_const,
+ native_bt.component_filter_borrow_output_port_by_index_const,
+ native_bt.component_filter_get_output_port_count,
+ bt2.port._OutputPort)
@property
def input_ports(self):
- return _ComponentPorts(False, self,
- native_bt.component_filter_get_input_port_by_name,
- native_bt.component_filter_get_input_port_by_index,
- native_bt.component_filter_get_input_port_count)
+ return _ComponentPorts(self._ptr,
+ native_bt.component_filter_borrow_input_port_by_name_const,
+ native_bt.component_filter_borrow_input_port_by_index_const,
+ native_bt.component_filter_get_input_port_count,
+ bt2.port._InputPort)
# This is analogous to _GenericSinkComponentClass, but for sink
@property
def input_ports(self):
- return _ComponentPorts(False, self,
- native_bt.component_sink_get_input_port_by_name,
- native_bt.component_sink_get_input_port_by_index,
- native_bt.component_sink_get_input_port_count)
+ return _ComponentPorts(self._ptr,
+ native_bt.component_sink_borrow_input_port_by_name_const,
+ native_bt.component_sink_borrow_input_port_by_index_const,
+ native_bt.component_sink_get_input_port_count,
+ bt2.port._InputPort)
_COMP_CLS_TYPE_TO_GENERIC_COMP_PYCLS = {
def _accept_port_connection(self, port, other_port):
return True
- def _accept_port_connection_from_native(self, port_ptr, other_port_ptr):
- native_bt.get(port_ptr)
- native_bt.get(other_port_ptr)
- port = bt2.port._create_private_from_ptr(port_ptr)
- other_port = bt2.port._create_from_ptr(other_port_ptr)
+ def _accept_port_connection_from_native(self, self_port_ptr, self_port_type, other_port_ptr):
+ port = bt2.port._create_self_from_ptr_and_get_ref(
+ self_port_ptr, self_port_type)
+ other_port_type = native_bt.PORT_TYPE_INPUT if self_port_type == native_bt.PORT_TYPE_OUTPUT else native_bt.PORT_TYPE_OUTPUT
+ other_port = bt2.port._create_from_ptr_and_get_ref(
+ other_port_ptr, other_port_type)
res = self._accept_port_connection(port, other_port_ptr)
if type(res) is not bool:
def _port_connected(self, port, other_port):
pass
- def _port_connected_from_native(self, port_ptr, other_port_ptr):
- native_bt.get(port_ptr)
- native_bt.get(other_port_ptr)
- port = bt2.port._create_private_from_ptr(port_ptr)
- other_port = bt2.port._create_from_ptr(other_port_ptr)
-
- try:
- self._port_connected(port, other_port)
- except:
- if not _NO_PRINT_TRACEBACK:
- traceback.print_exc()
-
- def _port_disconnected(self, port):
- pass
-
- def _port_disconnected_from_native(self, port_ptr):
- native_bt.get(port_ptr)
- port = bt2.port._create_private_from_ptr(port_ptr)
-
- try:
- self._port_disconnected(port)
- except:
- if not _NO_PRINT_TRACEBACK:
- traceback.print_exc()
+ def _port_connected_from_native(self, self_port_ptr, self_port_type, other_port_ptr):
+ port = bt2.port._create_self_from_ptr_and_get_ref(
+ self_port_ptr, self_port_type)
+ other_port_type = native_bt.PORT_TYPE_INPUT if self_port_type == native_bt.PORT_TYPE_OUTPUT else native_bt.PORT_TYPE_OUTPUT
+ other_port = bt2.port._create_from_ptr_and_get_ref(
+ other_port_ptr, other_port_type)
+ self._port_connected(port, other_port)
class _UserSourceComponent(_UserComponent, _SourceComponent):
@property
def _output_ports(self):
- return _ComponentPorts(True, self,
- native_bt.private_component_source_get_output_private_port_by_name,
- native_bt.private_component_source_get_output_private_port_by_index,
- native_bt.component_source_get_output_port_count)
+ def get_output_port_count(self_ptr):
+ ptr = self._as_not_self_specific_component_ptr(self_ptr)
+ return native_bt.component_source_get_output_port_count(ptr)
+
+ return _ComponentPorts(self._ptr,
+ native_bt.self_component_source_borrow_output_port_by_name,
+ native_bt.self_component_source_borrow_output_port_by_index,
+ get_output_port_count,
+ bt2.port._UserComponentOutputPort)
def _add_output_port(self, name):
utils._check_str(name)
- fn = native_bt.private_component_source_add_output_private_port
- comp_status, priv_port_ptr = fn(self._ptr, name, None)
+ fn = native_bt.self_component_source_add_output_port
+ comp_status, self_port_ptr = fn(self._ptr, name, None)
_handle_component_status(comp_status,
'cannot add output port to source component object')
- assert(priv_port_ptr)
- return bt2.port._create_private_from_ptr(priv_port_ptr)
+ assert self_port_ptr is not None
+ return bt2.port._UserComponentOutputPort._create_from_ptr(self_port_ptr)
class _UserFilterComponent(_UserComponent, _FilterComponent):
@property
def _output_ports(self):
- return _ComponentPorts(True, self,
- native_bt.private_component_filter_get_output_private_port_by_name,
- native_bt.private_component_filter_get_output_private_port_by_index,
- native_bt.component_filter_get_output_port_count)
+ def get_output_port_count(self_ptr):
+ ptr = self._as_not_self_specific_component_ptr(self_ptr)
+ return native_bt.component_filter_get_output_port_count(ptr)
+
+ return _ComponentPorts(self._ptr,
+ native_bt.self_component_filter_borrow_output_port_by_name,
+ native_bt.self_component_filter_borrow_output_port_by_index,
+ get_output_port_count,
+ bt2.port._UserComponentOutputPort)
@property
def _input_ports(self):
- return _ComponentPorts(True, self,
- native_bt.private_component_filter_get_input_private_port_by_name,
- native_bt.private_component_filter_get_input_private_port_by_index,
- native_bt.component_filter_get_input_port_count)
+ def get_input_port_count(self_ptr):
+ ptr = self._as_not_self_specific_component_ptr(self_ptr)
+ return native_bt.component_filter_get_input_port_count(ptr)
+
+ return _ComponentPorts(self._ptr,
+ native_bt.self_component_filter_borrow_input_port_by_name,
+ native_bt.self_component_filter_borrow_input_port_by_index,
+ get_input_port_count,
+ bt2.port._UserComponentInputPort)
def _add_output_port(self, name):
utils._check_str(name)
- fn = native_bt.private_component_filter_add_output_private_port
- comp_status, priv_port_ptr = fn(self._ptr, name, None)
+ fn = native_bt.self_component_filter_add_output_port
+ comp_status, self_port_ptr = fn(self._ptr, name, None)
_handle_component_status(comp_status,
'cannot add output port to filter component object')
- assert(priv_port_ptr)
- return bt2.port._create_private_from_ptr(priv_port_ptr)
+ assert self_port_ptr
+ return bt2.port._UserComponentOutputPort._create_from_ptr(self_port_ptr)
def _add_input_port(self, name):
utils._check_str(name)
- fn = native_bt.private_component_filter_add_input_private_port
- comp_status, priv_port_ptr = fn(self._ptr, name, None)
+ fn = native_bt.self_component_filter_add_input_port
+ comp_status, self_port_ptr = fn(self._ptr, name, None)
_handle_component_status(comp_status,
'cannot add input port to filter component object')
- assert(priv_port_ptr)
- return bt2.port._create_private_from_ptr(priv_port_ptr)
+ assert self_port_ptr
+ return bt2.port._UserComponentInputPort._create_from_ptr(self_port_ptr)
class _UserSinkComponent(_UserComponent, _SinkComponent):
@property
def _input_ports(self):
- return _ComponentPorts(True, self,
- native_bt.private_component_sink_get_input_private_port_by_name,
- native_bt.private_component_sink_get_input_private_port_by_index,
- native_bt.component_sink_get_input_port_count)
+ def get_input_port_count(self_ptr):
+ ptr = self._as_not_self_specific_component_ptr(self_ptr)
+ return native_bt.component_sink_get_input_port_count(ptr)
+
+ return _ComponentPorts(self._ptr,
+ native_bt.self_component_sink_borrow_input_port_by_name,
+ native_bt.self_component_sink_borrow_input_port_by_index,
+ get_input_port_count,
+ bt2.port._UserComponentInputPort)
def _add_input_port(self, name):
utils._check_str(name)
- fn = native_bt.private_component_sink_add_input_private_port
+ fn = native_bt.self_component_sink_add_input_port
comp_status, priv_port_ptr = fn(self._ptr, name, None)
_handle_component_status(comp_status,
'cannot add input port to sink component object')
- assert(priv_port_ptr)
- return bt2.port._create_private_from_ptr(priv_port_ptr)
+ assert priv_port_ptr
+ return bt2.port._UserComponentInputPort._create_from_ptr(priv_port_ptr)
elif status < 0:
raise bt2.Error(gen_error_msg)
- def add_sink_component(self, component_class, name, params=None):
- if issubclass(component_class, bt2.component._UserSinkComponent):
+ def add_component(self, component_class, name, params=None):
+ if issubclass(component_class, bt2.component._UserSourceComponent):
cc_ptr = component_class._cc_ptr
+ add_fn = native_bt.graph_add_source_component
+ cc_type = native_bt.COMPONENT_CLASS_TYPE_SOURCE
+ elif issubclass(component_class, bt2.component._UserFilterComponent):
+ cc_ptr = component_class._cc_ptr
+ add_fn = native_bt.graph_add_filter_component
+ cc_type = native_bt.COMPONENT_CLASS_TYPE_FILTER
+ elif issubclass(component_class, bt2.component._UserSinkComponent):
+ cc_ptr = component_class._cc_ptr
+ add_fn = native_bt.graph_add_sink_component
+ cc_type = native_bt.COMPONENT_CLASS_TYPE_SINK
else:
- raise TypeError("'{}' is not a sink component class".format(
+ raise TypeError("'{}' is not a component class".format(
component_class.__class__.__name__))
utils._check_str(name)
params_ptr = params._ptr if params is not None else None
- status, comp_ptr = native_bt.graph_add_sink_component(self._ptr, cc_ptr,
- name, params_ptr)
- self._handle_status(status, 'cannot add sink component to graph')
- assert(comp_ptr)
- return bt2.component._create_component_from_ptr(comp_ptr, native_bt.COMPONENT_CLASS_TYPE_SINK)
+ status, comp_ptr = add_fn(self._ptr, cc_ptr, name, params_ptr)
+ self._handle_status(status, 'cannot add component to graph')
+ assert comp_ptr
+ return bt2.component._create_component_from_ptr(comp_ptr, cc_type)
def connect_ports(self, upstream_port, downstream_port):
utils._check_type(upstream_port, bt2.port._OutputPort)
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
-from bt2 import native_bt, object, utils
-import collections.abc
+from bt2 import native_bt, object
import bt2.component
import bt2.connection
import bt2.message_iterator
import bt2.message
-import copy
import bt2
-def _create_from_ptr(ptr):
- port_type = native_bt.port_get_type(ptr)
-
- if port_type == native_bt.PORT_TYPE_INPUT:
- cls = _InputPort
- elif port_type == native_bt.PORT_TYPE_OUTPUT:
- cls = _OutputPort
- else:
- raise bt2.Error('unknown port type: {}'.format(port_type))
-
- return cls._create_from_ptr(ptr)
-
-
-def _create_private_from_ptr(ptr):
- pub_ptr = native_bt.port_from_private(ptr)
- utils._handle_ptr(pub_ptr, 'cannot get port object from private port object')
- port_type = native_bt.port_get_type(pub_ptr)
- assert(port_type == native_bt.PORT_TYPE_INPUT or port_type == native_bt.PORT_TYPE_OUTPUT)
-
- if port_type == native_bt.PORT_TYPE_INPUT:
- cls = _PrivateInputPort
- elif port_type == native_bt.PORT_TYPE_OUTPUT:
- cls = _PrivateOutputPort
-
- obj = cls._create_from_ptr(ptr)
- obj._pub_ptr = pub_ptr
- return obj
-
-
class _Port(object._SharedObject):
- @staticmethod
- def _name(ptr):
- name = native_bt.port_get_name(ptr)
- assert(name is not None)
- return name
-
- @staticmethod
- def _disconnect(ptr):
- status = native_bt.port_disconnect(ptr)
+ @classmethod
+ def _get_ref(cls, ptr):
+ ptr = cls._as_port_ptr(ptr)
+ return native_bt.port_get_ref(ptr)
- if status < 0:
- raise bt2.Error('cannot disconnect port')
+ @classmethod
+ def _put_ref(cls, ptr):
+ ptr = cls._as_port_ptr(ptr)
+ return native_bt.port_put_ref(ptr)
@property
def name(self):
- return self._name(self._ptr)
-
- @property
- def component(self):
- comp_ptr = native_bt.port_get_component(self._ptr)
-
- if comp_ptr is None:
- return
-
- return bt2.component._create_generic_component_from_ptr(comp_ptr)
+ ptr = self._as_port_ptr(self._ptr)
+ name = native_bt.port_get_name(ptr)
+ assert name is not None
+ return name
@property
def connection(self):
- conn_ptr = native_bt.port_get_connection(self._ptr)
+ ptr = self._as_port_ptr(self._ptr)
+ conn_ptr = native_bt.port_borrow_connection_const(ptr)
if conn_ptr is None:
return
- return bt2.connection._Connection._create_from_ptr(conn_ptr)
+ return bt2.connection._Connection._create_from_ptr_and_get_ref(conn_ptr)
@property
def is_connected(self):
return self.connection is not None
- def disconnect(self):
- self._disconnect(self._ptr)
-
- def __eq__(self, other):
- if type(other) is not type(self):
- return False
-
- return self.addr == other.addr
-
class _InputPort(_Port):
- pass
+ _as_port_ptr = staticmethod(native_bt.port_input_as_port_const)
class _OutputPort(_Port):
- def create_message_iterator(self, message_types=None,
- colander_component_name=None):
- msg_types = bt2.message._msg_types_from_msg_classes(message_types)
-
- if colander_component_name is not None:
- utils._check_str(colander_component_name)
-
- msg_iter_ptr = native_bt.py3_create_output_port_msg_iter(int(self._ptr),
- colander_component_name,
- msg_types)
+ _as_port_ptr = staticmethod(native_bt.port_output_as_port_const)
- if msg_iter_ptr is None:
- raise bt2.CreationError('cannot create output port message iterator')
- return bt2.message_iterator._OutputPortMessageIterator._create_from_ptr(msg_iter_ptr)
-
-
-class _PrivatePort(_Port):
- @property
- def name(self):
- return self._name(self._pub_ptr)
-
- @property
- def component(self):
- comp_ptr = native_bt.private_port_get_private_component(self._ptr)
-
- if comp_ptr is None:
- return
-
- pub_comp_ptr = native_bt.component_from_private(comp_ptr)
- assert(pub_comp_ptr)
- comp = bt2.component._create_generic_component_from_ptr(pub_comp_ptr)
- native_bt.put(comp_ptr)
- return comp
+class _UserComponentPort(_Port):
+ @classmethod
+ def _as_port_ptr(cls, ptr):
+ ptr = cls._as_self_port_ptr(ptr)
+ return native_bt.self_component_port_as_port(ptr)
@property
def connection(self):
- conn_ptr = native_bt.private_port_get_private_connection(self._ptr)
+ ptr = self._as_port_ptr(self._ptr)
+ conn_ptr = native_bt.port_borrow_connection_const(ptr)
if conn_ptr is None:
return
- return bt2.connection._create_private_from_ptr(conn_ptr)
-
- def remove_from_component(self):
- status = native_bt.private_port_remove_from_component(self._ptr)
-
- if status < 0:
- raise bt2.Error("cannot remove port from component")
-
- def disconnect(self):
- self._disconnect(self._pub_ptr)
+ return bt2.connection._Connection._create_from_ptr_and_get_ref(conn_ptr)
-class _PrivateInputPort(_PrivatePort, _InputPort):
- pass
+class _UserComponentInputPort(_UserComponentPort, _InputPort):
+ _as_self_port_ptr = staticmethod(native_bt.self_component_port_input_as_self_component_port)
-class _PrivateOutputPort(_PrivatePort, _OutputPort):
- pass
+class _UserComponentOutputPort(_UserComponentPort, _OutputPort):
+ _as_self_port_ptr = staticmethod(native_bt.self_component_port_output_as_self_component_port)
if name is None:
name = 'comp'
- return graph.add_sink_component(comp_cls, name)
+ return graph.add_component(comp_cls, name)
def test_name(self):
class MySink(bt2._UserSinkComponent):
finalized = True
graph = bt2.Graph()
- comp = graph.add_sink_component(MySink, 'lel')
+ comp = graph.add_component(MySink, 'lel')
del graph
del comp
if name is None:
name = 'comp'
- return graph.add_sink_component(comp_cls, name)
+ return graph.add_component(comp_cls, name)
def test_name(self):
class MySink(bt2._UserSinkComponent):
self._py_comp_cls = MySink
graph = bt2.Graph()
- comp = graph.add_sink_component(MySink, 'salut')
+ comp = graph.add_component(MySink, 'salut')
self._comp_cls = comp.component_class
self.assertTrue(issubclass(type(self._comp_cls),
bt2.component._GenericComponentClass))
-from bt2 import value
import unittest
-import copy
import bt2
-@unittest.skip("this is broken")
class PortTestCase(unittest.TestCase):
@staticmethod
def _create_comp(comp_cls, name=None):
comp = self._create_comp(MySource)
self.assertEqual(len(comp.output_ports), 1)
-
def test_flt_add_output_port(self):
class MyIter(bt2._UserMessageIterator):
def __next__(self):
self.assertEqual(port2.addr, comp_self._output_ports['print'].addr)
self.assertEqual(port1.addr, comp_self._output_ports['clear'].addr)
- comp = self._create_comp(MySource)
+ self._create_comp(MySource)
def test_user_flt_output_ports_getitem(self):
class MyIter(bt2._UserMessageIterator):
self.assertEqual(port2.addr, comp_self._output_ports['print'].addr)
self.assertEqual(port1.addr, comp_self._output_ports['clear'].addr)
- comp = self._create_comp(MyFilter)
+ self._create_comp(MyFilter)
def test_user_flt_input_ports_getitem(self):
class MyIter(bt2._UserMessageIterator):
self.assertEqual(port2.addr, comp_self._input_ports['print'].addr)
self.assertEqual(port1.addr, comp_self._input_ports['clear'].addr)
- comp = self._create_comp(MyFilter)
+ self._create_comp(MyFilter)
def test_user_sink_input_ports_getitem(self):
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
- comp = self._create_comp(MySink)
+ self._create_comp(MySink)
def test_user_src_output_ports_getitem_invalid_key(self):
class MyIter(bt2._UserMessageIterator):
with self.assertRaises(KeyError):
comp_self._output_ports['hello']
- comp = self._create_comp(MySource)
+ self._create_comp(MySource)
def test_user_flt_output_ports_getitem_invalid_key(self):
class MyIter(bt2._UserMessageIterator):
with self.assertRaises(KeyError):
comp_self._output_ports['hello']
- comp = self._create_comp(MyFilter)
+ self._create_comp(MyFilter)
def test_user_flt_input_ports_getitem_invalid_key(self):
class MyIter(bt2._UserMessageIterator):
with self.assertRaises(KeyError):
comp_self._input_ports['hello']
- comp = self._create_comp(MyFilter)
+ self._create_comp(MyFilter)
def test_user_sink_input_ports_getitem_invalid_key(self):
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
- comp = self._create_comp(MySink)
+ self._create_comp(MySink)
def test_user_src_output_ports_len(self):
class MyIter(bt2._UserMessageIterator):
comp_self._add_output_port('insert')
self.assertEqual(len(comp_self._output_ports), 3)
- comp = self._create_comp(MySource)
+ self._create_comp(MySource)
def test_user_flt_output_ports_len(self):
class MyIter(bt2._UserMessageIterator):
comp_self._add_output_port('insert')
self.assertEqual(len(comp_self._output_ports), 3)
- comp = self._create_comp(MyFilter)
+ self._create_comp(MyFilter)
def test_user_flt_input_ports_len(self):
class MyIter(bt2._UserMessageIterator):
comp_self._add_input_port('insert')
self.assertEqual(len(comp_self._input_ports), 3)
- comp = self._create_comp(MyFilter)
+ self._create_comp(MyFilter)
def test_user_sink_input_ports_len(self):
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
- comp = self._create_comp(MySink)
+ self._create_comp(MySink)
def test_user_src_output_ports_iter(self):
class MyIter(bt2._UserMessageIterator):
self.assertEqual(ports[2][0], 'insert')
self.assertEqual(ports[2][1].addr, port3.addr)
- comp = self._create_comp(MySource)
+ self._create_comp(MySource)
def test_user_flt_output_ports_iter(self):
class MyIter(bt2._UserMessageIterator):
self.assertEqual(ports[2][0], 'insert')
self.assertEqual(ports[2][1].addr, port3.addr)
- comp = self._create_comp(MyFilter)
+ self._create_comp(MyFilter)
def test_user_flt_input_ports_iter(self):
class MyIter(bt2._UserMessageIterator):
self.assertEqual(ports[2][0], 'insert')
self.assertEqual(ports[2][1].addr, port3.addr)
- comp = self._create_comp(MyFilter)
+ self._create_comp(MyFilter)
def test_user_sink_input_ports_iter(self):
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
- comp = self._create_comp(MySink)
+ self._create_comp(MySink)
def test_gen_src_output_ports_getitem(self):
class MyIter(bt2._UserMessageIterator):
comp = self._create_comp(MySink)
self.assertEqual(comp.input_ports['clear'].name, 'clear')
- def test_component(self):
- class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
- comp_self._add_input_port('clear')
-
- def _consume(self):
- pass
-
- comp = self._create_comp(MySink)
- self.assertEqual(comp.input_ports['clear'].component.addr, comp.addr)
-
def test_connection_none(self):
class MySink(bt2._UserSinkComponent):
def __init__(comp_self, params):
comp = self._create_comp(MySink)
self.assertFalse(comp.input_ports['clear'].is_connected)
- def test_eq(self):
- class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
- comp_self._add_input_port('clear')
-
- def _consume(self):
- pass
-
- comp = self._create_comp(MySink)
- self.assertEqual(comp.input_ports['clear'],
- comp.input_ports['clear'])
-
- def test_eq_invalid(self):
- class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
- comp_self._add_input_port('clear')
-
- def _consume(self):
- pass
-
- comp = self._create_comp(MySink)
- self.assertNotEqual(comp.input_ports['clear'], 23)
-
- def test_disconnect_no_connection(self):
- class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
- port = comp_self._add_input_port('clear')
-
- def _consume(self):
- pass
-
- comp = self._create_comp(MySink)
- comp.input_ports['clear'].disconnect()
-
- def test_priv_name(self):
+ def test_self_name(self):
class MySink(bt2._UserSinkComponent):
def __init__(comp_self, params):
port = comp_self._add_input_port('clear')
def _consume(self):
pass
- comp = self._create_comp(MySink)
+ self._create_comp(MySink)
- def test_priv_component(self):
- class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
- port = comp_self._add_input_port('clear')
- self.assertEqual(port.component, comp_self)
-
- def _consume(self):
- pass
-
- comp = self._create_comp(MySink)
-
- def test_priv_connection_none(self):
+ def test_self_connection_none(self):
class MySink(bt2._UserSinkComponent):
def __init__(comp_self, params):
port = comp_self._add_input_port('clear')
def _consume(self):
pass
- comp = self._create_comp(MySink)
+ self._create_comp(MySink)
- def test_priv_is_connected_false(self):
+ def test_self_is_connected_false(self):
class MySink(bt2._UserSinkComponent):
def __init__(comp_self, params):
port = comp_self._add_input_port('clear')
def _consume(self):
pass
- comp = self._create_comp(MySink)
-
- def test_priv_eq(self):
- class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
- port = comp_self._add_input_port('clear')
- self.assertEqual(port, port)
-
- def _consume(self):
- pass
-
- comp = self._create_comp(MySink)
-
- def test_priv_eq_invalid(self):
- class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
- port = comp_self._add_input_port('clear')
- self.assertNotEqual(port, 23)
-
- def _consume(self):
- pass
-
- comp = self._create_comp(MySink)
-
- def test_priv_disconnect_no_connection(self):
- class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
- port = comp_self._add_input_port('clear')
- port.disconnect()
-
- def _consume(self):
- pass
-
- comp = self._create_comp(MySink)
-
- def test_priv_remove_from_component(self):
- class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
- port = comp_self._add_input_port('clear')
- self.assertEqual(len(comp_self._input_ports), 1)
-
- try:
- port.remove_from_component()
- except:
- import traceback
- traceback.print_exc()
-
- self.assertEqual(len(comp_self._input_ports), 0)
- self.assertIsNone(port.component)
-
- def _consume(self):
- pass
-
- comp = self._create_comp(MySink)
+ self._create_comp(MySink)
projects / babeltrace.git / commitdiff
