# general configuration
needs_sphinx = '1.2'
-extensions = [
- 'sphinx.ext.autodoc',
- 'sphinx.ext.ifconfig',
-]
+extensions = ['sphinx.ext.autodoc', 'sphinx.ext.ifconfig']
templates_path = ['_templates']
source_suffix = '.rst'
source_encoding = 'utf-8-sig'
import atexit
atexit.register(_native_bt.bt2_cc_exit_handler)
- version = (_native_bt.version_get_major(), _native_bt.version_get_minor(),
- _native_bt.version_get_patch(), _native_bt.version_get_extra())
+ version = (
+ _native_bt.version_get_major(),
+ _native_bt.version_get_minor(),
+ _native_bt.version_get_patch(),
+ _native_bt.version_get_extra(),
+ )
_native_bt.bt2_cc_init_from_bt2()
def _name(self, name):
utils._check_str(name)
status = native_bt.clock_class_set_name(self._ptr, name)
- utils._handle_func_status(status,
- "cannot set clock class object's name")
+ utils._handle_func_status(status, "cannot set clock class object's name")
_name = property(fset=_name)
def _description(self, description):
utils._check_str(description)
status = native_bt.clock_class_set_description(self._ptr, description)
- utils._handle_func_status(status,
- "cannot set clock class object's description")
+ utils._handle_func_status(status, "cannot set clock class object's description")
_description = property(fset=_description)
def _origin_is_unix_epoch(self, origin_is_unix_epoch):
utils._check_bool(origin_is_unix_epoch)
- native_bt.clock_class_set_origin_is_unix_epoch(self._ptr, int(origin_is_unix_epoch))
+ native_bt.clock_class_set_origin_is_unix_epoch(
+ self._ptr, int(origin_is_unix_epoch)
+ )
_origin_is_unix_epoch = property(fset=_origin_is_unix_epoch)
@property
def ns_from_origin(self):
status, ns = native_bt.clock_snapshot_get_ns_from_origin(self._ptr)
- utils._handle_func_status(status,
- "cannot get clock snapshot's nanoseconds from origin")
+ utils._handle_func_status(
+ status, "cannot get clock snapshot's nanoseconds from origin"
+ )
return ns
def __eq__(self, other):
# - _bt_as_component_class_ptr: static method, convert the passed component class
# pointer to a 'bt_component_class *'.
+
class _GenericComponentClass(object._SharedObject):
@property
def name(self):
class _GenericSourceComponentClass(_GenericComponentClass):
_get_ref = staticmethod(native_bt.component_class_source_get_ref)
_put_ref = staticmethod(native_bt.component_class_source_put_ref)
- _bt_as_component_class_ptr = staticmethod(native_bt.component_class_source_as_component_class)
+ _bt_as_component_class_ptr = staticmethod(
+ native_bt.component_class_source_as_component_class
+ )
class _GenericFilterComponentClass(_GenericComponentClass):
_get_ref = staticmethod(native_bt.component_class_filter_get_ref)
_put_ref = staticmethod(native_bt.component_class_filter_put_ref)
- _bt_as_component_class_ptr = staticmethod(native_bt.component_class_filter_as_component_class)
+ _bt_as_component_class_ptr = staticmethod(
+ native_bt.component_class_filter_as_component_class
+ )
class _GenericSinkComponentClass(_GenericComponentClass):
_get_ref = staticmethod(native_bt.component_class_sink_get_ref)
_put_ref = staticmethod(native_bt.component_class_sink_put_ref)
- _bt_as_component_class_ptr = staticmethod(native_bt.component_class_sink_as_component_class)
+ _bt_as_component_class_ptr = staticmethod(
+ native_bt.component_class_sink_as_component_class
+ )
class _PortIterator(collections.abc.Iterator):
# 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):
+ 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
# - _bt_as_component_ptr: static method, must return the passed specialized
# component pointer (e.g. 'bt_component_sink *') as a 'bt_component *'.
+
class _Component:
@property
def name(self):
def cls(self):
cc_ptr = self._bt_borrow_component_class_ptr(self._ptr)
assert cc_ptr is not None
- return _create_component_class_from_ptr_and_get_ref(cc_ptr, self._bt_comp_cls_type)
+ return _create_component_class_from_ptr_and_get_ref(
+ cc_ptr, self._bt_comp_cls_type
+ )
def __eq__(self, other):
if not hasattr(other, 'addr'):
class _SourceComponent(_Component):
- _bt_borrow_component_class_ptr = staticmethod(native_bt.component_source_borrow_class_const)
+ _bt_borrow_component_class_ptr = staticmethod(
+ native_bt.component_source_borrow_class_const
+ )
_bt_comp_cls_type = native_bt.COMPONENT_CLASS_TYPE_SOURCE
- _bt_as_component_class_ptr = staticmethod(native_bt.component_class_source_as_component_class)
+ _bt_as_component_class_ptr = staticmethod(
+ native_bt.component_class_source_as_component_class
+ )
_bt_as_component_ptr = staticmethod(native_bt.component_source_as_component_const)
class _FilterComponent(_Component):
- _bt_borrow_component_class_ptr = staticmethod(native_bt.component_filter_borrow_class_const)
+ _bt_borrow_component_class_ptr = staticmethod(
+ native_bt.component_filter_borrow_class_const
+ )
_bt_comp_cls_type = native_bt.COMPONENT_CLASS_TYPE_FILTER
- _bt_as_component_class_ptr = staticmethod(native_bt.component_class_filter_as_component_class)
+ _bt_as_component_class_ptr = staticmethod(
+ native_bt.component_class_filter_as_component_class
+ )
_bt_as_component_ptr = staticmethod(native_bt.component_filter_as_component_const)
class _SinkComponent(_Component):
- _bt_borrow_component_class_ptr = staticmethod(native_bt.component_sink_borrow_class_const)
+ _bt_borrow_component_class_ptr = staticmethod(
+ native_bt.component_sink_borrow_class_const
+ )
_bt_comp_cls_type = native_bt.COMPONENT_CLASS_TYPE_SINK
- _bt_as_component_class_ptr = staticmethod(native_bt.component_class_sink_as_component_class)
+ _bt_as_component_class_ptr = staticmethod(
+ native_bt.component_class_sink_as_component_class
+ )
_bt_as_component_ptr = staticmethod(native_bt.component_sink_as_component_const)
@property
def output_ports(self):
- 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)
+ 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
@property
def output_ports(self):
- 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)
+ 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(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)
+ 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(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)
+ 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 = {
#
# Steals the reference to ptr from the caller.
+
def _create_component_from_ptr(ptr, comp_cls_type):
return _COMP_CLS_TYPE_TO_GENERIC_COMP_PYCLS[comp_cls_type]._create_from_ptr(ptr)
# Same as the above, but acquire a new reference instead of stealing the
# reference from the caller.
+
def _create_component_from_ptr_and_get_ref(ptr, comp_cls_type):
- return _COMP_CLS_TYPE_TO_GENERIC_COMP_PYCLS[comp_cls_type]._create_from_ptr_and_get_ref(ptr)
+ return _COMP_CLS_TYPE_TO_GENERIC_COMP_PYCLS[
+ comp_cls_type
+ ]._create_from_ptr_and_get_ref(ptr)
# Create a component class Python object of type
#
# Acquires a new reference to ptr.
+
def _create_component_class_from_ptr_and_get_ref(ptr, comp_cls_type):
- return _COMP_CLS_TYPE_TO_GENERIC_COMP_CLS_PYCLS[comp_cls_type]._create_from_ptr_and_get_ref(ptr)
+ return _COMP_CLS_TYPE_TO_GENERIC_COMP_CLS_PYCLS[
+ comp_cls_type
+ ]._create_from_ptr_and_get_ref(ptr)
def _trim_docstring(docstring):
if _UserSourceComponent in bases:
_UserComponentType._bt_set_iterator_class(cls, iter_cls)
- cc_ptr = native_bt.bt2_component_class_source_create(cls,
- comp_cls_name,
- comp_cls_descr,
- comp_cls_help)
+ cc_ptr = native_bt.bt2_component_class_source_create(
+ cls, comp_cls_name, comp_cls_descr, comp_cls_help
+ )
elif _UserFilterComponent in bases:
_UserComponentType._bt_set_iterator_class(cls, iter_cls)
- cc_ptr = native_bt.bt2_component_class_filter_create(cls,
- comp_cls_name,
- comp_cls_descr,
- comp_cls_help)
+ cc_ptr = native_bt.bt2_component_class_filter_create(
+ cls, comp_cls_name, comp_cls_descr, comp_cls_help
+ )
elif _UserSinkComponent in bases:
if not hasattr(cls, '_graph_is_configured'):
- raise bt2.IncompleteUserClass("cannot create component class '{}': missing a _graph_is_configured() method".format(class_name))
+ raise bt2.IncompleteUserClass(
+ "cannot create component class '{}': missing a _graph_is_configured() method".format(
+ class_name
+ )
+ )
if not hasattr(cls, '_consume'):
- raise bt2.IncompleteUserClass("cannot create component class '{}': missing a _consume() method".format(class_name))
-
- cc_ptr = native_bt.bt2_component_class_sink_create(cls,
- comp_cls_name,
- comp_cls_descr,
- comp_cls_help)
+ raise bt2.IncompleteUserClass(
+ "cannot create component class '{}': missing a _consume() method".format(
+ class_name
+ )
+ )
+
+ cc_ptr = native_bt.bt2_component_class_sink_create(
+ cls, comp_cls_name, comp_cls_descr, comp_cls_help
+ )
else:
- raise bt2.IncompleteUserClass("cannot find a known component class base in the bases of '{}'".format(class_name))
+ raise bt2.IncompleteUserClass(
+ "cannot find a known component class base in the bases of '{}'".format(
+ class_name
+ )
+ )
if cc_ptr is None:
- raise bt2.CreationError("cannot create component class '{}'".format(class_name))
+ raise bt2.CreationError(
+ "cannot create component class '{}'".format(class_name)
+ )
cls._bt_cc_ptr = cc_ptr
return self
def __call__(cls, *args, **kwargs):
- raise bt2.Error('cannot directly instantiate a user component from a Python module')
+ raise bt2.Error(
+ 'cannot directly instantiate a user component from a Python module'
+ )
@staticmethod
def _bt_set_iterator_class(cls, iter_cls):
if iter_cls is None:
- raise bt2.IncompleteUserClass("cannot create component class '{}': missing message iterator class".format(cls.__name__))
+ raise bt2.IncompleteUserClass(
+ "cannot create component class '{}': missing message iterator class".format(
+ cls.__name__
+ )
+ )
if not issubclass(iter_cls, bt2.message_iterator._UserMessageIterator):
- raise bt2.IncompleteUserClass("cannot create component class '{}': message iterator class does not inherit bt2._UserMessageIterator".format(cls.__name__))
+ raise bt2.IncompleteUserClass(
+ "cannot create component class '{}': message iterator class does not inherit bt2._UserMessageIterator".format(
+ cls.__name__
+ )
+ )
if not hasattr(iter_cls, '__next__'):
- raise bt2.IncompleteUserClass("cannot create component class '{}': message iterator class is missing a __next__() method".format(cls.__name__))
+ raise bt2.IncompleteUserClass(
+ "cannot create component class '{}': message iterator class is missing a __next__() method".format(
+ cls.__name__
+ )
+ )
cls._iter_cls = iter_cls
else:
params = None
- query_exec = bt2.QueryExecutor._create_from_ptr_and_get_ref(
- query_exec_ptr)
+ query_exec = bt2.QueryExecutor._create_from_ptr_and_get_ref(query_exec_ptr)
# this can raise, but the native side checks the exception
results = cls._query(query_exec, obj, params, log_level)
cc_ptr = cls._bt_as_component_class_ptr(cls._bt_cc_ptr)
native_bt.component_class_put_ref(cc_ptr)
+
# Subclasses must provide these methods or property:
#
# - _bt_as_not_self_specific_component_ptr: static method, must return the passed
# - _bt_comp_cls_type: property, one of the native_bt.COMPONENT_CLASS_TYPE_*
# constants.
+
class _UserComponent(metaclass=_UserComponentType):
@property
def name(self):
def cls(self):
comp_ptr = self._bt_as_not_self_specific_component_ptr(self._bt_ptr)
cc_ptr = self._bt_borrow_component_class_ptr(comp_ptr)
- return _create_component_class_from_ptr_and_get_ref(cc_ptr, self._bt_comp_cls_type)
+ return _create_component_class_from_ptr_and_get_ref(
+ cc_ptr, self._bt_comp_cls_type
+ )
@property
def addr(self):
def _port_connected(self, port, other_port):
pass
- def _bt_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)
+ def _bt_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)
if self_port_type == native_bt.PORT_TYPE_OUTPUT:
other_port_type = native_bt.PORT_TYPE_INPUT
other_port_type = native_bt.PORT_TYPE_OUTPUT
other_port = bt2.port._create_from_ptr_and_get_ref(
- other_port_ptr, other_port_type)
+ other_port_ptr, other_port_type
+ )
self._port_connected(port, other_port)
def _bt_graph_is_configured_from_native(self):
return tc
- def _create_clock_class(self, frequency=None, name=None, description=None,
- precision=None, offset=None, origin_is_unix_epoch=True,
- uuid=None):
+ def _create_clock_class(
+ self,
+ frequency=None,
+ name=None,
+ description=None,
+ precision=None,
+ offset=None,
+ origin_is_unix_epoch=True,
+ uuid=None,
+ ):
ptr = self._bt_as_self_component_ptr(self._bt_ptr)
cc_ptr = native_bt.clock_class_create(ptr)
class _UserSourceComponent(_UserComponent, _SourceComponent):
- _bt_as_not_self_specific_component_ptr = staticmethod(native_bt.self_component_source_as_component_source)
- _bt_as_self_component_ptr = staticmethod(native_bt.self_component_source_as_self_component)
+ _bt_as_not_self_specific_component_ptr = staticmethod(
+ native_bt.self_component_source_as_component_source
+ )
+ _bt_as_self_component_ptr = staticmethod(
+ native_bt.self_component_source_as_self_component
+ )
@property
def _output_ports(self):
ptr = self._bt_as_not_self_specific_component_ptr(self_ptr)
return native_bt.component_source_get_output_port_count(ptr)
- return _ComponentPorts(self._bt_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)
+ return _ComponentPorts(
+ self._bt_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, user_data=None):
utils._check_str(name)
fn = native_bt.self_component_source_add_output_port
comp_status, self_port_ptr = fn(self._bt_ptr, name, user_data)
- utils._handle_func_status(comp_status,
- 'cannot add output port to source component object')
+ utils._handle_func_status(
+ comp_status, 'cannot add output port to source component object'
+ )
assert self_port_ptr is not None
return bt2.port._UserComponentOutputPort._create_from_ptr(self_port_ptr)
class _UserFilterComponent(_UserComponent, _FilterComponent):
- _bt_as_not_self_specific_component_ptr = staticmethod(native_bt.self_component_filter_as_component_filter)
- _bt_as_self_component_ptr = staticmethod(native_bt.self_component_filter_as_self_component)
+ _bt_as_not_self_specific_component_ptr = staticmethod(
+ native_bt.self_component_filter_as_component_filter
+ )
+ _bt_as_self_component_ptr = staticmethod(
+ native_bt.self_component_filter_as_self_component
+ )
@property
def _output_ports(self):
ptr = self._bt_as_not_self_specific_component_ptr(self_ptr)
return native_bt.component_filter_get_output_port_count(ptr)
- return _ComponentPorts(self._bt_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)
+ return _ComponentPorts(
+ self._bt_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):
ptr = self._bt_as_not_self_specific_component_ptr(self_ptr)
return native_bt.component_filter_get_input_port_count(ptr)
- return _ComponentPorts(self._bt_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)
+ return _ComponentPorts(
+ self._bt_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, user_data=None):
utils._check_str(name)
fn = native_bt.self_component_filter_add_output_port
comp_status, self_port_ptr = fn(self._bt_ptr, name, user_data)
- utils._handle_func_status(comp_status,
- 'cannot add output port to filter component object')
+ utils._handle_func_status(
+ comp_status, 'cannot add output port to filter component object'
+ )
assert self_port_ptr
return bt2.port._UserComponentOutputPort._create_from_ptr(self_port_ptr)
utils._check_str(name)
fn = native_bt.self_component_filter_add_input_port
comp_status, self_port_ptr = fn(self._bt_ptr, name, user_data)
- utils._handle_func_status(comp_status,
- 'cannot add input port to filter component object')
+ utils._handle_func_status(
+ comp_status, 'cannot add input port to filter component object'
+ )
assert self_port_ptr
return bt2.port._UserComponentInputPort._create_from_ptr(self_port_ptr)
class _UserSinkComponent(_UserComponent, _SinkComponent):
- _bt_as_not_self_specific_component_ptr = staticmethod(native_bt.self_component_sink_as_component_sink)
- _bt_as_self_component_ptr = staticmethod(native_bt.self_component_sink_as_self_component)
+ _bt_as_not_self_specific_component_ptr = staticmethod(
+ native_bt.self_component_sink_as_component_sink
+ )
+ _bt_as_self_component_ptr = staticmethod(
+ native_bt.self_component_sink_as_self_component
+ )
@property
def _input_ports(self):
ptr = self._bt_as_not_self_specific_component_ptr(self_ptr)
return native_bt.component_sink_get_input_port_count(ptr)
- return _ComponentPorts(self._bt_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)
+ return _ComponentPorts(
+ self._bt_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, user_data=None):
utils._check_str(name)
fn = native_bt.self_component_sink_add_input_port
comp_status, self_port_ptr = fn(self._bt_ptr, name, user_data)
- utils._handle_func_status(comp_status,
- 'cannot add input port to sink component object')
+ utils._handle_func_status(
+ comp_status, 'cannot add input port to sink component object'
+ )
assert self_port_ptr
return bt2.port._UserComponentInputPort._create_from_ptr(self_port_ptr)
@property
def downstream_port(self):
port_ptr = native_bt.connection_borrow_downstream_port_const(self._ptr)
- utils._handle_ptr(port_ptr, "cannot get connection object's downstream port object")
- return bt2.port._create_from_ptr_and_get_ref(port_ptr, native_bt.PORT_TYPE_INPUT)
+ utils._handle_ptr(
+ port_ptr, "cannot get connection object's downstream port object"
+ )
+ return bt2.port._create_from_ptr_and_get_ref(
+ port_ptr, native_bt.PORT_TYPE_INPUT
+ )
@property
def upstream_port(self):
port_ptr = native_bt.connection_borrow_upstream_port_const(self._ptr)
- utils._handle_ptr(port_ptr, "cannot get connection object's upstream port object")
- return bt2.port._create_from_ptr_and_get_ref(port_ptr, native_bt.PORT_TYPE_OUTPUT)
+ utils._handle_ptr(
+ port_ptr, "cannot get connection object's upstream port object"
+ )
+ return bt2.port._create_from_ptr_and_get_ref(
+ port_ptr, native_bt.PORT_TYPE_OUTPUT
+ )
if field_ptr is None:
return
- return bt2.field._create_field_from_ptr(field_ptr, self._owner_ptr,
- self._owner_get_ref,
- self._owner_put_ref)
+ return bt2.field._create_field_from_ptr(
+ field_ptr, self._owner_ptr, self._owner_get_ref, self._owner_put_ref
+ )
@property
def specific_context_field(self):
if field_ptr is None:
return
- return bt2.field._create_field_from_ptr(field_ptr, self._owner_ptr,
- self._owner_get_ref,
- self._owner_put_ref)
+ return bt2.field._create_field_from_ptr(
+ field_ptr, self._owner_ptr, self._owner_get_ref, self._owner_put_ref
+ )
@property
def payload_field(self):
if field_ptr is None:
return
- return bt2.field._create_field_from_ptr(field_ptr, self._owner_ptr,
- self._owner_get_ref,
- self._owner_put_ref)
+ return bt2.field._create_field_from_ptr(
+ field_ptr, self._owner_ptr, self._owner_get_ref, self._owner_put_ref
+ )
def __getitem__(self, key):
utils._check_str(key)
def _emf_uri(self, emf_uri):
utils._check_str(emf_uri)
status = native_bt.event_class_set_emf_uri(self._ptr, emf_uri)
- utils._handle_func_status(status,
- "cannot set event class object's EMF URI")
+ utils._handle_func_status(status, "cannot set event class object's EMF URI")
_emf_uri = property(fset=_emf_uri)
@property
def specific_context_field_class(self):
- fc_ptr = native_bt.event_class_borrow_specific_context_field_class_const(self._ptr)
+ fc_ptr = native_bt.event_class_borrow_specific_context_field_class_const(
+ self._ptr
+ )
if fc_ptr is None:
return
def _specific_context_field_class(self, context_field_class):
if context_field_class is not None:
utils._check_type(context_field_class, bt2.field_class._StructureFieldClass)
- status = native_bt.event_class_set_specific_context_field_class(self._ptr, context_field_class._ptr)
- utils._handle_func_status(status,
- "cannot set event class object's context field class")
+ status = native_bt.event_class_set_specific_context_field_class(
+ self._ptr, context_field_class._ptr
+ )
+ utils._handle_func_status(
+ status, "cannot set event class object's context field class"
+ )
_specific_context_field_class = property(fset=_specific_context_field_class)
def _payload_field_class(self, payload_field_class):
if payload_field_class is not None:
utils._check_type(payload_field_class, bt2.field_class._StructureFieldClass)
- status = native_bt.event_class_set_payload_field_class(self._ptr, payload_field_class._ptr)
- utils._handle_func_status(status,
- "cannot set event class object's payload field class")
+ status = native_bt.event_class_set_payload_field_class(
+ self._ptr, payload_field_class._ptr
+ )
+ utils._handle_func_status(
+ status, "cannot set event class object's payload field class"
+ )
_payload_field_class = property(fset=_payload_field_class)
utils._handle_ptr(field_class_ptr, "cannot get field object's class")
typeid = native_bt.field_class_get_type(field_class_ptr)
field = _TYPE_ID_TO_OBJ[typeid]._create_from_ptr_and_get_ref(
- ptr, owner_ptr, owner_get_ref, owner_put_ref)
+ ptr, owner_ptr, owner_get_ref, owner_put_ref
+ )
return field
# currently selected field. If `field` is of any other type, return `field`
# directly.
+
def _get_leaf_field(field):
if not isinstance(field, _VariantField):
return field
if isinstance(other, numbers.Complex):
return complex(other)
- raise TypeError("'{}' object is not a number object".format(other.__class__.__name__))
+ raise TypeError(
+ "'{}' object is not a number object".format(other.__class__.__name__)
+ )
def __int__(self):
return int(self._value)
def __lt__(self, other):
if not isinstance(other, numbers.Number):
- raise TypeError('unorderable types: {}() < {}()'.format(self.__class__.__name__,
- other.__class__.__name__))
+ raise TypeError(
+ 'unorderable types: {}() < {}()'.format(
+ self.__class__.__name__, other.__class__.__name__
+ )
+ )
return self._value < self._extract_value(other)
@property
def labels(self):
status, labels = self._get_mapping_labels(self._ptr)
- utils._handle_func_status(status,
- "cannot get label for enumeration field")
+ utils._handle_func_status(status, "cannot get label for enumeration field")
assert labels is not None
return labels
class _UnsignedEnumerationField(_EnumerationField, _UnsignedIntegerField):
_NAME = 'Unsigned Enumeration'
- _get_mapping_labels = staticmethod(native_bt.field_unsigned_enumeration_get_mapping_labels)
+ _get_mapping_labels = staticmethod(
+ native_bt.field_unsigned_enumeration_get_mapping_labels
+ )
class _SignedEnumerationField(_EnumerationField, _SignedIntegerField):
_NAME = 'Signed Enumeration'
- _get_mapping_labels = staticmethod(native_bt.field_signed_enumeration_get_mapping_labels)
+ _get_mapping_labels = staticmethod(
+ native_bt.field_signed_enumeration_get_mapping_labels
+ )
@functools.total_ordering
def __iadd__(self, value):
value = self._value_to_str(value)
status = native_bt.field_string_append(self._ptr, value)
- utils._handle_func_status(status,
- "cannot append to string field object's value")
+ utils._handle_func_status(
+ status, "cannot append to string field object's value"
+ )
return self
def __getitem__(self, key):
utils._check_str(key)
- field_ptr = native_bt.field_structure_borrow_member_field_by_name(self._ptr, key)
+ field_ptr = native_bt.field_structure_borrow_member_field_by_name(
+ self._ptr, key
+ )
if field_ptr is None:
raise KeyError(key)
- return _create_field_from_ptr(field_ptr, self._owner_ptr,
- self._owner_get_ref,
- self._owner_put_ref)
+ return _create_field_from_ptr(
+ field_ptr, self._owner_ptr, self._owner_get_ref, self._owner_put_ref
+ )
def member_at_index(self, index):
utils._check_uint64(index)
if index >= len(self):
raise IndexError
- field_ptr = native_bt.field_structure_borrow_member_field_by_index(self._ptr, index)
+ field_ptr = native_bt.field_structure_borrow_member_field_by_index(
+ self._ptr, index
+ )
assert field_ptr is not None
- return _create_field_from_ptr(field_ptr, self._owner_ptr,
- self._owner_get_ref,
- self._owner_put_ref)
+ return _create_field_from_ptr(
+ field_ptr, self._owner_ptr, self._owner_get_ref, self._owner_put_ref
+ )
class _VariantField(_ContainerField, _Field):
field_ptr = native_bt.field_variant_borrow_selected_option_field(self._ptr)
utils._handle_ptr(field_ptr, "cannot get variant field's selected option")
- return _create_field_from_ptr(field_ptr, self._owner_ptr,
- self._owner_get_ref,
- self._owner_put_ref)
+ return _create_field_from_ptr(
+ field_ptr, self._owner_ptr, self._owner_get_ref, self._owner_put_ref
+ )
def _spec_eq(self, other):
return _get_leaf_field(self) == other
def __getitem__(self, index):
if not isinstance(index, numbers.Integral):
- raise TypeError("'{}' is not an integral number object: invalid index".format(index.__class__.__name__))
+ raise TypeError(
+ "'{}' is not an integral number object: invalid index".format(
+ index.__class__.__name__
+ )
+ )
index = int(index)
if index < 0 or index >= len(self):
raise IndexError('{} field object index is out of range'.format(self._NAME))
- field_ptr = native_bt.field_array_borrow_element_field_by_index(self._ptr, index)
- assert(field_ptr)
- return _create_field_from_ptr(field_ptr, self._owner_ptr,
- self._owner_get_ref,
- self._owner_put_ref)
+ field_ptr = native_bt.field_array_borrow_element_field_by_index(
+ self._ptr, index
+ )
+ assert field_ptr
+ return _create_field_from_ptr(
+ field_ptr, self._owner_ptr, self._owner_get_ref, self._owner_put_ref
+ )
def __setitem__(self, index, value):
# raises if index is somehow invalid
def _set_value(self, values):
if len(self) != len(values):
- raise ValueError(
- 'expected length of value and array field to match')
+ raise ValueError('expected length of value and array field to match')
for index, value in enumerate(values):
if value is not None:
def _check_create_status(self, ptr):
if ptr is None:
- raise bt2.CreationError('cannot create {} field class object'.format(self._NAME.lower()))
+ raise bt2.CreationError(
+ 'cannot create {} field class object'.format(self._NAME.lower())
+ )
class _IntegerFieldClass(_FieldClass):
@property
def field_value_range(self):
size = native_bt.field_class_integer_get_field_value_range(self._ptr)
- assert(size >= 1)
+ assert size >= 1
return size
def _field_value_range(self, size):
@property
def preferred_display_base(self):
- base = native_bt.field_class_integer_get_preferred_display_base(
- self._ptr)
- assert(base >= 0)
+ base = native_bt.field_class_integer_get_preferred_display_base(self._ptr)
+ assert base >= 0
return base
def _preferred_display_base(self, base):
utils._check_uint64(base)
- if base not in (IntegerDisplayBase.BINARY,
- IntegerDisplayBase.OCTAL,
- IntegerDisplayBase.DECIMAL,
- IntegerDisplayBase.HEXADECIMAL):
+ if base not in (
+ IntegerDisplayBase.BINARY,
+ IntegerDisplayBase.OCTAL,
+ IntegerDisplayBase.DECIMAL,
+ IntegerDisplayBase.HEXADECIMAL,
+ ):
raise ValueError("Display base is not a valid IntegerDisplayBase value")
- native_bt.field_class_integer_set_preferred_display_base(
- self._ptr, base)
+ native_bt.field_class_integer_set_preferred_display_base(self._ptr, base)
_preferred_display_base = property(fset=_preferred_display_base)
def _is_single_precision(self, is_single_precision):
utils._check_bool(is_single_precision)
native_bt.field_class_real_set_is_single_precision(
- self._ptr, is_single_precision)
+ self._ptr, is_single_precision
+ )
_is_single_precision = property(fset=_is_single_precision)
class _EnumerationFieldClassMapping:
def __init__(self, mapping_ptr):
base_mapping_ptr = self._as_enumeration_field_class_mapping_ptr(mapping_ptr)
- self._label = native_bt.field_class_enumeration_mapping_get_label(base_mapping_ptr)
+ self._label = native_bt.field_class_enumeration_mapping_get_label(
+ base_mapping_ptr
+ )
assert self._label is not None
ranges_ptr = self._mapping_borrow_ranges_ptr(mapping_ptr)
assert ranges_ptr is not None
class _UnsignedEnumerationFieldClassMapping(_EnumerationFieldClassMapping):
_ranges_type = bt2.integer_range_set.UnsignedIntegerRangeSet
- _as_enumeration_field_class_mapping_ptr = staticmethod(native_bt.field_class_unsigned_enumeration_mapping_as_mapping_const)
- _mapping_borrow_ranges_ptr = staticmethod(native_bt.field_class_unsigned_enumeration_mapping_borrow_ranges_const)
+ _as_enumeration_field_class_mapping_ptr = staticmethod(
+ native_bt.field_class_unsigned_enumeration_mapping_as_mapping_const
+ )
+ _mapping_borrow_ranges_ptr = staticmethod(
+ native_bt.field_class_unsigned_enumeration_mapping_borrow_ranges_const
+ )
class _SignedEnumerationFieldClassMapping(_EnumerationFieldClassMapping):
_ranges_type = bt2.integer_range_set.SignedIntegerRangeSet
- _as_enumeration_field_class_mapping_ptr = staticmethod(native_bt.field_class_signed_enumeration_mapping_as_mapping_const)
- _mapping_borrow_ranges_ptr = staticmethod(native_bt.field_class_signed_enumeration_mapping_borrow_ranges_const)
+ _as_enumeration_field_class_mapping_ptr = staticmethod(
+ native_bt.field_class_signed_enumeration_mapping_as_mapping_const
+ )
+ _mapping_borrow_ranges_ptr = staticmethod(
+ native_bt.field_class_signed_enumeration_mapping_borrow_ranges_const
+ )
class _EnumerationFieldClass(_IntegerFieldClass, collections.abc.Mapping):
raise bt2.Error("duplicate mapping label '{}'".format(label))
status = self._add_mapping(self._ptr, label, ranges._ptr)
- utils._handle_func_status(status,
- 'cannot add mapping to enumeration field class object')
+ utils._handle_func_status(
+ status, 'cannot add mapping to enumeration field class object'
+ )
def mappings_for_value(self, value):
status, labels = self._get_mapping_labels_for_value(self._ptr, value)
- utils._handle_func_status(status, 'cannot get mapping labels for value {}'.format(value))
+ utils._handle_func_status(
+ status, 'cannot get mapping labels for value {}'.format(value)
+ )
return [self[label] for label in labels]
def __iter__(self):
return self
-class _UnsignedEnumerationFieldClass(_EnumerationFieldClass, _UnsignedIntegerFieldClass):
+class _UnsignedEnumerationFieldClass(
+ _EnumerationFieldClass, _UnsignedIntegerFieldClass
+):
_NAME = 'Unsigned enumeration'
_range_set_type = bt2.integer_range_set.UnsignedIntegerRangeSet
_add_mapping = staticmethod(native_bt.field_class_unsigned_enumeration_add_mapping)
@staticmethod
def _get_mapping_by_index(enum_ptr, index):
- mapping_ptr = native_bt.field_class_unsigned_enumeration_borrow_mapping_by_index_const(enum_ptr, index)
+ mapping_ptr = native_bt.field_class_unsigned_enumeration_borrow_mapping_by_index_const(
+ enum_ptr, index
+ )
assert mapping_ptr is not None
return _UnsignedEnumerationFieldClassMapping(mapping_ptr)
@staticmethod
def _get_mapping_by_label(enum_ptr, label):
- mapping_ptr = native_bt.field_class_unsigned_enumeration_borrow_mapping_by_label_const(enum_ptr, label)
+ mapping_ptr = native_bt.field_class_unsigned_enumeration_borrow_mapping_by_label_const(
+ enum_ptr, label
+ )
if mapping_ptr is None:
return
@staticmethod
def _get_mapping_labels_for_value(enum_ptr, value):
utils._check_uint64(value)
- return native_bt.field_class_unsigned_enumeration_get_mapping_labels_for_value(enum_ptr, value)
+ return native_bt.field_class_unsigned_enumeration_get_mapping_labels_for_value(
+ enum_ptr, value
+ )
class _SignedEnumerationFieldClass(_EnumerationFieldClass, _SignedIntegerFieldClass):
@staticmethod
def _get_mapping_by_index(enum_ptr, index):
- mapping_ptr = native_bt.field_class_signed_enumeration_borrow_mapping_by_index_const(enum_ptr, index)
+ mapping_ptr = native_bt.field_class_signed_enumeration_borrow_mapping_by_index_const(
+ enum_ptr, index
+ )
assert mapping_ptr is not None
return _SignedEnumerationFieldClassMapping(mapping_ptr)
@staticmethod
def _get_mapping_by_label(enum_ptr, label):
- mapping_ptr = native_bt.field_class_signed_enumeration_borrow_mapping_by_label_const(enum_ptr, label)
+ mapping_ptr = native_bt.field_class_signed_enumeration_borrow_mapping_by_label_const(
+ enum_ptr, label
+ )
if mapping_ptr is None:
return
@staticmethod
def _get_mapping_labels_for_value(enum_ptr, value):
utils._check_int64(value)
- return native_bt.field_class_signed_enumeration_get_mapping_labels_for_value(enum_ptr, value)
+ return native_bt.field_class_signed_enumeration_get_mapping_labels_for_value(
+ enum_ptr, value
+ )
class _StringFieldClass(_FieldClass):
if name in self:
raise bt2.Error("duplicate member name '{}'".format(name))
- status = native_bt.field_class_structure_append_member(self._ptr, name, field_class._ptr)
- utils._handle_func_status(status,
- 'cannot append member to structure field class object')
+ status = native_bt.field_class_structure_append_member(
+ self._ptr, name, field_class._ptr
+ )
+ utils._handle_func_status(
+ status, 'cannot append member to structure field class object'
+ )
def __len__(self):
count = native_bt.field_class_structure_get_member_count(self._ptr)
def _create_member_from_ptr(member_ptr):
name = native_bt.field_class_structure_member_get_name(member_ptr)
assert name is not None
- fc_ptr = native_bt.field_class_structure_member_borrow_field_class_const(member_ptr)
+ fc_ptr = native_bt.field_class_structure_member_borrow_field_class_const(
+ member_ptr
+ )
assert fc_ptr is not None
fc = _create_field_class_from_ptr_and_get_ref(fc_ptr)
return _StructureFieldClassMember(name, fc)
def __getitem__(self, key):
if not isinstance(key, str):
- raise TypeError("key must be a 'str' object, got '{}'".format(key.__class__.__name__))
+ raise TypeError(
+ "key must be a 'str' object, got '{}'".format(key.__class__.__name__)
+ )
- member_ptr = native_bt.field_class_structure_borrow_member_by_name_const(self._ptr, key)
+ member_ptr = native_bt.field_class_structure_borrow_member_by_name_const(
+ self._ptr, key
+ )
if member_ptr is None:
raise KeyError(key)
def __iter__(self):
for idx in range(len(self)):
- member_ptr = native_bt.field_class_structure_borrow_member_by_index_const(self._ptr, idx)
+ member_ptr = native_bt.field_class_structure_borrow_member_by_index_const(
+ self._ptr, idx
+ )
assert member_ptr is not None
yield native_bt.field_class_structure_member_get_name(member_ptr)
if index >= len(self):
raise IndexError
- member_ptr = native_bt.field_class_structure_borrow_member_by_index_const(self._ptr, index)
+ member_ptr = native_bt.field_class_structure_borrow_member_by_index_const(
+ self._ptr, index
+ )
assert member_ptr is not None
return self._create_member_from_ptr(member_ptr)
class _VariantFieldClass(_FieldClass, collections.abc.Mapping):
_NAME = 'Variant'
- _borrow_option_by_name_ptr = staticmethod(native_bt.field_class_variant_borrow_option_by_name_const)
- _borrow_member_by_index_ptr = staticmethod(native_bt.field_class_variant_borrow_option_by_index_const)
+ _borrow_option_by_name_ptr = staticmethod(
+ native_bt.field_class_variant_borrow_option_by_name_const
+ )
+ _borrow_member_by_index_ptr = staticmethod(
+ native_bt.field_class_variant_borrow_option_by_index_const
+ )
@staticmethod
def _as_option_ptr(opt_ptr):
def __getitem__(self, key):
if not isinstance(key, str):
- raise TypeError("key must be a 'str' object, got '{}'".format(key.__class__.__name__))
+ raise TypeError(
+ "key must be a 'str' object, got '{}'".format(key.__class__.__name__)
+ )
opt_ptr = self._borrow_option_by_name_ptr(self._ptr, key)
if name in self:
raise bt2.Error("duplicate option name '{}'".format(name))
- status = native_bt.field_class_variant_without_selector_append_option(self._ptr, name, field_class._ptr)
- utils._handle_func_status(status,
- 'cannot append option to variant field class object')
+ status = native_bt.field_class_variant_without_selector_append_option(
+ self._ptr, name, field_class._ptr
+ )
+ utils._handle_func_status(
+ status, 'cannot append option to variant field class object'
+ )
def __iadd__(self, options):
for name, field_class in options:
base_opt_ptr = self._as_option_ptr(opt_ptr)
name = native_bt.field_class_variant_option_get_name(base_opt_ptr)
assert name is not None
- fc_ptr = native_bt.field_class_variant_option_borrow_field_class_const(base_opt_ptr)
+ fc_ptr = native_bt.field_class_variant_option_borrow_field_class_const(
+ base_opt_ptr
+ )
assert fc_ptr is not None
fc = _create_field_class_from_ptr_and_get_ref(fc_ptr)
range_set_ptr = self._option_borrow_ranges_ptr(opt_ptr)
@property
def selector_field_path(self):
- ptr = native_bt.field_class_variant_with_selector_borrow_selector_field_path_const(self._ptr)
+ ptr = native_bt.field_class_variant_with_selector_borrow_selector_field_path_const(
+ self._ptr
+ )
if ptr is None:
return
# TODO: check overlaps (precondition of self._append_option())
status = self._append_option(self._ptr, name, field_class._ptr, ranges._ptr)
- utils._handle_func_status(status,
- 'cannot append option to variant field class object')
+ utils._handle_func_status(
+ status, 'cannot append option to variant field class object'
+ )
def __iadd__(self, options):
for name, field_class, ranges in options:
class _VariantFieldClassWithUnsignedSelector(_VariantFieldClassWithSelector):
_NAME = 'Variant (with unsigned selector)'
- _borrow_option_by_name_ptr = staticmethod(native_bt.field_class_variant_with_unsigned_selector_borrow_option_by_name_const)
- _borrow_member_by_index_ptr = staticmethod(native_bt.field_class_variant_with_unsigned_selector_borrow_option_by_index_const)
- _as_option_ptr = staticmethod(native_bt.field_class_variant_with_unsigned_selector_option_as_option_const)
- _append_option = staticmethod(native_bt.field_class_variant_with_unsigned_selector_append_option)
- _option_borrow_ranges_ptr = staticmethod(native_bt.field_class_variant_with_unsigned_selector_option_borrow_ranges_const)
+ _borrow_option_by_name_ptr = staticmethod(
+ native_bt.field_class_variant_with_unsigned_selector_borrow_option_by_name_const
+ )
+ _borrow_member_by_index_ptr = staticmethod(
+ native_bt.field_class_variant_with_unsigned_selector_borrow_option_by_index_const
+ )
+ _as_option_ptr = staticmethod(
+ native_bt.field_class_variant_with_unsigned_selector_option_as_option_const
+ )
+ _append_option = staticmethod(
+ native_bt.field_class_variant_with_unsigned_selector_append_option
+ )
+ _option_borrow_ranges_ptr = staticmethod(
+ native_bt.field_class_variant_with_unsigned_selector_option_borrow_ranges_const
+ )
_range_set_type = bt2.integer_range_set.UnsignedIntegerRangeSet
class _VariantFieldClassWithSignedSelector(_VariantFieldClassWithSelector):
_NAME = 'Variant (with signed selector)'
- _borrow_option_by_name_ptr = staticmethod(native_bt.field_class_variant_with_signed_selector_borrow_option_by_name_const)
- _borrow_member_by_index_ptr = staticmethod(native_bt.field_class_variant_with_signed_selector_borrow_option_by_index_const)
- _as_option_ptr = staticmethod(native_bt.field_class_variant_with_signed_selector_option_as_option_const)
- _append_option = staticmethod(native_bt.field_class_variant_with_signed_selector_append_option)
- _option_borrow_ranges_ptr = staticmethod(native_bt.field_class_variant_with_signed_selector_option_borrow_ranges_const)
+ _borrow_option_by_name_ptr = staticmethod(
+ native_bt.field_class_variant_with_signed_selector_borrow_option_by_name_const
+ )
+ _borrow_member_by_index_ptr = staticmethod(
+ native_bt.field_class_variant_with_signed_selector_borrow_option_by_index_const
+ )
+ _as_option_ptr = staticmethod(
+ native_bt.field_class_variant_with_signed_selector_option_as_option_const
+ )
+ _append_option = staticmethod(
+ native_bt.field_class_variant_with_signed_selector_append_option
+ )
+ _option_borrow_ranges_ptr = staticmethod(
+ native_bt.field_class_variant_with_signed_selector_option_borrow_ranges_const
+ )
_range_set_type = bt2.integer_range_set.SignedIntegerRangeSet
class _ArrayFieldClass(_FieldClass):
@property
def element_field_class(self):
- elem_fc_ptr = native_bt.field_class_array_borrow_element_field_class_const(self._ptr)
+ elem_fc_ptr = native_bt.field_class_array_borrow_element_field_class_const(
+ self._ptr
+ )
return _create_field_class_from_ptr_and_get_ref(elem_fc_ptr)
class _DynamicArrayFieldClass(_ArrayFieldClass):
@property
def length_field_path(self):
- ptr = native_bt.field_class_dynamic_array_borrow_length_field_path_const(self._ptr)
+ ptr = native_bt.field_class_dynamic_array_borrow_length_field_path_const(
+ self._ptr
+ )
if ptr is None:
return
idx = native_bt.field_path_item_index_get_index(item_ptr)
yield _IndexFieldPathItem(idx)
elif item_type == native_bt.FIELD_PATH_ITEM_TYPE_CURRENT_ARRAY_ELEMENT:
- yield _CurrentArrayElementFieldPathItem()
+ yield _CurrentArrayElementFieldPathItem()
else:
assert False
native_bt.SCOPE_PACKET_CONTEXT: Scope.PACKET_CONTEXT,
native_bt.SCOPE_EVENT_COMMON_CONTEXT: Scope.EVENT_COMMON_CONTEXT,
native_bt.SCOPE_EVENT_SPECIFIC_CONTEXT: Scope.EVENT_SPECIFIC_CONTEXT,
- native_bt.SCOPE_EVENT_PAYLOAD: Scope.EVENT_PAYLOAD
+ native_bt.SCOPE_EVENT_PAYLOAD: Scope.EVENT_PAYLOAD,
}
import bt2
-def _graph_port_added_listener_from_native(user_listener,
- component_ptr, component_type,
- port_ptr, port_type):
- component = bt2.component._create_component_from_ptr_and_get_ref(component_ptr, component_type)
+def _graph_port_added_listener_from_native(
+ user_listener, component_ptr, component_type, port_ptr, port_type
+):
+ component = bt2.component._create_component_from_ptr_and_get_ref(
+ component_ptr, component_type
+ )
port = bt2.port._create_from_ptr_and_get_ref(port_ptr, port_type)
user_listener(component, port)
-def _graph_ports_connected_listener_from_native(user_listener,
- upstream_component_ptr, upstream_component_type,
- upstream_port_ptr,
- downstream_component_ptr, downstream_component_type,
- downstream_port_ptr):
+def _graph_ports_connected_listener_from_native(
+ user_listener,
+ upstream_component_ptr,
+ upstream_component_type,
+ upstream_port_ptr,
+ downstream_component_ptr,
+ downstream_component_type,
+ downstream_port_ptr,
+):
upstream_component = bt2.component._create_component_from_ptr_and_get_ref(
- upstream_component_ptr, upstream_component_type)
+ upstream_component_ptr, upstream_component_type
+ )
upstream_port = bt2.port._create_from_ptr_and_get_ref(
- upstream_port_ptr, native_bt.PORT_TYPE_OUTPUT)
+ upstream_port_ptr, native_bt.PORT_TYPE_OUTPUT
+ )
downstream_component = bt2.component._create_component_from_ptr_and_get_ref(
- downstream_component_ptr, downstream_component_type)
+ downstream_component_ptr, downstream_component_type
+ )
downstream_port = bt2.port._create_from_ptr_and_get_ref(
- downstream_port_ptr, native_bt.PORT_TYPE_INPUT)
- user_listener(upstream_component, upstream_port, downstream_component, downstream_port)
+ downstream_port_ptr, native_bt.PORT_TYPE_INPUT
+ )
+ user_listener(
+ upstream_component, upstream_port, downstream_component, downstream_port
+ )
class Graph(object._SharedObject):
super().__init__(ptr)
-
- def add_component(self, component_class, name, params=None,
- logging_level=bt2.logging.LoggingLevel.NONE):
+ def add_component(
+ self,
+ component_class,
+ name,
+ params=None,
+ logging_level=bt2.logging.LoggingLevel.NONE,
+ ):
if isinstance(component_class, bt2.component._GenericSourceComponentClass):
cc_ptr = component_class._ptr
add_fn = native_bt.graph_add_source_component
add_fn = native_bt.graph_add_filter_component
cc_type = native_bt.COMPONENT_CLASS_TYPE_FILTER
else:
- raise TypeError("'{}' is not a component class".format(
- component_class.__class__.__name__))
+ raise TypeError(
+ "'{}' is not a component class".format(
+ component_class.__class__.__name__
+ )
+ )
utils._check_str(name)
utils._check_log_level(logging_level)
params_ptr = params._ptr if params is not None else None
- status, comp_ptr = add_fn(self._ptr, cc_ptr, name,
- params_ptr, logging_level)
+ status, comp_ptr = add_fn(self._ptr, cc_ptr, name, params_ptr, logging_level)
utils._handle_func_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)
utils._check_type(downstream_port, bt2.port._InputPort)
- status, conn_ptr = native_bt.graph_connect_ports(self._ptr,
- upstream_port._ptr,
- downstream_port._ptr)
- utils._handle_func_status(status,
- 'cannot connect component ports within graph')
- assert(conn_ptr)
+ status, conn_ptr = native_bt.graph_connect_ports(
+ self._ptr, upstream_port._ptr, downstream_port._ptr
+ )
+ utils._handle_func_status(status, 'cannot connect component ports within graph')
+ assert conn_ptr
return bt2.connection._Connection._create_from_ptr(conn_ptr)
def add_port_added_listener(self, listener):
raise TypeError("'listener' parameter is not callable")
fn = native_bt.bt2_graph_add_port_added_listener
- listener_from_native = functools.partial(_graph_port_added_listener_from_native,
- listener)
+ listener_from_native = functools.partial(
+ _graph_port_added_listener_from_native, listener
+ )
listener_ids = fn(self._ptr, listener_from_native)
if listener_ids is None:
raise TypeError("'listener' parameter is not callable")
fn = native_bt.bt2_graph_add_ports_connected_listener
- listener_from_native = functools.partial(_graph_ports_connected_listener_from_native,
- listener)
+ listener_from_native = functools.partial(
+ _graph_ports_connected_listener_from_native, listener
+ )
listener_ids = fn(self._ptr, listener_from_native)
if listener_ids is None:
status = native_bt.graph_run(self._ptr)
try:
- utils._handle_func_status(status,
- 'graph object stopped running because of an unexpected error')
+ utils._handle_func_status(
+ status, 'graph object stopped running because of an unexpected error'
+ )
except bt2.Stop:
# done
return
@property
def is_canceled(self):
is_canceled = native_bt.graph_is_canceled(self._ptr)
- assert(is_canceled >= 0)
+ assert is_canceled >= 0
return is_canceled > 0
def create_output_port_message_iterator(self, output_port):
utils._check_type(output_port, bt2.port._OutputPort)
- msg_iter_ptr = native_bt.port_output_message_iterator_create(self._ptr, output_port._ptr)
+ msg_iter_ptr = native_bt.port_output_message_iterator_create(
+ self._ptr, output_port._ptr
+ )
if msg_iter_ptr is None:
raise bt2.CreationError('cannot create output port message iterator')
self._check_type(upper)
if lower > upper:
- raise ValueError("range's lower bound ({}) is greater than its upper bound ({})".format(lower, upper))
+ raise ValueError(
+ "range's lower bound ({}) is greater than its upper bound ({})".format(
+ lower, upper
+ )
+ )
self._lower = lower
self._upper = upper
rg = self._range_type(rg[0], rg[1])
status = self._add_range(self._ptr, rg.lower, rg.upper)
- utils._handle_func_status(status,
- 'cannot add range to range set object')
+ utils._handle_func_status(status, 'cannot add range to range set object')
def discard(self, rg):
raise NotImplementedError
class SignedIntegerRangeSet(_IntegerRangeSet):
_get_ref = staticmethod(native_bt.integer_range_set_signed_get_ref)
_put_ref = staticmethod(native_bt.integer_range_set_signed_put_ref)
- _as_range_set_ptr = staticmethod(native_bt.integer_range_set_signed_as_range_set_const)
+ _as_range_set_ptr = staticmethod(
+ native_bt.integer_range_set_signed_as_range_set_const
+ )
_create_range_set = staticmethod(native_bt.integer_range_set_signed_create)
- _borrow_range_by_index_ptr = staticmethod(native_bt.integer_range_set_signed_borrow_range_by_index_const)
+ _borrow_range_by_index_ptr = staticmethod(
+ native_bt.integer_range_set_signed_borrow_range_by_index_const
+ )
_range_get_lower = staticmethod(native_bt.integer_range_signed_get_lower)
_range_get_upper = staticmethod(native_bt.integer_range_signed_get_upper)
_add_range = staticmethod(native_bt.integer_range_set_signed_add_range)
class UnsignedIntegerRangeSet(_IntegerRangeSet):
_get_ref = staticmethod(native_bt.integer_range_set_unsigned_get_ref)
_put_ref = staticmethod(native_bt.integer_range_set_unsigned_put_ref)
- _as_range_set_ptr = staticmethod(native_bt.integer_range_set_unsigned_as_range_set_const)
+ _as_range_set_ptr = staticmethod(
+ native_bt.integer_range_set_unsigned_as_range_set_const
+ )
_create_range_set = staticmethod(native_bt.integer_range_set_unsigned_create)
- _borrow_range_by_index_ptr = staticmethod(native_bt.integer_range_set_unsigned_borrow_range_by_index_const)
+ _borrow_range_by_index_ptr = staticmethod(
+ native_bt.integer_range_set_unsigned_borrow_range_by_index_const
+ )
_range_get_lower = staticmethod(native_bt.integer_range_unsigned_get_lower)
_range_get_upper = staticmethod(native_bt.integer_range_unsigned_get_upper)
_add_range = staticmethod(native_bt.integer_range_set_unsigned_add_range)
@staticmethod
def _check_has_default_clock_class(clock_class):
if clock_class is None:
- raise bt2.NonexistentClockSnapshot('cannot get default clock snapshot: stream class has no default clock class')
+ raise bt2.NonexistentClockSnapshot(
+ 'cannot get default clock snapshot: stream class has no default clock class'
+ )
class _MessageWithDefaultClockSnapshot:
snapshot_ptr = borrow_clock_snapshot_ptr(self._ptr)
return bt2.clock_snapshot._ClockSnapshot._create_from_ptr_and_get_ref(
- snapshot_ptr, self._ptr, self._get_ref, self._put_ref)
+ snapshot_ptr, self._ptr, self._get_ref, self._put_ref
+ )
class _EventMessage(_Message, _MessageWithDefaultClockSnapshot):
- _borrow_default_clock_snapshot_ptr = staticmethod(native_bt.message_event_borrow_default_clock_snapshot_const)
+ _borrow_default_clock_snapshot_ptr = staticmethod(
+ native_bt.message_event_borrow_default_clock_snapshot_const
+ )
@property
def default_clock_snapshot(self):
event_ptr = native_bt.message_event_borrow_event(self._ptr)
assert event_ptr is not None
return bt2.event._Event._create_from_ptr_and_get_ref(
- event_ptr, self._ptr, self._get_ref, self._put_ref)
+ event_ptr, self._ptr, self._get_ref, self._put_ref
+ )
class _PacketMessage(_Message, _MessageWithDefaultClockSnapshot):
class _PacketBeginningMessage(_PacketMessage):
_borrow_packet_ptr = staticmethod(native_bt.message_packet_beginning_borrow_packet)
- _borrow_default_clock_snapshot_ptr = staticmethod(native_bt.message_packet_beginning_borrow_default_clock_snapshot_const)
+ _borrow_default_clock_snapshot_ptr = staticmethod(
+ native_bt.message_packet_beginning_borrow_default_clock_snapshot_const
+ )
class _PacketEndMessage(_PacketMessage):
_borrow_packet_ptr = staticmethod(native_bt.message_packet_end_borrow_packet)
- _borrow_default_clock_snapshot_ptr = staticmethod(native_bt.message_packet_end_borrow_default_clock_snapshot_const)
+ _borrow_default_clock_snapshot_ptr = staticmethod(
+ native_bt.message_packet_end_borrow_default_clock_snapshot_const
+ )
class _StreamMessage(_Message, _MessageWithDefaultClockSnapshot):
return bt2.clock_snapshot._UnknownClockSnapshot()
return bt2.clock_snapshot._ClockSnapshot._create_from_ptr_and_get_ref(
- snapshot_ptr, self._ptr, self._get_ref, self._put_ref)
+ snapshot_ptr, self._ptr, self._get_ref, self._put_ref
+ )
def _default_clock_snapshot(self, raw_value):
utils._check_uint64(raw_value)
class _StreamBeginningMessage(_StreamMessage):
_borrow_stream_ptr = staticmethod(native_bt.message_stream_beginning_borrow_stream)
- _borrow_default_clock_snapshot_ptr = staticmethod(native_bt.message_stream_beginning_borrow_default_clock_snapshot_const)
- _set_default_clock_snapshot = staticmethod(native_bt.message_stream_beginning_set_default_clock_snapshot)
+ _borrow_default_clock_snapshot_ptr = staticmethod(
+ native_bt.message_stream_beginning_borrow_default_clock_snapshot_const
+ )
+ _set_default_clock_snapshot = staticmethod(
+ native_bt.message_stream_beginning_set_default_clock_snapshot
+ )
class _StreamEndMessage(_StreamMessage):
_borrow_stream_ptr = staticmethod(native_bt.message_stream_end_borrow_stream)
- _borrow_default_clock_snapshot_ptr = staticmethod(native_bt.message_stream_end_borrow_default_clock_snapshot_const)
- _set_default_clock_snapshot = staticmethod(native_bt.message_stream_end_set_default_clock_snapshot)
+ _borrow_default_clock_snapshot_ptr = staticmethod(
+ native_bt.message_stream_end_borrow_default_clock_snapshot_const
+ )
+ _set_default_clock_snapshot = staticmethod(
+ native_bt.message_stream_end_set_default_clock_snapshot
+ )
class _MessageIteratorInactivityMessage(_Message, _MessageWithDefaultClockSnapshot):
- _borrow_default_clock_snapshot_ptr = staticmethod(native_bt.message_message_iterator_inactivity_borrow_default_clock_snapshot_const)
+ _borrow_default_clock_snapshot_ptr = staticmethod(
+ native_bt.message_message_iterator_inactivity_borrow_default_clock_snapshot_const
+ )
@property
def default_clock_snapshot(self):
def _check_has_default_clock_snapshots(self):
if not self._has_default_clock_snapshots:
- raise bt2.NonexistentClockSnapshot('cannot get default clock snapshot: such a message has no clock snapshots for this stream class')
+ raise bt2.NonexistentClockSnapshot(
+ 'cannot get default clock snapshot: such a message has no clock snapshots for this stream class'
+ )
@property
def beginning_default_clock_snapshot(self):
self._check_has_default_clock_snapshots()
- return self._get_default_clock_snapshot(self._borrow_beginning_clock_snapshot_ptr)
+ return self._get_default_clock_snapshot(
+ self._borrow_beginning_clock_snapshot_ptr
+ )
@property
def end_default_clock_snapshot(self):
class _DiscardedEventsMessage(_DiscardedMessage):
- _borrow_stream_ptr = staticmethod(native_bt.message_discarded_events_borrow_stream_const)
+ _borrow_stream_ptr = staticmethod(
+ native_bt.message_discarded_events_borrow_stream_const
+ )
_get_count = staticmethod(native_bt.message_discarded_events_get_count)
_set_count = staticmethod(native_bt.message_discarded_events_set_count)
- _borrow_beginning_clock_snapshot_ptr = staticmethod(native_bt.message_discarded_events_borrow_beginning_default_clock_snapshot_const)
- _borrow_end_clock_snapshot_ptr = staticmethod(native_bt.message_discarded_events_borrow_end_default_clock_snapshot_const)
+ _borrow_beginning_clock_snapshot_ptr = staticmethod(
+ native_bt.message_discarded_events_borrow_beginning_default_clock_snapshot_const
+ )
+ _borrow_end_clock_snapshot_ptr = staticmethod(
+ native_bt.message_discarded_events_borrow_end_default_clock_snapshot_const
+ )
@property
def _has_default_clock_snapshots(self):
class _DiscardedPacketsMessage(_DiscardedMessage):
- _borrow_stream_ptr = staticmethod(native_bt.message_discarded_packets_borrow_stream_const)
+ _borrow_stream_ptr = staticmethod(
+ native_bt.message_discarded_packets_borrow_stream_const
+ )
_get_count = staticmethod(native_bt.message_discarded_packets_get_count)
_set_count = staticmethod(native_bt.message_discarded_packets_set_count)
- _borrow_beginning_clock_snapshot_ptr = staticmethod(native_bt.message_discarded_packets_borrow_beginning_default_clock_snapshot_const)
- _borrow_end_clock_snapshot_ptr = staticmethod(native_bt.message_discarded_packets_borrow_end_default_clock_snapshot_const)
+ _borrow_beginning_clock_snapshot_ptr = staticmethod(
+ native_bt.message_discarded_packets_borrow_beginning_default_clock_snapshot_const
+ )
+ _borrow_end_clock_snapshot_ptr = staticmethod(
+ native_bt.message_discarded_packets_borrow_end_default_clock_snapshot_const
+ )
@property
def _has_default_clock_snapshots(self):
class _GenericMessageIterator(object._SharedObject, _MessageIterator):
def __init__(self, ptr):
- self._current_msgs = []
- self._at = 0
- super().__init__(ptr)
+ self._current_msgs = []
+ self._at = 0
+ super().__init__(ptr)
def __next__(self):
if len(self._current_msgs) == self._at:
status, msgs = self._get_msg_range(self._ptr)
- utils._handle_func_status(status,
- 'unexpected error: cannot advance the message iterator')
+ utils._handle_func_status(
+ status, 'unexpected error: cannot advance the message iterator'
+ )
self._current_msgs = msgs
self._at = 0
self._at = 0
status = self._seek_beginning(self._ptr)
- utils._handle_func_status(status,
- 'cannot seek message iterator beginning')
+ utils._handle_func_status(status, 'cannot seek message iterator beginning')
# This is created when a component wants to iterate on one of its input ports.
class _UserComponentInputPortMessageIterator(_GenericMessageIterator):
_get_msg_range = staticmethod(native_bt.bt2_self_component_port_input_get_msg_range)
- _get_ref = staticmethod(native_bt.self_component_port_input_message_iterator_get_ref)
- _put_ref = staticmethod(native_bt.self_component_port_input_message_iterator_put_ref)
- _can_seek_beginning = staticmethod(native_bt.self_component_port_input_message_iterator_can_seek_beginning)
- _seek_beginning = staticmethod(native_bt.self_component_port_input_message_iterator_seek_beginning)
+ _get_ref = staticmethod(
+ native_bt.self_component_port_input_message_iterator_get_ref
+ )
+ _put_ref = staticmethod(
+ native_bt.self_component_port_input_message_iterator_put_ref
+ )
+ _can_seek_beginning = staticmethod(
+ native_bt.self_component_port_input_message_iterator_can_seek_beginning
+ )
+ _seek_beginning = staticmethod(
+ native_bt.self_component_port_input_message_iterator_seek_beginning
+ )
# This is created when the user wants to iterate on a component's output port,
_get_msg_range = staticmethod(native_bt.bt2_port_output_get_msg_range)
_get_ref = staticmethod(native_bt.port_output_message_iterator_get_ref)
_put_ref = staticmethod(native_bt.port_output_message_iterator_put_ref)
- _can_seek_beginning = staticmethod(native_bt.port_output_message_iterator_can_seek_beginning)
- _seek_beginning = staticmethod(native_bt.port_output_message_iterator_seek_beginning)
+ _can_seek_beginning = staticmethod(
+ native_bt.port_output_message_iterator_can_seek_beginning
+ )
+ _seek_beginning = staticmethod(
+ native_bt.port_output_message_iterator_seek_beginning
+ )
# This is extended by the user to implement component classes in Python. It
def _bt_init_from_native(self, self_output_port_ptr):
self_output_port = bt2.port._create_self_from_ptr_and_get_ref(
- self_output_port_ptr, native_bt.PORT_TYPE_OUTPUT)
+ self_output_port_ptr, native_bt.PORT_TYPE_OUTPUT
+ )
self.__init__(self_output_port)
def __init__(self, output_port):
def _bt_seek_beginning_from_native(self):
self._seek_beginning()
- def _create_event_message(self, event_class, parent=None,
- default_clock_snapshot=None):
+ def _create_event_message(
+ self, event_class, parent=None, default_clock_snapshot=None
+ ):
utils._check_type(event_class, bt2.event_class._EventClass)
if event_class.stream_class.supports_packets:
if default_clock_snapshot is not None:
if event_class.stream_class.default_clock_class is None:
- raise ValueError('event messages in this stream must not have a default clock snapshot')
+ raise ValueError(
+ 'event messages in this stream must not have a default clock snapshot'
+ )
utils._check_uint64(default_clock_snapshot)
if event_class.stream_class.supports_packets:
ptr = native_bt.message_event_create_with_packet_and_default_clock_snapshot(
- self._bt_ptr, event_class._ptr, parent._ptr, default_clock_snapshot)
+ self._bt_ptr, event_class._ptr, parent._ptr, default_clock_snapshot
+ )
else:
ptr = native_bt.message_event_create_with_default_clock_snapshot(
- self._bt_ptr, event_class._ptr, parent._ptr, default_clock_snapshot)
+ self._bt_ptr, event_class._ptr, parent._ptr, default_clock_snapshot
+ )
else:
if event_class.stream_class.default_clock_class is not None:
- raise ValueError('event messages in this stream must have a default clock snapshot')
+ raise ValueError(
+ 'event messages in this stream must have a default clock snapshot'
+ )
if event_class.stream_class.supports_packets:
ptr = native_bt.message_event_create_with_packet(
- self._bt_ptr, event_class._ptr, parent._ptr)
+ self._bt_ptr, event_class._ptr, parent._ptr
+ )
else:
ptr = native_bt.message_event_create(
- self._bt_ptr, event_class._ptr, parent._ptr)
+ self._bt_ptr, event_class._ptr, parent._ptr
+ )
if ptr is None:
raise bt2.CreationError('cannot create event message object')
def _create_message_iterator_inactivity_message(self, clock_class, clock_snapshot):
utils._check_type(clock_class, bt2.clock_class._ClockClass)
ptr = native_bt.message_message_iterator_inactivity_create(
- self._bt_ptr, clock_class._ptr, clock_snapshot)
+ self._bt_ptr, clock_class._ptr, clock_snapshot
+ )
if ptr is None:
raise bt2.CreationError('cannot create inactivity message object')
if packet.stream.cls.packets_have_beginning_default_clock_snapshot:
if default_clock_snapshot is None:
- raise ValueError("packet beginning messages in this stream must have a default clock snapshot")
+ raise ValueError(
+ "packet beginning messages in this stream must have a default clock snapshot"
+ )
utils._check_uint64(default_clock_snapshot)
ptr = native_bt.message_packet_beginning_create_with_default_clock_snapshot(
- self._bt_ptr, packet._ptr, default_clock_snapshot)
+ self._bt_ptr, packet._ptr, default_clock_snapshot
+ )
else:
if default_clock_snapshot is not None:
- raise ValueError("packet beginning messages in this stream must not have a default clock snapshot")
+ raise ValueError(
+ "packet beginning messages in this stream must not have a default clock snapshot"
+ )
ptr = native_bt.message_packet_beginning_create(self._bt_ptr, packet._ptr)
if packet.stream.cls.packets_have_end_default_clock_snapshot:
if default_clock_snapshot is None:
- raise ValueError("packet end messages in this stream must have a default clock snapshot")
+ raise ValueError(
+ "packet end messages in this stream must have a default clock snapshot"
+ )
utils._check_uint64(default_clock_snapshot)
ptr = native_bt.message_packet_end_create_with_default_clock_snapshot(
- self._bt_ptr, packet._ptr, default_clock_snapshot)
+ self._bt_ptr, packet._ptr, default_clock_snapshot
+ )
else:
if default_clock_snapshot is not None:
- raise ValueError("packet end messages in this stream must not have a default clock snapshot")
+ raise ValueError(
+ "packet end messages in this stream must not have a default clock snapshot"
+ )
ptr = native_bt.message_packet_end_create(self._bt_ptr, packet._ptr)
return bt2.message._PacketEndMessage(ptr)
- def _create_discarded_events_message(self, stream, count=None,
- beg_clock_snapshot=None,
- end_clock_snapshot=None):
+ def _create_discarded_events_message(
+ self, stream, count=None, beg_clock_snapshot=None, end_clock_snapshot=None
+ ):
utils._check_type(stream, bt2.stream._Stream)
if not stream.cls.supports_discarded_events:
if stream.cls.discarded_events_have_default_clock_snapshots:
if beg_clock_snapshot is None or end_clock_snapshot is None:
- raise ValueError('discarded events have default clock snapshots for this stream class')
+ raise ValueError(
+ 'discarded events have default clock snapshots for this stream class'
+ )
utils._check_uint64(beg_clock_snapshot)
utils._check_uint64(end_clock_snapshot)
ptr = native_bt.message_discarded_events_create_with_default_clock_snapshots(
- self._bt_ptr, stream._ptr, beg_clock_snapshot, end_clock_snapshot)
+ self._bt_ptr, stream._ptr, beg_clock_snapshot, end_clock_snapshot
+ )
else:
if beg_clock_snapshot is not None or end_clock_snapshot is not None:
- raise ValueError('discarded events have no default clock snapshots for this stream class')
+ raise ValueError(
+ 'discarded events have no default clock snapshots for this stream class'
+ )
- ptr = native_bt.message_discarded_events_create(
- self._bt_ptr, stream._ptr)
+ ptr = native_bt.message_discarded_events_create(self._bt_ptr, stream._ptr)
if ptr is None:
raise bt2.CreationError('cannot discarded events message object')
return msg
- def _create_discarded_packets_message(self, stream, count=None, beg_clock_snapshot=None, end_clock_snapshot=None):
+ def _create_discarded_packets_message(
+ self, stream, count=None, beg_clock_snapshot=None, end_clock_snapshot=None
+ ):
utils._check_type(stream, bt2.stream._Stream)
if not stream.cls.supports_discarded_packets:
if stream.cls.discarded_packets_have_default_clock_snapshots:
if beg_clock_snapshot is None or end_clock_snapshot is None:
- raise ValueError('discarded packets have default clock snapshots for this stream class')
+ raise ValueError(
+ 'discarded packets have default clock snapshots for this stream class'
+ )
utils._check_uint64(beg_clock_snapshot)
utils._check_uint64(end_clock_snapshot)
ptr = native_bt.message_discarded_packets_create_with_default_clock_snapshots(
- self._bt_ptr, stream._ptr, beg_clock_snapshot, end_clock_snapshot)
+ self._bt_ptr, stream._ptr, beg_clock_snapshot, end_clock_snapshot
+ )
else:
if beg_clock_snapshot is not None or end_clock_snapshot is not None:
- raise ValueError('discarded packets have no default clock snapshots for this stream class')
+ raise ValueError(
+ 'discarded packets have no default clock snapshots for this stream class'
+ )
- ptr = native_bt.message_discarded_packets_create(
- self._bt_ptr, stream._ptr)
+ ptr = native_bt.message_discarded_packets_create(self._bt_ptr, stream._ptr)
if ptr is None:
raise bt2.CreationError('cannot discarded packets message object')
msg._count = count
return msg
-
return int(self._ptr)
def __repr__(self):
- return '<{}.{} object @ {}>'.format(self.__class__.__module__,
- self.__class__.__name__,
- hex(self.addr))
+ return '<{}.{} object @ {}>'.format(
+ self.__class__.__module__, self.__class__.__name__, hex(self.addr)
+ )
def __copy__(self):
raise NotImplementedError
# of unique objects, we make it so acquiring a reference on a unique object
# acquires a reference on its owner.
+
class _UniqueObject(_BaseObject):
# Create a _UniqueObject.
# - owner_put_ref: Callback to put a reference on the owner.
@classmethod
- def _create_from_ptr_and_get_ref(cls, ptr, owner_ptr,
- owner_get_ref, owner_put_ref):
+ def _create_from_ptr_and_get_ref(cls, ptr, owner_ptr, owner_get_ref, owner_put_ref):
obj = cls.__new__(cls)
obj._ptr = ptr
if field_ptr is None:
return
- return bt2.field._create_field_from_ptr(field_ptr, self._ptr,
- self._get_ref,
- self._put_ref)
+ return bt2.field._create_field_from_ptr(
+ field_ptr, self._ptr, self._get_ref, self._put_ref
+ )
plugin_set_ptr = None
if os.path.isfile(path):
- status, plugin_set_ptr = native_bt.bt2_plugin_find_all_from_file(path, fail_on_load_error)
+ status, plugin_set_ptr = native_bt.bt2_plugin_find_all_from_file(
+ path, fail_on_load_error
+ )
elif os.path.isdir(path):
- status, plugin_set_ptr = native_bt.bt2_plugin_find_all_from_dir(path, int(recurse), int(fail_on_load_error))
+ status, plugin_set_ptr = native_bt.bt2_plugin_find_all_from_dir(
+ path, int(recurse), int(fail_on_load_error)
+ )
else:
raise bt2.Error("invalid path: '{}'".format(path))
def __len__(self):
count = native_bt.plugin_set_get_plugin_count(self._ptr)
- assert(count >= 0)
+ assert count >= 0
return count
def __getitem__(self, index):
if self._at == total:
raise StopIteration
- comp_cls_ptr = self._plugin_comp_cls._borrow_component_class_by_index(plugin_ptr, self._at)
+ comp_cls_ptr = self._plugin_comp_cls._borrow_component_class_by_index(
+ plugin_ptr, self._at
+ )
assert comp_cls_ptr is not None
self._at += 1
comp_cls_type = self._plugin_comp_cls._comp_cls_type
- comp_cls_pycls = bt2.component._COMP_CLS_TYPE_TO_GENERIC_COMP_CLS_PYCLS[comp_cls_type]
+ comp_cls_pycls = bt2.component._COMP_CLS_TYPE_TO_GENERIC_COMP_CLS_PYCLS[
+ comp_cls_type
+ ]
comp_cls_ptr = comp_cls_pycls._bt_as_component_class_ptr(comp_cls_ptr)
name = native_bt.component_class_get_name(comp_cls_ptr)
assert name is not None
if cc_ptr is None:
raise KeyError(key)
- return bt2.component._create_component_class_from_ptr_and_get_ref(cc_ptr, self._comp_cls_type)
+ return bt2.component._create_component_class_from_ptr_and_get_ref(
+ cc_ptr, self._comp_cls_type
+ )
def __len__(self):
return self._component_class_count(self._plugin._ptr)
class _PluginSourceComponentClasses(_PluginComponentClasses):
- _component_class_count = staticmethod(native_bt.plugin_get_source_component_class_count)
- _borrow_component_class_by_name = staticmethod(native_bt.plugin_borrow_source_component_class_by_name_const)
- _borrow_component_class_by_index = staticmethod(native_bt.plugin_borrow_source_component_class_by_index_const)
+ _component_class_count = staticmethod(
+ native_bt.plugin_get_source_component_class_count
+ )
+ _borrow_component_class_by_name = staticmethod(
+ native_bt.plugin_borrow_source_component_class_by_name_const
+ )
+ _borrow_component_class_by_index = staticmethod(
+ native_bt.plugin_borrow_source_component_class_by_index_const
+ )
_comp_cls_type = native_bt.COMPONENT_CLASS_TYPE_SOURCE
class _PluginFilterComponentClasses(_PluginComponentClasses):
- _component_class_count = staticmethod(native_bt.plugin_get_filter_component_class_count)
- _borrow_component_class_by_name = staticmethod(native_bt.plugin_borrow_filter_component_class_by_name_const)
- _borrow_component_class_by_index = staticmethod(native_bt.plugin_borrow_filter_component_class_by_index_const)
+ _component_class_count = staticmethod(
+ native_bt.plugin_get_filter_component_class_count
+ )
+ _borrow_component_class_by_name = staticmethod(
+ native_bt.plugin_borrow_filter_component_class_by_name_const
+ )
+ _borrow_component_class_by_index = staticmethod(
+ native_bt.plugin_borrow_filter_component_class_by_index_const
+ )
_comp_cls_type = native_bt.COMPONENT_CLASS_TYPE_FILTER
class _PluginSinkComponentClasses(_PluginComponentClasses):
- _component_class_count = staticmethod(native_bt.plugin_get_sink_component_class_count)
- _borrow_component_class_by_name = staticmethod(native_bt.plugin_borrow_sink_component_class_by_name_const)
- _borrow_component_class_by_index = staticmethod(native_bt.plugin_borrow_sink_component_class_by_index_const)
+ _component_class_count = staticmethod(
+ native_bt.plugin_get_sink_component_class_count
+ )
+ _borrow_component_class_by_name = staticmethod(
+ native_bt.plugin_borrow_sink_component_class_by_name_const
+ )
+ _borrow_component_class_by_index = staticmethod(
+ native_bt.plugin_borrow_sink_component_class_by_index_const
+ )
_comp_cls_type = native_bt.COMPONENT_CLASS_TYPE_SINK
@property
def name(self):
name = native_bt.plugin_get_name(self._ptr)
- assert(name is not None)
+ assert name is not None
return name
@property
class _UserComponentInputPort(_UserComponentPort, _InputPort):
- _as_self_port_ptr = staticmethod(native_bt.self_component_port_input_as_self_component_port)
+ _as_self_port_ptr = staticmethod(
+ native_bt.self_component_port_input_as_self_component_port
+ )
def create_message_iterator(self):
- msg_iter_ptr = native_bt.self_component_port_input_message_iterator_create(self._ptr)
+ msg_iter_ptr = native_bt.self_component_port_input_message_iterator_create(
+ self._ptr
+ )
if msg_iter_ptr is None:
raise bt2.CreationError('cannot create message iterator object')
class _UserComponentOutputPort(_UserComponentPort, _OutputPort):
- _as_self_port_ptr = staticmethod(native_bt.self_component_port_output_as_self_component_port)
+ _as_self_port_ptr = staticmethod(
+ native_bt.self_component_port_output_as_self_component_port
+ )
_PORT_TYPE_TO_PYCLS = {
return component_class
-def register_plugin(module_name, name, description=None, author=None,
- license=None, version=None):
+def register_plugin(
+ module_name, name, description=None, author=None, license=None, version=None
+):
import sys
if module_name not in sys.modules:
- raise RuntimeError("cannot find module '{}' in loaded modules".format(module_name))
+ raise RuntimeError(
+ "cannot find module '{}' in loaded modules".format(module_name)
+ )
utils._check_str(name)
if version is not None:
if not _validate_version(version):
- raise ValueError('wrong version: expecting a tuple: (major, minor, patch) or (major, minor, patch, extra)')
+ raise ValueError(
+ 'wrong version: expecting a tuple: (major, minor, patch) or (major, minor, patch, extra)'
+ )
- sys.modules[module_name]._bt_plugin_info = _PluginInfo(name, description,
- author, license,
- version)
+ sys.modules[module_name]._bt_plugin_info = _PluginInfo(
+ name, description, author, license, version
+ )
def _validate_version(version):
def cancel(self):
status = native_bt.query_executor_cancel(self._ptr)
- utils._handle_func_status(status,
- 'cannot cancel query executor object')
+ utils._handle_func_status(status, 'cannot cancel query executor object')
@property
def is_canceled(self):
is_canceled = native_bt.query_executor_is_canceled(self._ptr)
- assert(is_canceled >= 0)
+ assert is_canceled >= 0
return is_canceled > 0
- def query(self, component_class, object, params=None,
- logging_level=bt2.logging.LoggingLevel.NONE):
+ def query(
+ self,
+ component_class,
+ object,
+ params=None,
+ logging_level=bt2.logging.LoggingLevel.NONE,
+ ):
if self.is_canceled:
raise bt2.Canceled
utils._check_log_level(logging_level)
cc_ptr = component_class._bt_component_class_ptr()
- status, result_ptr = native_bt.query_executor_query(self._ptr, cc_ptr,
- object, params_ptr,
- logging_level)
+ status, result_ptr = native_bt.query_executor_query(
+ self._ptr, cc_ptr, object, params_ptr, logging_level
+ )
utils._handle_func_status(status, 'cannot query component class')
- assert(result_ptr)
+ assert result_ptr
return bt2.value._create_from_ptr(result_ptr)
def cls(self):
stream_class_ptr = native_bt.stream_borrow_class(self._ptr)
assert stream_class_ptr is not None
- return bt2.stream_class._StreamClass._create_from_ptr_and_get_ref(stream_class_ptr)
+ return bt2.stream_class._StreamClass._create_from_ptr_and_get_ref(
+ stream_class_ptr
+ )
@property
def name(self):
def create_packet(self):
if not self.cls.supports_packets:
- raise bt2.Error('cannot create packet: stream class does not support packets')
+ raise bt2.Error(
+ 'cannot create packet: stream class does not support packets'
+ )
packet_ptr = native_bt.packet_create(self._ptr)
def __iter__(self):
for idx in range(len(self)):
- ec_ptr = native_bt.stream_class_borrow_event_class_by_index_const(self._ptr, idx)
+ ec_ptr = native_bt.stream_class_borrow_event_class_by_index_const(
+ self._ptr, idx
+ )
assert ec_ptr is not None
id = native_bt.event_class_get_id(ec_ptr)
yield id
- def create_event_class(self, id=None, name=None, log_level=None, emf_uri=None,
- specific_context_field_class=None,
- payload_field_class=None):
+ def create_event_class(
+ self,
+ id=None,
+ name=None,
+ log_level=None,
+ emf_uri=None,
+ specific_context_field_class=None,
+ payload_field_class=None,
+ ):
if self.assigns_automatic_event_class_id:
if id is not None:
- raise ValueError('id provided, but stream class assigns automatic event class ids')
+ raise ValueError(
+ 'id provided, but stream class assigns automatic event class ids'
+ )
ec_ptr = native_bt.event_class_create(self._ptr)
else:
if id is None:
- raise ValueError('id not provided, but stream class does not assign automatic event class ids')
+ raise ValueError(
+ 'id not provided, but stream class does not assign automatic event class ids'
+ )
utils._check_uint64(id)
ec_ptr = native_bt.event_class_create_with_id(self._ptr, id)
def _name(self, name):
utils._check_str(name)
status = native_bt.stream_class_set_name(self._ptr, name)
- utils._handle_func_status(status,
- "cannot set stream class object's name")
+ utils._handle_func_status(status, "cannot set stream class object's name")
_name = property(fset=_name)
def _assigns_automatic_event_class_id(self, auto_id):
utils._check_bool(auto_id)
- return native_bt.stream_class_set_assigns_automatic_event_class_id(self._ptr, auto_id)
+ return native_bt.stream_class_set_assigns_automatic_event_class_id(
+ self._ptr, auto_id
+ )
_assigns_automatic_event_class_id = property(fset=_assigns_automatic_event_class_id)
def _assigns_automatic_stream_id(self, auto_id):
utils._check_bool(auto_id)
- return native_bt.stream_class_set_assigns_automatic_stream_id(self._ptr, auto_id)
+ return native_bt.stream_class_set_assigns_automatic_stream_id(
+ self._ptr, auto_id
+ )
_assigns_automatic_stream_id = property(fset=_assigns_automatic_stream_id)
@property
def packets_have_beginning_default_clock_snapshot(self):
- return native_bt.stream_class_packets_have_beginning_default_clock_snapshot(self._ptr)
+ return native_bt.stream_class_packets_have_beginning_default_clock_snapshot(
+ self._ptr
+ )
@property
def packets_have_end_default_clock_snapshot(self):
utils._check_bool(with_end_cs)
if not supports and (with_begin_cs or with_end_cs):
- raise ValueError('cannot not support packets, but have default clock snapshots')
+ raise ValueError(
+ 'cannot not support packets, but have default clock snapshots'
+ )
if not supports and self.packet_context_field_class is not None:
raise ValueError('stream class already has a packet context field class')
- native_bt.stream_class_set_supports_packets(self._ptr, supports, with_begin_cs, with_end_cs)
+ native_bt.stream_class_set_supports_packets(
+ self._ptr, supports, with_begin_cs, with_end_cs
+ )
@property
def supports_discarded_events(self):
utils._check_bool(with_cs)
if not supports and with_cs:
- raise ValueError('cannot not support discarded events, but have default clock snapshots')
+ raise ValueError(
+ 'cannot not support discarded events, but have default clock snapshots'
+ )
- native_bt.stream_class_set_supports_discarded_events(self._ptr, supports, with_cs)
+ native_bt.stream_class_set_supports_discarded_events(
+ self._ptr, supports, with_cs
+ )
@property
def discarded_events_have_default_clock_snapshots(self):
- return native_bt.stream_class_discarded_events_have_default_clock_snapshots(self._ptr)
+ return native_bt.stream_class_discarded_events_have_default_clock_snapshots(
+ self._ptr
+ )
@property
def supports_discarded_packets(self):
utils._check_bool(with_cs)
if supports and not self.supports_packets:
- raise ValueError('cannot support discarded packets, but not support packets')
+ raise ValueError(
+ 'cannot support discarded packets, but not support packets'
+ )
if not supports and with_cs:
- raise ValueError('cannot not support discarded packets, but have default clock snapshots')
+ raise ValueError(
+ 'cannot not support discarded packets, but have default clock snapshots'
+ )
- native_bt.stream_class_set_supports_discarded_packets(self._ptr, supports, with_cs)
+ native_bt.stream_class_set_supports_discarded_packets(
+ self._ptr, supports, with_cs
+ )
@property
def discarded_packets_have_default_clock_snapshots(self):
- return native_bt.stream_class_discarded_packets_have_default_clock_snapshots(self._ptr)
+ return native_bt.stream_class_discarded_packets_have_default_clock_snapshots(
+ self._ptr
+ )
@property
def id(self):
@property
def packet_context_field_class(self):
- fc_ptr = native_bt.stream_class_borrow_packet_context_field_class_const(self._ptr)
+ fc_ptr = native_bt.stream_class_borrow_packet_context_field_class_const(
+ self._ptr
+ )
if fc_ptr is None:
return
def _packet_context_field_class(self, packet_context_field_class):
if packet_context_field_class is not None:
- utils._check_type(packet_context_field_class,
- bt2.field_class._StructureFieldClass)
+ utils._check_type(
+ packet_context_field_class, bt2.field_class._StructureFieldClass
+ )
if not self.supports_packets:
raise ValueError('stream class does not support packets')
- status = native_bt.stream_class_set_packet_context_field_class(self._ptr,
- packet_context_field_class._ptr)
- utils._handle_func_status(status,
- "cannot set stream class' packet context field class")
+ status = native_bt.stream_class_set_packet_context_field_class(
+ self._ptr, packet_context_field_class._ptr
+ )
+ utils._handle_func_status(
+ status, "cannot set stream class' packet context field class"
+ )
_packet_context_field_class = property(fset=_packet_context_field_class)
@property
def event_common_context_field_class(self):
- fc_ptr = native_bt.stream_class_borrow_event_common_context_field_class_const(self._ptr)
+ fc_ptr = native_bt.stream_class_borrow_event_common_context_field_class_const(
+ self._ptr
+ )
if fc_ptr is None:
return
def _event_common_context_field_class(self, event_common_context_field_class):
if event_common_context_field_class is not None:
- utils._check_type(event_common_context_field_class,
- bt2.field_class._StructureFieldClass)
+ utils._check_type(
+ event_common_context_field_class, bt2.field_class._StructureFieldClass
+ )
set_context_fn = native_bt.stream_class_set_event_common_context_field_class
status = set_context_fn(self._ptr, event_common_context_field_class._ptr)
- utils._handle_func_status(status,
- "cannot set stream class' event context field type")
+ utils._handle_func_status(
+ status, "cannot set stream class' event context field type"
+ )
_event_common_context_field_class = property(fset=_event_common_context_field_class)
def _default_clock_class(self, clock_class):
utils._check_type(clock_class, bt2.clock_class._ClockClass)
- native_bt.stream_class_set_default_clock_class(
- self._ptr, clock_class._ptr)
+ native_bt.stream_class_set_default_clock_class(self._ptr, clock_class._ptr)
_default_clock_class = property(fset=_default_clock_class)
raise TypeError('expected str or int, got {}'.format(type(value)))
status = set_env_entry_fn(self._trace._ptr, key, value)
- utils._handle_func_status(status,
- "cannot set trace object's environment entry")
+ utils._handle_func_status(status, "cannot set trace object's environment entry")
def __delitem__(self, key):
raise NotImplementedError
def _name(self, name):
utils._check_str(name)
status = native_bt.trace_set_name(self._ptr, name)
- utils._handle_func_status(status,
- "cannot set trace class object's name")
+ utils._handle_func_status(status, "cannot set trace class object's name")
_name = property(fset=_name)
if stream_class.assigns_automatic_stream_id:
if id is not None:
- raise ValueError("id provided, but stream class assigns automatic stream ids")
+ raise ValueError(
+ "id provided, but stream class assigns automatic stream ids"
+ )
stream_ptr = native_bt.stream_create(stream_class._ptr, self._ptr)
else:
if id is None:
- raise ValueError("id not provided, but stream class does not assign automatic stream ids")
+ raise ValueError(
+ "id not provided, but stream class does not assign automatic stream ids"
+ )
utils._check_uint64(id)
- stream_ptr = native_bt.stream_create_with_id(stream_class._ptr, self._ptr, id)
+ stream_ptr = native_bt.stream_create_with_id(
+ stream_class._ptr, self._ptr, id
+ )
if stream_ptr is None:
raise bt2.CreationError('cannot create stream object')
raise TypeError("'listener' parameter is not callable")
fn = native_bt.bt2_trace_add_destruction_listener
- listener_from_native = functools.partial(_trace_destruction_listener_from_native,
- listener)
+ listener_from_native = functools.partial(
+ _trace_destruction_listener_from_native, listener
+ )
listener_id = fn(self._ptr, listener_from_native)
if listener_id is None:
if self._at == len(self._trace_class):
raise StopIteration
- borrow_stream_class_fn = native_bt.trace_class_borrow_stream_class_by_index_const
+ borrow_stream_class_fn = (
+ native_bt.trace_class_borrow_stream_class_by_index_const
+ )
sc_ptr = borrow_stream_class_fn(self._trace_class._ptr, self._at)
assert sc_ptr
id = native_bt.stream_class_get_id(sc_ptr)
def _trace_class_destruction_listener_from_native(user_listener, trace_class_ptr):
- trace_class = bt2.trace_class._TraceClass._create_from_ptr_and_get_ref(trace_class_ptr)
+ trace_class = bt2.trace_class._TraceClass._create_from_ptr_and_get_ref(
+ trace_class_ptr
+ )
user_listener(trace_class)
def __iter__(self):
for idx in range(len(self)):
- sc_ptr = native_bt.trace_class_borrow_stream_class_by_index_const(self._ptr, idx)
+ sc_ptr = native_bt.trace_class_borrow_stream_class_by_index_const(
+ self._ptr, idx
+ )
assert sc_ptr is not None
id = native_bt.stream_class_get_id(sc_ptr)
yield id
- def create_stream_class(self, id=None,
- name=None,
- packet_context_field_class=None,
- event_common_context_field_class=None,
- default_clock_class=None,
- assigns_automatic_event_class_id=True,
- assigns_automatic_stream_id=True,
- supports_packets=False,
- packets_have_beginning_default_clock_snapshot=False,
- packets_have_end_default_clock_snapshot=False,
- supports_discarded_events=False,
- discarded_events_have_default_clock_snapshots=False,
- supports_discarded_packets=False,
- discarded_packets_have_default_clock_snapshots=False):
+ def create_stream_class(
+ self,
+ id=None,
+ name=None,
+ packet_context_field_class=None,
+ event_common_context_field_class=None,
+ default_clock_class=None,
+ assigns_automatic_event_class_id=True,
+ assigns_automatic_stream_id=True,
+ supports_packets=False,
+ packets_have_beginning_default_clock_snapshot=False,
+ packets_have_end_default_clock_snapshot=False,
+ supports_discarded_events=False,
+ discarded_events_have_default_clock_snapshots=False,
+ supports_discarded_packets=False,
+ discarded_packets_have_default_clock_snapshots=False,
+ ):
if self.assigns_automatic_stream_class_id:
if id is not None:
- raise ValueError('id provided, but trace class assigns automatic stream class ids')
+ raise ValueError(
+ 'id provided, but trace class assigns automatic stream class ids'
+ )
sc_ptr = native_bt.stream_class_create(self._ptr)
else:
if id is None:
- raise ValueError('id not provided, but trace class does not assign automatic stream class ids')
+ raise ValueError(
+ 'id not provided, but trace class does not assign automatic stream class ids'
+ )
utils._check_uint64(id)
sc_ptr = native_bt.stream_class_create_with_id(self._ptr, id)
# `packets_have_beginning_default_clock_snapshot` or
# `packets_have_end_default_clock_snapshot` is true, then this
# stream class needs a default clock class already.
- sc._set_supports_packets(supports_packets,
- packets_have_beginning_default_clock_snapshot,
- packets_have_end_default_clock_snapshot)
+ sc._set_supports_packets(
+ supports_packets,
+ packets_have_beginning_default_clock_snapshot,
+ packets_have_end_default_clock_snapshot,
+ )
# call after sc._set_supports_packets() because, if
# `packet_context_field_class` is not `None`, then this stream
sc._assigns_automatic_event_class_id = assigns_automatic_event_class_id
sc._assigns_automatic_stream_id = assigns_automatic_stream_id
- sc._set_supports_discarded_events(supports_discarded_events,
- discarded_events_have_default_clock_snapshots)
- sc._set_supports_discarded_packets(supports_discarded_packets,
- discarded_packets_have_default_clock_snapshots)
+ sc._set_supports_discarded_events(
+ supports_discarded_events, discarded_events_have_default_clock_snapshots
+ )
+ sc._set_supports_discarded_packets(
+ supports_discarded_packets, discarded_packets_have_default_clock_snapshots
+ )
return sc
@property
def _assigns_automatic_stream_class_id(self, auto_id):
utils._check_bool(auto_id)
- return native_bt.trace_class_set_assigns_automatic_stream_class_id(self._ptr, auto_id)
+ return native_bt.trace_class_set_assigns_automatic_stream_class_id(
+ self._ptr, auto_id
+ )
- _assigns_automatic_stream_class_id = property(fset=_assigns_automatic_stream_class_id)
+ _assigns_automatic_stream_class_id = property(
+ fset=_assigns_automatic_stream_class_id
+ )
# Field class creation methods.
def _check_create_status(self, ptr, type_name):
if ptr is None:
- raise bt2.CreationError(
- 'cannot create {} field class'.format(type_name))
+ raise bt2.CreationError('cannot create {} field class'.format(type_name))
- def _create_integer_field_class(self, create_func, py_cls, type_name, field_value_range, preferred_display_base):
+ def _create_integer_field_class(
+ self, create_func, py_cls, type_name, field_value_range, preferred_display_base
+ ):
field_class_ptr = create_func(self._ptr)
self._check_create_status(field_class_ptr, type_name)
return field_class
- def create_signed_integer_field_class(self, field_value_range=None, preferred_display_base=None):
- return self._create_integer_field_class(native_bt.field_class_signed_integer_create,
- bt2.field_class._SignedIntegerFieldClass,
- 'signed integer', field_value_range, preferred_display_base)
-
- def create_unsigned_integer_field_class(self, field_value_range=None, preferred_display_base=None):
- return self._create_integer_field_class(native_bt.field_class_unsigned_integer_create,
- bt2.field_class._UnsignedIntegerFieldClass,
- 'unsigned integer', field_value_range, preferred_display_base)
-
- def create_signed_enumeration_field_class(self, field_value_range=None, preferred_display_base=None):
- return self._create_integer_field_class(native_bt.field_class_signed_enumeration_create,
- bt2.field_class._SignedEnumerationFieldClass,
- 'signed enumeration', field_value_range, preferred_display_base)
-
- def create_unsigned_enumeration_field_class(self, field_value_range=None, preferred_display_base=None):
- return self._create_integer_field_class(native_bt.field_class_unsigned_enumeration_create,
- bt2.field_class._UnsignedEnumerationFieldClass,
- 'unsigned enumeration', field_value_range, preferred_display_base)
+ def create_signed_integer_field_class(
+ self, field_value_range=None, preferred_display_base=None
+ ):
+ return self._create_integer_field_class(
+ native_bt.field_class_signed_integer_create,
+ bt2.field_class._SignedIntegerFieldClass,
+ 'signed integer',
+ field_value_range,
+ preferred_display_base,
+ )
+
+ def create_unsigned_integer_field_class(
+ self, field_value_range=None, preferred_display_base=None
+ ):
+ return self._create_integer_field_class(
+ native_bt.field_class_unsigned_integer_create,
+ bt2.field_class._UnsignedIntegerFieldClass,
+ 'unsigned integer',
+ field_value_range,
+ preferred_display_base,
+ )
+
+ def create_signed_enumeration_field_class(
+ self, field_value_range=None, preferred_display_base=None
+ ):
+ return self._create_integer_field_class(
+ native_bt.field_class_signed_enumeration_create,
+ bt2.field_class._SignedEnumerationFieldClass,
+ 'signed enumeration',
+ field_value_range,
+ preferred_display_base,
+ )
+
+ def create_unsigned_enumeration_field_class(
+ self, field_value_range=None, preferred_display_base=None
+ ):
+ return self._create_integer_field_class(
+ native_bt.field_class_unsigned_enumeration_create,
+ bt2.field_class._UnsignedEnumerationFieldClass,
+ 'unsigned enumeration',
+ field_value_range,
+ preferred_display_base,
+ )
def create_real_field_class(self, is_single_precision=False):
field_class_ptr = native_bt.field_class_real_create(self._ptr)
utils._check_type(length_fc, bt2.field_class._UnsignedIntegerFieldClass)
length_fc_ptr = length_fc._ptr
- ptr = native_bt.field_class_dynamic_array_create(self._ptr, elem_fc._ptr, length_fc_ptr)
+ ptr = native_bt.field_class_dynamic_array_create(
+ self._ptr, elem_fc._ptr, length_fc_ptr
+ )
self._check_create_status(ptr, 'dynamic array')
return bt2.field_class._DynamicArrayFieldClass._create_from_ptr(ptr)
raise TypeError("'listener' parameter is not callable")
fn = native_bt.bt2_trace_class_add_destruction_listener
- listener_from_native = functools.partial(_trace_class_destruction_listener_from_native,
- listener)
+ listener_from_native = functools.partial(
+ _trace_class_destruction_listener_from_native, listener
+ )
listener_id = fn(self._ptr, listener_from_native)
if listener_id is None:
- utils._raise_bt2_error('cannot add destruction listener to trace class object')
+ utils._raise_bt2_error(
+ 'cannot add destruction listener to trace class object'
+ )
return bt2._ListenerHandle(listener_id, self)
class ComponentSpec:
- def __init__(self, plugin_name, class_name, params=None,
- logging_level=bt2.logging.LoggingLevel.NONE):
+ def __init__(
+ self,
+ plugin_name,
+ class_name,
+ params=None,
+ logging_level=bt2.logging.LoggingLevel.NONE,
+ ):
utils._check_str(plugin_name)
utils._check_str(class_name)
utils._check_log_level(logging_level)
# s -> ns
s = obj.timestamp()
else:
- raise TypeError('"{}" is not an integral number or a datetime.datetime object'.format(obj))
+ raise TypeError(
+ '"{}" is not an integral number or a datetime.datetime object'.format(obj)
+ )
return int(s * 1e9)
class TraceCollectionMessageIterator(bt2.message_iterator._MessageIterator):
- def __init__(self, source_component_specs, filter_component_specs=None,
- stream_intersection_mode=False, begin=None, end=None):
+ def __init__(
+ self,
+ source_component_specs,
+ filter_component_specs=None,
+ stream_intersection_mode=False,
+ begin=None,
+ end=None,
+ ):
utils._check_bool(stream_intersection_mode)
self._stream_intersection_mode = stream_intersection_mode
self._begin_ns = _get_ns(begin)
def _validate_component_specs(self, comp_specs):
for comp_spec in comp_specs:
if type(comp_spec) is not ComponentSpec:
- raise TypeError('"{}" object is not a ComponentSpec'.format(type(comp_spec)))
+ raise TypeError(
+ '"{}" object is not a ComponentSpec'.format(type(comp_spec))
+ )
def __next__(self):
return next(self._msg_iter)
try:
paths = src_comp_and_spec.spec.params['paths']
except Exception as e:
- raise bt2.Error('all source components must be created with a "paths" parameter in stream intersection mode') from e
+ raise bt2.Error(
+ 'all source components must be created with a "paths" parameter in stream intersection mode'
+ ) from e
params = {'paths': paths}
# contains the stream intersection range for each exposed
# trace
query_exec = bt2.QueryExecutor()
- trace_info_res = query_exec.query(src_comp_and_spec.comp.cls,
- 'trace-info', params)
+ trace_info_res = query_exec.query(
+ src_comp_and_spec.comp.cls, 'trace-info', params
+ )
begin = None
end = None
pass
if begin is None or end is None:
- raise bt2.Error('cannot find stream intersection range for port "{}"'.format(port.name))
+ raise bt2.Error(
+ 'cannot find stream intersection range for port "{}"'.format(port.name)
+ )
name = 'trimmer-{}-{}'.format(src_comp_and_spec.comp.name, port.name)
return self._create_trimmer(begin, end, name)
raise bt2.Error('cannot find "utils" plugin (needed for the muxer)')
if 'muxer' not in plugin.filter_component_classes:
- raise bt2.Error('cannot find "muxer" filter component class in "utils" plugin')
+ raise bt2.Error(
+ 'cannot find "muxer" filter component class in "utils" plugin'
+ )
comp_cls = plugin.filter_component_classes['muxer']
return self._graph.add_component(comp_cls, 'muxer')
raise bt2.Error('cannot find "utils" plugin (needed for the trimmer)')
if 'trimmer' not in plugin.filter_component_classes:
- raise bt2.Error('cannot find "trimmer" filter component class in "utils" plugin')
+ raise bt2.Error(
+ 'cannot find "trimmer" filter component class in "utils" plugin'
+ )
params = {}
return self._graph.add_component(comp_cls, name, params)
def _get_unique_comp_name(self, comp_spec):
- name = '{}-{}'.format(comp_spec.plugin_name,
- comp_spec.class_name)
- comps_and_specs = itertools.chain(self._src_comps_and_specs,
- self._flt_comps_and_specs)
+ name = '{}-{}'.format(comp_spec.plugin_name, comp_spec.class_name)
+ comps_and_specs = itertools.chain(
+ self._src_comps_and_specs, self._flt_comps_and_specs
+ )
if name in [comp_and_spec.comp.name for comp_and_spec in comps_and_specs]:
name += '-{}'.format(self._next_suffix)
if comp_spec.class_name not in comp_classes:
cc_type = 'source' if comp_cls_type == _CompClsType.SOURCE else 'filter'
- raise bt2.Error('no such {} component class in "{}" plugin: {}'.format(cc_type,
- comp_spec.plugin_name,
- comp_spec.class_name))
+ raise bt2.Error(
+ 'no such {} component class in "{}" plugin: {}'.format(
+ cc_type, comp_spec.plugin_name, comp_spec.class_name
+ )
+ )
comp_cls = comp_classes[comp_spec.class_name]
name = self._get_unique_comp_name(comp_spec)
- comp = self._graph.add_component(comp_cls, name, comp_spec.params,
- comp_spec.logging_level)
+ comp = self._graph.add_component(
+ comp_cls, name, comp_spec.params, comp_spec.logging_level
+ )
return comp
def _get_free_muxer_input_port(self):
self._muxer_comp = self._create_muxer()
if self._begin_ns is not None or self._end_ns is not None:
- trimmer_comp = self._create_trimmer(self._begin_ns,
- self._end_ns, 'trimmer')
- self._graph.connect_ports(self._muxer_comp.output_ports['out'],
- trimmer_comp.input_ports['in'])
+ trimmer_comp = self._create_trimmer(self._begin_ns, self._end_ns, 'trimmer')
+ self._graph.connect_ports(
+ self._muxer_comp.output_ports['out'], trimmer_comp.input_ports['in']
+ )
msg_iter_port = trimmer_comp.output_ports['out']
else:
msg_iter_port = self._muxer_comp.output_ports['out']
def _check_type(o, expected_type):
if not isinstance(o, expected_type):
- raise TypeError("'{}' is not a '{}' object".format(o.__class__.__name__,
- expected_type))
+ raise TypeError(
+ "'{}' is not a '{}' object".format(o.__class__.__name__, expected_type)
+ )
def _is_in_int64_range(v):
- assert(isinstance(v, int))
- return v >= -(2**63) and v <= (2**63 - 1)
+ assert isinstance(v, int)
+ return v >= -(2 ** 63) and v <= (2 ** 63 - 1)
def _is_int64(v):
def _is_in_uint64_range(v):
- assert(isinstance(v, int))
- return v >= 0 and v <= (2**64 - 1)
+ assert isinstance(v, int)
+ return v >= 0 and v <= (2 ** 64 - 1)
def _is_uint64(v):
# no error
return
- if status == native_bt.__BT_FUNC_STATUS_ERROR or status == native_bt.__BT_FUNC_STATUS_MEMORY_ERROR:
+ if (
+ status == native_bt.__BT_FUNC_STATUS_ERROR
+ or status == native_bt.__BT_FUNC_STATUS_MEMORY_ERROR
+ ):
if msg is None:
raise bt2.Error
else:
if isinstance(value, collections.abc.Mapping):
return MapValue(value)
- raise TypeError("cannot create value object from '{}' object".format(value.__class__.__name__))
+ raise TypeError(
+ "cannot create value object from '{}' object".format(value.__class__.__name__)
+ )
class _Value(object._SharedObject, metaclass=abc.ABCMeta):
def _check_create_status(self, ptr):
if ptr is None:
raise bt2.CreationError(
- 'cannot create {} value object'.format(self._NAME.lower()))
+ 'cannot create {} value object'.format(self._NAME.lower())
+ )
@functools.total_ordering
if isinstance(other, numbers.Complex):
return complex(other)
- raise TypeError("'{}' object is not a number object".format(other.__class__.__name__))
+ raise TypeError(
+ "'{}' object is not a number object".format(other.__class__.__name__)
+ )
def __int__(self):
return int(self._value)
value = value._value
if not isinstance(value, bool):
- raise TypeError("'{}' object is not a 'bool' or 'BoolValue' object".format(value.__class__))
+ raise TypeError(
+ "'{}' object is not a 'bool' or 'BoolValue' object".format(
+ value.__class__
+ )
+ )
return value
def __len__(self):
size = native_bt.value_array_get_size(self._ptr)
- assert(size >= 0)
+ assert size >= 0
return size
def _check_index(self, index):
# TODO: support slices also
if not isinstance(index, numbers.Integral):
- raise TypeError("'{}' object is not an integral number object: invalid index".format(index.__class__.__name__))
+ raise TypeError(
+ "'{}' object is not an integral number object: invalid index".format(
+ index.__class__.__name__
+ )
+ )
index = int(index)
def __getitem__(self, index):
self._check_index(index)
ptr = native_bt.value_array_borrow_element_by_index(self._ptr, index)
- assert(ptr)
+ assert ptr
return _create_from_ptr_and_get_ref(ptr)
def __setitem__(self, index, value):
else:
ptr = value._ptr
- status = native_bt.value_array_set_element_by_index(
- self._ptr, index, ptr)
+ status = native_bt.value_array_set_element_by_index(self._ptr, index, ptr)
utils._handle_func_status(status)
def append(self, value):
def __len__(self):
size = native_bt.value_map_get_size(self._ptr)
- assert(size >= 0)
+ assert size >= 0
return size
def __contains__(self, key):
def __getitem__(self, key):
self._check_key(key)
ptr = native_bt.value_map_borrow_entry_value(self._ptr, key)
- assert(ptr)
+ assert ptr
return _create_from_ptr_and_get_ref(ptr)
def __iter__(self):
--------------------------------------------------------------------------------
"""
+
def main():
- babeltrace_ext = Extension('bt2._native_bt',
- sources=['bt2/native_bt.c', '@srcdir@/bt2/logging.c'],
- libraries=['babeltrace2', 'glib-2.0'],
- extra_objects=['@top_builddir@/src/logging/.libs/libbabeltrace2-logging.a',
- '@top_builddir@/src/common/.libs/libbabeltrace2-common.a',
- '@top_builddir@/src/py-common/.libs/libbabeltrace2-py-common.a'])
+ babeltrace_ext = Extension(
+ 'bt2._native_bt',
+ sources=['bt2/native_bt.c', '@srcdir@/bt2/logging.c'],
+ libraries=['babeltrace2', 'glib-2.0'],
+ extra_objects=[
+ '@top_builddir@/src/logging/.libs/libbabeltrace2-logging.a',
+ '@top_builddir@/src/common/.libs/libbabeltrace2-common.a',
+ '@top_builddir@/src/py-common/.libs/libbabeltrace2-py-common.a',
+ ],
+ )
- dist = setup(name='bt2',
- version='@PACKAGE_VERSION@',
- description='Babeltrace 2 Python Bindings',
- packages=['bt2'],
- package_dir={'bt2': 'bt2'},
- options={'build':
- {
- 'build_base': 'build',
- 'build_lib': 'build/build_lib'
- },
- 'build_ext':
- {
- 'build_lib': 'build/build_lib'
- }
- },
- url='http://diamon.org/babeltrace',
- ext_modules=[babeltrace_ext],
- license='MIT',
- classifiers=[
- 'Development Status :: 5 - Production/Stable',
- 'Intended Audience :: Developers',
- 'License :: OSI Approved :: The MIT License',
- 'Programming Language :: Python :: 3'
- 'Topic :: System :: Logging',
- ])
+ dist = setup(
+ name='bt2',
+ version='@PACKAGE_VERSION@',
+ description='Babeltrace 2 Python Bindings',
+ packages=['bt2'],
+ package_dir={'bt2': 'bt2'},
+ options={
+ 'build': {'build_base': 'build', 'build_lib': 'build/build_lib'},
+ 'build_ext': {'build_lib': 'build/build_lib'},
+ },
+ url='http://diamon.org/babeltrace',
+ ext_modules=[babeltrace_ext],
+ license='MIT',
+ classifiers=[
+ 'Development Status :: 5 - Production/Stable',
+ 'Intended Audience :: Developers',
+ 'License :: OSI Approved :: The MIT License',
+ 'Programming Language :: Python :: 3' 'Topic :: System :: Logging',
+ ],
+ )
-# After the installation, we check that the install directory is included in
-# the Python search path and we print a warning message when it's not.
-# We need to do this because Python search path differs depending on the distro
-# and some distros don't include any /usr/local/ in the search path. This is
-# also useful for out-of-tree installs and tests.
-# It's only relevant to make this check on the `install` command.
+ # After the installation, we check that the install directory is included in
+ # the Python search path and we print a warning message when it's not.
+ # We need to do this because Python search path differs depending on the distro
+ # and some distros don't include any /usr/local/ in the search path. This is
+ # also useful for out-of-tree installs and tests.
+ # It's only relevant to make this check on the `install` command.
if 'install' in dist.command_obj:
install_dir = dist.command_obj['install'].install_libbase
# extra configuration is needed to use the bindings
print(PY_PATH_WARN_MSG.format(install_dir, install_dir))
+
if __name__ == "__main__":
main()
def test_cycles_to_ns_from_origin(self):
def f(comp_self):
- return comp_self._create_clock_class(frequency=10**8, origin_is_unix_epoch=True)
+ return comp_self._create_clock_class(
+ frequency=10 ** 8, origin_is_unix_epoch=True
+ )
cc = run_in_component_init(f)
self.assertEqual(cc.cycles_to_ns_from_origin(112), 1120)
cc = run_in_component_init(f)
with self.assertRaises(bt2.OverflowError):
- cc.cycles_to_ns_from_origin(2**63)
+ cc.cycles_to_ns_from_origin(2 ** 63)
def test_create_uuid(self):
def f(comp_self):
- return comp_self._create_clock_class(uuid=uuid.UUID('b43372c32ef0be28444dfc1c5cdafd33'))
+ return comp_self._create_clock_class(
+ uuid=uuid.UUID('b43372c32ef0be28444dfc1c5cdafd33')
+ )
cc = run_in_component_init(f)
self.assertEqual(cc.uuid, uuid.UUID('b43372c32ef0be28444dfc1c5cdafd33'))
class ClockSnapshotTestCase(unittest.TestCase):
def setUp(self):
def f(comp_self):
- cc = comp_self._create_clock_class(1000, 'my_cc',
- offset=bt2.ClockClassOffset(45, 354))
+ cc = comp_self._create_clock_class(
+ 1000, 'my_cc', offset=bt2.ClockClassOffset(45, 354)
+ )
tc = comp_self._create_trace_class()
return (cc, tc)
elif self._at == 1:
notif = self._create_event_message(_ec, _stream, 123)
elif self._at == 2:
- notif = self._create_event_message(_ec, _stream, 2**63)
+ notif = self._create_event_message(_ec, _stream, 2 ** 63)
elif self._at == 3:
notif = self._create_stream_end_message(_stream)
else:
self._graph = bt2.Graph()
self._src_comp = self._graph.add_component(MySrc, 'my_source')
self._msg_iter = self._graph.create_output_port_message_iterator(
- self._src_comp.output_ports['out'])
+ self._src_comp.output_ports['out']
+ )
for i, msg in enumerate(self._msg_iter):
if i == 1:
def test_create_default(self):
self.assertEqual(
- self._msg.default_clock_snapshot.clock_class.addr, self._cc.addr)
+ self._msg.default_clock_snapshot.clock_class.addr, self._cc.addr
+ )
self.assertEqual(self._msg.default_clock_snapshot.value, 123)
def test_clock_class(self):
self.assertEqual(
- self._msg.default_clock_snapshot.clock_class.addr, self._cc.addr)
+ self._msg.default_clock_snapshot.clock_class.addr, self._cc.addr
+ )
def test_ns_from_origin(self):
s_from_origin = 45 + ((354 + 123) / 1000)
ns_from_origin = int(s_from_origin * 1e9)
self.assertEqual(
- self._msg.default_clock_snapshot.ns_from_origin, ns_from_origin)
+ self._msg.default_clock_snapshot.ns_from_origin, ns_from_origin
+ )
def test_ns_from_origin_overflow(self):
with self.assertRaises(bt2.OverflowError):
def __next__(self):
raise bt2.Stop
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
pass
self._test_no_init(MySource)
def __next__(self):
raise bt2.Stop
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
pass
self._test_no_init(MyFilter)
pass
with self.assertRaises(bt2.IncompleteUserClass):
+
class MySource(bt2._UserSourceComponent):
pass
pass
with self.assertRaises(bt2.IncompleteUserClass):
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=int):
+
+ class MySource(bt2._UserSourceComponent, message_iterator_class=int):
pass
def test_incomplete_filter_no_msg_iter_cls(self):
pass
with self.assertRaises(bt2.IncompleteUserClass):
+
class MyFilter(bt2._UserFilterComponent):
pass
pass
with self.assertRaises(bt2.IncompleteUserClass):
+
class MySink(bt2._UserSinkComponent):
pass
class MyIter(bt2._UserMessageIterator):
pass
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
pass
def test_minimal_filter(self):
class MyIter(bt2._UserMessageIterator):
pass
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
pass
def test_minimal_sink(self):
def test_invalid_custom_name(self):
with self.assertRaises(TypeError):
+
class MySink(bt2._UserSinkComponent, name=23):
def _consume(self):
pass
query_log_level = log_level
query_log_level = None
- res = bt2.QueryExecutor().query(MySink, 'obj', None,
- bt2.LoggingLevel.WARNING)
+ res = bt2.QueryExecutor().query(MySink, 'obj', None, bt2.LoggingLevel.WARNING)
self.assertEqual(query_log_level, bt2.LoggingLevel.WARNING)
del query_log_level
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_params
query_params = params
- return {
- 'null': None,
- 'bt2': 'BT2',
- }
+ return {'null': None, 'bt2': 'BT2'}
query_params = None
params = {
'array': ['coucou', 23, None],
- 'other_map': {
- 'yes': 'yeah',
- '19': 19,
- 'minus 1.5': -1.5,
- },
+ 'other_map': {'yes': 'yeah', '19': 19, 'minus 1.5': -1.5},
'null': None,
}
res = bt2.QueryExecutor().query(MySink, 'obj', params)
self.assertEqual(query_params, params)
- self.assertEqual(res, {
- 'null': None,
- 'bt2': 'BT2',
- })
+ self.assertEqual(res, {'null': None, 'bt2': 'BT2'})
del query_params
def test_eq(self):
The help.
'''
+
def _consume(self):
pass
graph = bt2.Graph()
comp = graph.add_component(MySink, 'salut')
self._comp_cls = comp.cls
- self.assertTrue(issubclass(type(self._comp_cls),
- bt2.component._GenericComponentClass))
+ self.assertTrue(
+ issubclass(type(self._comp_cls), bt2.component._GenericComponentClass)
+ )
def tearDown(self):
del self._py_comp_cls
self.assertEqual(self._py_comp_cls, self._comp_cls)
def test_query(self):
- res = bt2.QueryExecutor().query(self._comp_cls, 'an object',
- {'yes': 'no', 'book': -17})
+ res = bt2.QueryExecutor().query(
+ self._comp_cls, 'an object', {'yes': 'no', 'book': -17}
+ )
expected = ['an object', {'yes': 'no', 'book': -17}, 23]
self.assertEqual(res, expected)
def __next__(self):
raise bt2.Stop
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
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.assertIsInstance(conn, bt2._Connection)
def test_downstream_port(self):
def __next__(self):
raise bt2.Stop
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
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):
def __next__(self):
raise bt2.Stop
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
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)
def test_create_full(self):
my_uuid = uuid.uuid1()
- cc = bt2.CtfWriterClock(name='name', description='some description',
- frequency=1001, precision=176,
- offset=bt2.ClockClassOffset(45, 3003),
- is_absolute=True, uuid=my_uuid)
+ cc = bt2.CtfWriterClock(
+ name='name',
+ description='some description',
+ frequency=1001,
+ precision=176,
+ offset=bt2.ClockClassOffset(45, 3003),
+ is_absolute=True,
+ uuid=my_uuid,
+ )
self.assertEqual(cc.name, 'name')
self.assertEqual(cc.description, 'some description')
self.assertEqual(cc.frequency, 1001)
def test_eq(self):
my_uuid = uuid.uuid1()
- cc1 = bt2.CtfWriterClock(name='name', description='some description',
- frequency=1001, precision=176,
- offset=bt2.ClockClassOffset(45, 3003),
- is_absolute=True, uuid=my_uuid)
- cc2 = bt2.CtfWriterClock(name='name', description='some description',
- frequency=1001, precision=176,
- offset=bt2.ClockClassOffset(45, 3003),
- is_absolute=True, uuid=my_uuid)
+ cc1 = bt2.CtfWriterClock(
+ name='name',
+ description='some description',
+ frequency=1001,
+ precision=176,
+ offset=bt2.ClockClassOffset(45, 3003),
+ is_absolute=True,
+ uuid=my_uuid,
+ )
+ cc2 = bt2.CtfWriterClock(
+ name='name',
+ description='some description',
+ frequency=1001,
+ precision=176,
+ offset=bt2.ClockClassOffset(45, 3003),
+ is_absolute=True,
+ uuid=my_uuid,
+ )
self.assertEqual(cc1, cc2)
def test_ne_name(self):
my_uuid = uuid.uuid1()
- cc1 = bt2.CtfWriterClock(name='mane', description='some description',
- frequency=1001, precision=176,
- offset=bt2.ClockClassOffset(45, 3003),
- is_absolute=True, uuid=my_uuid)
- cc2 = bt2.CtfWriterClock(name='name', description='some description',
- frequency=1001, precision=176,
- offset=bt2.ClockClassOffset(45, 3003),
- is_absolute=True, uuid=my_uuid)
+ cc1 = bt2.CtfWriterClock(
+ name='mane',
+ description='some description',
+ frequency=1001,
+ precision=176,
+ offset=bt2.ClockClassOffset(45, 3003),
+ is_absolute=True,
+ uuid=my_uuid,
+ )
+ cc2 = bt2.CtfWriterClock(
+ name='name',
+ description='some description',
+ frequency=1001,
+ precision=176,
+ offset=bt2.ClockClassOffset(45, 3003),
+ is_absolute=True,
+ uuid=my_uuid,
+ )
self.assertNotEqual(cc1, cc2)
def test_ne_description(self):
my_uuid = uuid.uuid1()
- cc1 = bt2.CtfWriterClock(name='name', description='some descripti2',
- frequency=1001, precision=176,
- offset=bt2.ClockClassOffset(45, 3003),
- is_absolute=True, uuid=my_uuid)
- cc2 = bt2.CtfWriterClock(name='name', description='some description',
- frequency=1001, precision=176,
- offset=bt2.ClockClassOffset(45, 3003),
- is_absolute=True, uuid=my_uuid)
+ cc1 = bt2.CtfWriterClock(
+ name='name',
+ description='some descripti2',
+ frequency=1001,
+ precision=176,
+ offset=bt2.ClockClassOffset(45, 3003),
+ is_absolute=True,
+ uuid=my_uuid,
+ )
+ cc2 = bt2.CtfWriterClock(
+ name='name',
+ description='some description',
+ frequency=1001,
+ precision=176,
+ offset=bt2.ClockClassOffset(45, 3003),
+ is_absolute=True,
+ uuid=my_uuid,
+ )
self.assertNotEqual(cc1, cc2)
def test_ne_frequency(self):
my_uuid = uuid.uuid1()
- cc1 = bt2.CtfWriterClock(name='name', description='some description',
- frequency=1003, precision=176,
- offset=bt2.ClockClassOffset(45, 3003),
- is_absolute=True, uuid=my_uuid)
- cc2 = bt2.CtfWriterClock(name='name', description='some description',
- frequency=1001, precision=176,
- offset=bt2.ClockClassOffset(45, 3003),
- is_absolute=True, uuid=my_uuid)
+ cc1 = bt2.CtfWriterClock(
+ name='name',
+ description='some description',
+ frequency=1003,
+ precision=176,
+ offset=bt2.ClockClassOffset(45, 3003),
+ is_absolute=True,
+ uuid=my_uuid,
+ )
+ cc2 = bt2.CtfWriterClock(
+ name='name',
+ description='some description',
+ frequency=1001,
+ precision=176,
+ offset=bt2.ClockClassOffset(45, 3003),
+ is_absolute=True,
+ uuid=my_uuid,
+ )
self.assertNotEqual(cc1, cc2)
def test_ne_precision(self):
my_uuid = uuid.uuid1()
- cc1 = bt2.CtfWriterClock(name='name', description='some description',
- frequency=1001, precision=171,
- offset=bt2.ClockClassOffset(45, 3003),
- is_absolute=True, uuid=my_uuid)
- cc2 = bt2.CtfWriterClock(name='name', description='some description',
- frequency=1001, precision=176,
- offset=bt2.ClockClassOffset(45, 3003),
- is_absolute=True, uuid=my_uuid)
+ cc1 = bt2.CtfWriterClock(
+ name='name',
+ description='some description',
+ frequency=1001,
+ precision=171,
+ offset=bt2.ClockClassOffset(45, 3003),
+ is_absolute=True,
+ uuid=my_uuid,
+ )
+ cc2 = bt2.CtfWriterClock(
+ name='name',
+ description='some description',
+ frequency=1001,
+ precision=176,
+ offset=bt2.ClockClassOffset(45, 3003),
+ is_absolute=True,
+ uuid=my_uuid,
+ )
self.assertNotEqual(cc1, cc2)
def test_ne_offset(self):
my_uuid = uuid.uuid1()
- cc1 = bt2.CtfWriterClock(name='name', description='some description',
- frequency=1001, precision=176,
- offset=bt2.ClockClassOffset(45, 3001),
- is_absolute=True, uuid=my_uuid)
- cc2 = bt2.CtfWriterClock(name='name', description='some description',
- frequency=1001, precision=176,
- offset=bt2.ClockClassOffset(45, 3003),
- is_absolute=True, uuid=my_uuid)
+ cc1 = bt2.CtfWriterClock(
+ name='name',
+ description='some description',
+ frequency=1001,
+ precision=176,
+ offset=bt2.ClockClassOffset(45, 3001),
+ is_absolute=True,
+ uuid=my_uuid,
+ )
+ cc2 = bt2.CtfWriterClock(
+ name='name',
+ description='some description',
+ frequency=1001,
+ precision=176,
+ offset=bt2.ClockClassOffset(45, 3003),
+ is_absolute=True,
+ uuid=my_uuid,
+ )
self.assertNotEqual(cc1, cc2)
def test_ne_absolute(self):
my_uuid = uuid.uuid1()
- cc1 = bt2.CtfWriterClock(name='name', description='some description',
- frequency=1001, precision=176,
- offset=bt2.ClockClassOffset(45, 3003),
- is_absolute=True, uuid=my_uuid)
- cc2 = bt2.CtfWriterClock(name='name', description='some description',
- frequency=1001, precision=176,
- offset=bt2.ClockClassOffset(45, 3003),
- is_absolute=False, uuid=my_uuid)
+ cc1 = bt2.CtfWriterClock(
+ name='name',
+ description='some description',
+ frequency=1001,
+ precision=176,
+ offset=bt2.ClockClassOffset(45, 3003),
+ is_absolute=True,
+ uuid=my_uuid,
+ )
+ cc2 = bt2.CtfWriterClock(
+ name='name',
+ description='some description',
+ frequency=1001,
+ precision=176,
+ offset=bt2.ClockClassOffset(45, 3003),
+ is_absolute=False,
+ uuid=my_uuid,
+ )
self.assertNotEqual(cc1, cc2)
def test_ne_uuid(self):
- cc1 = bt2.CtfWriterClock(name='name', description='some description',
- frequency=1001, precision=176,
- offset=bt2.ClockClassOffset(45, 3003),
- is_absolute=True, uuid=uuid.uuid1())
- cc2 = bt2.CtfWriterClock(name='name', description='some description',
- frequency=1001, precision=176,
- offset=bt2.ClockClassOffset(45, 3003),
- is_absolute=True, uuid=uuid.uuid1())
+ cc1 = bt2.CtfWriterClock(
+ name='name',
+ description='some description',
+ frequency=1001,
+ precision=176,
+ offset=bt2.ClockClassOffset(45, 3003),
+ is_absolute=True,
+ uuid=uuid.uuid1(),
+ )
+ cc2 = bt2.CtfWriterClock(
+ name='name',
+ description='some description',
+ frequency=1001,
+ precision=176,
+ offset=bt2.ClockClassOffset(45, 3003),
+ is_absolute=True,
+ uuid=uuid.uuid1(),
+ )
self.assertNotEqual(cc1, cc2)
def test_eq_invalid(self):
class EventTestCase(unittest.TestCase):
- def _create_test_event_message(self, packet_fields_config=None,
- event_fields_config=None,
- with_clockclass=False,
- with_cc=False, with_sc=False,
- with_ep=False, with_packet=False):
-
+ def _create_test_event_message(
+ self,
+ packet_fields_config=None,
+ event_fields_config=None,
+ with_clockclass=False,
+ with_cc=False,
+ with_sc=False,
+ with_ep=False,
+ with_packet=False,
+ ):
class MyIter(bt2._UserMessageIterator):
def __init__(self, self_output_port):
self._at = 0
- self._msgs = [
- self._create_stream_beginning_message(test_obj.stream),
- ]
+ self._msgs = [self._create_stream_beginning_message(test_obj.stream)]
if with_packet:
assert test_obj.packet
- self._msgs.append(self._create_packet_beginning_message(test_obj.packet))
+ self._msgs.append(
+ self._create_packet_beginning_message(test_obj.packet)
+ )
default_clock_snapshot = 789 if with_clockclass else None
assert test_obj.stream
ev_parent = test_obj.stream
- msg = self._create_event_message(test_obj.event_class, ev_parent, default_clock_snapshot)
+ msg = self._create_event_message(
+ test_obj.event_class, ev_parent, default_clock_snapshot
+ )
if event_fields_config is not None:
event_fields_config(msg.event)
('something_else', tc.create_real_field_class()),
]
- stream_class = tc.create_stream_class(default_clock_class=clock_class,
- event_common_context_field_class=cc,
- packet_context_field_class=pc,
- supports_packets=with_packet)
+ stream_class = tc.create_stream_class(
+ default_clock_class=clock_class,
+ event_common_context_field_class=cc,
+ packet_context_field_class=pc,
+ supports_packets=with_packet,
+ )
# specific context (event-class-defined)
sc = None
('mosquito', tc.create_signed_integer_field_class(8)),
]
- event_class = stream_class.create_event_class(name='garou',
- specific_context_field_class=sc,
- payload_field_class=ep)
+ event_class = stream_class.create_event_class(
+ name='garou',
+ specific_context_field_class=sc,
+ payload_field_class=ep,
+ )
trace = tc()
stream = trace.create_stream(stream_class)
test_obj = self
self._graph = bt2.Graph()
self._src_comp = self._graph.add_component(MySrc, 'my_source')
- self._msg_iter = self._graph.create_output_port_message_iterator(self._src_comp.output_ports['out'])
+ self._msg_iter = self._graph.create_output_port_message_iterator(
+ self._src_comp.output_ports['out']
+ )
for msg in self._msg_iter:
if type(msg) is bt2._EventMessage:
event.common_context_field['cpu_id'] = 1
event.common_context_field['stuff'] = 13.194
- msg = self._create_test_event_message(event_fields_config=event_fields_config, with_cc=True)
+ msg = self._create_test_event_message(
+ event_fields_config=event_fields_config, with_cc=True
+ )
self.assertEqual(msg.event.common_context_field['cpu_id'], 1)
self.assertEqual(msg.event.common_context_field['stuff'], 13.194)
event.specific_context_field['ant'] = -1
event.specific_context_field['msg'] = 'hellooo'
- msg = self._create_test_event_message(event_fields_config=event_fields_config, with_sc=True)
+ msg = self._create_test_event_message(
+ event_fields_config=event_fields_config, with_sc=True
+ )
self.assertEqual(msg.event.specific_context_field['ant'], -1)
self.assertEqual(msg.event.specific_context_field['msg'], 'hellooo')
event.payload_field['gnu'] = 23
event.payload_field['mosquito'] = 42
- msg = self._create_test_event_message(event_fields_config=event_fields_config, with_ep=True)
+ msg = self._create_test_event_message(
+ event_fields_config=event_fields_config, with_ep=True
+ )
self.assertEqual(msg.event.payload_field['giraffe'], 1)
self.assertEqual(msg.event.payload_field['gnu'], 23)
packet.context_field['something'] = 154
packet.context_field['something_else'] = 17.2
- msg = self._create_test_event_message(packet_fields_config=packet_fields_config,
- event_fields_config=event_fields_config,
- with_cc=True, with_sc=True, with_ep=True,
- with_packet=True)
+ msg = self._create_test_event_message(
+ packet_fields_config=packet_fields_config,
+ event_fields_config=event_fields_config,
+ with_cc=True,
+ with_sc=True,
+ with_ep=True,
+ with_packet=True,
+ )
ev = msg.event
# Test event fields
self._context_fc = self._tc.create_structure_field_class()
self._context_fc.append_member('allo', self._tc.create_string_field_class())
- self._context_fc.append_member('zola', self._tc.create_signed_integer_field_class(18))
+ self._context_fc.append_member(
+ 'zola', self._tc.create_signed_integer_field_class(18)
+ )
self._payload_fc = self._tc.create_structure_field_class()
self._payload_fc.append_member('zoom', self._tc.create_string_field_class())
- self._stream_class = self._tc.create_stream_class(assigns_automatic_event_class_id=True)
+ self._stream_class = self._tc.create_stream_class(
+ assigns_automatic_event_class_id=True
+ )
def test_create_default(self):
ec = self._stream_class.create_event_class()
self._stream_class.create_event_class(emf_uri=23)
def test_create_log_level(self):
- ec = self._stream_class.create_event_class(log_level=bt2.EventClassLogLevel.EMERGENCY)
+ ec = self._stream_class.create_event_class(
+ log_level=bt2.EventClassLogLevel.EMERGENCY
+ )
self.assertEqual(ec.log_level, bt2.EventClassLogLevel.EMERGENCY)
def test_create_invalid_log_level(self):
from utils import get_default_trace_class
-_COMP_BINOPS = (
- operator.eq,
- operator.ne,
-)
+_COMP_BINOPS = (operator.eq, operator.ne)
# Create and return a stream with the field classes part of its stream packet
#
# The stream is part of a dummy trace created from trace class `tc`.
+
def _create_stream(tc, ctx_field_classes):
packet_context_fc = tc.create_structure_field_class()
for name, fc in ctx_field_classes:
packet_context_fc.append_member(name, fc)
trace = tc()
- stream_class = tc.create_stream_class(packet_context_field_class=packet_context_fc,
- supports_packets=True)
+ stream_class = tc.create_stream_class(
+ packet_context_field_class=packet_context_fc, supports_packets=True
+ )
stream = trace.create_stream(stream_class)
return stream
# The field is part of a dummy stream, itself part of a dummy trace created
# from trace class `tc`.
+
def _create_field(tc, field_class):
field_name = 'field'
stream = _create_stream(tc, [(field_name, field_class)])
# The field is part of a dummy stream, itself part of a dummy trace created
# from trace class `tc`. It is made out of a dummy string field class.
+
def _create_string_field(tc):
field_name = 'string_field'
stream = _create_stream(tc, [(field_name, tc.create_string_field_class())])
# from trace class `tc`. It is made out of a dummy static array field class,
# with a dummy integer field class as element class.
+
def _create_int_array_field(tc, length):
elem_fc = tc.create_signed_integer_field_class(32)
fc = tc.create_static_array_field_class(elem_fc, length)
# from trace class `tc`. It is made out of a dummy static array field class,
# with a dummy integer field class as element and length classes.
+
def _create_dynamic_array(tc):
elem_fc = tc.create_signed_integer_field_class(32)
len_fc = tc.create_signed_integer_field_class(32)
# from trace class `tc`. It is made out of a dummy static array field class,
# with a dummy struct field class as element class.
+
def _create_struct_array_field(tc, length):
elem_fc = tc.create_structure_field_class()
fc = tc.create_static_array_field_class(elem_fc, length)
test_cb(op, bt2.create_value(0.0))
def _test_binop_rhs_complex(self, test_cb, op):
- test_cb(op, -23+19j)
+ test_cb(op, -23 + 19j)
def _test_binop_rhs_zero_complex(self, test_cb, op):
test_cb(op, 0j)
# Ignore this lint error:
# E711 comparison to None should be 'if cond is None:'
# since this is what we want to test (even though not good practice).
- self.assertFalse(self._def == None) # noqa: E711
+ self.assertFalse(self._def == None) # noqa: E711
def test_ne_none(self):
# Ignore this lint error:
# E711 comparison to None should be 'if cond is not None:'
# since this is what we want to test (even though not good practice).
- self.assertTrue(self._def != None) # noqa: E711
+ self.assertTrue(self._def != None) # noqa: E711
# This is a list of binary operators used for
# inject testing methods for each binary operation
for name, binop in _BINOPS:
- setattr(cls, test_binop_name('invalid_unknown'), partialmethod(_TestNumericField._test_binop_invalid_unknown, op=binop))
- setattr(cls, test_binop_name('invalid_none'), partialmethod(_TestNumericField._test_binop_invalid_none, op=binop))
- setattr(cls, test_binop_name('type_true'), partialmethod(_TestNumericField._test_binop_type_true, op=binop))
- setattr(cls, test_binop_name('type_pos_int'), partialmethod(_TestNumericField._test_binop_type_pos_int, op=binop))
- setattr(cls, test_binop_name('type_pos_vint'), partialmethod(_TestNumericField._test_binop_type_pos_vint, op=binop))
- setattr(cls, test_binop_name('value_true'), partialmethod(_TestNumericField._test_binop_value_true, op=binop))
- setattr(cls, test_binop_name('value_pos_int'), partialmethod(_TestNumericField._test_binop_value_pos_int, op=binop))
- setattr(cls, test_binop_name('value_pos_vint'), partialmethod(_TestNumericField._test_binop_value_pos_vint, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_true'), partialmethod(_TestNumericField._test_binop_lhs_addr_same_true, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_pos_int'), partialmethod(_TestNumericField._test_binop_lhs_addr_same_pos_int, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_pos_vint'), partialmethod(_TestNumericField._test_binop_lhs_addr_same_pos_vint, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_true'), partialmethod(_TestNumericField._test_binop_lhs_value_same_true, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_pos_int'), partialmethod(_TestNumericField._test_binop_lhs_value_same_pos_int, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_pos_vint'), partialmethod(_TestNumericField._test_binop_lhs_value_same_pos_vint, op=binop))
- setattr(cls, test_binop_name('type_neg_int'), partialmethod(_TestNumericField._test_binop_type_neg_int, op=binop))
- setattr(cls, test_binop_name('type_neg_vint'), partialmethod(_TestNumericField._test_binop_type_neg_vint, op=binop))
- setattr(cls, test_binop_name('value_neg_int'), partialmethod(_TestNumericField._test_binop_value_neg_int, op=binop))
- setattr(cls, test_binop_name('value_neg_vint'), partialmethod(_TestNumericField._test_binop_value_neg_vint, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_neg_int'), partialmethod(_TestNumericField._test_binop_lhs_addr_same_neg_int, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_neg_vint'), partialmethod(_TestNumericField._test_binop_lhs_addr_same_neg_vint, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_neg_int'), partialmethod(_TestNumericField._test_binop_lhs_value_same_neg_int, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_neg_vint'), partialmethod(_TestNumericField._test_binop_lhs_value_same_neg_vint, op=binop))
- setattr(cls, test_binop_name('type_false'), partialmethod(_TestNumericField._test_binop_type_false, op=binop))
- setattr(cls, test_binop_name('type_zero_int'), partialmethod(_TestNumericField._test_binop_type_zero_int, op=binop))
- setattr(cls, test_binop_name('type_zero_vint'), partialmethod(_TestNumericField._test_binop_type_zero_vint, op=binop))
- setattr(cls, test_binop_name('value_false'), partialmethod(_TestNumericField._test_binop_value_false, op=binop))
- setattr(cls, test_binop_name('value_zero_int'), partialmethod(_TestNumericField._test_binop_value_zero_int, op=binop))
- setattr(cls, test_binop_name('value_zero_vint'), partialmethod(_TestNumericField._test_binop_value_zero_vint, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_false'), partialmethod(_TestNumericField._test_binop_lhs_addr_same_false, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_zero_int'), partialmethod(_TestNumericField._test_binop_lhs_addr_same_zero_int, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_zero_vint'), partialmethod(_TestNumericField._test_binop_lhs_addr_same_zero_vint, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_false'), partialmethod(_TestNumericField._test_binop_lhs_value_same_false, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_zero_int'), partialmethod(_TestNumericField._test_binop_lhs_value_same_zero_int, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_zero_vint'), partialmethod(_TestNumericField._test_binop_lhs_value_same_zero_vint, op=binop))
- setattr(cls, test_binop_name('type_pos_float'), partialmethod(_TestNumericField._test_binop_type_pos_float, op=binop))
- setattr(cls, test_binop_name('type_neg_float'), partialmethod(_TestNumericField._test_binop_type_neg_float, op=binop))
- setattr(cls, test_binop_name('type_pos_vfloat'), partialmethod(_TestNumericField._test_binop_type_pos_vfloat, op=binop))
- setattr(cls, test_binop_name('type_neg_vfloat'), partialmethod(_TestNumericField._test_binop_type_neg_vfloat, op=binop))
- setattr(cls, test_binop_name('value_pos_float'), partialmethod(_TestNumericField._test_binop_value_pos_float, op=binop))
- setattr(cls, test_binop_name('value_neg_float'), partialmethod(_TestNumericField._test_binop_value_neg_float, op=binop))
- setattr(cls, test_binop_name('value_pos_vfloat'), partialmethod(_TestNumericField._test_binop_value_pos_vfloat, op=binop))
- setattr(cls, test_binop_name('value_neg_vfloat'), partialmethod(_TestNumericField._test_binop_value_neg_vfloat, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_pos_float'), partialmethod(_TestNumericField._test_binop_lhs_addr_same_pos_float, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_neg_float'), partialmethod(_TestNumericField._test_binop_lhs_addr_same_neg_float, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_pos_vfloat'), partialmethod(_TestNumericField._test_binop_lhs_addr_same_pos_vfloat, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_neg_vfloat'), partialmethod(_TestNumericField._test_binop_lhs_addr_same_neg_vfloat, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_pos_float'), partialmethod(_TestNumericField._test_binop_lhs_value_same_pos_float, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_neg_float'), partialmethod(_TestNumericField._test_binop_lhs_value_same_neg_float, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_pos_vfloat'), partialmethod(_TestNumericField._test_binop_lhs_value_same_pos_vfloat, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_neg_vfloat'), partialmethod(_TestNumericField._test_binop_lhs_value_same_neg_vfloat, op=binop))
- setattr(cls, test_binop_name('type_zero_float'), partialmethod(_TestNumericField._test_binop_type_zero_float, op=binop))
- setattr(cls, test_binop_name('type_zero_vfloat'), partialmethod(_TestNumericField._test_binop_type_zero_vfloat, op=binop))
- setattr(cls, test_binop_name('value_zero_float'), partialmethod(_TestNumericField._test_binop_value_zero_float, op=binop))
- setattr(cls, test_binop_name('value_zero_vfloat'), partialmethod(_TestNumericField._test_binop_value_zero_vfloat, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_zero_float'), partialmethod(_TestNumericField._test_binop_lhs_addr_same_zero_float, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_zero_vfloat'), partialmethod(_TestNumericField._test_binop_lhs_addr_same_zero_vfloat, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_zero_float'), partialmethod(_TestNumericField._test_binop_lhs_value_same_zero_float, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_zero_vfloat'), partialmethod(_TestNumericField._test_binop_lhs_value_same_zero_vfloat, op=binop))
- setattr(cls, test_binop_name('type_complex'), partialmethod(_TestNumericField._test_binop_type_complex, op=binop))
- setattr(cls, test_binop_name('type_zero_complex'), partialmethod(_TestNumericField._test_binop_type_zero_complex, op=binop))
- setattr(cls, test_binop_name('value_complex'), partialmethod(_TestNumericField._test_binop_value_complex, op=binop))
- setattr(cls, test_binop_name('value_zero_complex'), partialmethod(_TestNumericField._test_binop_value_zero_complex, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_complex'), partialmethod(_TestNumericField._test_binop_lhs_addr_same_complex, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_zero_complex'), partialmethod(_TestNumericField._test_binop_lhs_addr_same_zero_complex, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_complex'), partialmethod(_TestNumericField._test_binop_lhs_value_same_complex, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_zero_complex'), partialmethod(_TestNumericField._test_binop_lhs_value_same_zero_complex, op=binop))
+ setattr(
+ cls,
+ test_binop_name('invalid_unknown'),
+ partialmethod(_TestNumericField._test_binop_invalid_unknown, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('invalid_none'),
+ partialmethod(_TestNumericField._test_binop_invalid_none, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_true'),
+ partialmethod(_TestNumericField._test_binop_type_true, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_pos_int'),
+ partialmethod(_TestNumericField._test_binop_type_pos_int, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_pos_vint'),
+ partialmethod(_TestNumericField._test_binop_type_pos_vint, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_true'),
+ partialmethod(_TestNumericField._test_binop_value_true, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_pos_int'),
+ partialmethod(_TestNumericField._test_binop_value_pos_int, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_pos_vint'),
+ partialmethod(_TestNumericField._test_binop_value_pos_vint, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_true'),
+ partialmethod(_TestNumericField._test_binop_lhs_addr_same_true, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_pos_int'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_addr_same_pos_int, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_pos_vint'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_addr_same_pos_vint, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_true'),
+ partialmethod(_TestNumericField._test_binop_lhs_value_same_true, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_pos_int'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_value_same_pos_int, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_pos_vint'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_value_same_pos_vint, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_neg_int'),
+ partialmethod(_TestNumericField._test_binop_type_neg_int, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_neg_vint'),
+ partialmethod(_TestNumericField._test_binop_type_neg_vint, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_neg_int'),
+ partialmethod(_TestNumericField._test_binop_value_neg_int, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_neg_vint'),
+ partialmethod(_TestNumericField._test_binop_value_neg_vint, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_neg_int'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_addr_same_neg_int, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_neg_vint'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_addr_same_neg_vint, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_neg_int'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_value_same_neg_int, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_neg_vint'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_value_same_neg_vint, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_false'),
+ partialmethod(_TestNumericField._test_binop_type_false, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_zero_int'),
+ partialmethod(_TestNumericField._test_binop_type_zero_int, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_zero_vint'),
+ partialmethod(_TestNumericField._test_binop_type_zero_vint, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_false'),
+ partialmethod(_TestNumericField._test_binop_value_false, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_zero_int'),
+ partialmethod(_TestNumericField._test_binop_value_zero_int, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_zero_vint'),
+ partialmethod(_TestNumericField._test_binop_value_zero_vint, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_false'),
+ partialmethod(_TestNumericField._test_binop_lhs_addr_same_false, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_zero_int'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_addr_same_zero_int, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_zero_vint'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_addr_same_zero_vint, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_false'),
+ partialmethod(_TestNumericField._test_binop_lhs_value_same_false, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_zero_int'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_value_same_zero_int, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_zero_vint'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_value_same_zero_vint, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_pos_float'),
+ partialmethod(_TestNumericField._test_binop_type_pos_float, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_neg_float'),
+ partialmethod(_TestNumericField._test_binop_type_neg_float, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_pos_vfloat'),
+ partialmethod(_TestNumericField._test_binop_type_pos_vfloat, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_neg_vfloat'),
+ partialmethod(_TestNumericField._test_binop_type_neg_vfloat, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_pos_float'),
+ partialmethod(_TestNumericField._test_binop_value_pos_float, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_neg_float'),
+ partialmethod(_TestNumericField._test_binop_value_neg_float, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_pos_vfloat'),
+ partialmethod(_TestNumericField._test_binop_value_pos_vfloat, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_neg_vfloat'),
+ partialmethod(_TestNumericField._test_binop_value_neg_vfloat, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_pos_float'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_addr_same_pos_float, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_neg_float'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_addr_same_neg_float, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_pos_vfloat'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_addr_same_pos_vfloat, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_neg_vfloat'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_addr_same_neg_vfloat, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_pos_float'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_value_same_pos_float, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_neg_float'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_value_same_neg_float, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_pos_vfloat'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_value_same_pos_vfloat, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_neg_vfloat'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_value_same_neg_vfloat, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_zero_float'),
+ partialmethod(_TestNumericField._test_binop_type_zero_float, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_zero_vfloat'),
+ partialmethod(_TestNumericField._test_binop_type_zero_vfloat, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_zero_float'),
+ partialmethod(_TestNumericField._test_binop_value_zero_float, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_zero_vfloat'),
+ partialmethod(_TestNumericField._test_binop_value_zero_vfloat, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_zero_float'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_addr_same_zero_float, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_zero_vfloat'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_addr_same_zero_vfloat, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_zero_float'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_value_same_zero_float, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_zero_vfloat'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_value_same_zero_vfloat, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_complex'),
+ partialmethod(_TestNumericField._test_binop_type_complex, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_zero_complex'),
+ partialmethod(_TestNumericField._test_binop_type_zero_complex, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_complex'),
+ partialmethod(_TestNumericField._test_binop_value_complex, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_zero_complex'),
+ partialmethod(_TestNumericField._test_binop_value_zero_complex, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_complex'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_addr_same_complex, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_zero_complex'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_addr_same_zero_complex, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_complex'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_value_same_complex, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_zero_complex'),
+ partialmethod(
+ _TestNumericField._test_binop_lhs_value_same_zero_complex, op=binop
+ ),
+ )
# inject testing methods for each unary operation
for name, unaryop in _UNARYOPS:
- setattr(cls, test_unaryop_name('type'), partialmethod(_TestNumericField._test_unaryop_type, op=unaryop))
- setattr(cls, test_unaryop_name('value'), partialmethod(_TestNumericField._test_unaryop_value, op=unaryop))
- setattr(cls, test_unaryop_name('addr_same'), partialmethod(_TestNumericField._test_unaryop_addr_same, op=unaryop))
- setattr(cls, test_unaryop_name('value_same'), partialmethod(_TestNumericField._test_unaryop_value_same, op=unaryop))
+ setattr(
+ cls,
+ test_unaryop_name('type'),
+ partialmethod(_TestNumericField._test_unaryop_type, op=unaryop),
+ )
+ setattr(
+ cls,
+ test_unaryop_name('value'),
+ partialmethod(_TestNumericField._test_unaryop_value, op=unaryop),
+ )
+ setattr(
+ cls,
+ test_unaryop_name('addr_same'),
+ partialmethod(_TestNumericField._test_unaryop_addr_same, op=unaryop),
+ )
+ setattr(
+ cls,
+ test_unaryop_name('value_same'),
+ partialmethod(_TestNumericField._test_unaryop_value_same, op=unaryop),
+ )
class _TestIntegerFieldCommon(_TestNumericField):
field = _create_field(self._tc, uint_fc)
# Larger than the IEEE 754 double-precision exact representation of
# integers.
- raw = (2**53) + 1
- field.value = (2**53) + 1
+ raw = (2 ** 53) + 1
+ field.value = (2 ** 53) + 1
self.assertEqual(field, raw)
def test_assign_uint_invalid_neg(self):
fc.add_mapping('something', bt2.SignedIntegerRangeSet([(17, 17)]))
fc.add_mapping('speaker', bt2.SignedIntegerRangeSet([(12, 16)]))
fc.add_mapping('can', bt2.SignedIntegerRangeSet([(18, 2540)]))
- fc.add_mapping('whole range', bt2.SignedIntegerRangeSet([(-(2 ** 31), (2 ** 31) - 1)]))
+ fc.add_mapping(
+ 'whole range', bt2.SignedIntegerRangeSet([(-(2 ** 31), (2 ** 31) - 1)])
+ )
fc.add_mapping('zip', bt2.SignedIntegerRangeSet([(-45, 1001)]))
return fc
# Establish all permutations of the three expected matches since
# the order in which mappings are enumerated is not explicitly part of
# the API.
- for p in itertools.permutations(['whole range', 'something',
- 'zip']):
+ for p in itertools.permutations(['whole range', 'something', 'zip']):
candidate = '{} ({})'.format(self._def_value, ', '.join(p))
if candidate == s:
expected_string_found = True
array_fc = self._tc.create_static_array_field_class(struct_fc, 3)
stream = _create_stream(self._tc, [('array_field', array_fc)])
values = [
- {
- 'an_int': 42,
- 'a_string': 'hello',
- 'another_int': 66
- },
- {
- 'an_int': 1,
- 'a_string': 'goodbye',
- 'another_int': 488
- },
- {
- 'an_int': 156,
- 'a_string': 'or not',
- 'another_int': 4648
- },
+ {'an_int': 42, 'a_string': 'hello', 'another_int': 66},
+ {'an_int': 1, 'a_string': 'goodbye', 'another_int': 488},
+ {'an_int': 156, 'a_string': 'or not', 'another_int': 4648},
]
array = stream.create_packet().context_field['array_field']
def test_str_op(self):
s = str(self._def)
- expected_string = '[{}]'.format(', '.join(
- [repr(v) for v in self._def_value]))
+ expected_string = '[{}]'.format(', '.join([repr(v) for v in self._def_value]))
self.assertEqual(expected_string, s)
'C': 17.5,
'D': 16497,
'E': {},
- 'F': {'F_1': 52}
+ 'F': {'F_1': 52},
}
def _modify_def(self):
self._def.member_at_index(len(self._def_value))
def test_eq(self):
- field = _create_field(self._tc, self._create_fc(self._tc, ))
+ field = _create_field(self._tc, self._create_fc(self._tc))
field['A'] = -1872
field['B'] = 'salut'
field['C'] = 17.5
'C': 17.5,
'D': 16497,
'E': {},
- 'F': {'F_1': 52}
+ 'F': {'F_1': 52},
}
for vkey, vval in self._def.items():
'C': 17.5,
'D': 16497,
'E': {},
- 'F': {'F_1': 52}
+ 'F': {'F_1': 52},
}
self.assertEqual(self._def, orig_values)
struct_fc.append_member(field_class=int_fc, name='an_int')
struct_fc.append_member(field_class=str_fc, name='a_string')
struct_fc.append_member(field_class=another_int_fc, name='another_int')
- values = {
- 'an_int': 42,
- 'a_string': 'hello',
- 'another_int': 66
- }
+ values = {'an_int': 42, 'a_string': 'hello', 'another_int': 66}
struct = _create_field(self._tc, struct_fc)
struct.value = values
self._create_func(field_value_range=0)
def test_create_base(self):
- fc = self._create_func(preferred_display_base=bt2.IntegerDisplayBase.HEXADECIMAL)
+ fc = self._create_func(
+ preferred_display_base=bt2.IntegerDisplayBase.HEXADECIMAL
+ )
self.assertEqual(fc.preferred_display_base, bt2.IntegerDisplayBase.HEXADECIMAL)
def test_create_invalid_base_type(self):
#
# [(lower0, upper0), (lower1, upper1), ...]
+
def enum_mapping_to_set(mapping):
return {(x.lower, x.upper) for x in mapping.ranges}
def test_iadd(self):
self._fc.add_mapping('c', self._ranges1)
- self._fc += [
- ('d', self._ranges2),
- ('e', self._ranges3),
- ]
+ self._fc += [('d', self._ranges2), ('e', self._ranges3)]
self.assertEqual(len(self._fc), 3)
self.assertEqual(self._fc['c'].label, 'c')
self.assertEqual(labels, expected_labels)
-class UnsignedEnumerationFieldClassTestCase(_EnumerationFieldClassTestCase, unittest.TestCase):
+class UnsignedEnumerationFieldClassTestCase(
+ _EnumerationFieldClassTestCase, unittest.TestCase
+):
def _spec_set_up(self):
self._ranges1 = bt2.UnsignedIntegerRangeSet([(1, 4), (18, 47)])
self._ranges2 = bt2.UnsignedIntegerRangeSet([(5, 5)])
self._create_func = self._tc.create_unsigned_enumeration_field_class
-class SignedEnumerationFieldClassTestCase(_EnumerationFieldClassTestCase, unittest.TestCase):
+class SignedEnumerationFieldClassTestCase(
+ _EnumerationFieldClassTestCase, unittest.TestCase
+):
def _spec_set_up(self):
self._ranges1 = bt2.SignedIntegerRangeSet([(-10, -4), (18, 47)])
self._ranges2 = bt2.SignedIntegerRangeSet([(-3, -3)])
self.assertIsNotNone(self._fc)
-class _TestElementContainer():
+class _TestElementContainer:
def setUp(self):
self._tc = get_default_trace_class()
self._fc = self._create_default_fc()
self._append_element_method(self._fc, 'a_float', a_field_class)
self._append_element_method(self._fc, 'b_int', b_field_class)
c_field_class = self._tc.create_string_field_class()
- d_field_class = self._tc.create_signed_enumeration_field_class(field_value_range=32)
+ d_field_class = self._tc.create_signed_enumeration_field_class(
+ field_value_range=32
+ )
e_field_class = self._tc.create_structure_field_class()
self._fc += [
('c_string', c_field_class),
a_fc = self._tc.create_signed_integer_field_class(32)
b_fc = self._tc.create_string_field_class()
c_fc = self._tc.create_real_field_class()
- elements = (
- ('a', a_fc),
- ('b', b_fc),
- ('c', c_fc),
- )
+ elements = (('a', a_fc), ('b', b_fc), ('c', c_fc))
for elem in elements:
self._append_element_method(self._fc, *elem)
self.assertEqual(elem.name, 'a')
def test_at_index_invalid(self):
- self._append_element_method(self._fc, 'c', self._tc.create_signed_integer_field_class(32))
+ self._append_element_method(
+ self._fc, 'c', self._tc.create_signed_integer_field_class(32)
+ )
with self.assertRaises(TypeError):
self._at_index_method(self._fc, 'yes')
def test_at_index_out_of_bounds_after(self):
- self._append_element_method(self._fc, 'c', self._tc.create_signed_integer_field_class(32))
+ self._append_element_method(
+ self._fc, 'c', self._tc.create_signed_integer_field_class(32)
+ )
with self.assertRaises(IndexError):
self._at_index_method(self._fc, len(self._fc))
class StructureFieldClassTestCase(_TestElementContainer, unittest.TestCase):
- _append_element_method = staticmethod(bt2.field_class._StructureFieldClass.append_member)
- _at_index_method = staticmethod(bt2.field_class._StructureFieldClass.member_at_index)
+ _append_element_method = staticmethod(
+ bt2.field_class._StructureFieldClass.append_member
+ )
+ _at_index_method = staticmethod(
+ bt2.field_class._StructureFieldClass.member_at_index
+ )
def _create_default_fc(self):
return self._tc.create_structure_field_class()
-class VariantFieldClassWithoutSelectorTestCase(_TestElementContainer, unittest.TestCase):
- _append_element_method = staticmethod(bt2.field_class._VariantFieldClassWithoutSelector.append_option)
- _at_index_method = staticmethod(bt2.field_class._VariantFieldClassWithoutSelector.option_at_index)
+class VariantFieldClassWithoutSelectorTestCase(
+ _TestElementContainer, unittest.TestCase
+):
+ _append_element_method = staticmethod(
+ bt2.field_class._VariantFieldClassWithoutSelector.append_option
+ )
+ _at_index_method = staticmethod(
+ bt2.field_class._VariantFieldClassWithoutSelector.option_at_index
+ )
def _create_default_fc(self):
return self._tc.create_variant_field_class()
def test_append_element_kwargs(self):
int_field_class = self._tc.create_signed_integer_field_class(32)
- self._fc.append_option(name='int32', field_class=int_field_class,
- ranges=self._ranges1)
+ self._fc.append_option(
+ name='int32', field_class=int_field_class, ranges=self._ranges1
+ )
opt = self._fc['int32']
self.assertEqual(opt.field_class.addr, int_field_class.addr)
self.assertEqual(opt.name, 'int32')
a_field_class = self._tc.create_real_field_class()
self._fc.append_option('a_float', a_field_class, self._ranges1)
c_field_class = self._tc.create_string_field_class()
- d_field_class = self._tc.create_signed_enumeration_field_class(field_value_range=32)
+ d_field_class = self._tc.create_signed_enumeration_field_class(
+ field_value_range=32
+ )
self._fc += [
('c_string', c_field_class, self._ranges2),
('d_enum', d_field_class, self._ranges3),
self.assertEqual(self._fc.option_at_index(1).ranges.addr, self._ranges2.addr)
def test_at_index_invalid(self):
- self._fc.append_option('c', self._tc.create_signed_integer_field_class(32), self._ranges3)
+ self._fc.append_option(
+ 'c', self._tc.create_signed_integer_field_class(32), self._ranges3
+ )
with self.assertRaises(TypeError):
self._fc.option_at_index('yes')
def test_at_index_out_of_bounds_after(self):
- self._fc.append_option('c', self._tc.create_signed_integer_field_class(32), self._ranges3)
+ self._fc.append_option(
+ 'c', self._tc.create_signed_integer_field_class(32), self._ranges3
+ )
with self.assertRaises(IndexError):
self._fc.option_at_index(len(self._fc))
# } inner_struct[2];
# };
self._fc.append_option('a', self._tc.create_real_field_class(), self._ranges1)
- self._fc.append_option('b', self._tc.create_signed_integer_field_class(21), self._ranges2)
- self._fc.append_option('c', self._tc.create_unsigned_integer_field_class(34), self._ranges3)
+ self._fc.append_option(
+ 'b', self._tc.create_signed_integer_field_class(21), self._ranges2
+ )
+ self._fc.append_option(
+ 'c', self._tc.create_unsigned_integer_field_class(34), self._ranges3
+ )
foo_fc = self._tc.create_real_field_class()
bar_fc = self._tc.create_string_field_class()
inner_struct_fc.append_member('baz', baz_fc)
inner_struct_fc.append_member('variant', self._fc)
- inner_struct_array_fc = self._tc.create_static_array_field_class(inner_struct_fc, 2)
+ inner_struct_array_fc = self._tc.create_static_array_field_class(
+ inner_struct_fc, 2
+ )
outer_struct_fc = self._tc.create_structure_field_class()
outer_struct_fc.append_member('foo', foo_fc)
# The path to the selector field is resolved when the sequence is
# actually used, for example in a packet context.
- self._tc.create_stream_class(supports_packets=True,
- packet_context_field_class=outer_struct_fc)
+ self._tc.create_stream_class(
+ supports_packets=True, packet_context_field_class=outer_struct_fc
+ )
def test_selector_field_path_length(self):
self._fill_default_fc_for_field_path_test()
self.assertIsInstance(path_items[0], bt2.field_path._IndexFieldPathItem)
self.assertEqual(path_items[0].index, 1)
- self.assertIsInstance(path_items[1], bt2.field_path._CurrentArrayElementFieldPathItem)
+ self.assertIsInstance(
+ path_items[1], bt2.field_path._CurrentArrayElementFieldPathItem
+ )
self.assertIsInstance(path_items[2], bt2.field_path._IndexFieldPathItem)
self.assertEqual(path_items[2].index, 0)
def test_selector_field_path_root_scope(self):
self._fill_default_fc_for_field_path_test()
- self.assertEqual(self._fc.selector_field_path.root_scope, bt2.field_path.Scope.PACKET_CONTEXT)
+ self.assertEqual(
+ self._fc.selector_field_path.root_scope, bt2.field_path.Scope.PACKET_CONTEXT
+ )
-class VariantFieldClassWithUnsignedSelectorTestCase(_VariantFieldClassWithSelectorTestCase, unittest.TestCase):
+class VariantFieldClassWithUnsignedSelectorTestCase(
+ _VariantFieldClassWithSelectorTestCase, unittest.TestCase
+):
def _spec_set_up(self):
self._ranges1 = bt2.UnsignedIntegerRangeSet([(1, 4), (18, 47)])
self._ranges2 = bt2.UnsignedIntegerRangeSet([(5, 5)])
self._selector_fc = self._tc.create_unsigned_integer_field_class()
-class VariantFieldClassWithSignedSelectorTestCase(_VariantFieldClassWithSelectorTestCase, unittest.TestCase):
+class VariantFieldClassWithSignedSelectorTestCase(
+ _VariantFieldClassWithSelectorTestCase, unittest.TestCase
+):
def _spec_set_up(self):
self._ranges1 = bt2.SignedIntegerRangeSet([(-10, -4), (18, 47)])
self._ranges2 = bt2.SignedIntegerRangeSet([(-3, -3)])
def test_create_invalid_length(self):
with self.assertRaises(ValueError):
- self._tc.create_static_array_field_class(self._tc.create_string_field_class(), -17)
+ self._tc.create_static_array_field_class(
+ self._tc.create_string_field_class(), -17
+ )
def test_create_invalid_length_type(self):
with self.assertRaises(TypeError):
- self._tc.create_static_array_field_class(self._tc.create_string_field_class(), 'the length')
+ self._tc.create_static_array_field_class(
+ self._tc.create_string_field_class(), 'the length'
+ )
class DynamicArrayFieldClassTestCase(unittest.TestCase):
inner_struct_fc.append_member('len', self._len_fc)
inner_struct_fc.append_member('dyn_array', fc)
- inner_struct_array_fc = self._tc.create_static_array_field_class(inner_struct_fc, 2)
+ inner_struct_array_fc = self._tc.create_static_array_field_class(
+ inner_struct_fc, 2
+ )
outer_struct_fc = self._tc.create_structure_field_class()
outer_struct_fc.append_member('foo', foo_fc)
# The path to the length field is resolved when the sequence is
# actually used, for example in a packet context.
- self._tc.create_stream_class(packet_context_field_class=outer_struct_fc,
- supports_packets=True)
+ self._tc.create_stream_class(
+ packet_context_field_class=outer_struct_fc, supports_packets=True
+ )
return fc
self.assertIsInstance(path_items[0], bt2.field_path._IndexFieldPathItem)
self.assertEqual(path_items[0].index, 1)
- self.assertIsInstance(path_items[1], bt2.field_path._CurrentArrayElementFieldPathItem)
+ self.assertIsInstance(
+ path_items[1], bt2.field_path._CurrentArrayElementFieldPathItem
+ )
self.assertIsInstance(path_items[2], bt2.field_path._IndexFieldPathItem)
self.assertEqual(path_items[2].index, 2)
def test_field_path_root_scope(self):
fc = self._create_field_class_for_field_path_test()
- self.assertEqual(fc.length_field_path.root_scope, bt2.field_path.Scope.PACKET_CONTEXT)
+ self.assertEqual(
+ fc.length_field_path.root_scope, bt2.field_path.Scope.PACKET_CONTEXT
+ )
def test_create_invalid_field_class(self):
with self.assertRaises(TypeError):
def test_create_invalid_length_type(self):
with self.assertRaises(TypeError):
- self._tc.create_dynamic_array_field_class(self._tc.create_string_field_class(), 17)
+ self._tc.create_dynamic_array_field_class(
+ self._tc.create_string_field_class(), 17
+ )
if __name__ == "__main__":
self._ec = self._sc.create_event_class(name='salut')
self._my_int_ft = self._tc.create_signed_integer_field_class(32)
payload_ft = self._tc.create_structure_field_class()
- payload_ft += [
- ('my_int', self._my_int_ft),
- ]
+ payload_ft += [('my_int', self._my_int_ft)]
self._ec.payload_field_type = payload_ft
self._stream = self._t.create_stream(self._sc)
self._packet = self._stream.create_packet()
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
+
def _graph_is_configured(self):
pass
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
+
def _graph_is_configured(self):
pass
def _consume(self):
pass
+
def _graph_is_configured(self):
pass
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
+
def _graph_is_configured(self):
pass
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
+
def _graph_is_configured(self):
pass
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
+
def _graph_is_configured(self):
pass
- comp = self._graph.add_component(MySink, 'salut',
- logging_level=bt2.LoggingLevel.DEBUG)
+ comp = self._graph.add_component(
+ MySink, 'salut', logging_level=bt2.LoggingLevel.DEBUG
+ )
self.assertEqual(comp.logging_level, bt2.LoggingLevel.DEBUG)
def test_connect_ports(self):
def __next__(self):
raise bt2.Stop
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
- conn = self._graph.connect_ports(src.output_ports['out'],
- sink.input_ports['in'])
+ conn = self._graph.connect_ports(
+ src.output_ports['out'], sink.input_ports['in']
+ )
self.assertTrue(src.output_ports['out'].is_connected)
self.assertTrue(sink.input_ports['in'].is_connected)
self.assertEqual(src.output_ports['out'].connection.addr, conn.addr)
def __next__(self):
raise bt2.Stop
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
sink = self._graph.add_component(MySink, 'sink')
with self.assertRaises(TypeError):
- conn = self._graph.connect_ports(sink.input_ports['in'],
- src.output_ports['out'])
+ conn = self._graph.connect_ports(
+ sink.input_ports['in'], src.output_ports['out']
+ )
def test_cancel(self):
self.assertFalse(self._graph.is_canceled)
def __next__(self):
return self._create_stream_beginning_message(self._stream)
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
self._at += 1
return msg
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
- conn = self._graph.connect_ports(src.output_ports['out'],
- sink.input_ports['in'])
+ conn = self._graph.connect_ports(
+ src.output_ports['out'], sink.input_ports['in']
+ )
self._graph.run()
def test_run_again(self):
self._at += 1
return msg
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
- conn = self._graph.connect_ports(src.output_ports['out'],
- sink.input_ports['in'])
+ conn = self._graph.connect_ports(
+ src.output_ports['out'], sink.input_ports['in']
+ )
with self.assertRaises(bt2.TryAgain):
self._graph.run()
self._at += 1
return msg
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
- conn = self._graph.connect_ports(src.output_ports['out'],
- sink.input_ports['in'])
+ conn = self._graph.connect_ports(
+ src.output_ports['out'], sink.input_ports['in']
+ )
with self.assertRaises(bt2.Error):
self._graph.run()
def __next__(self):
raise bt2.Stop
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
self._add_output_port('zero')
nonlocal calls
calls.append((port_added_listener, component, port))
- def ports_connected_listener(upstream_component, upstream_port,
- downstream_component, downstream_port):
+ def ports_connected_listener(
+ upstream_component, upstream_port, downstream_component, downstream_port
+ ):
nonlocal calls
- calls.append((ports_connected_listener,
- upstream_component, upstream_port,
- downstream_component, downstream_port))
+ calls.append(
+ (
+ ports_connected_listener,
+ upstream_component,
+ upstream_port,
+ downstream_component,
+ downstream_port,
+ )
+ )
calls = []
self._graph.add_port_added_listener(port_added_listener)
self._graph.add_ports_connected_listener(ports_connected_listener)
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
- self._graph.connect_ports(src.output_ports['out'],
- sink.input_ports['in'])
+ self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
self.assertEqual(len(calls), 5)
def __next__(self):
raise bt2.Stop
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
self._add_output_port('zero')
def __next__(self):
raise bt2.Stop
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
def _graph_is_configured(self):
pass
- def ports_connected_listener(upstream_component, upstream_port,
- downstream_component, downstream_port):
+ def ports_connected_listener(
+ upstream_component, upstream_port, downstream_component, downstream_port
+ ):
raise ValueError('oh noes!')
graph = bt2.Graph()
self.assertIn(self._range3, self._rs)
def test_create_tuples(self):
- rs = self._CLS((
- (self._range1.lower, self._range1.upper),
- (self._range2.lower, self._range2.upper),
- (self._range3.lower, self._range3.upper),
- ))
+ rs = self._CLS(
+ (
+ (self._range1.lower, self._range1.upper),
+ (self._range2.lower, self._range2.upper),
+ (self._range3.lower, self._range3.upper),
+ )
+ )
self.assertEqual(len(rs), 3)
self.assertIn(self._range1, rs)
self.assertIn(self._range2, rs)
class AllMessagesTestCase(unittest.TestCase):
def setUp(self):
-
class MyIter(bt2._UserMessageIterator):
def __init__(self, self_port_output):
self._at = 0
- self._with_stream_msgs_clock_snapshots = self_port_output.user_data.get('with_stream_msgs_clock_snapshots', False)
+ self._with_stream_msgs_clock_snapshots = self_port_output.user_data.get(
+ 'with_stream_msgs_clock_snapshots', False
+ )
def __next__(self):
if test_obj._clock_class:
if self._at == 0:
if self._with_stream_msgs_clock_snapshots:
- msg = self._create_stream_beginning_message(test_obj._stream, default_clock_snapshot=self._at)
+ msg = self._create_stream_beginning_message(
+ test_obj._stream, default_clock_snapshot=self._at
+ )
else:
- msg = self._create_stream_beginning_message(test_obj._stream)
+ msg = self._create_stream_beginning_message(
+ test_obj._stream
+ )
elif self._at == 1:
- msg = self._create_packet_beginning_message(test_obj._packet, self._at)
+ msg = self._create_packet_beginning_message(
+ test_obj._packet, self._at
+ )
elif self._at == 2:
- msg = self._create_event_message(test_obj._event_class, test_obj._packet, self._at)
+ msg = self._create_event_message(
+ test_obj._event_class, test_obj._packet, self._at
+ )
elif self._at == 3:
- msg = self._create_message_iterator_inactivity_message(test_obj._clock_class, self._at)
+ msg = self._create_message_iterator_inactivity_message(
+ test_obj._clock_class, self._at
+ )
elif self._at == 4:
- msg = self._create_discarded_events_message(test_obj._stream, 890, self._at, self._at)
+ msg = self._create_discarded_events_message(
+ test_obj._stream, 890, self._at, self._at
+ )
elif self._at == 5:
- msg = self._create_packet_end_message(test_obj._packet, self._at)
+ msg = self._create_packet_end_message(
+ test_obj._packet, self._at
+ )
elif self._at == 6:
- msg = self._create_discarded_packets_message(test_obj._stream, 678, self._at, self._at)
+ msg = self._create_discarded_packets_message(
+ test_obj._stream, 678, self._at, self._at
+ )
elif self._at == 7:
if self._with_stream_msgs_clock_snapshots:
- msg = self._create_stream_end_message(test_obj._stream, default_clock_snapshot=self._at)
+ msg = self._create_stream_end_message(
+ test_obj._stream, default_clock_snapshot=self._at
+ )
else:
msg = self._create_stream_end_message(test_obj._stream)
elif self._at >= 8:
elif self._at == 1:
msg = self._create_packet_beginning_message(test_obj._packet)
elif self._at == 2:
- msg = self._create_event_message(test_obj._event_class, test_obj._packet)
+ msg = self._create_event_message(
+ test_obj._event_class, test_obj._packet
+ )
elif self._at == 3:
- msg = self._create_discarded_events_message(test_obj._stream, 890)
+ msg = self._create_discarded_events_message(
+ test_obj._stream, 890
+ )
elif self._at == 4:
msg = self._create_packet_end_message(test_obj._packet)
elif self._at == 5:
- msg = self._create_discarded_packets_message(test_obj._stream, 678)
+ msg = self._create_discarded_packets_message(
+ test_obj._stream, 678
+ )
elif self._at == 6:
msg = self._create_stream_end_message(test_obj._stream)
elif self._at >= 7:
else:
cc = None
- sc = tc.create_stream_class(default_clock_class=cc,
- supports_packets=True,
- packets_have_beginning_default_clock_snapshot=with_cc,
- packets_have_end_default_clock_snapshot=with_cc,
- supports_discarded_events=True,
- discarded_events_have_default_clock_snapshots=with_cc,
- supports_discarded_packets=True,
- discarded_packets_have_default_clock_snapshots=with_cc)
+ sc = tc.create_stream_class(
+ default_clock_class=cc,
+ supports_packets=True,
+ packets_have_beginning_default_clock_snapshot=with_cc,
+ packets_have_end_default_clock_snapshot=with_cc,
+ supports_discarded_events=True,
+ discarded_events_have_default_clock_snapshots=with_cc,
+ supports_discarded_packets=True,
+ discarded_packets_have_default_clock_snapshots=with_cc,
+ )
# Create payload field class
my_int_fc = tc.create_signed_integer_field_class(32)
payload_fc = tc.create_structure_field_class()
- payload_fc += [
- ('my_int', my_int_fc),
- ]
+ payload_fc += [('my_int', my_int_fc)]
ec = sc.create_event_class(name='salut', payload_field_class=payload_fc)
def test_all_msg_with_cc(self):
params = {'with_cc': True}
self._src_comp = self._graph.add_component(self._src, 'my_source', params)
- self._msg_iter = self._graph.create_output_port_message_iterator(self._src_comp.output_ports['out'])
+ self._msg_iter = self._graph.create_output_port_message_iterator(
+ self._src_comp.output_ports['out']
+ )
for i, msg in enumerate(self._msg_iter):
if i == 0:
self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
self.assertEqual(msg.stream.addr, self._stream.addr)
- self.assertIsInstance(msg.default_clock_snapshot, bt2.clock_snapshot._UnknownClockSnapshot)
+ self.assertIsInstance(
+ msg.default_clock_snapshot, bt2.clock_snapshot._UnknownClockSnapshot
+ )
elif i == 1:
self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
self.assertEqual(msg.packet.addr, self._packet.addr)
self.assertEqual(msg.event.cls.addr, self._event_class.addr)
self.assertEqual(msg.default_clock_snapshot.value, i)
elif i == 3:
- self.assertIsInstance(msg, bt2.message._MessageIteratorInactivityMessage)
+ self.assertIsInstance(
+ msg, bt2.message._MessageIteratorInactivityMessage
+ )
self.assertEqual(msg.default_clock_snapshot.value, i)
elif i == 4:
self.assertIsInstance(msg, bt2.message._DiscardedEventsMessage)
self.assertEqual(msg.stream.addr, self._stream.addr)
self.assertEqual(msg.count, 890)
- self.assertEqual(msg.stream.cls.default_clock_class.addr, self._clock_class.addr)
+ self.assertEqual(
+ msg.stream.cls.default_clock_class.addr, self._clock_class.addr
+ )
self.assertEqual(msg.beginning_default_clock_snapshot.value, i)
self.assertEqual(msg.end_default_clock_snapshot.value, i)
elif i == 5:
self.assertIsInstance(msg, bt2.message._DiscardedPacketsMessage)
self.assertEqual(msg.stream.addr, self._stream.addr)
self.assertEqual(msg.count, 678)
- self.assertEqual(msg.stream.cls.default_clock_class.addr, self._clock_class.addr)
+ self.assertEqual(
+ msg.stream.cls.default_clock_class.addr, self._clock_class.addr
+ )
self.assertEqual(msg.beginning_default_clock_snapshot.value, i)
self.assertEqual(msg.end_default_clock_snapshot.value, i)
elif i == 7:
self.assertIsInstance(msg, bt2.message._StreamEndMessage)
self.assertEqual(msg.stream.addr, self._stream.addr)
- self.assertIsInstance(msg.default_clock_snapshot, bt2.clock_snapshot._UnknownClockSnapshot)
+ self.assertIsInstance(
+ msg.default_clock_snapshot, bt2.clock_snapshot._UnknownClockSnapshot
+ )
else:
raise Exception
def test_all_msg_without_cc(self):
params = {'with_cc': False}
self._src_comp = self._graph.add_component(self._src, 'my_source', params)
- self._msg_iter = self._graph.create_output_port_message_iterator(self._src_comp.output_ports['out'])
+ self._msg_iter = self._graph.create_output_port_message_iterator(
+ self._src_comp.output_ports['out']
+ )
for i, msg in enumerate(self._msg_iter):
if i == 0:
raise Exception
def test_msg_stream_with_clock_snapshots(self):
- params = {
- 'with_cc': True,
- 'with_stream_msgs_clock_snapshots': True,
- }
+ params = {'with_cc': True, 'with_stream_msgs_clock_snapshots': True}
self._src_comp = self._graph.add_component(self._src, 'my_source', params)
- self._msg_iter = self._graph.create_output_port_message_iterator(self._src_comp.output_ports['out'])
+ self._msg_iter = self._graph.create_output_port_message_iterator(
+ self._src_comp.output_ports['out']
+ )
msgs = list(self._msg_iter)
msg_stream_beg = msgs[0]
-
#
# Copyright (C) 2019 EfficiOS Inc.
#
graph = bt2.Graph()
src_comp = graph.add_component(src_comp_cls, 'src')
sink_comp = graph.add_component(MySink, 'sink')
- graph.connect_ports(src_comp.output_ports['out'],
- sink_comp.input_ports['in'])
+ graph.connect_ports(src_comp.output_ports['out'], sink_comp.input_ports['in'])
return graph
def test_init(self):
initialized = True
the_output_port_from_iter = self_port_output
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(self, params):
nonlocal the_output_port_from_source
the_output_port_from_source = self._add_output_port('out', 'user data')
graph = self._create_graph(MySource)
graph.run()
self.assertTrue(initialized)
- self.assertEqual(the_output_port_from_source.addr, the_output_port_from_iter.addr)
+ self.assertEqual(
+ the_output_port_from_source.addr, the_output_port_from_iter.addr
+ )
self.assertEqual(the_output_port_from_iter.user_data, 'user data')
def test_finalize(self):
nonlocal finalized
finalized = True
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
nonlocal salut
salut = self._component._salut
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
self._salut = 23
nonlocal addr
addr = self.addr
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
else:
raise StopIteration
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MySourceIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MySourceIter):
def __init__(self, params):
tc = self._create_trace_class()
sc = tc.create_stream_class(supports_packets=True)
input_port = self._add_input_port('in')
self._add_output_port('out', input_port)
-
graph = bt2.Graph()
src = graph.add_component(MySource, 'src')
flt = graph.add_component(MyFilter, 'flt')
it, MySourceIter = self._setup_seek_beginning_test()
def _seek_beginning_error(self):
- raise ValueError('ouch')
+ raise ValueError('ouch')
MySourceIter._seek_beginning = _seek_beginning_error
it.seek_beginning()
-
class OutputPortMessageIteratorTestCase(unittest.TestCase):
def test_component(self):
class MyIter(bt2._UserMessageIterator):
elif self._at == 6:
msg = self._create_stream_end_message(test_obj._stream)
else:
- msg = self._create_event_message(test_obj._event_class, test_obj._packet)
+ msg = self._create_event_message(
+ test_obj._event_class, test_obj._packet
+ )
msg.event.payload_field['my_int'] = self._at * 3
self._at += 1
return msg
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
# Create payload field class
my_int_ft = trace_class.create_signed_integer_field_class(32)
payload_ft = trace_class.create_structure_field_class()
- payload_ft += [
- ('my_int', my_int_ft),
- ]
+ payload_ft += [('my_int', my_int_ft)]
- event_class = stream_class.create_event_class(name='salut', payload_field_class=payload_ft)
+ event_class = stream_class.create_event_class(
+ name='salut', payload_field_class=payload_ft
+ )
trace = trace_class()
stream = trace.create_stream(stream_class)
field = msg.event.payload_field['my_int']
self.assertEqual(field, at * 3)
+
if __name__ == '__main__':
unittest.main()
]
# stream class
- sc = tc.create_stream_class(default_clock_class=clock_class,
- event_common_context_field_class=sec,
- packet_context_field_class=pc,
- supports_packets=True)
+ sc = tc.create_stream_class(
+ default_clock_class=clock_class,
+ event_common_context_field_class=sec,
+ packet_context_field_class=pc,
+ supports_packets=True,
+ )
# event context
ec = tc.create_structure_field_class()
class FindPluginsTestCase(unittest.TestCase):
def test_find_nonexistent_dir(self):
with self.assertRaises(bt2.Error):
- bt2.find_plugins('/this/does/not/exist/246703df-cb85-46d5-8406-5e8dc4a88b41')
+ bt2.find_plugins(
+ '/this/does/not/exist/246703df-cb85-46d5-8406-5e8dc4a88b41'
+ )
def test_find_none_existing_dir(self):
plugins = bt2.find_plugins(_TEST_PLUGIN_PLUGINS_PATH, recurse=False)
self.assertTrue(len(pset) >= 3)
def test_find_file(self):
- extension = _TEST_PLUGIN_PLUGIN_EXTENSION_BY_OS.get(os.environ['BT_OS_TYPE'], 'so')
+ extension = _TEST_PLUGIN_PLUGIN_EXTENSION_BY_OS.get(
+ os.environ['BT_OS_TYPE'], 'so'
+ )
plugin_name = 'babeltrace-plugin-utils.{}'.format(extension)
path = os.path.join(_TEST_PLUGIN_PLUGINS_PATH, 'utils', '.libs', plugin_name)
pset = bt2.find_plugins(path)
class FindPluginTestCase(unittest.TestCase):
def test_find_none(self):
- plugin = bt2.find_plugin('this-does-not-exist-246703df-cb85-46d5-8406-5e8dc4a88b41')
+ plugin = bt2.find_plugin(
+ 'this-does-not-exist-246703df-cb85-46d5-8406-5e8dc4a88b41'
+ )
self.assertIsNone(plugin)
def test_find_existing(self):
def __next__(self):
raise bt2.Stop
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
port = comp_self._add_output_port('out')
self.assertEqual(port.name, 'out')
def __next__(self):
raise bt2.Stop
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
port = comp_self._add_output_port('out')
self.assertEqual(port.name, 'out')
def __next__(self):
raise bt2.Stop
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
port = comp_self._add_input_port('in')
self.assertEqual(port.name, 'in')
def __next__(self):
raise bt2.Stop
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
def __next__(self):
raise bt2.Stop
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
def __next__(self):
raise bt2.Stop
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
port1 = comp_self._add_input_port('clear')
port2 = comp_self._add_input_port('print')
def __next__(self):
raise bt2.Stop
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
def __next__(self):
raise bt2.Stop
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
def __next__(self):
raise bt2.Stop
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
def __next__(self):
raise bt2.Stop
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
def __next__(self):
raise bt2.Stop
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
def __next__(self):
raise bt2.Stop
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
def __next__(self):
raise bt2.Stop
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
def __next__(self):
raise bt2.Stop
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
def __next__(self):
raise bt2.Stop
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
port1 = comp_self._add_input_port('clear')
port2 = comp_self._add_input_port('print')
port2 = None
port3 = None
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
nonlocal port1, port2, port3
port1 = comp_self._add_output_port('clear')
port2 = None
port3 = None
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
nonlocal port1, port2, port3
port1 = comp_self._add_output_port('clear')
port2 = None
port3 = None
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
nonlocal port1, port2, port3
port1 = comp_self._add_input_port('clear')
def __next__(self):
raise bt2.Stop
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
def __next__(self):
raise bt2.Stop
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
def __next__(self):
raise bt2.Stop
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
def __next__(self):
raise bt2.Stop
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
def __next__(self):
raise bt2.Stop
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
def __next__(self):
raise bt2.Stop
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
port2 = None
port3 = None
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
nonlocal port1, port2, port3
port1 = comp_self._add_output_port('clear')
port2 = None
port3 = None
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
nonlocal port1, port2, port3
port1 = comp_self._add_output_port('clear')
port2 = None
port3 = None
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
nonlocal port1, port2, port3
port1 = comp_self._add_input_port('clear')
def __next__(self):
raise bt2.Stop
- class MySource(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MySource(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
nonlocal user_datas
def __next__(self):
raise bt2.Stop
- class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
nonlocal user_datas
user_datas = []
comp = self._create_comp(MyFilter)
- self.assertEqual(user_datas,
- [None, 'user data string', None, {'user data': 'dict'}])
+ self.assertEqual(
+ user_datas, [None, 'user data string', None, {'user data': 'dict'}]
+ )
def test_sink_self_port_user_data(self):
class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
- class MySink(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+ class MySink(bt2._UserFilterComponent, message_iterator_class=MyIter):
def __init__(comp_self, params):
nonlocal user_datas
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_params
query_params = params
- return {
- 'null': None,
- 'bt2': 'BT2',
- }
+ return {'null': None, 'bt2': 'BT2'}
query_params = None
params = {
'array': ['coucou', 23, None],
- 'other_map': {
- 'yes': 'yeah',
- '19': 19,
- 'minus 1.5': -1.5,
- },
+ 'other_map': {'yes': 'yeah', '19': 19, 'minus 1.5': -1.5},
'null': None,
}
res = bt2.QueryExecutor().query(MySink, 'obj', params)
self.assertEqual(query_params, params)
- self.assertEqual(res, {
- 'null': None,
- 'bt2': 'BT2',
- })
+ self.assertEqual(res, {'null': None, 'bt2': 'BT2'})
del query_params
def test_query_params_none(self):
query_log_level = log_level
query_log_level = None
- res = bt2.QueryExecutor().query(MySink, 'obj', None,
- bt2.LoggingLevel.INFO)
+ res = bt2.QueryExecutor().query(MySink, 'obj', None, bt2.LoggingLevel.INFO)
self.assertEqual(query_log_level, bt2.LoggingLevel.INFO)
del query_log_level
def test_create_packet_context_field_class(self):
fc = self._tc.create_structure_field_class()
- sc = self._tc.create_stream_class(packet_context_field_class=fc,
- supports_packets=True)
+ sc = self._tc.create_stream_class(
+ packet_context_field_class=fc, supports_packets=True
+ )
self.assertEqual(sc.packet_context_field_class, fc)
def test_create_invalid_packet_context_field_class(self):
sc.create_event_class()
def test_supports_packets_without_cs(self):
- sc = self._tc.create_stream_class(default_clock_class=self._cc,
- supports_packets=True)
+ sc = self._tc.create_stream_class(
+ default_clock_class=self._cc, supports_packets=True
+ )
self.assertTrue(sc.supports_packets)
self.assertFalse(sc.packets_have_beginning_default_clock_snapshot)
self.assertFalse(sc.packets_have_end_default_clock_snapshot)
def test_supports_packets_with_begin_cs(self):
- sc = self._tc.create_stream_class(default_clock_class=self._cc,
- supports_packets=True,
- packets_have_beginning_default_clock_snapshot=True)
+ sc = self._tc.create_stream_class(
+ default_clock_class=self._cc,
+ supports_packets=True,
+ packets_have_beginning_default_clock_snapshot=True,
+ )
self.assertTrue(sc.supports_packets)
self.assertTrue(sc.packets_have_beginning_default_clock_snapshot)
self.assertFalse(sc.packets_have_end_default_clock_snapshot)
def test_supports_packets_with_end_cs(self):
- sc = self._tc.create_stream_class(default_clock_class=self._cc,
- supports_packets=True,
- packets_have_end_default_clock_snapshot=True)
+ sc = self._tc.create_stream_class(
+ default_clock_class=self._cc,
+ supports_packets=True,
+ packets_have_end_default_clock_snapshot=True,
+ )
self.assertTrue(sc.supports_packets)
self.assertFalse(sc.packets_have_beginning_default_clock_snapshot)
self.assertTrue(sc.packets_have_end_default_clock_snapshot)
def test_supports_packets_raises_type_error(self):
with self.assertRaises(TypeError):
- sc = self._tc.create_stream_class(default_clock_class=self._cc,
- supports_packets=23)
+ sc = self._tc.create_stream_class(
+ default_clock_class=self._cc, supports_packets=23
+ )
def test_packets_have_begin_default_cs_raises_type_error(self):
with self.assertRaises(TypeError):
- sc = self._tc.create_stream_class(default_clock_class=self._cc,
- packets_have_beginning_default_clock_snapshot=23)
+ sc = self._tc.create_stream_class(
+ default_clock_class=self._cc,
+ packets_have_beginning_default_clock_snapshot=23,
+ )
def test_packets_have_end_default_cs_raises_type_error(self):
with self.assertRaises(TypeError):
- sc = self._tc.create_stream_class(default_clock_class=self._cc,
- packets_have_end_default_clock_snapshot=23)
+ sc = self._tc.create_stream_class(
+ default_clock_class=self._cc, packets_have_end_default_clock_snapshot=23
+ )
def test_does_not_support_packets_raises_with_begin_cs(self):
with self.assertRaises(ValueError):
- sc = self._tc.create_stream_class(default_clock_class=self._cc,
- packets_have_beginning_default_clock_snapshot=True)
+ sc = self._tc.create_stream_class(
+ default_clock_class=self._cc,
+ packets_have_beginning_default_clock_snapshot=True,
+ )
def test_does_not_support_packets_raises_with_end_cs(self):
with self.assertRaises(ValueError):
- sc = self._tc.create_stream_class(default_clock_class=self._cc,
- packets_have_end_default_clock_snapshot=True)
+ sc = self._tc.create_stream_class(
+ default_clock_class=self._cc,
+ packets_have_end_default_clock_snapshot=True,
+ )
def test_supports_discarded_events_without_cs(self):
- sc = self._tc.create_stream_class(default_clock_class=self._cc,
- supports_discarded_events=True)
+ sc = self._tc.create_stream_class(
+ default_clock_class=self._cc, supports_discarded_events=True
+ )
self.assertTrue(sc.supports_discarded_events)
self.assertFalse(sc.discarded_events_have_default_clock_snapshots)
def test_supports_discarded_events_with_cs(self):
- sc = self._tc.create_stream_class(default_clock_class=self._cc,
- supports_discarded_events=True,
- discarded_events_have_default_clock_snapshots=True)
+ sc = self._tc.create_stream_class(
+ default_clock_class=self._cc,
+ supports_discarded_events=True,
+ discarded_events_have_default_clock_snapshots=True,
+ )
self.assertTrue(sc.supports_discarded_events)
self.assertTrue(sc.discarded_events_have_default_clock_snapshots)
def test_supports_discarded_events_raises_type_error(self):
with self.assertRaises(TypeError):
- sc = self._tc.create_stream_class(default_clock_class=self._cc,
- supports_discarded_events=23)
+ sc = self._tc.create_stream_class(
+ default_clock_class=self._cc, supports_discarded_events=23
+ )
def test_discarded_events_have_default_cs_raises_type_error(self):
with self.assertRaises(TypeError):
- sc = self._tc.create_stream_class(default_clock_class=self._cc,
- discarded_events_have_default_clock_snapshots=23)
+ sc = self._tc.create_stream_class(
+ default_clock_class=self._cc,
+ discarded_events_have_default_clock_snapshots=23,
+ )
def test_does_not_support_discarded_events_raises_with_cs(self):
with self.assertRaises(ValueError):
- sc = self._tc.create_stream_class(default_clock_class=self._cc,
- discarded_events_have_default_clock_snapshots=True)
+ sc = self._tc.create_stream_class(
+ default_clock_class=self._cc,
+ discarded_events_have_default_clock_snapshots=True,
+ )
def test_supports_discarded_packets_without_cs(self):
- sc = self._tc.create_stream_class(default_clock_class=self._cc,
- supports_discarded_packets=True,
- supports_packets=True)
+ sc = self._tc.create_stream_class(
+ default_clock_class=self._cc,
+ supports_discarded_packets=True,
+ supports_packets=True,
+ )
self.assertTrue(sc.supports_discarded_packets)
self.assertFalse(sc.discarded_packets_have_default_clock_snapshots)
def test_supports_discarded_packets_with_cs(self):
- sc = self._tc.create_stream_class(default_clock_class=self._cc,
- supports_discarded_packets=True,
- discarded_packets_have_default_clock_snapshots=True,
- supports_packets=True)
+ sc = self._tc.create_stream_class(
+ default_clock_class=self._cc,
+ supports_discarded_packets=True,
+ discarded_packets_have_default_clock_snapshots=True,
+ supports_packets=True,
+ )
self.assertTrue(sc.supports_discarded_packets)
self.assertTrue(sc.discarded_packets_have_default_clock_snapshots)
def test_supports_discarded_packets_raises_without_packet_support(self):
with self.assertRaises(ValueError):
- sc = self._tc.create_stream_class(default_clock_class=self._cc,
- supports_discarded_packets=True)
+ sc = self._tc.create_stream_class(
+ default_clock_class=self._cc, supports_discarded_packets=True
+ )
def test_supports_discarded_packets_raises_type_error(self):
with self.assertRaises(TypeError):
- sc = self._tc.create_stream_class(default_clock_class=self._cc,
- supports_discarded_packets=23,
- supports_packets=True)
+ sc = self._tc.create_stream_class(
+ default_clock_class=self._cc,
+ supports_discarded_packets=23,
+ supports_packets=True,
+ )
def test_discarded_packets_have_default_cs_raises_type_error(self):
with self.assertRaises(TypeError):
- sc = self._tc.create_stream_class(default_clock_class=self._cc,
- discarded_packets_have_default_clock_snapshots=23,
- supports_packets=True)
+ sc = self._tc.create_stream_class(
+ default_clock_class=self._cc,
+ discarded_packets_have_default_clock_snapshots=23,
+ supports_packets=True,
+ )
def test_does_not_support_discarded_packets_raises_with_cs(self):
with self.assertRaises(ValueError):
- sc = self._tc.create_stream_class(default_clock_class=self._cc,
- discarded_packets_have_default_clock_snapshots=True,
- supports_packets=True)
+ sc = self._tc.create_stream_class(
+ default_clock_class=self._cc,
+ discarded_packets_have_default_clock_snapshots=True,
+ supports_packets=True,
+ )
def test_trace_class(self):
sc = self._tc.create_stream_class()
class TraceClassTestCase(unittest.TestCase):
-
def test_create_default(self):
def f(comp_self):
return comp_self._create_trace_class()
def test_no_assigns_automatic_stream_class_id(self):
def f(comp_self):
- return comp_self._create_trace_class(assigns_automatic_stream_class_id=False)
+ return comp_self._create_trace_class(
+ assigns_automatic_stream_class_id=False
+ )
tc = run_in_component_init(f)
self.assertFalse(tc.assigns_automatic_stream_class_id)
def test_no_assigns_automatic_stream_class_id_raises(self):
def f(comp_self):
- return comp_self._create_trace_class(assigns_automatic_stream_class_id=False)
+ return comp_self._create_trace_class(
+ assigns_automatic_stream_class_id=False
+ )
tc = run_in_component_init(f)
self.assertFalse(tc.assigns_automatic_stream_class_id)
@staticmethod
def _create_trace_class_with_some_stream_classes():
def f(comp_self):
- return comp_self._create_trace_class(assigns_automatic_stream_class_id=False)
+ return comp_self._create_trace_class(
+ assigns_automatic_stream_class_id=False
+ )
tc = run_in_component_init(f)
sc1 = tc.create_stream_class(id=12)
_BT_CTF_TRACES_PATH = os.environ['BT_CTF_TRACES_PATH']
-_3EVENTS_INTERSECT_TRACE_PATH = os.path.join(_BT_CTF_TRACES_PATH,
- 'intersection',
- '3eventsintersect')
+_3EVENTS_INTERSECT_TRACE_PATH = os.path.join(
+ _BT_CTF_TRACES_PATH, 'intersection', '3eventsintersect'
+)
class ComponentSpecTestCase(unittest.TestCase):
# Return a map, msg type -> number of messages of this type.
+
def _count_msgs_by_type(msgs):
res = {}
def test_iter_custom_filter(self):
src_spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
- flt_spec = bt2.ComponentSpec('utils', 'trimmer', {
- 'end': '13515309.000000075',
- })
+ flt_spec = bt2.ComponentSpec('utils', 'trimmer', {'end': '13515309.000000075'})
msg_iter = bt2.TraceCollectionMessageIterator(src_spec, flt_spec)
hist = _count_msgs_by_type(msg_iter)
self.assertEqual(hist[bt2.message._EventMessage], 5)
def test_iter_intersection(self):
specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
- msg_iter = bt2.TraceCollectionMessageIterator(specs, stream_intersection_mode=True)
+ msg_iter = bt2.TraceCollectionMessageIterator(
+ specs, stream_intersection_mode=True
+ )
msgs = list(msg_iter)
self.assertEqual(len(msgs), 15)
hist = _count_msgs_by_type(msgs)
copy.deepcopy(self._def)
-_COMP_BINOPS = (
- operator.eq,
- operator.ne,
-)
+_COMP_BINOPS = (operator.eq, operator.ne)
# Base class for numeric value test cases.
test_cb(op, 0.0)
def _test_binop_rhs_complex(self, test_cb, op):
- test_cb(op, -23+19j)
+ test_cb(op, -23 + 19j)
def _test_binop_rhs_zero_complex(self, test_cb, op):
test_cb(op, 0j)
# inject testing methods for each binary operation
for name, binop in _BINOPS:
- setattr(cls, test_binop_name('invalid_unknown'), partialmethod(_TestNumericValue._test_binop_invalid_unknown, op=binop))
- setattr(cls, test_binop_name('invalid_none'), partialmethod(_TestNumericValue._test_binop_invalid_none, op=binop))
- setattr(cls, test_binop_name('type_true'), partialmethod(_TestNumericValue._test_binop_type_true, op=binop))
- setattr(cls, test_binop_name('type_pos_int'), partialmethod(_TestNumericValue._test_binop_type_pos_int, op=binop))
- setattr(cls, test_binop_name('type_pos_vint'), partialmethod(_TestNumericValue._test_binop_type_pos_vint, op=binop))
- setattr(cls, test_binop_name('value_true'), partialmethod(_TestNumericValue._test_binop_value_true, op=binop))
- setattr(cls, test_binop_name('value_pos_int'), partialmethod(_TestNumericValue._test_binop_value_pos_int, op=binop))
- setattr(cls, test_binop_name('value_pos_vint'), partialmethod(_TestNumericValue._test_binop_value_pos_vint, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_true'), partialmethod(_TestNumericValue._test_binop_lhs_addr_same_true, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_pos_int'), partialmethod(_TestNumericValue._test_binop_lhs_addr_same_pos_int, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_pos_vint'), partialmethod(_TestNumericValue._test_binop_lhs_addr_same_pos_vint, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_true'), partialmethod(_TestNumericValue._test_binop_lhs_value_same_true, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_pos_int'), partialmethod(_TestNumericValue._test_binop_lhs_value_same_pos_int, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_pos_vint'), partialmethod(_TestNumericValue._test_binop_lhs_value_same_pos_vint, op=binop))
- setattr(cls, test_binop_name('type_neg_int'), partialmethod(_TestNumericValue._test_binop_type_neg_int, op=binop))
- setattr(cls, test_binop_name('type_neg_vint'), partialmethod(_TestNumericValue._test_binop_type_neg_vint, op=binop))
- setattr(cls, test_binop_name('value_neg_int'), partialmethod(_TestNumericValue._test_binop_value_neg_int, op=binop))
- setattr(cls, test_binop_name('value_neg_vint'), partialmethod(_TestNumericValue._test_binop_value_neg_vint, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_neg_int'), partialmethod(_TestNumericValue._test_binop_lhs_addr_same_neg_int, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_neg_vint'), partialmethod(_TestNumericValue._test_binop_lhs_addr_same_neg_vint, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_neg_int'), partialmethod(_TestNumericValue._test_binop_lhs_value_same_neg_int, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_neg_vint'), partialmethod(_TestNumericValue._test_binop_lhs_value_same_neg_vint, op=binop))
- setattr(cls, test_binop_name('type_false'), partialmethod(_TestNumericValue._test_binop_type_false, op=binop))
- setattr(cls, test_binop_name('type_zero_int'), partialmethod(_TestNumericValue._test_binop_type_zero_int, op=binop))
- setattr(cls, test_binop_name('type_zero_vint'), partialmethod(_TestNumericValue._test_binop_type_zero_vint, op=binop))
- setattr(cls, test_binop_name('value_false'), partialmethod(_TestNumericValue._test_binop_value_false, op=binop))
- setattr(cls, test_binop_name('value_zero_int'), partialmethod(_TestNumericValue._test_binop_value_zero_int, op=binop))
- setattr(cls, test_binop_name('value_zero_vint'), partialmethod(_TestNumericValue._test_binop_value_zero_vint, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_false'), partialmethod(_TestNumericValue._test_binop_lhs_addr_same_false, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_zero_int'), partialmethod(_TestNumericValue._test_binop_lhs_addr_same_zero_int, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_zero_vint'), partialmethod(_TestNumericValue._test_binop_lhs_addr_same_zero_vint, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_false'), partialmethod(_TestNumericValue._test_binop_lhs_value_same_false, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_zero_int'), partialmethod(_TestNumericValue._test_binop_lhs_value_same_zero_int, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_zero_vint'), partialmethod(_TestNumericValue._test_binop_lhs_value_same_zero_vint, op=binop))
- setattr(cls, test_binop_name('type_neg_float'), partialmethod(_TestNumericValue._test_binop_type_neg_float, op=binop))
- setattr(cls, test_binop_name('type_neg_vfloat'), partialmethod(_TestNumericValue._test_binop_type_neg_vfloat, op=binop))
- setattr(cls, test_binop_name('value_neg_float'), partialmethod(_TestNumericValue._test_binop_value_neg_float, op=binop))
- setattr(cls, test_binop_name('value_neg_vfloat'), partialmethod(_TestNumericValue._test_binop_value_neg_vfloat, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_neg_float'), partialmethod(_TestNumericValue._test_binop_lhs_addr_same_neg_float, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_neg_vfloat'), partialmethod(_TestNumericValue._test_binop_lhs_addr_same_neg_vfloat, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_neg_float'), partialmethod(_TestNumericValue._test_binop_lhs_value_same_neg_float, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_neg_vfloat'), partialmethod(_TestNumericValue._test_binop_lhs_value_same_neg_vfloat, op=binop))
- setattr(cls, test_binop_name('type_pos_float'), partialmethod(_TestNumericValue._test_binop_type_pos_float, op=binop))
- setattr(cls, test_binop_name('type_pos_vfloat'), partialmethod(_TestNumericValue._test_binop_type_pos_vfloat, op=binop))
- setattr(cls, test_binop_name('value_pos_float'), partialmethod(_TestNumericValue._test_binop_value_pos_float, op=binop))
- setattr(cls, test_binop_name('value_pos_vfloat'), partialmethod(_TestNumericValue._test_binop_value_pos_vfloat, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_pos_float'), partialmethod(_TestNumericValue._test_binop_lhs_addr_same_pos_float, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_pos_vfloat'), partialmethod(_TestNumericValue._test_binop_lhs_addr_same_pos_vfloat, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_pos_float'), partialmethod(_TestNumericValue._test_binop_lhs_value_same_pos_float, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_pos_vfloat'), partialmethod(_TestNumericValue._test_binop_lhs_value_same_pos_vfloat, op=binop))
- setattr(cls, test_binop_name('type_zero_float'), partialmethod(_TestNumericValue._test_binop_type_zero_float, op=binop))
- setattr(cls, test_binop_name('type_zero_vfloat'), partialmethod(_TestNumericValue._test_binop_type_zero_vfloat, op=binop))
- setattr(cls, test_binop_name('value_zero_float'), partialmethod(_TestNumericValue._test_binop_value_zero_float, op=binop))
- setattr(cls, test_binop_name('value_zero_vfloat'), partialmethod(_TestNumericValue._test_binop_value_zero_vfloat, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_zero_float'), partialmethod(_TestNumericValue._test_binop_lhs_addr_same_zero_float, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_zero_vfloat'), partialmethod(_TestNumericValue._test_binop_lhs_addr_same_zero_vfloat, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_zero_float'), partialmethod(_TestNumericValue._test_binop_lhs_value_same_zero_float, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_zero_vfloat'), partialmethod(_TestNumericValue._test_binop_lhs_value_same_zero_vfloat, op=binop))
- setattr(cls, test_binop_name('type_complex'), partialmethod(_TestNumericValue._test_binop_type_complex, op=binop))
- setattr(cls, test_binop_name('type_zero_complex'), partialmethod(_TestNumericValue._test_binop_type_zero_complex, op=binop))
- setattr(cls, test_binop_name('value_complex'), partialmethod(_TestNumericValue._test_binop_value_complex, op=binop))
- setattr(cls, test_binop_name('value_zero_complex'), partialmethod(_TestNumericValue._test_binop_value_zero_complex, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_complex'), partialmethod(_TestNumericValue._test_binop_lhs_addr_same_complex, op=binop))
- setattr(cls, test_binop_name('lhs_addr_same_zero_complex'), partialmethod(_TestNumericValue._test_binop_lhs_addr_same_zero_complex, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_complex'), partialmethod(_TestNumericValue._test_binop_lhs_value_same_complex, op=binop))
- setattr(cls, test_binop_name('lhs_value_same_zero_complex'), partialmethod(_TestNumericValue._test_binop_lhs_value_same_zero_complex, op=binop))
+ setattr(
+ cls,
+ test_binop_name('invalid_unknown'),
+ partialmethod(_TestNumericValue._test_binop_invalid_unknown, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('invalid_none'),
+ partialmethod(_TestNumericValue._test_binop_invalid_none, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_true'),
+ partialmethod(_TestNumericValue._test_binop_type_true, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_pos_int'),
+ partialmethod(_TestNumericValue._test_binop_type_pos_int, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_pos_vint'),
+ partialmethod(_TestNumericValue._test_binop_type_pos_vint, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_true'),
+ partialmethod(_TestNumericValue._test_binop_value_true, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_pos_int'),
+ partialmethod(_TestNumericValue._test_binop_value_pos_int, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_pos_vint'),
+ partialmethod(_TestNumericValue._test_binop_value_pos_vint, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_true'),
+ partialmethod(_TestNumericValue._test_binop_lhs_addr_same_true, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_pos_int'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_addr_same_pos_int, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_pos_vint'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_addr_same_pos_vint, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_true'),
+ partialmethod(_TestNumericValue._test_binop_lhs_value_same_true, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_pos_int'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_value_same_pos_int, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_pos_vint'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_value_same_pos_vint, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_neg_int'),
+ partialmethod(_TestNumericValue._test_binop_type_neg_int, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_neg_vint'),
+ partialmethod(_TestNumericValue._test_binop_type_neg_vint, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_neg_int'),
+ partialmethod(_TestNumericValue._test_binop_value_neg_int, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_neg_vint'),
+ partialmethod(_TestNumericValue._test_binop_value_neg_vint, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_neg_int'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_addr_same_neg_int, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_neg_vint'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_addr_same_neg_vint, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_neg_int'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_value_same_neg_int, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_neg_vint'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_value_same_neg_vint, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_false'),
+ partialmethod(_TestNumericValue._test_binop_type_false, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_zero_int'),
+ partialmethod(_TestNumericValue._test_binop_type_zero_int, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_zero_vint'),
+ partialmethod(_TestNumericValue._test_binop_type_zero_vint, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_false'),
+ partialmethod(_TestNumericValue._test_binop_value_false, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_zero_int'),
+ partialmethod(_TestNumericValue._test_binop_value_zero_int, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_zero_vint'),
+ partialmethod(_TestNumericValue._test_binop_value_zero_vint, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_false'),
+ partialmethod(_TestNumericValue._test_binop_lhs_addr_same_false, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_zero_int'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_addr_same_zero_int, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_zero_vint'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_addr_same_zero_vint, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_false'),
+ partialmethod(_TestNumericValue._test_binop_lhs_value_same_false, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_zero_int'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_value_same_zero_int, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_zero_vint'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_value_same_zero_vint, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_neg_float'),
+ partialmethod(_TestNumericValue._test_binop_type_neg_float, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_neg_vfloat'),
+ partialmethod(_TestNumericValue._test_binop_type_neg_vfloat, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_neg_float'),
+ partialmethod(_TestNumericValue._test_binop_value_neg_float, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_neg_vfloat'),
+ partialmethod(_TestNumericValue._test_binop_value_neg_vfloat, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_neg_float'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_addr_same_neg_float, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_neg_vfloat'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_addr_same_neg_vfloat, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_neg_float'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_value_same_neg_float, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_neg_vfloat'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_value_same_neg_vfloat, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_pos_float'),
+ partialmethod(_TestNumericValue._test_binop_type_pos_float, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_pos_vfloat'),
+ partialmethod(_TestNumericValue._test_binop_type_pos_vfloat, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_pos_float'),
+ partialmethod(_TestNumericValue._test_binop_value_pos_float, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_pos_vfloat'),
+ partialmethod(_TestNumericValue._test_binop_value_pos_vfloat, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_pos_float'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_addr_same_pos_float, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_pos_vfloat'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_addr_same_pos_vfloat, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_pos_float'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_value_same_pos_float, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_pos_vfloat'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_value_same_pos_vfloat, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_zero_float'),
+ partialmethod(_TestNumericValue._test_binop_type_zero_float, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_zero_vfloat'),
+ partialmethod(_TestNumericValue._test_binop_type_zero_vfloat, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_zero_float'),
+ partialmethod(_TestNumericValue._test_binop_value_zero_float, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_zero_vfloat'),
+ partialmethod(_TestNumericValue._test_binop_value_zero_vfloat, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_zero_float'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_addr_same_zero_float, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_zero_vfloat'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_addr_same_zero_vfloat, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_zero_float'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_value_same_zero_float, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_zero_vfloat'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_value_same_zero_vfloat, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_complex'),
+ partialmethod(_TestNumericValue._test_binop_type_complex, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('type_zero_complex'),
+ partialmethod(_TestNumericValue._test_binop_type_zero_complex, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_complex'),
+ partialmethod(_TestNumericValue._test_binop_value_complex, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('value_zero_complex'),
+ partialmethod(_TestNumericValue._test_binop_value_zero_complex, op=binop),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_complex'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_addr_same_complex, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_addr_same_zero_complex'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_addr_same_zero_complex, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_complex'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_value_same_complex, op=binop
+ ),
+ )
+ setattr(
+ cls,
+ test_binop_name('lhs_value_same_zero_complex'),
+ partialmethod(
+ _TestNumericValue._test_binop_lhs_value_same_zero_complex, op=binop
+ ),
+ )
# inject testing methods for each unary operation
for name, unaryop in _UNARYOPS:
- setattr(cls, test_unaryop_name('type'), partialmethod(_TestNumericValue._test_unaryop_type, op=unaryop))
- setattr(cls, test_unaryop_name('value'), partialmethod(_TestNumericValue._test_unaryop_value, op=unaryop))
- setattr(cls, test_unaryop_name('addr_same'), partialmethod(_TestNumericValue._test_unaryop_addr_same, op=unaryop))
- setattr(cls, test_unaryop_name('value_same'), partialmethod(_TestNumericValue._test_unaryop_value_same, op=unaryop))
+ setattr(
+ cls,
+ test_unaryop_name('type'),
+ partialmethod(_TestNumericValue._test_unaryop_type, op=unaryop),
+ )
+ setattr(
+ cls,
+ test_unaryop_name('value'),
+ partialmethod(_TestNumericValue._test_unaryop_value, op=unaryop),
+ )
+ setattr(
+ cls,
+ test_unaryop_name('addr_same'),
+ partialmethod(_TestNumericValue._test_unaryop_addr_same, op=unaryop),
+ )
+ setattr(
+ cls,
+ test_unaryop_name('value_same'),
+ partialmethod(_TestNumericValue._test_unaryop_value_same, op=unaryop),
+ )
class CreateValueFuncTestCase(unittest.TestCase):
a = A()
- with self.assertRaisesRegex(TypeError, "cannot create value object from 'A' object") as cm:
+ with self.assertRaisesRegex(
+ TypeError, "cannot create value object from 'A' object"
+ ) as cm:
v = bt2.create_value(a)
return self.assertRaisesRegex(TypeError, r'expecting an integral number object')
def _assert_expecting_int64(self):
- return self.assertRaisesRegex(ValueError, r"expecting a signed 64-bit integral value")
+ return self.assertRaisesRegex(
+ ValueError, r"expecting a signed 64-bit integral value"
+ )
def _assert_expecting_uint64(self):
- return self.assertRaisesRegex(ValueError, r"expecting an unsigned 64-bit integral value")
+ return self.assertRaisesRegex(
+ ValueError, r"expecting an unsigned 64-bit integral value"
+ )
def test_create_default(self):
i = self._CLS()
def test_compare_big_int(self):
# Larger than the IEEE 754 double-precision exact representation of
# integers.
- raw = (2**53) + 1
+ raw = (2 ** 53) + 1
v = bt2.create_value(raw)
self.assertEqual(v, raw)
'pos-int': 42,
'neg-float': -42.4,
'pos-float': 23.17,
- 'str': 'yes'
+ 'str': 'yes',
}
self._def = bt2.MapValue(copy.deepcopy(self._def_value))
self.assertNotEqual(a1, a2)
def test_eq_same_content_same_len(self):
- raw = {
- '3': 3,
- 'True': True,
- 'array': [1, 2.5, None, {'a': 17.6, 'b': None}]
- }
+ raw = {'3': 3, 'True': True, 'array': [1, 2.5, None, {'a': 17.6, 'b': None}]}
a1 = bt2.MapValue(raw)
a2 = bt2.MapValue(copy.deepcopy(raw))
self.assertEqual(a1, a2)
#
# The value returned by the callable is returned by run_in_component_init.
+
def run_in_component_init(func):
class MySink(bt2._UserSinkComponent):
def __init__(self, params):
del res_bound
return res
+
# Create an empty trace class with default values.
+
def get_default_trace_class():
def f(comp_self):
return comp_self._create_trace_class()
else:
raise StopIteration
+
@bt2.plugin_component_class
-class TheSourceOfAllEvil(bt2._UserSourceComponent,
- message_iterator_class=TheIteratorOfAllEvil):
+class TheSourceOfAllEvil(
+ bt2._UserSourceComponent, message_iterator_class=TheIteratorOfAllEvil
+):
def __init__(self, params):
tc = self._create_trace_class()
# smaller than this offset (in other words, a time that it's not
# possible to represent with this clock class).
cc = self._create_clock_class(frequency=1, offset=bt2.ClockClassOffset(10000))
- sc = tc.create_stream_class(default_clock_class=cc,
- supports_packets=True,
- packets_have_beginning_default_clock_snapshot=True,
- packets_have_end_default_clock_snapshot=True)
+ sc = tc.create_stream_class(
+ default_clock_class=cc,
+ supports_packets=True,
+ packets_have_beginning_default_clock_snapshot=True,
+ packets_have_end_default_clock_snapshot=True,
+ )
ec1 = sc.create_event_class(name='event 1')
ec2 = sc.create_event_class(name='event 2')
self._add_output_port('out', (tc, sc, ec1, ec2, params))
trace_path = os.path.join(debug_info_data_path, 'trace')
target_prefix = os.path.join(debug_info_data_path, '..', '..')
src = bt2.ComponentSpec('ctf', 'fs', trace_path)
- flt = bt2.ComponentSpec('lttng-utils', 'debug-info', {
- 'target-prefix': target_prefix,
- })
- it = bt2.TraceCollectionNotificationIterator(src, flt,
- [bt2.EventNotification])
+ flt = bt2.ComponentSpec(
+ 'lttng-utils', 'debug-info', {'target-prefix': target_prefix}
+ )
+ it = bt2.TraceCollectionNotificationIterator(src, flt, [bt2.EventNotification])
notifs = list(it)
debug_info = notifs[2].event['debug_info']
self.assertEqual(debug_info['bin'], 'libhello_so+0x14d4')
class QueryTraceInfoClockOffsetTestCase(unittest.TestCase):
-
def setUp(self):
ctf = bt2.find_plugin('ctf')
self._fs = ctf.source_component_classes['fs']
- self._paths = [os.path.join(test_ctf_traces_path, 'intersection', '3eventsintersect')]
+ self._paths = [
+ os.path.join(test_ctf_traces_path, 'intersection', '3eventsintersect')
+ ]
self._executor = bt2.QueryExecutor()
def _check(self, trace, offset):
self.assertEqual(trace['range-ns']['begin'], 13515309000000000 + offset)
self.assertEqual(trace['range-ns']['end'], 13515309000000120 + offset)
- self.assertEqual(trace['intersection-range-ns']['begin'], 13515309000000070 + offset)
- self.assertEqual(trace['intersection-range-ns']['end'], 13515309000000100 + offset)
+ self.assertEqual(
+ trace['intersection-range-ns']['begin'], 13515309000000070 + offset
+ )
+ self.assertEqual(
+ trace['intersection-range-ns']['end'], 13515309000000100 + offset
+ )
streams = sorted(trace['streams'], key=sort_predictably)
self.assertEqual(streams[0]['range-ns']['begin'], 13515309000000000 + offset)
# Without clock class offset
def test_no_clock_class_offset(self):
- res = self._executor.query(self._fs, 'trace-info', {
- 'paths': self._paths,
- })
+ res = self._executor.query(self._fs, 'trace-info', {'paths': self._paths})
trace = res[0]
self._check(trace, 0)
# With clock-class-offset-s
def test_clock_class_offset_s(self):
- res = self._executor.query(self._fs, 'trace-info', {
- 'paths': self._paths,
- 'clock-class-offset-s': 2,
- })
+ res = self._executor.query(
+ self._fs, 'trace-info', {'paths': self._paths, 'clock-class-offset-s': 2}
+ )
trace = res[0]
self._check(trace, 2000000000)
# With clock-class-offset-ns
def test_clock_class_offset_ns(self):
- res = self._executor.query(self._fs, 'trace-info', {
- 'paths': self._paths,
- 'clock-class-offset-ns': 2,
- })
+ res = self._executor.query(
+ self._fs, 'trace-info', {'paths': self._paths, 'clock-class-offset-ns': 2}
+ )
trace = res[0]
self._check(trace, 2)
# With both, negative
def test_clock_class_offset_both(self):
- res = self._executor.query(self._fs, 'trace-info', {
- 'paths': self._paths,
- 'clock-class-offset-s': -2,
- 'clock-class-offset-ns': -2,
- })
+ res = self._executor.query(
+ self._fs,
+ 'trace-info',
+ {
+ 'paths': self._paths,
+ 'clock-class-offset-s': -2,
+ 'clock-class-offset-ns': -2,
+ },
+ )
trace = res[0]
self._check(trace, -2000000002)
def test_clock_class_offset_s_wrong_type(self):
with self.assertRaises(bt2.InvalidParams):
- self._executor.query(self._fs, 'trace-info', {
- 'paths': self._paths,
- 'clock-class-offset-s': "2",
- })
+ self._executor.query(
+ self._fs,
+ 'trace-info',
+ {'paths': self._paths, 'clock-class-offset-s': "2"},
+ )
def test_clock_class_offset_s_wrong_type_none(self):
with self.assertRaises(bt2.InvalidParams):
- self._executor.query(self._fs, 'trace-info', {
- 'paths': self._paths,
- 'clock-class-offset-s': None,
- })
+ self._executor.query(
+ self._fs,
+ 'trace-info',
+ {'paths': self._paths, 'clock-class-offset-s': None},
+ )
def test_clock_class_offset_ns_wrong_type(self):
with self.assertRaises(bt2.InvalidParams):
- self._executor.query(self._fs, 'trace-info', {
- 'paths': self._paths,
- 'clock-class-offset-ns': "2",
- })
+ self._executor.query(
+ self._fs,
+ 'trace-info',
+ {'paths': self._paths, 'clock-class-offset-ns': "2"},
+ )
def test_clock_class_offset_ns_wrong_type_none(self):
with self.assertRaises(bt2.InvalidParams):
- self._executor.query(self._fs, 'trace-info', {
- 'paths': self._paths,
- 'clock-class-offset-ns': None,
- })
+ self._executor.query(
+ self._fs,
+ 'trace-info',
+ {'paths': self._paths, 'clock-class-offset-ns': None},
+ )
class QueryTraceInfoPortNameTestCase(unittest.TestCase):
@bt2.plugin_component_class
-class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
+class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
pass
@bt2.plugin_component_class
-class MyFilter(bt2._UserFilterComponent,
- message_iterator_class=MyIter):
+class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
pass
pass
-bt2.register_plugin(__name__, 'sparkling', author='Philippe Proulx',
- description='A delicious plugin.',
- version=(1, 2, 3, 'EXTRA'),
- license='MIT')
+bt2.register_plugin(
+ __name__,
+ 'sparkling',
+ author='Philippe Proulx',
+ description='A delicious plugin.',
+ version=(1, 2, 3, 'EXTRA'),
+ license='MIT',
+)
if __name__ == '__main__':
argparser = argparse.ArgumentParser()
- argparser.add_argument('-f', '--failfast',
- help='Stop on first fail or error',
- action='store_true')
+ argparser.add_argument(
+ '-f', '--failfast', help='Stop on first fail or error', action='store_true'
+ )
- argparser.add_argument('start_dir',
- help='Base directory where to search for tests',
- type=str)
+ argparser.add_argument(
+ 'start_dir', help='Base directory where to search for tests', type=str
+ )
mut_exclu_group = argparser.add_mutually_exclusive_group(required=True)
- mut_exclu_group.add_argument('-p', '--pattern',
- help='Glob-style pattern of test files to run '
- '(e.g. test_event*.py)',
- type=str)
-
- mut_exclu_group.add_argument('-t', '--test-case',
- help='Run a specfic test module name, test class '
- 'name, or test method name '
- '(e.g. test_event.EventTestCase.test_clock_value)',
- type=str)
+ mut_exclu_group.add_argument(
+ '-p',
+ '--pattern',
+ help='Glob-style pattern of test files to run ' '(e.g. test_event*.py)',
+ type=str,
+ )
+ mut_exclu_group.add_argument(
+ '-t',
+ '--test-case',
+ help='Run a specfic test module name, test class '
+ 'name, or test method name '
+ '(e.g. test_event.EventTestCase.test_clock_value)',
+ type=str,
+ )
args = argparser.parse_args()
# else it will fail to parse.
sys.exit(1)
-
if tests.countTestCases() < 1:
print("No tests matching '%s' found in '%s'" % (pattern, start_dir))
sys.exit(1)
projects / babeltrace.git / commitdiff
