specs = [bt2.ComponentSpec('ctf', 'fs', th.path) for th in self._trace_handles]
try:
- iter_cls = bt2.TraceCollectionNotificationIterator
+ iter_cls = bt2.TraceCollectionMessageIterator
tc_iter = iter_cls(specs,
stream_intersection_mode=self._intersect_mode,
begin=begin_s, end=end_s,
- notification_types=[bt2.EventNotification])
+ message_types=[bt2.EventMessage])
return map(reader_event._create_event, tc_iter)
except:
raise ValueError
bt2/field_types.py \
bt2/graph.py \
bt2/logging.py \
- bt2/notification_iterator.py \
- bt2/notification.py \
+ bt2/message_iterator.py \
+ bt2/message.py \
bt2/object.py \
bt2/packet.py \
bt2/plugin.py \
bt2/stream_class.py \
bt2/stream.py \
bt2/trace.py \
- bt2/trace_collection_notification_iterator.py \
+ bt2/trace_collection_message_iterator.py \
bt2/utils.py \
bt2/values.py
from bt2.fields import _VariantField
from bt2.graph import *
from bt2.logging import *
-from bt2.notification import *
-from bt2.notification import _DiscardedEventsNotification
-from bt2.notification import _DiscardedPacketsNotification
-from bt2.notification_iterator import *
-from bt2.notification_iterator import _UserNotificationIterator
+from bt2.message import *
+from bt2.message import _DiscardedEventsMessage
+from bt2.message import _DiscardedPacketsMessage
+from bt2.message_iterator import *
+from bt2.message_iterator import _UserMessageIterator
from bt2.packet import _Packet
from bt2.plugin import *
from bt2.port import *
from bt2.stream import _Stream
from bt2.stream_class import *
from bt2.trace import *
-from bt2.trace_collection_notification_iterator import *
+from bt2.trace_collection_message_iterator import *
from bt2.values import *
from bt2.values import _Value
pass
-class NotificationIteratorCanceled(Exception):
+class MessageIteratorCanceled(Exception):
pass
# THE SOFTWARE.
from bt2 import native_bt, object, utils
-import bt2.notification_iterator
+import bt2.message_iterator
import collections.abc
import bt2.values
import traceback
# finalized.
#
# User-defined source and filter component classes must use the
-# `notification_iterator_class` class parameter to specify the
-# notification iterator class to use for this component class:
+# `message_iterator_class` class parameter to specify the
+# message iterator class to use for this component class:
#
-# class MyNotificationIterator(bt2._UserNotificationIterator):
+# class MyMessageIterator(bt2._UserMessageIterator):
# ...
#
# class MySource(bt2._UserSourceComponent,
-# notification_iterator_class=MyNotificationIterator):
+# message_iterator_class=MyMessageIterator):
# ...
#
-# This notification iterator class must inherit
-# bt2._UserNotificationIterator, and it must define the _get() and
-# _next() methods. The notification iterator class can also define an
+# This message iterator class must inherit
+# bt2._UserMessageIterator, and it must define the _get() and
+# _next() methods. The message iterator class can also define an
# __init__() method: this method has access to the original Python
# component object which was used to create it as the `component`
-# property. The notification iterator class can also define a
+# property. The message iterator class can also define a
# _finalize() method (again, do NOT use __del__()): this is called when
-# the notification iterator is (really) destroyed.
+# the message iterator is (really) destroyed.
#
# When the user-defined class is destroyed, this metaclass's __del__()
# method is called: the native BT component class pointer is put (not
if len(lines) >= 3:
comp_cls_help = '\n'.join(lines[2:])
- iter_cls = kwargs.get('notification_iterator_class')
+ iter_cls = kwargs.get('message_iterator_class')
if _UserSourceComponent in bases:
_UserComponentType._set_iterator_class(cls, iter_cls)
@staticmethod
def _set_iterator_class(cls, iter_cls):
if iter_cls is None:
- raise bt2.IncompleteUserClass("cannot create component class '{}': missing notification 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.notification_iterator._UserNotificationIterator):
- raise bt2.IncompleteUserClass("cannot create component class '{}': notification iterator class does not inherit bt2._UserNotificationIterator".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__))
if not hasattr(iter_cls, '__next__'):
- raise bt2.IncompleteUserClass("cannot create component class '{}': notification 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
# THE SOFTWARE.
from bt2 import native_bt, object, utils
-import bt2.notification_iterator
+import bt2.message_iterator
import collections.abc
import bt2.port
import copy
class _PrivateConnection(object._PrivateObject, _Connection):
- def create_notification_iterator(self, notification_types=None):
- notif_types = bt2.notification._notif_types_from_notif_classes(notification_types)
- status, notif_iter_ptr = native_bt.py3_create_priv_conn_notif_iter(int(self._ptr),
- notif_types)
- _handle_status(status, 'cannot create notification iterator object')
- assert(notif_iter_ptr)
- return bt2.notification_iterator._PrivateConnectionNotificationIterator._create_from_ptr(notif_iter_ptr)
+ def create_message_iterator(self, message_types=None):
+ msg_types = bt2.message._msg_types_from_msg_classes(message_types)
+ status, msg_iter_ptr = native_bt.py3_create_priv_conn_msg_iter(int(self._ptr),
+ msg_types)
+ _handle_status(status, 'cannot create message iterator object')
+ assert(msg_iter_ptr)
+ return bt2.message_iterator._PrivateConnectionMessageIterator._create_from_ptr(msg_iter_ptr)
@property
def is_ended(self):
--- /dev/null
+# The MIT License (MIT)
+#
+# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+# THE SOFTWARE.
+
+from bt2 import native_bt, object, utils
+import bt2.clock_class_priority_map
+import bt2.clock_value
+import collections
+import bt2.packet
+import bt2.stream
+import bt2.event
+import copy
+import bt2
+
+
+def _create_from_ptr(ptr):
+ msg_type = native_bt.message_get_type(ptr)
+ cls = None
+
+ if msg_type not in _MESSAGE_TYPE_TO_CLS:
+ raise bt2.Error('unknown message type: {}'.format(msg_type))
+
+ return _MESSAGE_TYPE_TO_CLS[msg_type]._create_from_ptr(ptr)
+
+
+def _msg_types_from_msg_classes(message_types):
+ if message_types is None:
+ msg_types = None
+ else:
+ for msg_cls in message_types:
+ if msg_cls not in _MESSAGE_TYPE_TO_CLS.values():
+ raise ValueError("'{}' is not a message class".format(msg_cls))
+
+ msg_types = [msg_cls._TYPE for msg_cls in message_types]
+
+ return msg_types
+
+
+class _Message(object._Object):
+ pass
+
+
+class _CopyableMessage(_Message):
+ def __copy__(self):
+ return self._copy(lambda obj: obj)
+
+ def __deepcopy__(self, memo):
+ cpy = self._copy(copy.deepcopy)
+ memo[id(self)] = cpy
+ return cpy
+
+
+class EventMessage(_CopyableMessage):
+ _TYPE = native_bt.MESSAGE_TYPE_EVENT
+
+ def __init__(self, event, cc_prio_map=None):
+ utils._check_type(event, bt2.event._Event)
+
+ if cc_prio_map is not None:
+ utils._check_type(cc_prio_map, bt2.clock_class_priority_map.ClockClassPriorityMap)
+ cc_prio_map_ptr = cc_prio_map._ptr
+ else:
+ cc_prio_map_ptr = None
+
+ ptr = native_bt.message_event_create(event._ptr, cc_prio_map_ptr)
+
+ if ptr is None:
+ raise bt2.CreationError('cannot create event message object')
+
+ super().__init__(ptr)
+
+ @property
+ def event(self):
+ event_ptr = native_bt.message_event_get_event(self._ptr)
+ assert(event_ptr)
+ return bt2.event._create_from_ptr(event_ptr)
+
+ @property
+ def clock_class_priority_map(self):
+ cc_prio_map_ptr = native_bt.message_event_get_clock_class_priority_map(self._ptr)
+ assert(cc_prio_map_ptr)
+ return bt2.clock_class_priority_map.ClockClassPriorityMap._create_from_ptr(cc_prio_map_ptr)
+
+ def __eq__(self, other):
+ if type(other) is not type(self):
+ return False
+
+ if self.addr == other.addr:
+ return True
+
+ self_props = (
+ self.event,
+ self.clock_class_priority_map,
+ )
+ other_props = (
+ other.event,
+ other.clock_class_priority_map,
+ )
+ return self_props == other_props
+
+ def _copy(self, copy_func):
+ # We can always use references here because those properties are
+ # frozen anyway if they are part of a message. Since the
+ # user cannot modify them after copying the message, it's
+ # useless to copy/deep-copy them.
+ return EventMessage(self.event, self.clock_class_priority_map)
+
+
+class PacketBeginningMessage(_CopyableMessage):
+ _TYPE = native_bt.MESSAGE_TYPE_PACKET_BEGINNING
+
+ def __init__(self, packet):
+ utils._check_type(packet, bt2.packet._Packet)
+ ptr = native_bt.message_packet_begin_create(packet._ptr)
+
+ if ptr is None:
+ raise bt2.CreationError('cannot create packet beginning message object')
+
+ super().__init__(ptr)
+
+ @property
+ def packet(self):
+ packet_ptr = native_bt.message_packet_begin_get_packet(self._ptr)
+ assert(packet_ptr)
+ return bt2.packet._Packet._create_from_ptr(packet_ptr)
+
+ def __eq__(self, other):
+ if type(other) is not type(self):
+ return False
+
+ if self.addr == other.addr:
+ return True
+
+ return self.packet == other.packet
+
+ def _copy(self, copy_func):
+ # We can always use references here because those properties are
+ # frozen anyway if they are part of a message. Since the
+ # user cannot modify them after copying the message, it's
+ # useless to copy/deep-copy them.
+ return PacketBeginningMessage(self.packet)
+
+
+class PacketEndMessage(_CopyableMessage):
+ _TYPE = native_bt.MESSAGE_TYPE_PACKET_END
+
+ def __init__(self, packet):
+ utils._check_type(packet, bt2.packet._Packet)
+ ptr = native_bt.message_packet_end_create(packet._ptr)
+
+ if ptr is None:
+ raise bt2.CreationError('cannot create packet end message object')
+
+ super().__init__(ptr)
+
+ @property
+ def packet(self):
+ packet_ptr = native_bt.message_packet_end_get_packet(self._ptr)
+ assert(packet_ptr)
+ return bt2.packet._Packet._create_from_ptr(packet_ptr)
+
+ def __eq__(self, other):
+ if type(other) is not type(self):
+ return False
+
+ if self.addr == other.addr:
+ return True
+
+ return self.packet == other.packet
+
+ def _copy(self, copy_func):
+ # We can always use references here because those properties are
+ # frozen anyway if they are part of a message. Since the
+ # user cannot modify them after copying the message, it's
+ # useless to copy/deep-copy them.
+ return PacketEndMessage(self.packet)
+
+
+class StreamBeginningMessage(_CopyableMessage):
+ _TYPE = native_bt.MESSAGE_TYPE_STREAM_BEGINNING
+
+ def __init__(self, stream):
+ utils._check_type(stream, bt2.stream._Stream)
+ ptr = native_bt.message_stream_begin_create(stream._ptr)
+
+ if ptr is None:
+ raise bt2.CreationError('cannot create stream beginning message object')
+
+ super().__init__(ptr)
+
+ @property
+ def stream(self):
+ stream_ptr = native_bt.message_stream_begin_get_stream(self._ptr)
+ assert(stream_ptr)
+ return bt2.stream._create_from_ptr(stream_ptr)
+
+ def __eq__(self, other):
+ if type(other) is not type(self):
+ return False
+
+ if self.addr == other.addr:
+ return True
+
+ return self.stream == other.stream
+
+ def _copy(self, copy_func):
+ # We can always use references here because those properties are
+ # frozen anyway if they are part of a message. Since the
+ # user cannot modify them after copying the message, it's
+ # useless to copy/deep-copy them.
+ return StreamBeginningMessage(self.stream)
+
+
+class StreamEndMessage(_CopyableMessage):
+ _TYPE = native_bt.MESSAGE_TYPE_STREAM_END
+
+ def __init__(self, stream):
+ utils._check_type(stream, bt2.stream._Stream)
+ ptr = native_bt.message_stream_end_create(stream._ptr)
+
+ if ptr is None:
+ raise bt2.CreationError('cannot create stream end message object')
+
+ super().__init__(ptr)
+
+ @property
+ def stream(self):
+ stream_ptr = native_bt.message_stream_end_get_stream(self._ptr)
+ assert(stream_ptr)
+ return bt2.stream._create_from_ptr(stream_ptr)
+
+ def __eq__(self, other):
+ if type(other) is not type(self):
+ return False
+
+ if self.addr == other.addr:
+ return True
+
+ return self.stream == other.stream
+
+ def _copy(self, copy_func):
+ # We can always use references here because those properties are
+ # frozen anyway if they are part of a message. Since the
+ # user cannot modify them after copying the message, it's
+ # useless to copy/deep-copy them.
+ return StreamEndMessage(self.stream)
+
+
+class _InactivityMessageClockValuesIterator(collections.abc.Iterator):
+ def __init__(self, msg_clock_values):
+ self._msg_clock_values = msg_clock_values
+ self._clock_classes = list(msg_clock_values._msg.clock_class_priority_map)
+ self._at = 0
+
+ def __next__(self):
+ if self._at == len(self._clock_classes):
+ raise StopIteration
+
+ self._at += 1
+ return self._clock_classes[at]
+
+
+class _InactivityMessageClockValues(collections.abc.Mapping):
+ def __init__(self, msg):
+ self._msg = msg
+
+ def __getitem__(self, clock_class):
+ utils._check_type(clock_class, bt2.ClockClass)
+ clock_value_ptr = native_bt.message_inactivity_get_clock_value(self._msg._ptr,
+ clock_class._ptr)
+
+ if clock_value_ptr is None:
+ return
+
+ clock_value = bt2.clock_value._create_clock_value_from_ptr(clock_value_ptr)
+ return clock_value
+
+ def add(self, clock_value):
+ utils._check_type(clock_value, bt2.clock_value._ClockValue)
+ ret = native_bt.message_inactivity_set_clock_value(self._msg._ptr,
+ clock_value._ptr)
+ utils._handle_ret(ret, "cannot set inactivity message object's clock value")
+
+ def __len__(self):
+ return len(self._msg.clock_class_priority_map)
+
+ def __iter__(self):
+ return _InactivityMessageClockValuesIterator(self)
+
+
+class InactivityMessage(_CopyableMessage):
+ _TYPE = native_bt.MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY
+
+ def __init__(self, cc_prio_map=None):
+ if cc_prio_map is not None:
+ utils._check_type(cc_prio_map, bt2.clock_class_priority_map.ClockClassPriorityMap)
+ cc_prio_map_ptr = cc_prio_map._ptr
+ else:
+ cc_prio_map_ptr = None
+
+ ptr = native_bt.message_inactivity_create(cc_prio_map_ptr)
+
+ if ptr is None:
+ raise bt2.CreationError('cannot create inactivity message object')
+
+ super().__init__(ptr)
+
+ @property
+ def clock_class_priority_map(self):
+ cc_prio_map_ptr = native_bt.message_inactivity_get_clock_class_priority_map(self._ptr)
+ assert(cc_prio_map_ptr)
+ return bt2.clock_class_priority_map.ClockClassPriorityMap._create_from_ptr(cc_prio_map_ptr)
+
+ @property
+ def clock_values(self):
+ return _InactivityMessageClockValues(self)
+
+ def _get_clock_values(self):
+ clock_values = {}
+
+ for clock_class, clock_value in self.clock_values.items():
+ if clock_value is None:
+ continue
+
+ clock_values[clock_class] = clock_value
+
+ return clock_values
+
+ def __eq__(self, other):
+ if type(other) is not type(self):
+ return False
+
+ if self.addr == other.addr:
+ return True
+
+ self_props = (
+ self.clock_class_priority_map,
+ self._get_clock_values(),
+ )
+ other_props = (
+ other.clock_class_priority_map,
+ other._get_clock_values(),
+ )
+ return self_props == other_props
+
+ def __copy__(self):
+ cpy = InactivityMessage(self.clock_class_priority_map)
+
+ for clock_class, clock_value in self.clock_values.items():
+ if clock_value is None:
+ continue
+
+ cpy.clock_values.add(clock_value)
+
+ return cpy
+
+ def __deepcopy__(self, memo):
+ cc_prio_map_cpy = copy.deepcopy(self.clock_class_priority_map)
+ cpy = InactivityMessage(cc_prio_map_cpy)
+
+ # copy clock values
+ for orig_clock_class in self.clock_class_priority_map:
+ orig_clock_value = self.clock_value(orig_clock_class)
+
+ if orig_clock_value is None:
+ continue
+
+ # find equivalent, copied clock class in CC priority map copy
+ for cpy_clock_class in cc_prio_map_cpy:
+ if cpy_clock_class == orig_clock_class:
+ break
+
+ # create copy of clock value from copied clock class
+ clock_value_cpy = cpy_clock_class(orig_clock_value.cycles)
+
+ # set copied clock value in message copy
+ cpy.clock_values.add(clock_value_cpy)
+
+ memo[id(self)] = cpy
+ return cpy
+
+
+class _DiscardedElementsMessage(_Message):
+ def __eq__(self, other):
+ if type(other) is not type(self):
+ return False
+
+ if self.addr == other.addr:
+ return True
+
+ self_props = (
+ self.count,
+ self.stream,
+ self.beginning_clock_value,
+ self.end_clock_value,
+ )
+ other_props = (
+ other.count,
+ other.stream,
+ other.beginning_clock_value,
+ other.end_clock_value,
+ )
+ return self_props == other_props
+
+
+class _DiscardedPacketsMessage(_DiscardedElementsMessage):
+ _TYPE = native_bt.MESSAGE_TYPE_DISCARDED_PACKETS
+
+ @property
+ def count(self):
+ count = native_bt.message_discarded_packets_get_count(self._ptr)
+ assert(count >= 0)
+ return count
+
+ @property
+ def stream(self):
+ stream_ptr = native_bt.message_discarded_packets_get_stream(self._ptr)
+ assert(stream_ptr)
+ return bt2.stream._create_from_ptr(stream_ptr)
+
+ @property
+ def beginning_clock_value(self):
+ clock_value_ptr = native_bt.message_discarded_packets_get_begin_clock_value(self._ptr)
+
+ if clock_value_ptr is None:
+ return
+
+ clock_value = bt2.clock_value._create_clock_value_from_ptr(clock_value_ptr)
+ return clock_value
+
+ @property
+ def end_clock_value(self):
+ clock_value_ptr = native_bt.message_discarded_packets_get_end_clock_value(self._ptr)
+
+ if clock_value_ptr is None:
+ return
+
+ clock_value = bt2.clock_value._create_clock_value_from_ptr(clock_value_ptr)
+ return clock_value
+
+
+class _DiscardedEventsMessage(_DiscardedElementsMessage):
+ _TYPE = native_bt.MESSAGE_TYPE_DISCARDED_EVENTS
+
+ @property
+ def count(self):
+ count = native_bt.message_discarded_events_get_count(self._ptr)
+ assert(count >= 0)
+ return count
+
+ @property
+ def stream(self):
+ stream_ptr = native_bt.message_discarded_events_get_stream(self._ptr)
+ assert(stream_ptr)
+ return bt2.stream._create_from_ptr(stream_ptr)
+
+ @property
+ def beginning_clock_value(self):
+ clock_value_ptr = native_bt.message_discarded_events_get_begin_clock_value(self._ptr)
+
+ if clock_value_ptr is None:
+ return
+
+ clock_value = bt2.clock_value._create_clock_value_from_ptr(clock_value_ptr)
+ return clock_value
+
+ @property
+ def end_clock_value(self):
+ clock_value_ptr = native_bt.message_discarded_events_get_end_clock_value(self._ptr)
+
+ if clock_value_ptr is None:
+ return
+
+ clock_value = bt2.clock_value._create_clock_value_from_ptr(clock_value_ptr)
+ return clock_value
+
+
+_MESSAGE_TYPE_TO_CLS = {
+ native_bt.MESSAGE_TYPE_EVENT: EventMessage,
+ native_bt.MESSAGE_TYPE_PACKET_BEGINNING: PacketBeginningMessage,
+ native_bt.MESSAGE_TYPE_PACKET_END: PacketEndMessage,
+ native_bt.MESSAGE_TYPE_STREAM_BEGINNING: StreamBeginningMessage,
+ native_bt.MESSAGE_TYPE_STREAM_END: StreamEndMessage,
+ native_bt.MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY: InactivityMessage,
+ native_bt.MESSAGE_TYPE_DISCARDED_PACKETS: _DiscardedPacketsMessage,
+ native_bt.MESSAGE_TYPE_DISCARDED_EVENTS: _DiscardedEventsMessage,
+}
--- /dev/null
+# The MIT License (MIT)
+#
+# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+# THE SOFTWARE.
+
+from bt2 import native_bt, object, utils
+import bt2.message
+import collections.abc
+import bt2.component
+import bt2
+
+
+class _MessageIterator(collections.abc.Iterator):
+ def _handle_status(self, status, gen_error_msg):
+ if status == native_bt.MESSAGE_ITERATOR_STATUS_CANCELED:
+ raise bt2.MessageIteratorCanceled
+ elif status == native_bt.MESSAGE_ITERATOR_STATUS_AGAIN:
+ raise bt2.TryAgain
+ elif status == native_bt.MESSAGE_ITERATOR_STATUS_END:
+ raise bt2.Stop
+ elif status == native_bt.MESSAGE_ITERATOR_STATUS_UNSUPPORTED:
+ raise bt2.UnsupportedFeature
+ elif status < 0:
+ raise bt2.Error(gen_error_msg)
+
+ def __next__(self):
+ raise NotImplementedError
+
+
+class _GenericMessageIterator(object._Object, _MessageIterator):
+ def _get_msg(self):
+ msg_ptr = native_bt.message_iterator_get_message(self._ptr)
+ utils._handle_ptr(msg_ptr, "cannot get message iterator object's current message object")
+ return bt2.message._create_from_ptr(msg_ptr)
+
+ def _next(self):
+ status = native_bt.message_iterator_next(self._ptr)
+ self._handle_status(status,
+ 'unexpected error: cannot advance the message iterator')
+
+ def __next__(self):
+ self._next()
+ return self._get_msg()
+
+
+class _PrivateConnectionMessageIterator(_GenericMessageIterator):
+ @property
+ def component(self):
+ comp_ptr = native_bt.private_connection_message_iterator_get_component(self._ptr)
+ assert(comp_ptr)
+ return bt2.component._create_generic_component_from_ptr(comp_ptr)
+
+
+class _OutputPortMessageIterator(_GenericMessageIterator):
+ pass
+
+
+class _UserMessageIterator(_MessageIterator):
+ def __new__(cls, ptr):
+ # User iterator objects are always created by the native side,
+ # that is, never instantiated directly by Python code.
+ #
+ # The native code calls this, then manually calls
+ # self.__init__() without the `ptr` argument. The user has
+ # access to self.component during this call, thanks to this
+ # self._ptr argument being set.
+ #
+ # self._ptr is NOT owned by this object here, so there's nothing
+ # to do in __del__().
+ self = super().__new__(cls)
+ self._ptr = ptr
+ return self
+
+ def __init__(self):
+ pass
+
+ @property
+ def _component(self):
+ return native_bt.py3_get_user_component_from_user_msg_iter(self._ptr)
+
+ @property
+ def addr(self):
+ return int(self._ptr)
+
+ def _finalize(self):
+ pass
+
+ def __next__(self):
+ raise bt2.Stop
+
+ def _next_from_native(self):
+ # this can raise anything: it's catched by the native part
+ try:
+ msg = next(self)
+ except StopIteration:
+ raise bt2.Stop
+ except:
+ raise
+
+ utils._check_type(msg, bt2.message._Message)
+
+ # take a new reference for the native part
+ msg._get()
+ return int(msg._ptr)
static PyObject *py_mod_bt2_exc_try_again_type = NULL;
static PyObject *py_mod_bt2_exc_stop_type = NULL;
static PyObject *py_mod_bt2_exc_port_connection_refused_type = NULL;
-static PyObject *py_mod_bt2_exc_notif_iter_canceled_type = NULL;
+static PyObject *py_mod_bt2_exc_msg_iter_canceled_type = NULL;
static PyObject *py_mod_bt2_exc_invalid_query_object_type = NULL;
static PyObject *py_mod_bt2_exc_invalid_query_params_type = NULL;
Py_XDECREF(py_mod_bt2_exc_try_again_type);
Py_XDECREF(py_mod_bt2_exc_stop_type);
Py_XDECREF(py_mod_bt2_exc_port_connection_refused_type);
- Py_XDECREF(py_mod_bt2_exc_notif_iter_canceled_type);
+ Py_XDECREF(py_mod_bt2_exc_msg_iter_canceled_type);
Py_XDECREF(py_mod_bt2_exc_invalid_query_object_type);
Py_XDECREF(py_mod_bt2_exc_invalid_query_params_type);
}
py_comp = bt_self_component_get_data(self_component);
- /* Find user's Python notification iterator class */
+ /* Find user's Python message iterator class */
py_comp_cls = PyObject_GetAttrString(py_comp, "__class__");
if (!py_comp_cls) {
BT_LOGE_STR("Cannot get Python object's `__class__` attribute.");
}
/*
- * Create object with borrowed native notification iterator
+ * Create object with borrowed native message iterator
* reference:
*
* py_iter = py_iter_cls.__new__(py_iter_cls, py_iter_ptr)
/*
* Since the Python code can never instantiate a user-defined
- * notification iterator class, the native notification iterator
- * object does NOT belong to a user Python notification iterator
+ * message iterator class, the native message iterator
+ * object does NOT belong to a user Python message iterator
* object (borrowed reference). However this Python object is
- * owned by this native notification iterator object.
+ * owned by this native message iterator object.
*
- * In the Python world, the lifetime of the native notification
- * iterator is managed by a _GenericNotificationIterator
+ * In the Python world, the lifetime of the native message
+ * iterator is managed by a _GenericMessageIterator
* instance:
*
- * _GenericNotificationIterator instance:
- * owns a native bt_notification_iterator object (iter)
+ * _GenericMessageIterator instance:
+ * owns a native bt_message_iterator object (iter)
* owns a _UserMessageIterator instance (py_iter)
* self._ptr is a borrowed reference to the
- * native bt_private_connection_private_notification_iterator
+ * native bt_private_connection_private_message_iterator
* object (iter)
*/
bt_self_message_iterator_set_data(self_message_iterator, py_iter);
/*
* The returned object, on success, is an integer object
- * (PyLong) containing the address of a native notification
+ * (PyLong) containing the address of a native message
* object (which is now ours).
*/
msgs[0] =
+++ /dev/null
-# The MIT License (MIT)
-#
-# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-#
-# Permission is hereby granted, free of charge, to any person obtaining a copy
-# of this software and associated documentation files (the "Software"), to deal
-# in the Software without restriction, including without limitation the rights
-# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-# copies of the Software, and to permit persons to whom the Software is
-# furnished to do so, subject to the following conditions:
-#
-# The above copyright notice and this permission notice shall be included in
-# all copies or substantial portions of the Software.
-#
-# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
-# THE SOFTWARE.
-
-from bt2 import native_bt, object, utils
-import bt2.clock_class_priority_map
-import bt2.clock_value
-import collections
-import bt2.packet
-import bt2.stream
-import bt2.event
-import copy
-import bt2
-
-
-def _create_from_ptr(ptr):
- notif_type = native_bt.notification_get_type(ptr)
- cls = None
-
- if notif_type not in _NOTIF_TYPE_TO_CLS:
- raise bt2.Error('unknown notification type: {}'.format(notif_type))
-
- return _NOTIF_TYPE_TO_CLS[notif_type]._create_from_ptr(ptr)
-
-
-def _notif_types_from_notif_classes(notification_types):
- if notification_types is None:
- notif_types = None
- else:
- for notif_cls in notification_types:
- if notif_cls not in _NOTIF_TYPE_TO_CLS.values():
- raise ValueError("'{}' is not a notification class".format(notif_cls))
-
- notif_types = [notif_cls._TYPE for notif_cls in notification_types]
-
- return notif_types
-
-
-class _Notification(object._Object):
- pass
-
-
-class _CopyableNotification(_Notification):
- def __copy__(self):
- return self._copy(lambda obj: obj)
-
- def __deepcopy__(self, memo):
- cpy = self._copy(copy.deepcopy)
- memo[id(self)] = cpy
- return cpy
-
-
-class EventNotification(_CopyableNotification):
- _TYPE = native_bt.MESSAGE_TYPE_EVENT
-
- def __init__(self, event, cc_prio_map=None):
- utils._check_type(event, bt2.event._Event)
-
- if cc_prio_map is not None:
- utils._check_type(cc_prio_map, bt2.clock_class_priority_map.ClockClassPriorityMap)
- cc_prio_map_ptr = cc_prio_map._ptr
- else:
- cc_prio_map_ptr = None
-
- ptr = native_bt.notification_event_create(event._ptr, cc_prio_map_ptr)
-
- if ptr is None:
- raise bt2.CreationError('cannot create event notification object')
-
- super().__init__(ptr)
-
- @property
- def event(self):
- event_ptr = native_bt.notification_event_get_event(self._ptr)
- assert(event_ptr)
- return bt2.event._create_from_ptr(event_ptr)
-
- @property
- def clock_class_priority_map(self):
- cc_prio_map_ptr = native_bt.notification_event_get_clock_class_priority_map(self._ptr)
- assert(cc_prio_map_ptr)
- return bt2.clock_class_priority_map.ClockClassPriorityMap._create_from_ptr(cc_prio_map_ptr)
-
- def __eq__(self, other):
- if type(other) is not type(self):
- return False
-
- if self.addr == other.addr:
- return True
-
- self_props = (
- self.event,
- self.clock_class_priority_map,
- )
- other_props = (
- other.event,
- other.clock_class_priority_map,
- )
- return self_props == other_props
-
- def _copy(self, copy_func):
- # We can always use references here because those properties are
- # frozen anyway if they are part of a notification. Since the
- # user cannot modify them after copying the notification, it's
- # useless to copy/deep-copy them.
- return EventNotification(self.event, self.clock_class_priority_map)
-
-
-class PacketBeginningNotification(_CopyableNotification):
- _TYPE = native_bt.MESSAGE_TYPE_PACKET_BEGINNING
-
- def __init__(self, packet):
- utils._check_type(packet, bt2.packet._Packet)
- ptr = native_bt.notification_packet_begin_create(packet._ptr)
-
- if ptr is None:
- raise bt2.CreationError('cannot create packet beginning notification object')
-
- super().__init__(ptr)
-
- @property
- def packet(self):
- packet_ptr = native_bt.notification_packet_begin_get_packet(self._ptr)
- assert(packet_ptr)
- return bt2.packet._Packet._create_from_ptr(packet_ptr)
-
- def __eq__(self, other):
- if type(other) is not type(self):
- return False
-
- if self.addr == other.addr:
- return True
-
- return self.packet == other.packet
-
- def _copy(self, copy_func):
- # We can always use references here because those properties are
- # frozen anyway if they are part of a notification. Since the
- # user cannot modify them after copying the notification, it's
- # useless to copy/deep-copy them.
- return PacketBeginningNotification(self.packet)
-
-
-class PacketEndNotification(_CopyableNotification):
- _TYPE = native_bt.MESSAGE_TYPE_PACKET_END
-
- def __init__(self, packet):
- utils._check_type(packet, bt2.packet._Packet)
- ptr = native_bt.notification_packet_end_create(packet._ptr)
-
- if ptr is None:
- raise bt2.CreationError('cannot create packet end notification object')
-
- super().__init__(ptr)
-
- @property
- def packet(self):
- packet_ptr = native_bt.notification_packet_end_get_packet(self._ptr)
- assert(packet_ptr)
- return bt2.packet._Packet._create_from_ptr(packet_ptr)
-
- def __eq__(self, other):
- if type(other) is not type(self):
- return False
-
- if self.addr == other.addr:
- return True
-
- return self.packet == other.packet
-
- def _copy(self, copy_func):
- # We can always use references here because those properties are
- # frozen anyway if they are part of a notification. Since the
- # user cannot modify them after copying the notification, it's
- # useless to copy/deep-copy them.
- return PacketEndNotification(self.packet)
-
-
-class StreamBeginningNotification(_CopyableNotification):
- _TYPE = native_bt.MESSAGE_TYPE_STREAM_BEGINNING
-
- def __init__(self, stream):
- utils._check_type(stream, bt2.stream._Stream)
- ptr = native_bt.notification_stream_begin_create(stream._ptr)
-
- if ptr is None:
- raise bt2.CreationError('cannot create stream beginning notification object')
-
- super().__init__(ptr)
-
- @property
- def stream(self):
- stream_ptr = native_bt.notification_stream_begin_get_stream(self._ptr)
- assert(stream_ptr)
- return bt2.stream._create_from_ptr(stream_ptr)
-
- def __eq__(self, other):
- if type(other) is not type(self):
- return False
-
- if self.addr == other.addr:
- return True
-
- return self.stream == other.stream
-
- def _copy(self, copy_func):
- # We can always use references here because those properties are
- # frozen anyway if they are part of a notification. Since the
- # user cannot modify them after copying the notification, it's
- # useless to copy/deep-copy them.
- return StreamBeginningNotification(self.stream)
-
-
-class StreamEndNotification(_CopyableNotification):
- _TYPE = native_bt.MESSAGE_TYPE_STREAM_END
-
- def __init__(self, stream):
- utils._check_type(stream, bt2.stream._Stream)
- ptr = native_bt.notification_stream_end_create(stream._ptr)
-
- if ptr is None:
- raise bt2.CreationError('cannot create stream end notification object')
-
- super().__init__(ptr)
-
- @property
- def stream(self):
- stream_ptr = native_bt.notification_stream_end_get_stream(self._ptr)
- assert(stream_ptr)
- return bt2.stream._create_from_ptr(stream_ptr)
-
- def __eq__(self, other):
- if type(other) is not type(self):
- return False
-
- if self.addr == other.addr:
- return True
-
- return self.stream == other.stream
-
- def _copy(self, copy_func):
- # We can always use references here because those properties are
- # frozen anyway if they are part of a notification. Since the
- # user cannot modify them after copying the notification, it's
- # useless to copy/deep-copy them.
- return StreamEndNotification(self.stream)
-
-
-class _InactivityNotificationClockValuesIterator(collections.abc.Iterator):
- def __init__(self, notif_clock_values):
- self._notif_clock_values = notif_clock_values
- self._clock_classes = list(notif_clock_values._notif.clock_class_priority_map)
- self._at = 0
-
- def __next__(self):
- if self._at == len(self._clock_classes):
- raise StopIteration
-
- self._at += 1
- return self._clock_classes[at]
-
-
-class _InactivityNotificationClockValues(collections.abc.Mapping):
- def __init__(self, notif):
- self._notif = notif
-
- def __getitem__(self, clock_class):
- utils._check_type(clock_class, bt2.ClockClass)
- clock_value_ptr = native_bt.notification_inactivity_get_clock_value(self._notif._ptr,
- clock_class._ptr)
-
- if clock_value_ptr is None:
- return
-
- clock_value = bt2.clock_value._create_clock_value_from_ptr(clock_value_ptr)
- return clock_value
-
- def add(self, clock_value):
- utils._check_type(clock_value, bt2.clock_value._ClockValue)
- ret = native_bt.notification_inactivity_set_clock_value(self._notif._ptr,
- clock_value._ptr)
- utils._handle_ret(ret, "cannot set inactivity notification object's clock value")
-
- def __len__(self):
- return len(self._notif.clock_class_priority_map)
-
- def __iter__(self):
- return _InactivityNotificationClockValuesIterator(self)
-
-
-class InactivityNotification(_CopyableNotification):
- _TYPE = native_bt.MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY
-
- def __init__(self, cc_prio_map=None):
- if cc_prio_map is not None:
- utils._check_type(cc_prio_map, bt2.clock_class_priority_map.ClockClassPriorityMap)
- cc_prio_map_ptr = cc_prio_map._ptr
- else:
- cc_prio_map_ptr = None
-
- ptr = native_bt.notification_inactivity_create(cc_prio_map_ptr)
-
- if ptr is None:
- raise bt2.CreationError('cannot create inactivity notification object')
-
- super().__init__(ptr)
-
- @property
- def clock_class_priority_map(self):
- cc_prio_map_ptr = native_bt.notification_inactivity_get_clock_class_priority_map(self._ptr)
- assert(cc_prio_map_ptr)
- return bt2.clock_class_priority_map.ClockClassPriorityMap._create_from_ptr(cc_prio_map_ptr)
-
- @property
- def clock_values(self):
- return _InactivityNotificationClockValues(self)
-
- def _get_clock_values(self):
- clock_values = {}
-
- for clock_class, clock_value in self.clock_values.items():
- if clock_value is None:
- continue
-
- clock_values[clock_class] = clock_value
-
- return clock_values
-
- def __eq__(self, other):
- if type(other) is not type(self):
- return False
-
- if self.addr == other.addr:
- return True
-
- self_props = (
- self.clock_class_priority_map,
- self._get_clock_values(),
- )
- other_props = (
- other.clock_class_priority_map,
- other._get_clock_values(),
- )
- return self_props == other_props
-
- def __copy__(self):
- cpy = InactivityNotification(self.clock_class_priority_map)
-
- for clock_class, clock_value in self.clock_values.items():
- if clock_value is None:
- continue
-
- cpy.clock_values.add(clock_value)
-
- return cpy
-
- def __deepcopy__(self, memo):
- cc_prio_map_cpy = copy.deepcopy(self.clock_class_priority_map)
- cpy = InactivityNotification(cc_prio_map_cpy)
-
- # copy clock values
- for orig_clock_class in self.clock_class_priority_map:
- orig_clock_value = self.clock_value(orig_clock_class)
-
- if orig_clock_value is None:
- continue
-
- # find equivalent, copied clock class in CC priority map copy
- for cpy_clock_class in cc_prio_map_cpy:
- if cpy_clock_class == orig_clock_class:
- break
-
- # create copy of clock value from copied clock class
- clock_value_cpy = cpy_clock_class(orig_clock_value.cycles)
-
- # set copied clock value in notification copy
- cpy.clock_values.add(clock_value_cpy)
-
- memo[id(self)] = cpy
- return cpy
-
-
-class _DiscardedElementsNotification(_Notification):
- def __eq__(self, other):
- if type(other) is not type(self):
- return False
-
- if self.addr == other.addr:
- return True
-
- self_props = (
- self.count,
- self.stream,
- self.beginning_clock_value,
- self.end_clock_value,
- )
- other_props = (
- other.count,
- other.stream,
- other.beginning_clock_value,
- other.end_clock_value,
- )
- return self_props == other_props
-
-
-class _DiscardedPacketsNotification(_DiscardedElementsNotification):
- _TYPE = native_bt.MESSAGE_TYPE_DISCARDED_PACKETS
-
- @property
- def count(self):
- count = native_bt.notification_discarded_packets_get_count(self._ptr)
- assert(count >= 0)
- return count
-
- @property
- def stream(self):
- stream_ptr = native_bt.notification_discarded_packets_get_stream(self._ptr)
- assert(stream_ptr)
- return bt2.stream._create_from_ptr(stream_ptr)
-
- @property
- def beginning_clock_value(self):
- clock_value_ptr = native_bt.notification_discarded_packets_get_begin_clock_value(self._ptr)
-
- if clock_value_ptr is None:
- return
-
- clock_value = bt2.clock_value._create_clock_value_from_ptr(clock_value_ptr)
- return clock_value
-
- @property
- def end_clock_value(self):
- clock_value_ptr = native_bt.notification_discarded_packets_get_end_clock_value(self._ptr)
-
- if clock_value_ptr is None:
- return
-
- clock_value = bt2.clock_value._create_clock_value_from_ptr(clock_value_ptr)
- return clock_value
-
-
-class _DiscardedEventsNotification(_DiscardedElementsNotification):
- _TYPE = native_bt.MESSAGE_TYPE_DISCARDED_EVENTS
-
- @property
- def count(self):
- count = native_bt.notification_discarded_events_get_count(self._ptr)
- assert(count >= 0)
- return count
-
- @property
- def stream(self):
- stream_ptr = native_bt.notification_discarded_events_get_stream(self._ptr)
- assert(stream_ptr)
- return bt2.stream._create_from_ptr(stream_ptr)
-
- @property
- def beginning_clock_value(self):
- clock_value_ptr = native_bt.notification_discarded_events_get_begin_clock_value(self._ptr)
-
- if clock_value_ptr is None:
- return
-
- clock_value = bt2.clock_value._create_clock_value_from_ptr(clock_value_ptr)
- return clock_value
-
- @property
- def end_clock_value(self):
- clock_value_ptr = native_bt.notification_discarded_events_get_end_clock_value(self._ptr)
-
- if clock_value_ptr is None:
- return
-
- clock_value = bt2.clock_value._create_clock_value_from_ptr(clock_value_ptr)
- return clock_value
-
-
-_NOTIF_TYPE_TO_CLS = {
- native_bt.MESSAGE_TYPE_EVENT: EventNotification,
- native_bt.MESSAGE_TYPE_PACKET_BEGINNING: PacketBeginningNotification,
- native_bt.MESSAGE_TYPE_PACKET_END: PacketEndNotification,
- native_bt.MESSAGE_TYPE_STREAM_BEGINNING: StreamBeginningNotification,
- native_bt.MESSAGE_TYPE_STREAM_END: StreamEndNotification,
- native_bt.MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY: InactivityNotification,
- native_bt.MESSAGE_TYPE_DISCARDED_PACKETS: _DiscardedPacketsNotification,
- native_bt.MESSAGE_TYPE_DISCARDED_EVENTS: _DiscardedEventsNotification,
-}
+++ /dev/null
-# The MIT License (MIT)
-#
-# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-#
-# Permission is hereby granted, free of charge, to any person obtaining a copy
-# of this software and associated documentation files (the "Software"), to deal
-# in the Software without restriction, including without limitation the rights
-# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-# copies of the Software, and to permit persons to whom the Software is
-# furnished to do so, subject to the following conditions:
-#
-# The above copyright notice and this permission notice shall be included in
-# all copies or substantial portions of the Software.
-#
-# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
-# THE SOFTWARE.
-
-from bt2 import native_bt, object, utils
-import bt2.notification
-import collections.abc
-import bt2.component
-import bt2
-
-
-class _NotificationIterator(collections.abc.Iterator):
- def _handle_status(self, status, gen_error_msg):
- if status == native_bt.NOTIFICATION_ITERATOR_STATUS_CANCELED:
- raise bt2.NotificationIteratorCanceled
- elif status == native_bt.NOTIFICATION_ITERATOR_STATUS_AGAIN:
- raise bt2.TryAgain
- elif status == native_bt.NOTIFICATION_ITERATOR_STATUS_END:
- raise bt2.Stop
- elif status == native_bt.NOTIFICATION_ITERATOR_STATUS_UNSUPPORTED:
- raise bt2.UnsupportedFeature
- elif status < 0:
- raise bt2.Error(gen_error_msg)
-
- def __next__(self):
- raise NotImplementedError
-
-
-class _GenericNotificationIterator(object._Object, _NotificationIterator):
- def _get_notif(self):
- notif_ptr = native_bt.notification_iterator_get_notification(self._ptr)
- utils._handle_ptr(notif_ptr, "cannot get notification iterator object's current notification object")
- return bt2.notification._create_from_ptr(notif_ptr)
-
- def _next(self):
- status = native_bt.notification_iterator_next(self._ptr)
- self._handle_status(status,
- 'unexpected error: cannot advance the notification iterator')
-
- def __next__(self):
- self._next()
- return self._get_notif()
-
-
-class _PrivateConnectionNotificationIterator(_GenericNotificationIterator):
- @property
- def component(self):
- comp_ptr = native_bt.private_connection_notification_iterator_get_component(self._ptr)
- assert(comp_ptr)
- return bt2.component._create_generic_component_from_ptr(comp_ptr)
-
-
-class _OutputPortNotificationIterator(_GenericNotificationIterator):
- pass
-
-
-class _UserNotificationIterator(_NotificationIterator):
- def __new__(cls, ptr):
- # User iterator objects are always created by the native side,
- # that is, never instantiated directly by Python code.
- #
- # The native code calls this, then manually calls
- # self.__init__() without the `ptr` argument. The user has
- # access to self.component during this call, thanks to this
- # self._ptr argument being set.
- #
- # self._ptr is NOT owned by this object here, so there's nothing
- # to do in __del__().
- self = super().__new__(cls)
- self._ptr = ptr
- return self
-
- def __init__(self):
- pass
-
- @property
- def _component(self):
- return native_bt.py3_get_user_component_from_user_notif_iter(self._ptr)
-
- @property
- def addr(self):
- return int(self._ptr)
-
- def _finalize(self):
- pass
-
- def __next__(self):
- raise bt2.Stop
-
- def _next_from_native(self):
- # this can raise anything: it's catched by the native part
- try:
- notif = next(self)
- except StopIteration:
- raise bt2.Stop
- except:
- raise
-
- utils._check_type(notif, bt2.notification._Notification)
-
- # take a new reference for the native part
- notif._get()
- return int(notif._ptr)
import collections.abc
import bt2.component
import bt2.connection
-import bt2.notification_iterator
-import bt2.notification
+import bt2.message_iterator
+import bt2.message
import copy
import bt2
class _OutputPort(_Port):
- def create_notification_iterator(self, notification_types=None,
+ def create_message_iterator(self, message_types=None,
colander_component_name=None):
- notif_types = bt2.notification._notif_types_from_notif_classes(notification_types)
+ msg_types = bt2.message._msg_types_from_msg_classes(message_types)
if colander_component_name is not None:
utils._check_str(colander_component_name)
- notif_iter_ptr = native_bt.py3_create_output_port_notif_iter(int(self._ptr),
+ msg_iter_ptr = native_bt.py3_create_output_port_msg_iter(int(self._ptr),
colander_component_name,
- notif_types)
+ msg_types)
- if notif_iter_ptr is None:
- raise bt2.CreationError('cannot create output port notification iterator')
+ if msg_iter_ptr is None:
+ raise bt2.CreationError('cannot create output port message iterator')
- return bt2.notification_iterator._OutputPortNotificationIterator._create_from_ptr(notif_iter_ptr)
+ return bt2.message_iterator._OutputPortMessageIterator._create_from_ptr(msg_iter_ptr)
class _PrivatePort(object._PrivateObject, _Port):
--- /dev/null
+# The MIT License (MIT)
+#
+# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+# THE SOFTWARE.
+
+from bt2 import utils
+import bt2
+import itertools
+import bt2.message_iterator
+import datetime
+import collections.abc
+from collections import namedtuple
+import numbers
+
+
+# a pair of component and ComponentSpec
+_ComponentAndSpec = namedtuple('_ComponentAndSpec', ['comp', 'spec'])
+
+
+class ComponentSpec:
+ def __init__(self, plugin_name, component_class_name, params=None):
+ utils._check_str(plugin_name)
+ utils._check_str(component_class_name)
+ self._plugin_name = plugin_name
+ self._component_class_name = component_class_name
+
+ if type(params) is str:
+ self._params = bt2.create_value({'path': params})
+ else:
+ self._params = bt2.create_value(params)
+
+ @property
+ def plugin_name(self):
+ return self._plugin_name
+
+ @property
+ def component_class_name(self):
+ return self._component_class_name
+
+ @property
+ def params(self):
+ return self._params
+
+
+# datetime.datetime or integral to nanoseconds
+def _get_ns(obj):
+ if obj is None:
+ return
+
+ if isinstance(obj, numbers.Real):
+ # consider that it's already in seconds
+ s = obj
+ elif isinstance(obj, datetime.datetime):
+ # s -> ns
+ s = obj.timestamp()
+ else:
+ raise TypeError('"{}" is not an integral number or a datetime.datetime object'.format(obj))
+
+ return int(s * 1e9)
+
+
+class _CompClsType:
+ SOURCE = 0
+ FILTER = 1
+
+
+class TraceCollectionMessageIterator(bt2.message_iterator._MessageIterator):
+ def __init__(self, source_component_specs, filter_component_specs=None,
+ message_types=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)
+ self._end_ns = _get_ns(end)
+ self._message_types = message_types
+
+ if type(source_component_specs) is ComponentSpec:
+ source_component_specs = [source_component_specs]
+
+ if type(filter_component_specs) is ComponentSpec:
+ filter_component_specs = [filter_component_specs]
+ elif filter_component_specs is None:
+ filter_component_specs = []
+
+ self._src_comp_specs = source_component_specs
+ self._flt_comp_specs = filter_component_specs
+ self._next_suffix = 1
+ self._connect_ports = False
+
+ # lists of _ComponentAndSpec
+ self._src_comps_and_specs = []
+ self._flt_comps_and_specs = []
+
+ self._validate_component_specs(source_component_specs)
+ self._validate_component_specs(filter_component_specs)
+ self._build_graph()
+
+ 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)))
+
+ def __next__(self):
+ return next(self._msg_iter)
+
+ def _create_stream_intersection_trimmer(self, port):
+ # find the original parameters specified by the user to create
+ # this port's component to get the `path` parameter
+ for src_comp_and_spec in self._src_comps_and_specs:
+ if port.component == src_comp_and_spec.comp:
+ params = src_comp_and_spec.spec.params
+ break
+
+ try:
+ path = params['path']
+ except:
+ raise bt2.Error('all source components must be created with a "path" parameter in stream intersection mode')
+
+ params = {'path': str(path)}
+
+ # query the port's component for the `trace-info` object which
+ # contains the stream intersection range for each exposed
+ # trace
+ query_exec = bt2.QueryExecutor()
+ trace_info_res = query_exec.query(port.component.component_class,
+ 'trace-info', params)
+ begin = None
+ end = None
+
+ # find the trace info for this port's trace by name's prefix
+ try:
+ for trace_info in trace_info_res:
+ if port.name.startswith(str(trace_info['path'])):
+ range_ns = trace_info['intersection-range-ns']
+ begin = range_ns['begin']
+ end = range_ns['end']
+ break
+ except:
+ pass
+
+ if begin is None or end is None:
+ raise bt2.Error('cannot find stream intersection range for port "{}"'.format(port.name))
+
+ name = 'trimmer-{}-{}'.format(port.component.name, port.name)
+ return self._create_trimmer(begin, end, name)
+
+ def _create_muxer(self):
+ plugin = bt2.find_plugin('utils')
+
+ if plugin is None:
+ 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')
+
+ comp_cls = plugin.filter_component_classes['muxer']
+ return self._graph.add_component(comp_cls, 'muxer')
+
+ def _create_trimmer(self, begin, end, name):
+ plugin = bt2.find_plugin('utils')
+
+ if plugin is None:
+ 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')
+
+ params = {}
+
+ if begin is not None:
+ params['begin'] = begin
+
+ if end is not None:
+ params['end'] = end
+
+ comp_cls = plugin.filter_component_classes['trimmer']
+ 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.component_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)
+ self._next_suffix += 1
+
+ return name
+
+ def _create_comp(self, comp_spec, comp_cls_type):
+ plugin = bt2.find_plugin(comp_spec.plugin_name)
+
+ if plugin is None:
+ raise bt2.Error('no such plugin: {}'.format(comp_spec.plugin_name))
+
+ if comp_cls_type == _CompClsType.SOURCE:
+ comp_classes = plugin.source_component_classes
+ else:
+ comp_classes = plugin.filter_component_classes
+
+ if comp_spec.component_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.component_class_name))
+
+ comp_cls = comp_classes[comp_spec.component_class_name]
+ name = self._get_unique_comp_name(comp_spec)
+ comp = self._graph.add_component(comp_cls, name, comp_spec.params)
+ return comp
+
+ def _get_free_muxer_input_port(self):
+ for port in self._muxer_comp.input_ports.values():
+ if not port.is_connected:
+ return port
+
+ def _connect_src_comp_port(self, port):
+ # if this trace collection iterator is in stream intersection
+ # mode, we need this connection:
+ #
+ # port -> trimmer -> muxer
+ #
+ # otherwise, simply:
+ #
+ # port -> muxer
+ if self._stream_intersection_mode:
+ trimmer_comp = self._create_stream_intersection_trimmer(port)
+ self._graph.connect_ports(port, trimmer_comp.input_ports['in'])
+ port_to_muxer = trimmer_comp.output_ports['out']
+ else:
+ port_to_muxer = port
+
+ self._graph.connect_ports(port_to_muxer, self._get_free_muxer_input_port())
+
+ def _graph_port_added(self, port):
+ if not self._connect_ports:
+ return
+
+ if type(port) is bt2._InputPort:
+ return
+
+ if port.component not in [comp.comp for comp in self._src_comps_and_specs]:
+ # do not care about non-source components (muxer, trimmer, etc.)
+ return
+
+ self._connect_src_comp_port(port)
+
+ def _build_graph(self):
+ self._graph = bt2.Graph()
+ self._graph.add_listener(bt2.GraphListenerType.PORT_ADDED,
+ self._graph_port_added)
+ 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'])
+ msg_iter_port = trimmer_comp.output_ports['out']
+ else:
+ msg_iter_port = self._muxer_comp.output_ports['out']
+
+ # create extra filter components (chained)
+ for comp_spec in self._flt_comp_specs:
+ comp = self._create_comp(comp_spec, _CompClsType.FILTER)
+ self._flt_comps_and_specs.append(_ComponentAndSpec(comp, comp_spec))
+
+ # connect the extra filter chain
+ for comp_and_spec in self._flt_comps_and_specs:
+ in_port = list(comp_and_spec.comp.input_ports.values())[0]
+ out_port = list(comp_and_spec.comp.output_ports.values())[0]
+ self._graph.connect_ports(msg_iter_port, in_port)
+ msg_iter_port = out_port
+
+ # Here we create the components, self._graph_port_added() is
+ # called when they add ports, but the callback returns early
+ # because self._connect_ports is False. This is because the
+ # self._graph_port_added() could not find the associated source
+ # component specification in self._src_comps_and_specs because
+ # it does not exist yet (it needs the created component to
+ # exist).
+ for comp_spec in self._src_comp_specs:
+ comp = self._create_comp(comp_spec, _CompClsType.SOURCE)
+ self._src_comps_and_specs.append(_ComponentAndSpec(comp, comp_spec))
+
+ # Now we connect the ports which exist at this point. We allow
+ # self._graph_port_added() to automatically connect _new_ ports.
+ self._connect_ports = True
+
+ for comp_and_spec in self._src_comps_and_specs:
+ # Keep a separate list because comp_and_spec.output_ports
+ # could change during the connection of one of its ports.
+ # Any new port is handled by self._graph_port_added().
+ out_ports = [port for port in comp_and_spec.comp.output_ports.values()]
+
+ for out_port in out_ports:
+ if not out_port.component or out_port.is_connected:
+ continue
+
+ self._connect_src_comp_port(out_port)
+
+ # create this trace collection iterator's message iterator
+ self._msg_iter = msg_iter_port.create_message_iterator(self._message_types)
+++ /dev/null
-# The MIT License (MIT)
-#
-# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-#
-# Permission is hereby granted, free of charge, to any person obtaining a copy
-# of this software and associated documentation files (the "Software"), to deal
-# in the Software without restriction, including without limitation the rights
-# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-# copies of the Software, and to permit persons to whom the Software is
-# furnished to do so, subject to the following conditions:
-#
-# The above copyright notice and this permission notice shall be included in
-# all copies or substantial portions of the Software.
-#
-# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
-# THE SOFTWARE.
-
-from bt2 import utils
-import bt2
-import itertools
-import bt2.notification_iterator
-import datetime
-import collections.abc
-from collections import namedtuple
-import numbers
-
-
-# a pair of component and ComponentSpec
-_ComponentAndSpec = namedtuple('_ComponentAndSpec', ['comp', 'spec'])
-
-
-class ComponentSpec:
- def __init__(self, plugin_name, component_class_name, params=None):
- utils._check_str(plugin_name)
- utils._check_str(component_class_name)
- self._plugin_name = plugin_name
- self._component_class_name = component_class_name
-
- if type(params) is str:
- self._params = bt2.create_value({'path': params})
- else:
- self._params = bt2.create_value(params)
-
- @property
- def plugin_name(self):
- return self._plugin_name
-
- @property
- def component_class_name(self):
- return self._component_class_name
-
- @property
- def params(self):
- return self._params
-
-
-# datetime.datetime or integral to nanoseconds
-def _get_ns(obj):
- if obj is None:
- return
-
- if isinstance(obj, numbers.Real):
- # consider that it's already in seconds
- s = obj
- elif isinstance(obj, datetime.datetime):
- # s -> ns
- s = obj.timestamp()
- else:
- raise TypeError('"{}" is not an integral number or a datetime.datetime object'.format(obj))
-
- return int(s * 1e9)
-
-
-class _CompClsType:
- SOURCE = 0
- FILTER = 1
-
-
-class TraceCollectionNotificationIterator(bt2.notification_iterator._NotificationIterator):
- def __init__(self, source_component_specs, filter_component_specs=None,
- notification_types=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)
- self._end_ns = _get_ns(end)
- self._notification_types = notification_types
-
- if type(source_component_specs) is ComponentSpec:
- source_component_specs = [source_component_specs]
-
- if type(filter_component_specs) is ComponentSpec:
- filter_component_specs = [filter_component_specs]
- elif filter_component_specs is None:
- filter_component_specs = []
-
- self._src_comp_specs = source_component_specs
- self._flt_comp_specs = filter_component_specs
- self._next_suffix = 1
- self._connect_ports = False
-
- # lists of _ComponentAndSpec
- self._src_comps_and_specs = []
- self._flt_comps_and_specs = []
-
- self._validate_component_specs(source_component_specs)
- self._validate_component_specs(filter_component_specs)
- self._build_graph()
-
- 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)))
-
- def __next__(self):
- return next(self._notif_iter)
-
- def _create_stream_intersection_trimmer(self, port):
- # find the original parameters specified by the user to create
- # this port's component to get the `path` parameter
- for src_comp_and_spec in self._src_comps_and_specs:
- if port.component == src_comp_and_spec.comp:
- params = src_comp_and_spec.spec.params
- break
-
- try:
- path = params['path']
- except:
- raise bt2.Error('all source components must be created with a "path" parameter in stream intersection mode')
-
- params = {'path': str(path)}
-
- # query the port's component for the `trace-info` object which
- # contains the stream intersection range for each exposed
- # trace
- query_exec = bt2.QueryExecutor()
- trace_info_res = query_exec.query(port.component.component_class,
- 'trace-info', params)
- begin = None
- end = None
-
- # find the trace info for this port's trace by name's prefix
- try:
- for trace_info in trace_info_res:
- if port.name.startswith(str(trace_info['path'])):
- range_ns = trace_info['intersection-range-ns']
- begin = range_ns['begin']
- end = range_ns['end']
- break
- except:
- pass
-
- if begin is None or end is None:
- raise bt2.Error('cannot find stream intersection range for port "{}"'.format(port.name))
-
- name = 'trimmer-{}-{}'.format(port.component.name, port.name)
- return self._create_trimmer(begin, end, name)
-
- def _create_muxer(self):
- plugin = bt2.find_plugin('utils')
-
- if plugin is None:
- 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')
-
- comp_cls = plugin.filter_component_classes['muxer']
- return self._graph.add_component(comp_cls, 'muxer')
-
- def _create_trimmer(self, begin, end, name):
- plugin = bt2.find_plugin('utils')
-
- if plugin is None:
- 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')
-
- params = {}
-
- if begin is not None:
- params['begin'] = begin
-
- if end is not None:
- params['end'] = end
-
- comp_cls = plugin.filter_component_classes['trimmer']
- 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.component_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)
- self._next_suffix += 1
-
- return name
-
- def _create_comp(self, comp_spec, comp_cls_type):
- plugin = bt2.find_plugin(comp_spec.plugin_name)
-
- if plugin is None:
- raise bt2.Error('no such plugin: {}'.format(comp_spec.plugin_name))
-
- if comp_cls_type == _CompClsType.SOURCE:
- comp_classes = plugin.source_component_classes
- else:
- comp_classes = plugin.filter_component_classes
-
- if comp_spec.component_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.component_class_name))
-
- comp_cls = comp_classes[comp_spec.component_class_name]
- name = self._get_unique_comp_name(comp_spec)
- comp = self._graph.add_component(comp_cls, name, comp_spec.params)
- return comp
-
- def _get_free_muxer_input_port(self):
- for port in self._muxer_comp.input_ports.values():
- if not port.is_connected:
- return port
-
- def _connect_src_comp_port(self, port):
- # if this trace collection iterator is in stream intersection
- # mode, we need this connection:
- #
- # port -> trimmer -> muxer
- #
- # otherwise, simply:
- #
- # port -> muxer
- if self._stream_intersection_mode:
- trimmer_comp = self._create_stream_intersection_trimmer(port)
- self._graph.connect_ports(port, trimmer_comp.input_ports['in'])
- port_to_muxer = trimmer_comp.output_ports['out']
- else:
- port_to_muxer = port
-
- self._graph.connect_ports(port_to_muxer, self._get_free_muxer_input_port())
-
- def _graph_port_added(self, port):
- if not self._connect_ports:
- return
-
- if type(port) is bt2._InputPort:
- return
-
- if port.component not in [comp.comp for comp in self._src_comps_and_specs]:
- # do not care about non-source components (muxer, trimmer, etc.)
- return
-
- self._connect_src_comp_port(port)
-
- def _build_graph(self):
- self._graph = bt2.Graph()
- self._graph.add_listener(bt2.GraphListenerType.PORT_ADDED,
- self._graph_port_added)
- 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'])
- notif_iter_port = trimmer_comp.output_ports['out']
- else:
- notif_iter_port = self._muxer_comp.output_ports['out']
-
- # create extra filter components (chained)
- for comp_spec in self._flt_comp_specs:
- comp = self._create_comp(comp_spec, _CompClsType.FILTER)
- self._flt_comps_and_specs.append(_ComponentAndSpec(comp, comp_spec))
-
- # connect the extra filter chain
- for comp_and_spec in self._flt_comps_and_specs:
- in_port = list(comp_and_spec.comp.input_ports.values())[0]
- out_port = list(comp_and_spec.comp.output_ports.values())[0]
- self._graph.connect_ports(notif_iter_port, in_port)
- notif_iter_port = out_port
-
- # Here we create the components, self._graph_port_added() is
- # called when they add ports, but the callback returns early
- # because self._connect_ports is False. This is because the
- # self._graph_port_added() could not find the associated source
- # component specification in self._src_comps_and_specs because
- # it does not exist yet (it needs the created component to
- # exist).
- for comp_spec in self._src_comp_specs:
- comp = self._create_comp(comp_spec, _CompClsType.SOURCE)
- self._src_comps_and_specs.append(_ComponentAndSpec(comp, comp_spec))
-
- # Now we connect the ports which exist at this point. We allow
- # self._graph_port_added() to automatically connect _new_ ports.
- self._connect_ports = True
-
- for comp_and_spec in self._src_comps_and_specs:
- # Keep a separate list because comp_and_spec.output_ports
- # could change during the connection of one of its ports.
- # Any new port is handled by self._graph_port_added().
- out_ports = [port for port in comp_and_spec.comp.output_ports.values()]
-
- for out_port in out_ports:
- if not out_port.component or out_port.is_connected:
- continue
-
- self._connect_src_comp_port(out_port)
-
- # create this trace collection iterator's notification iterator
- self._notif_iter = notif_iter_port.create_notification_iterator(self._notification_types)
test_field_types.py \
test_fields.py \
test_graph.py \
- test_notification.py \
- test_notification_iterator.py \
+ test_message.py \
+ test_message_iterator.py \
test_packet.py \
test_plugin.py \
test_port.py \
test_stream.py \
test_stream_class.py \
test_trace.py \
- test_trace_collection_notification_iterator.py \
+ test_trace_collection_message_iterator.py \
test_values.py \
.coveragerc
cls()
def test_no_init_source(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
pass
self._test_no_init(MySource)
def test_no_init_filter(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
pass
self._test_no_init(MyFilter)
self._test_no_init(MySink)
- def test_incomplete_source_no_notif_iter_cls(self):
- class MyIter(bt2._UserNotificationIterator):
+ def test_incomplete_source_no_msg_iter_cls(self):
+ class MyIter(bt2._UserMessageIterator):
pass
with self.assertRaises(bt2.IncompleteUserClass):
class MySource(bt2._UserSourceComponent):
pass
- def test_incomplete_source_wrong_notif_iter_cls_type(self):
- class MyIter(bt2._UserNotificationIterator):
+ def test_incomplete_source_wrong_msg_iter_cls_type(self):
+ class MyIter(bt2._UserMessageIterator):
pass
with self.assertRaises(bt2.IncompleteUserClass):
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=int):
+ message_iterator_class=int):
pass
- def test_incomplete_filter_no_notif_iter_cls(self):
- class MyIter(bt2._UserNotificationIterator):
+ def test_incomplete_filter_no_msg_iter_cls(self):
+ class MyIter(bt2._UserMessageIterator):
pass
with self.assertRaises(bt2.IncompleteUserClass):
pass
def test_incomplete_sink_no_consume_method(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
pass
with self.assertRaises(bt2.IncompleteUserClass):
pass
def test_minimal_source(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
pass
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
pass
def test_minimal_filter(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
pass
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
pass
def test_minimal_sink(self):
@unittest.skip("this is broken")
class ConnectionTestCase(unittest.TestCase):
def test_create(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
self.assertNotIsInstance(conn, bt2._PrivateConnection)
def test_is_ended_false(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
self.assertFalse(conn.is_ended)
def test_is_ended_true(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
self.assertTrue(conn.is_ended)
def test_downstream_port(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
self.assertEqual(conn.downstream_port.addr, sink.input_ports['in'].addr)
def test_upstream_port(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
self.assertEqual(conn.upstream_port.addr, src.output_ports['out'].addr)
def test_eq(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
self.assertEqual(conn, conn)
def test_eq_invalid(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
@unittest.skip("this is broken")
class PrivateConnectionTestCase(unittest.TestCase):
def test_create(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
del priv_conn
def test_is_ended_false(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
del priv_conn
def test_is_ended_true(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
del priv_conn
def test_downstream_port(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
del priv_port
def test_upstream_port(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
del priv_port
def test_eq(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
del priv_conn
def test_eq_invalid(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
self._graph.add_component(int, 'salut')
def test_connect_ports(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
self.assertEqual(sink.input_ports['in'].connection, conn)
def test_connect_ports_invalid_direction(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
src.output_ports['out'])
def test_connect_ports_refused(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
sink.input_ports['in'])
def test_connect_ports_canceled(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
sink.input_ports['in'])
def test_connect_ports_cannot_consume_accept(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
self.assertIs(type(exc), bt2.CannotConsumeGraph)
def test_connect_ports_cannot_consume_connected(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
self.assertTrue(self._graph.is_canceled)
def test_run(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __init__(self):
self._build_meta()
self._at = 0
if self._at == 5:
raise bt2.Stop
- notif = bt2.EventNotification(self._create_event(self._at * 3))
+ msg = bt2.EventMessage(self._create_event(self._at * 3))
self._at += 1
- return notif
+ return msg
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
self._at = 0
def _consume(comp_self):
- notif = next(comp_self._notif_iter)
+ msg = next(comp_self._msg_iter)
if comp_self._at == 0:
- self.assertIsInstance(notif, bt2.StreamBeginningNotification)
+ self.assertIsInstance(msg, bt2.StreamBeginningMessage)
elif comp_self._at == 1:
- self.assertIsInstance(notif, bt2.PacketBeginningNotification)
+ self.assertIsInstance(msg, bt2.PacketBeginningMessage)
elif comp_self._at >= 2 and comp_self._at <= 6:
- self.assertIsInstance(notif, bt2.EventNotification)
- self.assertEqual(notif.event.event_class.name, 'salut')
- field = notif.event.payload_field['my_int']
+ self.assertIsInstance(msg, bt2.EventMessage)
+ self.assertEqual(msg.event.event_class.name, 'salut')
+ field = msg.event.payload_field['my_int']
self.assertEqual(field, (comp_self._at - 2) * 3)
elif comp_self._at == 7:
- self.assertIsInstance(notif, bt2.PacketEndNotification)
+ self.assertIsInstance(msg, bt2.PacketEndMessage)
elif comp_self._at == 8:
- self.assertIsInstance(notif, bt2.StreamEndNotification)
+ self.assertIsInstance(msg, bt2.StreamEndMessage)
comp_self._at += 1
def _port_connected(self, port, other_port):
- self._notif_iter = port.connection.create_notification_iterator()
+ self._msg_iter = port.connection.create_message_iterator()
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
self._graph.run()
def test_run_again(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __init__(self):
self._build_meta()
self._at = 0
if self._at == 1:
raise bt2.TryAgain
- notif = bt2.EventNotification(self._create_event(self._at * 3))
+ msg = bt2.EventMessage(self._create_event(self._at * 3))
self._at += 1
- return notif
+ return msg
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
def _consume(comp_self):
if comp_self._at == 0:
- notif = next(comp_self._notif_iter)
- self.assertIsInstance(notif, bt2.EventNotification)
+ msg = next(comp_self._msg_iter)
+ self.assertIsInstance(msg, bt2.EventMessage)
elif comp_self._at == 1:
with self.assertRaises(bt2.TryAgain):
- notif = next(comp_self._notif_iter)
+ msg = next(comp_self._msg_iter)
raise bt2.TryAgain
comp_self._at += 1
def _port_connected(self, port, other_port):
- types = [bt2.EventNotification]
- self._notif_iter = port.connection.create_notification_iterator(types)
+ types = [bt2.EventMessage]
+ self._msg_iter = port.connection.create_message_iterator(types)
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
self._graph.run()
def test_run_no_sink(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
pass
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
self._add_input_port('in')
self._graph.run()
def test_run_error(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __init__(self):
self._build_meta()
self._at = 0
if self._at == 1:
raise bt2.TryAgain
- notif = bt2.EventNotification(self._create_event(self._at * 3))
+ msg = bt2.EventMessage(self._create_event(self._at * 3))
self._at += 1
- return notif
+ return msg
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
def _consume(comp_self):
if comp_self._at == 0:
- notif = next(comp_self._notif_iter)
- self.assertIsInstance(notif, bt2.EventNotification)
+ msg = next(comp_self._msg_iter)
+ self.assertIsInstance(msg, bt2.EventMessage)
elif comp_self._at == 1:
raise RuntimeError('error!')
comp_self._at += 1
def _port_connected(self, port, other_port):
- types = [bt2.EventNotification]
- self._notif_iter = port.connection.create_notification_iterator(types)
+ types = [bt2.EventMessage]
+ self._msg_iter = port.connection.create_message_iterator(types)
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
self._graph.run()
def test_run_cannot_consume(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
pass
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
self.assertIs(type(exc), bt2.CannotConsumeGraph)
def test_listeners(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(self, params):
self._add_output_port('out')
self._add_output_port('zero')
--- /dev/null
+from bt2 import values
+import collections
+import unittest
+import copy
+import bt2
+
+
+class _MessageTestCase(unittest.TestCase):
+ def setUp(self):
+ self._trace = bt2.Trace()
+ self._sc = bt2.StreamClass()
+ self._ec = bt2.EventClass('salut')
+ self._my_int_ft = bt2.IntegerFieldType(32)
+ self._ec.payload_field_type = bt2.StructureFieldType()
+ self._ec.payload_field_type += collections.OrderedDict([
+ ('my_int', self._my_int_ft),
+ ])
+ self._sc.add_event_class(self._ec)
+ self._clock_class = bt2.ClockClass('allo', 1000)
+ self._trace.add_clock_class(self._clock_class)
+ self._trace.packet_header_field_type = bt2.StructureFieldType()
+ self._trace.packet_header_field_type += collections.OrderedDict([
+ ('hello', self._my_int_ft),
+ ])
+ self._trace.add_stream_class(self._sc)
+ self._cc_prio_map = bt2.ClockClassPriorityMap()
+ self._cc_prio_map[self._clock_class] = 231
+ self._stream = self._sc()
+ self._packet = self._stream.create_packet()
+ self._packet.header_field['hello'] = 19487
+ self._event = self._ec()
+ self._event.clock_values.add(self._clock_class(1772))
+ self._event.payload_field['my_int'] = 23
+ self._event.packet = self._packet
+
+ def tearDown(self):
+ del self._trace
+ del self._sc
+ del self._ec
+ del self._my_int_ft
+ del self._clock_class
+ del self._cc_prio_map
+ del self._stream
+ del self._packet
+ del self._event
+
+
+@unittest.skip("this is broken")
+class EventMessageTestCase(_MessageTestCase):
+ def test_create_no_cc_prio_map(self):
+ msg = bt2.EventMessage(self._event)
+ self.assertEqual(msg.event.addr, self._event.addr)
+ self.assertEqual(len(msg.clock_class_priority_map), 0)
+
+ def test_create_with_cc_prio_map(self):
+ msg = bt2.EventMessage(self._event, self._cc_prio_map)
+ self.assertEqual(msg.event.addr, self._event.addr)
+ self.assertEqual(len(msg.clock_class_priority_map), 1)
+ self.assertEqual(msg.clock_class_priority_map.highest_priority_clock_class.addr,
+ self._clock_class.addr)
+ self.assertEqual(msg.clock_class_priority_map[self._clock_class], 231)
+
+ def test_eq(self):
+ msg = bt2.EventMessage(self._event, self._cc_prio_map)
+ event_copy = copy.copy(self._event)
+ event_copy.packet = self._packet
+ cc_prio_map_copy = copy.copy(self._cc_prio_map)
+ msg2 = bt2.EventMessage(event_copy, cc_prio_map_copy)
+ self.assertEqual(msg, msg2)
+
+ def test_ne_event(self):
+ msg = bt2.EventMessage(self._event, self._cc_prio_map)
+ event_copy = copy.copy(self._event)
+ event_copy.payload_field['my_int'] = 17
+ event_copy.packet = self._packet
+ cc_prio_map_copy = copy.copy(self._cc_prio_map)
+ msg2 = bt2.EventMessage(event_copy, cc_prio_map_copy)
+ self.assertNotEqual(msg, msg2)
+
+ def test_ne_cc_prio_map(self):
+ msg = bt2.EventMessage(self._event)
+ event_copy = copy.copy(self._event)
+ event_copy.packet = self._packet
+ cc_prio_map_copy = copy.copy(self._cc_prio_map)
+ msg2 = bt2.EventMessage(event_copy, cc_prio_map_copy)
+ self.assertNotEqual(msg, msg2)
+
+ def test_eq_invalid(self):
+ msg = bt2.EventMessage(self._event)
+ self.assertNotEqual(msg, 23)
+
+ def test_copy(self):
+ msg = bt2.EventMessage(self._event, self._cc_prio_map)
+ msg2 = copy.copy(msg)
+ self.assertEqual(msg, msg2)
+
+ def test_deepcopy(self):
+ msg = bt2.EventMessage(self._event, self._cc_prio_map)
+ msg2 = copy.deepcopy(msg)
+ self.assertEqual(msg, msg2)
+
+
+@unittest.skip("this is broken")
+class PacketBeginningMessageTestCase(_MessageTestCase):
+ def test_create(self):
+ msg = bt2.PacketBeginningMessage(self._packet)
+ self.assertEqual(msg.packet.addr, self._packet.addr)
+
+ def test_eq(self):
+ msg = bt2.PacketBeginningMessage(self._packet)
+ packet_copy = copy.copy(self._packet)
+ msg2 = bt2.PacketBeginningMessage(packet_copy)
+ self.assertEqual(msg, msg2)
+
+ def test_ne_packet(self):
+ msg = bt2.PacketBeginningMessage(self._packet)
+ packet_copy = copy.copy(self._packet)
+ packet_copy.header_field['hello'] = 1847
+ msg2 = bt2.PacketBeginningMessage(packet_copy)
+ self.assertNotEqual(msg, msg2)
+
+ def test_eq_invalid(self):
+ msg = bt2.PacketBeginningMessage(self._packet)
+ self.assertNotEqual(msg, 23)
+
+ def test_copy(self):
+ msg = bt2.PacketBeginningMessage(self._packet)
+ msg2 = copy.copy(msg)
+ self.assertEqual(msg, msg2)
+
+ def test_deepcopy(self):
+ msg = bt2.PacketBeginningMessage(self._packet)
+ msg2 = copy.deepcopy(msg)
+ self.assertEqual(msg, msg2)
+
+
+@unittest.skip("this is broken")
+class PacketEndMessageTestCase(_MessageTestCase):
+ def test_create(self):
+ msg = bt2.PacketEndMessage(self._packet)
+ self.assertEqual(msg.packet.addr, self._packet.addr)
+
+ def test_eq(self):
+ msg = bt2.PacketEndMessage(self._packet)
+ packet_copy = copy.copy(self._packet)
+ msg2 = bt2.PacketEndMessage(packet_copy)
+ self.assertEqual(msg, msg2)
+
+ def test_ne_packet(self):
+ msg = bt2.PacketEndMessage(self._packet)
+ packet_copy = copy.copy(self._packet)
+ packet_copy.header_field['hello'] = 1847
+ msg2 = bt2.PacketEndMessage(packet_copy)
+ self.assertNotEqual(msg, msg2)
+
+ def test_eq_invalid(self):
+ msg = bt2.PacketEndMessage(self._packet)
+ self.assertNotEqual(msg, 23)
+
+ def test_copy(self):
+ msg = bt2.PacketEndMessage(self._packet)
+ msg2 = copy.copy(msg)
+ self.assertEqual(msg, msg2)
+
+ def test_deepcopy(self):
+ msg = bt2.PacketEndMessage(self._packet)
+ msg2 = copy.deepcopy(msg)
+ self.assertEqual(msg, msg2)
+
+
+@unittest.skip("this is broken")
+class StreamBeginningMessageTestCase(_MessageTestCase):
+ def test_create(self):
+ msg = bt2.StreamBeginningMessage(self._stream)
+ self.assertEqual(msg.stream.addr, self._stream.addr)
+
+ def test_eq(self):
+ msg = bt2.StreamBeginningMessage(self._stream)
+ stream_copy = copy.copy(self._stream)
+ msg2 = bt2.StreamBeginningMessage(stream_copy)
+ self.assertEqual(msg, msg2)
+
+ def test_ne_stream(self):
+ msg = bt2.StreamBeginningMessage(self._stream)
+ stream_copy = self._sc(name='salut')
+ msg2 = bt2.StreamBeginningMessage(stream_copy)
+ self.assertNotEqual(msg, msg2)
+
+ def test_eq_invalid(self):
+ msg = bt2.StreamBeginningMessage(self._stream)
+ self.assertNotEqual(msg, 23)
+
+ def test_copy(self):
+ msg = bt2.StreamBeginningMessage(self._stream)
+ msg2 = copy.copy(msg)
+ self.assertEqual(msg, msg2)
+
+ def test_deepcopy(self):
+ msg = bt2.StreamBeginningMessage(self._stream)
+ msg2 = copy.deepcopy(msg)
+ self.assertEqual(msg, msg2)
+
+
+@unittest.skip("this is broken")
+class StreamEndMessageTestCase(_MessageTestCase):
+ def test_create(self):
+ msg = bt2.StreamEndMessage(self._stream)
+ self.assertEqual(msg.stream.addr, self._stream.addr)
+
+ def test_eq(self):
+ msg = bt2.StreamEndMessage(self._stream)
+ stream_copy = copy.copy(self._stream)
+ msg2 = bt2.StreamEndMessage(stream_copy)
+ self.assertEqual(msg, msg2)
+
+ def test_ne_stream(self):
+ msg = bt2.StreamEndMessage(self._stream)
+ stream_copy = self._sc(name='salut')
+ msg2 = bt2.StreamEndMessage(stream_copy)
+ self.assertNotEqual(msg, msg2)
+
+ def test_eq_invalid(self):
+ msg = bt2.StreamEndMessage(self._stream)
+ self.assertNotEqual(msg, 23)
+
+ def test_copy(self):
+ msg = bt2.StreamEndMessage(self._stream)
+ msg2 = copy.copy(msg)
+ self.assertEqual(msg, msg2)
+
+ def test_deepcopy(self):
+ msg = bt2.StreamEndMessage(self._stream)
+ msg2 = copy.deepcopy(msg)
+ self.assertEqual(msg, msg2)
+
+
+@unittest.skip("this is broken")
+class InactivityMessageTestCase(unittest.TestCase):
+ def setUp(self):
+ self._cc1 = bt2.ClockClass('cc1', 1000)
+ self._cc2 = bt2.ClockClass('cc2', 2000)
+ self._cc_prio_map = bt2.ClockClassPriorityMap()
+ self._cc_prio_map[self._cc1] = 25
+ self._cc_prio_map[self._cc2] = 50
+
+ def tearDown(self):
+ del self._cc1
+ del self._cc2
+ del self._cc_prio_map
+
+ def test_create_no_cc_prio_map(self):
+ msg = bt2.InactivityMessage()
+ self.assertEqual(len(msg.clock_class_priority_map), 0)
+
+ def test_create_with_cc_prio_map(self):
+ msg = bt2.InactivityMessage(self._cc_prio_map)
+ msg.clock_values.add(self._cc1(123))
+ msg.clock_values.add(self._cc2(19487))
+ self.assertEqual(len(msg.clock_class_priority_map), 2)
+ self.assertEqual(msg.clock_class_priority_map, self._cc_prio_map)
+ self.assertEqual(msg.clock_values[self._cc1], 123)
+ self.assertEqual(msg.clock_values[self._cc2], 19487)
+
+ def test_eq(self):
+ msg = bt2.InactivityMessage(self._cc_prio_map)
+ msg.clock_values.add(self._cc1(123))
+ msg.clock_values.add(self._cc2(19487))
+ cc_prio_map_copy = copy.copy(self._cc_prio_map)
+ msg2 = bt2.InactivityMessage(cc_prio_map_copy)
+ msg2.clock_values.add(self._cc1(123))
+ msg2.clock_values.add(self._cc2(19487))
+ self.assertEqual(msg, msg2)
+
+ def test_ne_cc_prio_map(self):
+ msg = bt2.InactivityMessage(self._cc_prio_map)
+ msg.clock_values.add(self._cc1(123))
+ msg.clock_values.add(self._cc2(19487))
+ cc_prio_map_copy = copy.copy(self._cc_prio_map)
+ cc_prio_map_copy[self._cc2] = 23
+ msg2 = bt2.InactivityMessage(cc_prio_map_copy)
+ self.assertNotEqual(msg, msg2)
+
+ def test_ne_clock_value(self):
+ msg = bt2.InactivityMessage(self._cc_prio_map)
+ msg.clock_values.add(self._cc1(123))
+ msg.clock_values.add(self._cc2(19487))
+ msg2 = bt2.InactivityMessage(self._cc_prio_map)
+ msg.clock_values.add(self._cc1(123))
+ msg.clock_values.add(self._cc2(1847))
+ self.assertNotEqual(msg, msg2)
+
+ def test_eq_invalid(self):
+ msg = bt2.InactivityMessage(self._cc_prio_map)
+ self.assertNotEqual(msg, 23)
+
+ def test_copy(self):
+ msg = bt2.InactivityMessage(self._cc_prio_map)
+ msg.clock_values.add(self._cc1(123))
+ msg.clock_values.add(self._cc2(19487))
+ msg_copy = copy.copy(msg)
+ self.assertEqual(msg, msg_copy)
+ self.assertNotEqual(msg.addr, msg_copy.addr)
+ self.assertEqual(msg.clock_class_priority_map.addr,
+ msg_copy.clock_class_priority_map.addr)
+ self.assertEqual(msg_copy.clock_values[self._cc1], 123)
+ self.assertEqual(msg_copy.clock_values[self._cc2], 19487)
+
+ def test_deepcopy(self):
+ msg = bt2.InactivityMessage(self._cc_prio_map)
+ msg.clock_values.add(self._cc1(123))
+ msg.clock_values.add(self._cc2(19487))
+ msg_copy = copy.deepcopy(msg)
+ self.assertEqual(msg, msg_copy)
+ self.assertNotEqual(msg.addr, msg_copy.addr)
+ self.assertNotEqual(msg.clock_class_priority_map.addr,
+ msg_copy.clock_class_priority_map.addr)
+ self.assertEqual(msg.clock_class_priority_map,
+ msg_copy.clock_class_priority_map)
+ self.assertNotEqual(list(msg.clock_class_priority_map)[0].addr,
+ list(msg_copy.clock_class_priority_map)[0].addr)
+ self.assertIsNone(msg_copy.clock_values[self._cc1])
+ self.assertIsNone(msg_copy.clock_values[self._cc2])
+ self.assertEqual(msg_copy.clock_values[list(msg_copy.clock_class_priority_map)[0]], 123)
+ self.assertEqual(msg_copy.clock_values[list(msg_copy.clock_class_priority_map)[1]], 19487)
+
+
+@unittest.skip("this is broken")
+class DiscardedPacketsMessageTestCase(unittest.TestCase):
+ def setUp(self):
+ self._trace = bt2.Trace()
+ self._sc = bt2.StreamClass()
+ self._ec = bt2.EventClass('salut')
+ self._clock_class = bt2.ClockClass('yo', 1000)
+ self._uint64_int_ft = bt2.IntegerFieldType(64, mapped_clock_class=self._clock_class)
+ self._my_int_ft = bt2.IntegerFieldType(32)
+ self._ec.payload_field_type = bt2.StructureFieldType()
+ self._ec.payload_field_type += collections.OrderedDict([
+ ('my_int', self._my_int_ft),
+ ])
+ self._sc.add_event_class(self._ec)
+ self._sc.packet_context_field_type = bt2.StructureFieldType()
+ self._sc.packet_context_field_type += collections.OrderedDict([
+ ('packet_seq_num', self._my_int_ft),
+ ('timestamp_begin', self._uint64_int_ft),
+ ('timestamp_end', self._uint64_int_ft),
+ ])
+ self._trace.add_clock_class(self._clock_class)
+ self._trace.add_stream_class(self._sc)
+ self._stream = self._sc()
+
+ def tearDown(self):
+ del self._trace
+ del self._sc
+ del self._ec
+ del self._clock_class
+ del self._uint64_int_ft
+ del self._my_int_ft
+ del self._stream
+
+ def _create_event(self, packet):
+ event = self._ec()
+ event.payload_field['my_int'] = 23
+ event.packet = packet
+ return event
+
+ def _get_msg(self):
+ class MyIter(bt2._UserMessageIterator):
+ def __init__(iter_self):
+ packet1 = self._stream.create_packet()
+ packet1.context_field['packet_seq_num'] = 0
+ packet1.context_field['timestamp_begin'] = 3
+ packet1.context_field['timestamp_end'] = 6
+ packet2 = self._stream.create_packet()
+ packet2.context_field['packet_seq_num'] = 5
+ packet2.context_field['timestamp_begin'] = 7
+ packet2.context_field['timestamp_end'] = 10
+ iter_self._ev1 = self._create_event(packet1)
+ iter_self._ev2 = self._create_event(packet2)
+ iter_self._at = 0
+
+ def __next__(self):
+ if self._at == 0:
+ msg = bt2.EventMessage(self._ev1)
+ elif self._at == 1:
+ msg = bt2.EventMessage(self._ev2)
+ else:
+ raise bt2.Stop
+
+ self._at += 1
+ return msg
+
+ class MySource(bt2._UserSourceComponent,
+ message_iterator_class=MyIter):
+ def __init__(self, params):
+ self._add_output_port('out')
+
+ class MySink(bt2._UserSinkComponent):
+ def __init__(self, params):
+ self._add_input_port('in')
+
+ def _consume(comp_self):
+ nonlocal the_msg
+ msg = next(comp_self._msg_iter)
+
+ if type(msg) is bt2._DiscardedPacketsMessage:
+ the_msg = msg
+ raise bt2.Stop
+
+ def _port_connected(self, port, other_port):
+ self._msg_iter = port.connection.create_message_iterator()
+
+ the_msg = None
+ 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'])
+ graph.run()
+ return the_msg
+
+ def test_create(self):
+ self.assertIsInstance(self._get_msg(), bt2._DiscardedPacketsMessage)
+
+ def test_count(self):
+ self.assertEqual(self._get_msg().count, 4)
+
+ def test_stream(self):
+ self.assertEqual(self._get_msg().stream.addr, self._stream.addr)
+
+ def test_beginning_clock_value(self):
+ msg = self._get_msg()
+ beginning_clock_value = msg.beginning_clock_value
+ self.assertEqual(beginning_clock_value.clock_class, self._clock_class)
+ self.assertEqual(beginning_clock_value, 6)
+
+ def test_end_clock_value(self):
+ msg = self._get_msg()
+ end_clock_value = msg.end_clock_value
+ self.assertEqual(end_clock_value.clock_class, self._clock_class)
+ self.assertEqual(end_clock_value, 7)
+
+ def test_eq(self):
+ msg1 = self._get_msg()
+ msg2 = self._get_msg()
+ self.assertEqual(msg1, msg2)
+
+ def test_eq_invalid(self):
+ msg1 = self._get_msg()
+ self.assertNotEqual(msg1, 23)
+
+
+@unittest.skip("this is broken")
+class DiscardedEventsMessageTestCase(unittest.TestCase):
+ def setUp(self):
+ self._trace = bt2.Trace()
+ self._sc = bt2.StreamClass()
+ self._ec = bt2.EventClass('salut')
+ self._clock_class = bt2.ClockClass('yo', 1000)
+ self._uint64_int_ft = bt2.IntegerFieldType(64, mapped_clock_class=self._clock_class)
+ self._my_int_ft = bt2.IntegerFieldType(32)
+ self._ec.payload_field_type = bt2.StructureFieldType()
+ self._ec.payload_field_type += collections.OrderedDict([
+ ('my_int', self._my_int_ft),
+ ])
+ self._sc.add_event_class(self._ec)
+ self._sc.packet_context_field_type = bt2.StructureFieldType()
+ self._sc.packet_context_field_type += collections.OrderedDict([
+ ('events_discarded', self._my_int_ft),
+ ('timestamp_begin', self._uint64_int_ft),
+ ('timestamp_end', self._uint64_int_ft),
+ ])
+ self._trace.add_clock_class(self._clock_class)
+ self._trace.add_stream_class(self._sc)
+ self._stream = self._sc()
+
+ def tearDown(self):
+ del self._trace
+ del self._sc
+ del self._ec
+ del self._clock_class
+ del self._uint64_int_ft
+ del self._my_int_ft
+ del self._stream
+
+ def _create_event(self, packet):
+ event = self._ec()
+ event.payload_field['my_int'] = 23
+ event.packet = packet
+ return event
+
+ def _get_msg(self):
+ class MyIter(bt2._UserMessageIterator):
+ def __init__(iter_self):
+ packet1 = self._stream.create_packet()
+ packet1.context_field['events_discarded'] = 0
+ packet1.context_field['timestamp_begin'] = 3
+ packet1.context_field['timestamp_end'] = 6
+ packet2 = self._stream.create_packet()
+ packet2.context_field['events_discarded'] = 10
+ packet2.context_field['timestamp_begin'] = 7
+ packet2.context_field['timestamp_end'] = 10
+ iter_self._ev1 = self._create_event(packet1)
+ iter_self._ev2 = self._create_event(packet2)
+ iter_self._at = 0
+
+ def __next__(self):
+ if self._at == 0:
+ msg = bt2.EventMessage(self._ev1)
+ elif self._at == 1:
+ msg = bt2.EventMessage(self._ev2)
+ else:
+ raise bt2.Stop
+
+ self._at += 1
+ return msg
+
+ class MySource(bt2._UserSourceComponent,
+ message_iterator_class=MyIter):
+ def __init__(self, params):
+ self._add_output_port('out')
+
+ class MySink(bt2._UserSinkComponent):
+ def __init__(self, params):
+ self._add_input_port('in')
+
+ def _consume(comp_self):
+ nonlocal the_msg
+ msg = next(comp_self._msg_iter)
+
+ if type(msg) is bt2._DiscardedEventsMessage:
+ the_msg = msg
+ raise bt2.Stop
+
+ def _port_connected(self, port, other_port):
+ self._msg_iter = port.connection.create_message_iterator()
+
+ the_msg = None
+ 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'])
+ graph.run()
+ return the_msg
+
+ def test_create(self):
+ self.assertIsInstance(self._get_msg(), bt2._DiscardedEventsMessage)
+
+ def test_count(self):
+ self.assertEqual(self._get_msg().count, 10)
+
+ def test_stream(self):
+ self.assertEqual(self._get_msg().stream.addr, self._stream.addr)
+
+ def test_beginning_clock_value(self):
+ msg = self._get_msg()
+ beginning_clock_value = msg.beginning_clock_value
+ self.assertEqual(beginning_clock_value.clock_class, self._clock_class)
+ self.assertEqual(beginning_clock_value, 6)
+
+ def test_end_clock_value(self):
+ msg = self._get_msg()
+ end_clock_value = msg.end_clock_value
+ self.assertEqual(end_clock_value.clock_class, self._clock_class)
+ self.assertEqual(end_clock_value, 10)
+
+ def test_eq(self):
+ msg1 = self._get_msg()
+ msg2 = self._get_msg()
+ self.assertEqual(msg1, msg2)
+
+ def test_eq_invalid(self):
+ msg1 = self._get_msg()
+ self.assertNotEqual(msg1, 23)
--- /dev/null
+from bt2 import values
+import collections
+import unittest
+import copy
+import bt2
+
+
+@unittest.skip("this is broken")
+class UserMessageIteratorTestCase(unittest.TestCase):
+ @staticmethod
+ def _create_graph(src_comp_cls):
+ class MySink(bt2._UserSinkComponent):
+ def __init__(self, params):
+ self._add_input_port('in')
+
+ def _consume(self):
+ next(self._msg_iter)
+
+ def _port_connected(self, port, other_port):
+ self._msg_iter = port.connection.create_message_iterator()
+
+ 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'])
+ return graph
+
+ def test_init(self):
+ class MyIter(bt2._UserMessageIterator):
+ def __init__(self):
+ nonlocal initialized
+ initialized = True
+
+ class MySource(bt2._UserSourceComponent,
+ message_iterator_class=MyIter):
+ def __init__(self, params):
+ self._add_output_port('out')
+
+ initialized = False
+ graph = self._create_graph(MySource)
+ self.assertTrue(initialized)
+
+ def test_finalize(self):
+ class MyIter(bt2._UserMessageIterator):
+ def _finalize(self):
+ nonlocal finalized
+ finalized = True
+
+ class MySource(bt2._UserSourceComponent,
+ message_iterator_class=MyIter):
+ def __init__(self, params):
+ self._add_output_port('out')
+
+ finalized = False
+ graph = self._create_graph(MySource)
+ del graph
+ self.assertTrue(finalized)
+
+ def test_component(self):
+ class MyIter(bt2._UserMessageIterator):
+ def __init__(self):
+ nonlocal salut
+ salut = self._component._salut
+
+ class MySource(bt2._UserSourceComponent,
+ message_iterator_class=MyIter):
+ def __init__(self, params):
+ self._add_output_port('out')
+ self._salut = 23
+
+ salut = None
+ graph = self._create_graph(MySource)
+ self.assertEqual(salut, 23)
+
+ def test_addr(self):
+ class MyIter(bt2._UserMessageIterator):
+ def __init__(self):
+ nonlocal addr
+ addr = self.addr
+
+ class MySource(bt2._UserSourceComponent,
+ message_iterator_class=MyIter):
+ def __init__(self, params):
+ self._add_output_port('out')
+
+ addr = None
+ graph = self._create_graph(MySource)
+ self.assertIsNotNone(addr)
+ self.assertNotEqual(addr, 0)
+
+
+@unittest.skip("this is broken")
+class PrivateConnectionMessageIteratorTestCase(unittest.TestCase):
+ def test_component(self):
+ class MyIter(bt2._UserMessageIterator):
+ pass
+
+ class MySource(bt2._UserSourceComponent,
+ message_iterator_class=MyIter):
+ def __init__(self, params):
+ self._add_output_port('out')
+
+ class MySink(bt2._UserSinkComponent):
+ def __init__(self, params):
+ self._add_input_port('in')
+
+ def _consume(self):
+ next(self._msg_iter)
+
+ def _port_connected(self, port, other_port):
+ nonlocal upstream_comp
+ self._msg_iter = port.connection.create_message_iterator()
+ upstream_comp = self._msg_iter.component
+
+ upstream_comp = None
+ graph = bt2.Graph()
+ src_comp = graph.add_component(MySource, 'src')
+ sink_comp = graph.add_component(MySink, 'sink')
+ graph.connect_ports(src_comp.output_ports['out'],
+ sink_comp.input_ports['in'])
+ self.assertEqual(src_comp, upstream_comp)
+ del upstream_comp
+
+
+@unittest.skip("this is broken")
+class OutputPortMessageIteratorTestCase(unittest.TestCase):
+ def test_component(self):
+ class MyIter(bt2._UserMessageIterator):
+ def __init__(self):
+ self._build_meta()
+ self._at = 0
+
+ def _build_meta(self):
+ self._trace = bt2.Trace()
+ self._sc = bt2.StreamClass()
+ self._ec = bt2.EventClass('salut')
+ self._my_int_ft = bt2.IntegerFieldType(32)
+ self._ec.payload_field_type = bt2.StructureFieldType()
+ self._ec.payload_field_type += collections.OrderedDict([
+ ('my_int', self._my_int_ft),
+ ])
+ self._sc.add_event_class(self._ec)
+ self._trace.add_stream_class(self._sc)
+ self._stream = self._sc()
+ self._packet = self._stream.create_packet()
+
+ def _create_event(self, value):
+ ev = self._ec()
+ ev.payload_field['my_int'] = value
+ ev.packet = self._packet
+ return ev
+
+ def __next__(self):
+ if self._at == 5:
+ raise bt2.Stop
+
+ msg = bt2.EventMessage(self._create_event(self._at * 3))
+ self._at += 1
+ return msg
+
+ 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')
+ types = [bt2.EventMessage]
+ msg_iter = src.output_ports['out'].create_message_iterator(types)
+
+ for at, msg in enumerate(msg_iter):
+ self.assertIsInstance(msg, bt2.EventMessage)
+ self.assertEqual(msg.event.event_class.name, 'salut')
+ field = msg.event.payload_field['my_int']
+ self.assertEqual(field, at * 3)
+++ /dev/null
-from bt2 import values
-import collections
-import unittest
-import copy
-import bt2
-
-
-class _NotificationTestCase(unittest.TestCase):
- def setUp(self):
- self._trace = bt2.Trace()
- self._sc = bt2.StreamClass()
- self._ec = bt2.EventClass('salut')
- self._my_int_ft = bt2.IntegerFieldType(32)
- self._ec.payload_field_type = bt2.StructureFieldType()
- self._ec.payload_field_type += collections.OrderedDict([
- ('my_int', self._my_int_ft),
- ])
- self._sc.add_event_class(self._ec)
- self._clock_class = bt2.ClockClass('allo', 1000)
- self._trace.add_clock_class(self._clock_class)
- self._trace.packet_header_field_type = bt2.StructureFieldType()
- self._trace.packet_header_field_type += collections.OrderedDict([
- ('hello', self._my_int_ft),
- ])
- self._trace.add_stream_class(self._sc)
- self._cc_prio_map = bt2.ClockClassPriorityMap()
- self._cc_prio_map[self._clock_class] = 231
- self._stream = self._sc()
- self._packet = self._stream.create_packet()
- self._packet.header_field['hello'] = 19487
- self._event = self._ec()
- self._event.clock_values.add(self._clock_class(1772))
- self._event.payload_field['my_int'] = 23
- self._event.packet = self._packet
-
- def tearDown(self):
- del self._trace
- del self._sc
- del self._ec
- del self._my_int_ft
- del self._clock_class
- del self._cc_prio_map
- del self._stream
- del self._packet
- del self._event
-
-
-@unittest.skip("this is broken")
-class EventNotificationTestCase(_NotificationTestCase):
- def test_create_no_cc_prio_map(self):
- notif = bt2.EventNotification(self._event)
- self.assertEqual(notif.event.addr, self._event.addr)
- self.assertEqual(len(notif.clock_class_priority_map), 0)
-
- def test_create_with_cc_prio_map(self):
- notif = bt2.EventNotification(self._event, self._cc_prio_map)
- self.assertEqual(notif.event.addr, self._event.addr)
- self.assertEqual(len(notif.clock_class_priority_map), 1)
- self.assertEqual(notif.clock_class_priority_map.highest_priority_clock_class.addr,
- self._clock_class.addr)
- self.assertEqual(notif.clock_class_priority_map[self._clock_class], 231)
-
- def test_eq(self):
- notif = bt2.EventNotification(self._event, self._cc_prio_map)
- event_copy = copy.copy(self._event)
- event_copy.packet = self._packet
- cc_prio_map_copy = copy.copy(self._cc_prio_map)
- notif2 = bt2.EventNotification(event_copy, cc_prio_map_copy)
- self.assertEqual(notif, notif2)
-
- def test_ne_event(self):
- notif = bt2.EventNotification(self._event, self._cc_prio_map)
- event_copy = copy.copy(self._event)
- event_copy.payload_field['my_int'] = 17
- event_copy.packet = self._packet
- cc_prio_map_copy = copy.copy(self._cc_prio_map)
- notif2 = bt2.EventNotification(event_copy, cc_prio_map_copy)
- self.assertNotEqual(notif, notif2)
-
- def test_ne_cc_prio_map(self):
- notif = bt2.EventNotification(self._event)
- event_copy = copy.copy(self._event)
- event_copy.packet = self._packet
- cc_prio_map_copy = copy.copy(self._cc_prio_map)
- notif2 = bt2.EventNotification(event_copy, cc_prio_map_copy)
- self.assertNotEqual(notif, notif2)
-
- def test_eq_invalid(self):
- notif = bt2.EventNotification(self._event)
- self.assertNotEqual(notif, 23)
-
- def test_copy(self):
- notif = bt2.EventNotification(self._event, self._cc_prio_map)
- notif2 = copy.copy(notif)
- self.assertEqual(notif, notif2)
-
- def test_deepcopy(self):
- notif = bt2.EventNotification(self._event, self._cc_prio_map)
- notif2 = copy.deepcopy(notif)
- self.assertEqual(notif, notif2)
-
-
-@unittest.skip("this is broken")
-class PacketBeginningNotificationTestCase(_NotificationTestCase):
- def test_create(self):
- notif = bt2.PacketBeginningNotification(self._packet)
- self.assertEqual(notif.packet.addr, self._packet.addr)
-
- def test_eq(self):
- notif = bt2.PacketBeginningNotification(self._packet)
- packet_copy = copy.copy(self._packet)
- notif2 = bt2.PacketBeginningNotification(packet_copy)
- self.assertEqual(notif, notif2)
-
- def test_ne_packet(self):
- notif = bt2.PacketBeginningNotification(self._packet)
- packet_copy = copy.copy(self._packet)
- packet_copy.header_field['hello'] = 1847
- notif2 = bt2.PacketBeginningNotification(packet_copy)
- self.assertNotEqual(notif, notif2)
-
- def test_eq_invalid(self):
- notif = bt2.PacketBeginningNotification(self._packet)
- self.assertNotEqual(notif, 23)
-
- def test_copy(self):
- notif = bt2.PacketBeginningNotification(self._packet)
- notif2 = copy.copy(notif)
- self.assertEqual(notif, notif2)
-
- def test_deepcopy(self):
- notif = bt2.PacketBeginningNotification(self._packet)
- notif2 = copy.deepcopy(notif)
- self.assertEqual(notif, notif2)
-
-
-@unittest.skip("this is broken")
-class PacketEndNotificationTestCase(_NotificationTestCase):
- def test_create(self):
- notif = bt2.PacketEndNotification(self._packet)
- self.assertEqual(notif.packet.addr, self._packet.addr)
-
- def test_eq(self):
- notif = bt2.PacketEndNotification(self._packet)
- packet_copy = copy.copy(self._packet)
- notif2 = bt2.PacketEndNotification(packet_copy)
- self.assertEqual(notif, notif2)
-
- def test_ne_packet(self):
- notif = bt2.PacketEndNotification(self._packet)
- packet_copy = copy.copy(self._packet)
- packet_copy.header_field['hello'] = 1847
- notif2 = bt2.PacketEndNotification(packet_copy)
- self.assertNotEqual(notif, notif2)
-
- def test_eq_invalid(self):
- notif = bt2.PacketEndNotification(self._packet)
- self.assertNotEqual(notif, 23)
-
- def test_copy(self):
- notif = bt2.PacketEndNotification(self._packet)
- notif2 = copy.copy(notif)
- self.assertEqual(notif, notif2)
-
- def test_deepcopy(self):
- notif = bt2.PacketEndNotification(self._packet)
- notif2 = copy.deepcopy(notif)
- self.assertEqual(notif, notif2)
-
-
-@unittest.skip("this is broken")
-class StreamBeginningNotificationTestCase(_NotificationTestCase):
- def test_create(self):
- notif = bt2.StreamBeginningNotification(self._stream)
- self.assertEqual(notif.stream.addr, self._stream.addr)
-
- def test_eq(self):
- notif = bt2.StreamBeginningNotification(self._stream)
- stream_copy = copy.copy(self._stream)
- notif2 = bt2.StreamBeginningNotification(stream_copy)
- self.assertEqual(notif, notif2)
-
- def test_ne_stream(self):
- notif = bt2.StreamBeginningNotification(self._stream)
- stream_copy = self._sc(name='salut')
- notif2 = bt2.StreamBeginningNotification(stream_copy)
- self.assertNotEqual(notif, notif2)
-
- def test_eq_invalid(self):
- notif = bt2.StreamBeginningNotification(self._stream)
- self.assertNotEqual(notif, 23)
-
- def test_copy(self):
- notif = bt2.StreamBeginningNotification(self._stream)
- notif2 = copy.copy(notif)
- self.assertEqual(notif, notif2)
-
- def test_deepcopy(self):
- notif = bt2.StreamBeginningNotification(self._stream)
- notif2 = copy.deepcopy(notif)
- self.assertEqual(notif, notif2)
-
-
-@unittest.skip("this is broken")
-class StreamEndNotificationTestCase(_NotificationTestCase):
- def test_create(self):
- notif = bt2.StreamEndNotification(self._stream)
- self.assertEqual(notif.stream.addr, self._stream.addr)
-
- def test_eq(self):
- notif = bt2.StreamEndNotification(self._stream)
- stream_copy = copy.copy(self._stream)
- notif2 = bt2.StreamEndNotification(stream_copy)
- self.assertEqual(notif, notif2)
-
- def test_ne_stream(self):
- notif = bt2.StreamEndNotification(self._stream)
- stream_copy = self._sc(name='salut')
- notif2 = bt2.StreamEndNotification(stream_copy)
- self.assertNotEqual(notif, notif2)
-
- def test_eq_invalid(self):
- notif = bt2.StreamEndNotification(self._stream)
- self.assertNotEqual(notif, 23)
-
- def test_copy(self):
- notif = bt2.StreamEndNotification(self._stream)
- notif2 = copy.copy(notif)
- self.assertEqual(notif, notif2)
-
- def test_deepcopy(self):
- notif = bt2.StreamEndNotification(self._stream)
- notif2 = copy.deepcopy(notif)
- self.assertEqual(notif, notif2)
-
-
-@unittest.skip("this is broken")
-class InactivityNotificationTestCase(unittest.TestCase):
- def setUp(self):
- self._cc1 = bt2.ClockClass('cc1', 1000)
- self._cc2 = bt2.ClockClass('cc2', 2000)
- self._cc_prio_map = bt2.ClockClassPriorityMap()
- self._cc_prio_map[self._cc1] = 25
- self._cc_prio_map[self._cc2] = 50
-
- def tearDown(self):
- del self._cc1
- del self._cc2
- del self._cc_prio_map
-
- def test_create_no_cc_prio_map(self):
- notif = bt2.InactivityNotification()
- self.assertEqual(len(notif.clock_class_priority_map), 0)
-
- def test_create_with_cc_prio_map(self):
- notif = bt2.InactivityNotification(self._cc_prio_map)
- notif.clock_values.add(self._cc1(123))
- notif.clock_values.add(self._cc2(19487))
- self.assertEqual(len(notif.clock_class_priority_map), 2)
- self.assertEqual(notif.clock_class_priority_map, self._cc_prio_map)
- self.assertEqual(notif.clock_values[self._cc1], 123)
- self.assertEqual(notif.clock_values[self._cc2], 19487)
-
- def test_eq(self):
- notif = bt2.InactivityNotification(self._cc_prio_map)
- notif.clock_values.add(self._cc1(123))
- notif.clock_values.add(self._cc2(19487))
- cc_prio_map_copy = copy.copy(self._cc_prio_map)
- notif2 = bt2.InactivityNotification(cc_prio_map_copy)
- notif2.clock_values.add(self._cc1(123))
- notif2.clock_values.add(self._cc2(19487))
- self.assertEqual(notif, notif2)
-
- def test_ne_cc_prio_map(self):
- notif = bt2.InactivityNotification(self._cc_prio_map)
- notif.clock_values.add(self._cc1(123))
- notif.clock_values.add(self._cc2(19487))
- cc_prio_map_copy = copy.copy(self._cc_prio_map)
- cc_prio_map_copy[self._cc2] = 23
- notif2 = bt2.InactivityNotification(cc_prio_map_copy)
- self.assertNotEqual(notif, notif2)
-
- def test_ne_clock_value(self):
- notif = bt2.InactivityNotification(self._cc_prio_map)
- notif.clock_values.add(self._cc1(123))
- notif.clock_values.add(self._cc2(19487))
- notif2 = bt2.InactivityNotification(self._cc_prio_map)
- notif.clock_values.add(self._cc1(123))
- notif.clock_values.add(self._cc2(1847))
- self.assertNotEqual(notif, notif2)
-
- def test_eq_invalid(self):
- notif = bt2.InactivityNotification(self._cc_prio_map)
- self.assertNotEqual(notif, 23)
-
- def test_copy(self):
- notif = bt2.InactivityNotification(self._cc_prio_map)
- notif.clock_values.add(self._cc1(123))
- notif.clock_values.add(self._cc2(19487))
- notif_copy = copy.copy(notif)
- self.assertEqual(notif, notif_copy)
- self.assertNotEqual(notif.addr, notif_copy.addr)
- self.assertEqual(notif.clock_class_priority_map.addr,
- notif_copy.clock_class_priority_map.addr)
- self.assertEqual(notif_copy.clock_values[self._cc1], 123)
- self.assertEqual(notif_copy.clock_values[self._cc2], 19487)
-
- def test_deepcopy(self):
- notif = bt2.InactivityNotification(self._cc_prio_map)
- notif.clock_values.add(self._cc1(123))
- notif.clock_values.add(self._cc2(19487))
- notif_copy = copy.deepcopy(notif)
- self.assertEqual(notif, notif_copy)
- self.assertNotEqual(notif.addr, notif_copy.addr)
- self.assertNotEqual(notif.clock_class_priority_map.addr,
- notif_copy.clock_class_priority_map.addr)
- self.assertEqual(notif.clock_class_priority_map,
- notif_copy.clock_class_priority_map)
- self.assertNotEqual(list(notif.clock_class_priority_map)[0].addr,
- list(notif_copy.clock_class_priority_map)[0].addr)
- self.assertIsNone(notif_copy.clock_values[self._cc1])
- self.assertIsNone(notif_copy.clock_values[self._cc2])
- self.assertEqual(notif_copy.clock_values[list(notif_copy.clock_class_priority_map)[0]], 123)
- self.assertEqual(notif_copy.clock_values[list(notif_copy.clock_class_priority_map)[1]], 19487)
-
-
-@unittest.skip("this is broken")
-class DiscardedPacketsNotificationTestCase(unittest.TestCase):
- def setUp(self):
- self._trace = bt2.Trace()
- self._sc = bt2.StreamClass()
- self._ec = bt2.EventClass('salut')
- self._clock_class = bt2.ClockClass('yo', 1000)
- self._uint64_int_ft = bt2.IntegerFieldType(64, mapped_clock_class=self._clock_class)
- self._my_int_ft = bt2.IntegerFieldType(32)
- self._ec.payload_field_type = bt2.StructureFieldType()
- self._ec.payload_field_type += collections.OrderedDict([
- ('my_int', self._my_int_ft),
- ])
- self._sc.add_event_class(self._ec)
- self._sc.packet_context_field_type = bt2.StructureFieldType()
- self._sc.packet_context_field_type += collections.OrderedDict([
- ('packet_seq_num', self._my_int_ft),
- ('timestamp_begin', self._uint64_int_ft),
- ('timestamp_end', self._uint64_int_ft),
- ])
- self._trace.add_clock_class(self._clock_class)
- self._trace.add_stream_class(self._sc)
- self._stream = self._sc()
-
- def tearDown(self):
- del self._trace
- del self._sc
- del self._ec
- del self._clock_class
- del self._uint64_int_ft
- del self._my_int_ft
- del self._stream
-
- def _create_event(self, packet):
- event = self._ec()
- event.payload_field['my_int'] = 23
- event.packet = packet
- return event
-
- def _get_notif(self):
- class MyIter(bt2._UserNotificationIterator):
- def __init__(iter_self):
- packet1 = self._stream.create_packet()
- packet1.context_field['packet_seq_num'] = 0
- packet1.context_field['timestamp_begin'] = 3
- packet1.context_field['timestamp_end'] = 6
- packet2 = self._stream.create_packet()
- packet2.context_field['packet_seq_num'] = 5
- packet2.context_field['timestamp_begin'] = 7
- packet2.context_field['timestamp_end'] = 10
- iter_self._ev1 = self._create_event(packet1)
- iter_self._ev2 = self._create_event(packet2)
- iter_self._at = 0
-
- def __next__(self):
- if self._at == 0:
- notif = bt2.EventNotification(self._ev1)
- elif self._at == 1:
- notif = bt2.EventNotification(self._ev2)
- else:
- raise bt2.Stop
-
- self._at += 1
- return notif
-
- class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
- def __init__(self, params):
- self._add_output_port('out')
-
- class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
- self._add_input_port('in')
-
- def _consume(comp_self):
- nonlocal the_notif
- notif = next(comp_self._notif_iter)
-
- if type(notif) is bt2._DiscardedPacketsNotification:
- the_notif = notif
- raise bt2.Stop
-
- def _port_connected(self, port, other_port):
- self._notif_iter = port.connection.create_notification_iterator()
-
- the_notif = None
- 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'])
- graph.run()
- return the_notif
-
- def test_create(self):
- self.assertIsInstance(self._get_notif(), bt2._DiscardedPacketsNotification)
-
- def test_count(self):
- self.assertEqual(self._get_notif().count, 4)
-
- def test_stream(self):
- self.assertEqual(self._get_notif().stream.addr, self._stream.addr)
-
- def test_beginning_clock_value(self):
- notif = self._get_notif()
- beginning_clock_value = notif.beginning_clock_value
- self.assertEqual(beginning_clock_value.clock_class, self._clock_class)
- self.assertEqual(beginning_clock_value, 6)
-
- def test_end_clock_value(self):
- notif = self._get_notif()
- end_clock_value = notif.end_clock_value
- self.assertEqual(end_clock_value.clock_class, self._clock_class)
- self.assertEqual(end_clock_value, 7)
-
- def test_eq(self):
- notif1 = self._get_notif()
- notif2 = self._get_notif()
- self.assertEqual(notif1, notif2)
-
- def test_eq_invalid(self):
- notif1 = self._get_notif()
- self.assertNotEqual(notif1, 23)
-
-
-@unittest.skip("this is broken")
-class DiscardedEventsNotificationTestCase(unittest.TestCase):
- def setUp(self):
- self._trace = bt2.Trace()
- self._sc = bt2.StreamClass()
- self._ec = bt2.EventClass('salut')
- self._clock_class = bt2.ClockClass('yo', 1000)
- self._uint64_int_ft = bt2.IntegerFieldType(64, mapped_clock_class=self._clock_class)
- self._my_int_ft = bt2.IntegerFieldType(32)
- self._ec.payload_field_type = bt2.StructureFieldType()
- self._ec.payload_field_type += collections.OrderedDict([
- ('my_int', self._my_int_ft),
- ])
- self._sc.add_event_class(self._ec)
- self._sc.packet_context_field_type = bt2.StructureFieldType()
- self._sc.packet_context_field_type += collections.OrderedDict([
- ('events_discarded', self._my_int_ft),
- ('timestamp_begin', self._uint64_int_ft),
- ('timestamp_end', self._uint64_int_ft),
- ])
- self._trace.add_clock_class(self._clock_class)
- self._trace.add_stream_class(self._sc)
- self._stream = self._sc()
-
- def tearDown(self):
- del self._trace
- del self._sc
- del self._ec
- del self._clock_class
- del self._uint64_int_ft
- del self._my_int_ft
- del self._stream
-
- def _create_event(self, packet):
- event = self._ec()
- event.payload_field['my_int'] = 23
- event.packet = packet
- return event
-
- def _get_notif(self):
- class MyIter(bt2._UserNotificationIterator):
- def __init__(iter_self):
- packet1 = self._stream.create_packet()
- packet1.context_field['events_discarded'] = 0
- packet1.context_field['timestamp_begin'] = 3
- packet1.context_field['timestamp_end'] = 6
- packet2 = self._stream.create_packet()
- packet2.context_field['events_discarded'] = 10
- packet2.context_field['timestamp_begin'] = 7
- packet2.context_field['timestamp_end'] = 10
- iter_self._ev1 = self._create_event(packet1)
- iter_self._ev2 = self._create_event(packet2)
- iter_self._at = 0
-
- def __next__(self):
- if self._at == 0:
- notif = bt2.EventNotification(self._ev1)
- elif self._at == 1:
- notif = bt2.EventNotification(self._ev2)
- else:
- raise bt2.Stop
-
- self._at += 1
- return notif
-
- class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
- def __init__(self, params):
- self._add_output_port('out')
-
- class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
- self._add_input_port('in')
-
- def _consume(comp_self):
- nonlocal the_notif
- notif = next(comp_self._notif_iter)
-
- if type(notif) is bt2._DiscardedEventsNotification:
- the_notif = notif
- raise bt2.Stop
-
- def _port_connected(self, port, other_port):
- self._notif_iter = port.connection.create_notification_iterator()
-
- the_notif = None
- 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'])
- graph.run()
- return the_notif
-
- def test_create(self):
- self.assertIsInstance(self._get_notif(), bt2._DiscardedEventsNotification)
-
- def test_count(self):
- self.assertEqual(self._get_notif().count, 10)
-
- def test_stream(self):
- self.assertEqual(self._get_notif().stream.addr, self._stream.addr)
-
- def test_beginning_clock_value(self):
- notif = self._get_notif()
- beginning_clock_value = notif.beginning_clock_value
- self.assertEqual(beginning_clock_value.clock_class, self._clock_class)
- self.assertEqual(beginning_clock_value, 6)
-
- def test_end_clock_value(self):
- notif = self._get_notif()
- end_clock_value = notif.end_clock_value
- self.assertEqual(end_clock_value.clock_class, self._clock_class)
- self.assertEqual(end_clock_value, 10)
-
- def test_eq(self):
- notif1 = self._get_notif()
- notif2 = self._get_notif()
- self.assertEqual(notif1, notif2)
-
- def test_eq_invalid(self):
- notif1 = self._get_notif()
- self.assertNotEqual(notif1, 23)
+++ /dev/null
-from bt2 import values
-import collections
-import unittest
-import copy
-import bt2
-
-
-@unittest.skip("this is broken")
-class UserNotificationIteratorTestCase(unittest.TestCase):
- @staticmethod
- def _create_graph(src_comp_cls):
- class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
- self._add_input_port('in')
-
- def _consume(self):
- next(self._notif_iter)
-
- def _port_connected(self, port, other_port):
- self._notif_iter = port.connection.create_notification_iterator()
-
- 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'])
- return graph
-
- def test_init(self):
- class MyIter(bt2._UserNotificationIterator):
- def __init__(self):
- nonlocal initialized
- initialized = True
-
- class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
- def __init__(self, params):
- self._add_output_port('out')
-
- initialized = False
- graph = self._create_graph(MySource)
- self.assertTrue(initialized)
-
- def test_finalize(self):
- class MyIter(bt2._UserNotificationIterator):
- def _finalize(self):
- nonlocal finalized
- finalized = True
-
- class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
- def __init__(self, params):
- self._add_output_port('out')
-
- finalized = False
- graph = self._create_graph(MySource)
- del graph
- self.assertTrue(finalized)
-
- def test_component(self):
- class MyIter(bt2._UserNotificationIterator):
- def __init__(self):
- nonlocal salut
- salut = self._component._salut
-
- class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
- def __init__(self, params):
- self._add_output_port('out')
- self._salut = 23
-
- salut = None
- graph = self._create_graph(MySource)
- self.assertEqual(salut, 23)
-
- def test_addr(self):
- class MyIter(bt2._UserNotificationIterator):
- def __init__(self):
- nonlocal addr
- addr = self.addr
-
- class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
- def __init__(self, params):
- self._add_output_port('out')
-
- addr = None
- graph = self._create_graph(MySource)
- self.assertIsNotNone(addr)
- self.assertNotEqual(addr, 0)
-
-
-@unittest.skip("this is broken")
-class PrivateConnectionNotificationIteratorTestCase(unittest.TestCase):
- def test_component(self):
- class MyIter(bt2._UserNotificationIterator):
- pass
-
- class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
- def __init__(self, params):
- self._add_output_port('out')
-
- class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
- self._add_input_port('in')
-
- def _consume(self):
- next(self._notif_iter)
-
- def _port_connected(self, port, other_port):
- nonlocal upstream_comp
- self._notif_iter = port.connection.create_notification_iterator()
- upstream_comp = self._notif_iter.component
-
- upstream_comp = None
- graph = bt2.Graph()
- src_comp = graph.add_component(MySource, 'src')
- sink_comp = graph.add_component(MySink, 'sink')
- graph.connect_ports(src_comp.output_ports['out'],
- sink_comp.input_ports['in'])
- self.assertEqual(src_comp, upstream_comp)
- del upstream_comp
-
-
-@unittest.skip("this is broken")
-class OutputPortNotificationIteratorTestCase(unittest.TestCase):
- def test_component(self):
- class MyIter(bt2._UserNotificationIterator):
- def __init__(self):
- self._build_meta()
- self._at = 0
-
- def _build_meta(self):
- self._trace = bt2.Trace()
- self._sc = bt2.StreamClass()
- self._ec = bt2.EventClass('salut')
- self._my_int_ft = bt2.IntegerFieldType(32)
- self._ec.payload_field_type = bt2.StructureFieldType()
- self._ec.payload_field_type += collections.OrderedDict([
- ('my_int', self._my_int_ft),
- ])
- self._sc.add_event_class(self._ec)
- self._trace.add_stream_class(self._sc)
- self._stream = self._sc()
- self._packet = self._stream.create_packet()
-
- def _create_event(self, value):
- ev = self._ec()
- ev.payload_field['my_int'] = value
- ev.packet = self._packet
- return ev
-
- def __next__(self):
- if self._at == 5:
- raise bt2.Stop
-
- notif = bt2.EventNotification(self._create_event(self._at * 3))
- self._at += 1
- return notif
-
- class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
- def __init__(self, params):
- self._add_output_port('out')
-
- graph = bt2.Graph()
- src = graph.add_component(MySource, 'src')
- types = [bt2.EventNotification]
- notif_iter = src.output_ports['out'].create_notification_iterator(types)
-
- for at, notif in enumerate(notif_iter):
- self.assertIsInstance(notif, bt2.EventNotification)
- self.assertEqual(notif.event.event_class.name, 'salut')
- field = notif.event.payload_field['my_int']
- self.assertEqual(field, at * 3)
return graph.add_component(comp_cls, name)
def test_src_add_output_port(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
port = comp_self._add_output_port('out')
self.assertEqual(port.name, 'out')
def test_flt_add_output_port(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
port = comp_self._add_output_port('out')
self.assertEqual(port.name, 'out')
self.assertEqual(len(comp.output_ports), 1)
def test_flt_add_input_port(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
port = comp_self._add_input_port('in')
self.assertEqual(port.name, 'in')
self.assertEqual(len(comp.input_ports), 1)
def test_user_src_output_ports_getitem(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
comp = self._create_comp(MySource)
def test_user_flt_output_ports_getitem(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
comp = self._create_comp(MyFilter)
def test_user_flt_input_ports_getitem(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
port1 = comp_self._add_input_port('clear')
port2 = comp_self._add_input_port('print')
comp = self._create_comp(MySink)
def test_user_src_output_ports_getitem_invalid_key(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
comp = self._create_comp(MySource)
def test_user_flt_output_ports_getitem_invalid_key(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
comp = self._create_comp(MyFilter)
def test_user_flt_input_ports_getitem_invalid_key(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
comp = self._create_comp(MySink)
def test_user_src_output_ports_len(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
comp = self._create_comp(MySource)
def test_user_flt_output_ports_len(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
comp = self._create_comp(MyFilter)
def test_user_flt_input_ports_len(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
comp = self._create_comp(MySink)
def test_user_src_output_ports_iter(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
comp = self._create_comp(MySource)
def test_user_flt_output_ports_iter(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
comp = self._create_comp(MyFilter)
def test_user_flt_input_ports_iter(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
port1 = comp_self._add_input_port('clear')
port2 = comp_self._add_input_port('print')
comp = self._create_comp(MySink)
def test_gen_src_output_ports_getitem(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
port3 = None
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
nonlocal port1, port2, port3
port1 = comp_self._add_output_port('clear')
del port3
def test_gen_flt_output_ports_getitem(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
port3 = None
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
nonlocal port1, port2, port3
port1 = comp_self._add_output_port('clear')
del port3
def test_gen_flt_input_ports_getitem(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
port3 = None
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
nonlocal port1, port2, port3
port1 = comp_self._add_input_port('clear')
del port3
def test_gen_src_output_ports_getitem_invalid_key(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
comp.output_ports['hello']
def test_gen_flt_output_ports_getitem_invalid_key(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
comp.output_ports['hello']
def test_gen_flt_input_ports_getitem_invalid_key(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
comp.input_ports['hello']
def test_gen_src_output_ports_len(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
self.assertEqual(len(comp.output_ports), 3)
def test_gen_flt_output_ports_len(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
self.assertEqual(len(comp.output_ports), 3)
def test_gen_flt_input_ports_len(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
self.assertEqual(len(comp.input_ports), 3)
def test_gen_src_output_ports_iter(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
port3 = None
class MySource(bt2._UserSourceComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
nonlocal port1, port2, port3
port1 = comp_self._add_output_port('clear')
del port3
def test_gen_flt_output_ports_iter(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
port3 = None
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
nonlocal port1, port2, port3
port1 = comp_self._add_output_port('clear')
del port3
def test_gen_flt_input_ports_iter(self):
- class MyIter(bt2._UserNotificationIterator):
+ class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.Stop
port3 = None
class MyFilter(bt2._UserFilterComponent,
- notification_iterator_class=MyIter):
+ message_iterator_class=MyIter):
def __init__(comp_self, params):
nonlocal port1, port2, port3
port1 = comp_self._add_input_port('clear')
--- /dev/null
+import unittest
+import datetime
+import copy
+import uuid
+import bt2
+import os
+import os.path
+
+
+_TEST_CTF_TRACES_PATH = os.environ['TEST_CTF_TRACES_PATH']
+_3EVENTS_INTERSECT_TRACE_PATH = os.path.join(_TEST_CTF_TRACES_PATH,
+ 'intersection',
+ '3eventsintersect')
+
+
+@unittest.skip("this is broken")
+class ComponentSpecTestCase(unittest.TestCase):
+ def test_create_good_no_params(self):
+ spec = bt2.ComponentSpec('plugin', 'compcls')
+
+ def test_create_good_with_params(self):
+ spec = bt2.ComponentSpec('plugin', 'compcls', {'salut': 23})
+
+ def test_create_good_with_path_params(self):
+ spec = bt2.ComponentSpec('plugin', 'compcls', 'a path')
+ self.assertEqual(spec.params['path'], 'a path')
+
+ def test_create_wrong_plugin_name_type(self):
+ with self.assertRaises(TypeError):
+ spec = bt2.ComponentSpec(23, 'compcls')
+
+ def test_create_wrong_component_class_name_type(self):
+ with self.assertRaises(TypeError):
+ spec = bt2.ComponentSpec('plugin', 190)
+
+ def test_create_wrong_params_type(self):
+ with self.assertRaises(TypeError):
+ spec = bt2.ComponentSpec('dwdw', 'compcls', datetime.datetime.now())
+
+
+@unittest.skip("this is broken")
+class TraceCollectionMessageIteratorTestCase(unittest.TestCase):
+ def test_create_wrong_stream_intersection_mode_type(self):
+ specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+
+ with self.assertRaises(TypeError):
+ msg_iter = bt2.TraceCollectionMessageIterator(specs, stream_intersection_mode=23)
+
+ def test_create_wrong_begin_type(self):
+ specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+
+ with self.assertRaises(TypeError):
+ msg_iter = bt2.TraceCollectionMessageIterator(specs, begin='hi')
+
+ def test_create_wrong_end_type(self):
+ specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+
+ with self.assertRaises(TypeError):
+ msg_iter = bt2.TraceCollectionMessageIterator(specs, begin='lel')
+
+ def test_create_no_such_plugin(self):
+ specs = [bt2.ComponentSpec('77', '101', _3EVENTS_INTERSECT_TRACE_PATH)]
+
+ with self.assertRaises(bt2.Error):
+ msg_iter = bt2.TraceCollectionMessageIterator(specs)
+
+ def test_create_begin_s(self):
+ specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+ msg_iter = bt2.TraceCollectionMessageIterator(specs, begin=19457.918232)
+
+ def test_create_end_s(self):
+ specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+ msg_iter = bt2.TraceCollectionMessageIterator(specs, end=123.12312)
+
+ def test_create_begin_datetime(self):
+ specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+ msg_iter = bt2.TraceCollectionMessageIterator(specs, begin=datetime.datetime.now())
+
+ def test_create_end_datetime(self):
+ specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+ msg_iter = bt2.TraceCollectionMessageIterator(specs, end=datetime.datetime.now())
+
+ def test_iter_no_intersection(self):
+ specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+ msg_iter = bt2.TraceCollectionMessageIterator(specs)
+ self.assertEqual(len(list(msg_iter)), 28)
+
+ def test_iter_no_intersection_subscribe(self):
+ specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+ msg_iter = bt2.TraceCollectionMessageIterator(specs,
+ message_types=[bt2.EventMessage])
+ self.assertEqual(len(list(msg_iter)), 8)
+
+ def test_iter_specs_not_list(self):
+ spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
+ msg_iter = bt2.TraceCollectionMessageIterator(spec,
+ message_types=[bt2.EventMessage])
+ self.assertEqual(len(list(msg_iter)), 8)
+
+ def test_iter_custom_filter(self):
+ src_spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
+ flt_spec = bt2.ComponentSpec('utils', 'trimmer', {
+ 'end': 13515309000000075,
+ })
+ msg_iter = bt2.TraceCollectionMessageIterator(src_spec, flt_spec,
+ message_types=[bt2.EventMessage])
+ self.assertEqual(len(list(msg_iter)), 5)
+
+ def test_iter_intersection(self):
+ specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+ msg_iter = bt2.TraceCollectionMessageIterator(specs, stream_intersection_mode=True)
+ self.assertEqual(len(list(msg_iter)), 15)
+
+ def test_iter_intersection_subscribe(self):
+ specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+ msg_iter = bt2.TraceCollectionMessageIterator(specs, stream_intersection_mode=True,
+ message_types=[bt2.EventMessage])
+ self.assertEqual(len(list(msg_iter)), 3)
+
+ def test_iter_intersection_no_path_param(self):
+ specs = [bt2.ComponentSpec('text', 'dmesg', {'read-from-stdin': True})]
+
+ with self.assertRaises(bt2.Error):
+ msg_iter = bt2.TraceCollectionMessageIterator(specs, stream_intersection_mode=True,
+ message_types=[bt2.EventMessage])
+
+ def test_iter_no_intersection_two_traces(self):
+ spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
+ specs = [spec, spec]
+ msg_iter = bt2.TraceCollectionMessageIterator(specs)
+ self.assertEqual(len(list(msg_iter)), 56)
+
+ def test_iter_no_intersection_begin(self):
+ specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+ msg_iter = bt2.TraceCollectionMessageIterator(specs,
+ message_types=[bt2.EventMessage],
+ begin=13515309.000000023)
+ self.assertEqual(len(list(msg_iter)), 6)
+
+ def test_iter_no_intersection_end(self):
+ specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+ msg_iter = bt2.TraceCollectionMessageIterator(specs,
+ message_types=[bt2.EventMessage],
+ end=13515309.000000075)
+ self.assertEqual(len(list(msg_iter)), 5)
+++ /dev/null
-import unittest
-import datetime
-import copy
-import uuid
-import bt2
-import os
-import os.path
-
-
-_TEST_CTF_TRACES_PATH = os.environ['TEST_CTF_TRACES_PATH']
-_3EVENTS_INTERSECT_TRACE_PATH = os.path.join(_TEST_CTF_TRACES_PATH,
- 'intersection',
- '3eventsintersect')
-
-
-@unittest.skip("this is broken")
-class ComponentSpecTestCase(unittest.TestCase):
- def test_create_good_no_params(self):
- spec = bt2.ComponentSpec('plugin', 'compcls')
-
- def test_create_good_with_params(self):
- spec = bt2.ComponentSpec('plugin', 'compcls', {'salut': 23})
-
- def test_create_good_with_path_params(self):
- spec = bt2.ComponentSpec('plugin', 'compcls', 'a path')
- self.assertEqual(spec.params['path'], 'a path')
-
- def test_create_wrong_plugin_name_type(self):
- with self.assertRaises(TypeError):
- spec = bt2.ComponentSpec(23, 'compcls')
-
- def test_create_wrong_component_class_name_type(self):
- with self.assertRaises(TypeError):
- spec = bt2.ComponentSpec('plugin', 190)
-
- def test_create_wrong_params_type(self):
- with self.assertRaises(TypeError):
- spec = bt2.ComponentSpec('dwdw', 'compcls', datetime.datetime.now())
-
-
-@unittest.skip("this is broken")
-class TraceCollectionNotificationIteratorTestCase(unittest.TestCase):
- def test_create_wrong_stream_intersection_mode_type(self):
- specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
-
- with self.assertRaises(TypeError):
- notif_iter = bt2.TraceCollectionNotificationIterator(specs, stream_intersection_mode=23)
-
- def test_create_wrong_begin_type(self):
- specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
-
- with self.assertRaises(TypeError):
- notif_iter = bt2.TraceCollectionNotificationIterator(specs, begin='hi')
-
- def test_create_wrong_end_type(self):
- specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
-
- with self.assertRaises(TypeError):
- notif_iter = bt2.TraceCollectionNotificationIterator(specs, begin='lel')
-
- def test_create_no_such_plugin(self):
- specs = [bt2.ComponentSpec('77', '101', _3EVENTS_INTERSECT_TRACE_PATH)]
-
- with self.assertRaises(bt2.Error):
- notif_iter = bt2.TraceCollectionNotificationIterator(specs)
-
- def test_create_begin_s(self):
- specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
- notif_iter = bt2.TraceCollectionNotificationIterator(specs, begin=19457.918232)
-
- def test_create_end_s(self):
- specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
- notif_iter = bt2.TraceCollectionNotificationIterator(specs, end=123.12312)
-
- def test_create_begin_datetime(self):
- specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
- notif_iter = bt2.TraceCollectionNotificationIterator(specs, begin=datetime.datetime.now())
-
- def test_create_end_datetime(self):
- specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
- notif_iter = bt2.TraceCollectionNotificationIterator(specs, end=datetime.datetime.now())
-
- def test_iter_no_intersection(self):
- specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
- notif_iter = bt2.TraceCollectionNotificationIterator(specs)
- self.assertEqual(len(list(notif_iter)), 28)
-
- def test_iter_no_intersection_subscribe(self):
- specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
- notif_iter = bt2.TraceCollectionNotificationIterator(specs,
- notification_types=[bt2.EventNotification])
- self.assertEqual(len(list(notif_iter)), 8)
-
- def test_iter_specs_not_list(self):
- spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
- notif_iter = bt2.TraceCollectionNotificationIterator(spec,
- notification_types=[bt2.EventNotification])
- self.assertEqual(len(list(notif_iter)), 8)
-
- def test_iter_custom_filter(self):
- src_spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
- flt_spec = bt2.ComponentSpec('utils', 'trimmer', {
- 'end': 13515309000000075,
- })
- notif_iter = bt2.TraceCollectionNotificationIterator(src_spec, flt_spec,
- notification_types=[bt2.EventNotification])
- self.assertEqual(len(list(notif_iter)), 5)
-
- def test_iter_intersection(self):
- specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
- notif_iter = bt2.TraceCollectionNotificationIterator(specs, stream_intersection_mode=True)
- self.assertEqual(len(list(notif_iter)), 15)
-
- def test_iter_intersection_subscribe(self):
- specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
- notif_iter = bt2.TraceCollectionNotificationIterator(specs, stream_intersection_mode=True,
- notification_types=[bt2.EventNotification])
- self.assertEqual(len(list(notif_iter)), 3)
-
- def test_iter_intersection_no_path_param(self):
- specs = [bt2.ComponentSpec('text', 'dmesg', {'read-from-stdin': True})]
-
- with self.assertRaises(bt2.Error):
- notif_iter = bt2.TraceCollectionNotificationIterator(specs, stream_intersection_mode=True,
- notification_types=[bt2.EventNotification])
-
- def test_iter_no_intersection_two_traces(self):
- spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
- specs = [spec, spec]
- notif_iter = bt2.TraceCollectionNotificationIterator(specs)
- self.assertEqual(len(list(notif_iter)), 56)
-
- def test_iter_no_intersection_begin(self):
- specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
- notif_iter = bt2.TraceCollectionNotificationIterator(specs,
- notification_types=[bt2.EventNotification],
- begin=13515309.000000023)
- self.assertEqual(len(list(notif_iter)), 6)
-
- def test_iter_no_intersection_end(self):
- specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
- notif_iter = bt2.TraceCollectionNotificationIterator(specs,
- notification_types=[bt2.EventNotification],
- end=13515309.000000075)
- self.assertEqual(len(list(notif_iter)), 5)
projects / babeltrace.git / commitdiff
