from bt2.field import _ArrayField
from bt2.field import _EnumerationField
from bt2.field import _Field
-from bt2.field import _FloatingPointNumberField
+from bt2.field import _RealField
from bt2.field import _IntegerField
from bt2.field import _SequenceField
from bt2.field import _StringField
return bt2.ClockClass._create_from_ptr(ptr)
@property
- def cycles(self):
- ret, cycles = native_bt.clock_snapshot_get_value(self._ptr)
- assert(ret == 0)
- return cycles
+ def value(self):
+ return native_bt.clock_snapshot_get_value(self._ptr)
@property
def ns_from_epoch(self):
return tc
- def _create_clock_class(self):
+ def _create_clock_class(self, frequency=None):
ptr = self._as_self_component_ptr(self._ptr)
cc_ptr = native_bt.clock_class_create(ptr)
if cc_ptr is None:
raise bt2.CreationError('could not create clock class')
- return bt2.ClockClass._create_from_ptr(cc_ptr)
+ cc = bt2.ClockClass._create_from_ptr(cc_ptr)
+
+ if frequency is not None:
+ cc._frequency = frequency
+
+ return cc
class _UserSourceComponent(_UserComponent, _SourceComponent):
import bt2.stream
import bt2.field
import bt2.clock_snapshot
-import collections
-import numbers
-import copy
-import abc
import bt2
-def _create_from_ptr(ptr):
- # recreate the event class wrapper of this event's class (the
- # identity could be different, but the underlying address should be
- # the same)
- event_class_ptr = native_bt.event_get_class(ptr)
- utils._handle_ptr(event_class_ptr, "cannot get event object's class")
- event_class = bt2.EventClass._create_from_ptr(event_class_ptr)
- event = _Event._create_from_ptr(ptr)
- event._event_class = event_class
- return event
-
-
-class _EventClockSnapshotsIterator(collections.abc.Iterator):
- def __init__(self, event_clock_snapshots):
- self._event_clock_snapshots = event_clock_snapshots
- self._clock_classes = event_clock_snapshots._event._clock_classes
- self._at = 0
-
- def __next__(self):
- if self._at == len(self._clock_classes):
- raise StopIteration
-
- self._at += 1
- return self._clock_classes[at]
-
-
-class _EventClockSnapshots(collections.abc.Mapping):
- def __init__(self, event):
- self._event = event
-
- def __getitem__(self, clock_class):
- utils._check_type(clock_class, bt2.ClockClass)
- clock_snapshot_ptr = native_bt.event_get_clock_snapshot(self._event._ptr,
- clock_class._ptr)
-
- if clock_snapshot_ptr is None:
- return
-
- clock_snapshot = bt2.clock_snapshot._create_clock_snapshot_from_ptr(clock_snapshot_ptr)
- return clock_snapshot
-
- def add(self, clock_snapshot):
- utils._check_type(clock_snapshot, bt2.clock_snapshot._ClockSnapshot)
- ret = native_bt.event_set_clock_snapshot(self._ptr,
- clock_snapshot._ptr)
- utils._handle_ret(ret, "cannot set event object's clock value")
-
- def __len__(self):
- count = len(self._event._clock_classes)
- assert(count >= 0)
- return count
-
- def __iter__(self):
- return _EventClockSnapshotsIterator(self)
-
-
class _Event(object._UniqueObject):
@property
def event_class(self):
- return self._event_class
+ event_class_ptr = native_bt.event_borrow_class(self._ptr)
+ assert event_class_ptr is not None
+ return bt2.EventClass._create_from_ptr_and_get_ref(event_class_ptr)
@property
def name(self):
- return self._event_class.name
+ return self.event_class.name
@property
def id(self):
- return self._event_class.id
+ return self.event_class.id
@property
def packet(self):
- packet_ptr = native_bt.event_get_packet(self._ptr)
-
- if packet_ptr is None:
- return packet_ptr
-
- return bt2.packet._Packet._create_from_ptr(packet_ptr)
-
- @packet.setter
- def packet(self, packet):
- utils._check_type(packet, bt2.packet._Packet)
- ret = native_bt.event_set_packet(self._ptr, packet._ptr)
- utils._handle_ret(ret, "cannot set event object's packet object")
+ packet_ptr = native_bt.event_borrow_packet(self._ptr)
+ assert packet_ptr is not None
+ return bt2.packet._Packet._create_from_ptr_and_get_ref(packet_ptr)
@property
def stream(self):
- stream_ptr = native_bt.event_get_stream(self._ptr)
-
- if stream_ptr is None:
- return stream_ptr
-
- return bt2.stream._Stream._create_from_ptr(stream_ptr)
-
- @property
- def header_field(self):
- field_ptr = native_bt.event_get_header(self._ptr)
-
- if field_ptr is None:
- return
-
- return bt2.field._create_from_ptr(field_ptr)
-
- @header_field.setter
- def header_field(self, header_field):
- header_field_ptr = None
-
- if header_field is not None:
- utils._check_type(header_field, bt2.field._Field)
- header_field_ptr = header_field._ptr
-
- ret = native_bt.event_set_header(self._ptr, header_field_ptr)
- utils._handle_ret(ret, "cannot set event object's header field")
+ stream_ptr = native_bt.event_borrow_stream(self._ptr)
+ assert stream_ptr is not None
+ return bt2._Stream._create_from_ptr_and_get_ref(stream_ptr)
@property
- def stream_event_context_field(self):
- field_ptr = native_bt.event_get_stream_event_context(self._ptr)
+ def common_context_field(self):
+ field_ptr = native_bt.event_borrow_common_context_field(self._ptr)
if field_ptr is None:
return
- return bt2.field._create_from_ptr(field_ptr)
-
- @stream_event_context_field.setter
- def stream_event_context_field(self, stream_event_context):
- stream_event_context_ptr = None
-
- if stream_event_context is not None:
- utils._check_type(stream_event_context, bt2.field._Field)
- stream_event_context_ptr = stream_event_context._ptr
-
- ret = native_bt.event_set_stream_event_context(self._ptr,
- stream_event_context_ptr)
- utils._handle_ret(ret, "cannot set event object's stream event context field")
+ return bt2.field._create_field_from_ptr(field_ptr, self._owner_ptr,
+ self._owner_get_ref,
+ self._owner_put_ref)
@property
- def context_field(self):
- field_ptr = native_bt.event_get_event_context(self._ptr)
+ def specific_context_field(self):
+ field_ptr = native_bt.event_borrow_specific_context_field(self._ptr)
if field_ptr is None:
return
- return bt2.field._create_from_ptr(field_ptr)
-
- @context_field.setter
- def context_field(self, context):
- context_ptr = None
-
- if context is not None:
- utils._check_type(context, bt2.field._Field)
- context_ptr = context._ptr
-
- ret = native_bt.event_set_event_context(self._ptr, context_ptr)
- utils._handle_ret(ret, "cannot set event object's context field")
+ return bt2.field._create_field_from_ptr(field_ptr, self._owner_ptr,
+ self._owner_get_ref,
+ self._owner_put_ref)
@property
def payload_field(self):
- field_ptr = native_bt.event_get_event_payload(self._ptr)
+ field_ptr = native_bt.event_borrow_payload_field(self._ptr)
if field_ptr is None:
return
- return bt2.field._create_from_ptr(field_ptr)
-
- @payload_field.setter
- def payload_field(self, payload):
- payload_ptr = None
-
- if payload is not None:
- utils._check_type(payload, bt2.field._Field)
- payload_ptr = payload._ptr
-
- ret = native_bt.event_set_event_payload(self._ptr, payload_ptr)
- utils._handle_ret(ret, "cannot set event object's payload field")
-
- def _get_clock_snapshot_cycles(self, clock_class_ptr):
- clock_snapshot_ptr = native_bt.event_get_clock_snapshot(self._ptr,
- clock_class_ptr)
-
- if clock_snapshot_ptr is None:
- return
-
- ret, cycles = native_bt.clock_snapshot_get_value(clock_snapshot_ptr)
- native_bt.put(clock_snapshot_ptr)
- utils._handle_ret(ret, "cannot get clock value object's cycles")
- return cycles
-
- @property
- def clock_snapshots(self):
- return _EventClockSnapshots(self)
+ return bt2.field._create_field_from_ptr(field_ptr, self._owner_ptr,
+ self._owner_get_ref,
+ self._owner_put_ref)
def __getitem__(self, key):
utils._check_str(key)
if payload_field is not None and key in payload_field:
return payload_field[key]
- context_field = self.context_field
-
- if context_field is not None and key in context_field:
- return context_field[key]
-
- sec_field = self.stream_event_context_field
+ specific_context_field = self.specific_context_field
- if sec_field is not None and key in sec_field:
- return sec_field[key]
+ if specific_context_field is not None and key in specific_context_field:
+ return specific_context_field[key]
- header_field = self.header_field
+ common_context_field = self.common_context_field
- if header_field is not None and key in header_field:
- return header_field[key]
+ if common_context_field is not None and key in common_context_field:
+ return common_context_field[key]
- packet = self.packet
+ packet_context_field = self.packet.context_field
- if packet is None:
- raise KeyError(key)
-
- pkt_context_field = packet.context_field
-
- if pkt_context_field is not None and key in pkt_context_field:
- return pkt_context_field[key]
-
- pkt_header_field = packet.header_field
-
- if pkt_header_field is not None and key in pkt_header_field:
- return pkt_header_field[key]
+ if packet_context_field is not None and key in packet_context_field:
+ return packet_context_field[key]
raise KeyError(key)
-
- @property
- def _clock_classes(self):
- stream_class = self.event_class.stream_class
-
- if stream_class is None:
- return []
-
- trace = stream_class.trace
-
- if trace is None:
- return []
-
- clock_classes = []
-
- for clock_class in trace.clock_classes.values():
- clock_classes.append(clock_class)
-
- return clock_classes
-
- @property
- def _clock_class_ptrs(self):
- return [cc._ptr for cc in self._clock_classes]
-
- def __eq__(self, other):
- if type(other) is not type(self):
- return False
-
- if self.addr == other.addr:
- return True
-
- self_clock_snapshots = {}
- other_clock_snapshots = {}
-
- for clock_class_ptr in self._clock_class_ptrs:
- self_clock_snapshots[int(clock_class_ptr)] = self._get_clock_snapshot_cycles(clock_class_ptr)
-
- for clock_class_ptr in other._clock_class_ptrs:
- other_clock_snapshots[int(clock_class_ptr)] = self._get_clock_snapshot_cycles(clock_class_ptr)
-
- self_props = (
- self.header_field,
- self.stream_event_context_field,
- self.context_field,
- self.payload_field,
- self_clock_snapshots,
- )
- other_props = (
- other.header_field,
- other.stream_event_context_field,
- other.context_field,
- other.payload_field,
- other_clock_snapshots,
- )
- return self_props == other_props
-
- def _copy(self, copy_func):
- cpy = self.event_class()
-
- # copy fields
- cpy.header_field = copy_func(self.header_field)
- cpy.stream_event_context_field = copy_func(self.stream_event_context_field)
- cpy.context_field = copy_func(self.context_field)
- cpy.payload_field = copy_func(self.payload_field)
-
- # Copy known clock value references. It's not necessary to copy
- # clock class or clock value objects because once a clock value
- # is created from a clock class, the clock class is frozen.
- # Thus even if we copy the clock class, the user cannot modify
- # it, therefore it's useless to copy it.
- for clock_class in self._clock_classes:
- clock_snapshot = self.clock_snapshots[clock_class]
-
- if clock_snapshot is not None:
- cpy.clock_snapshots.add(clock_snapshot)
-
- return cpy
-
- def __copy__(self):
- return self._copy(copy.copy)
-
- def __deepcopy__(self, memo):
- cpy = self._copy(copy.deepcopy)
- memo[id(self)] = cpy
- return cpy
return _get_leaf_field(obj.selected_field)
-def _create_from_ptr(ptr):
- # recreate the field class wrapper of this field's type (the identity
- # could be different, but the underlying address should be the
- # same)
- field_class_ptr = native_bt.field_get_type(ptr)
- utils._handle_ptr(field_class_ptr, "cannot get field object's type")
- field_class = bt2.field_class._create_from_ptr(field_class_ptr)
- typeid = native_bt.field_class_get_type_id(field_class._ptr)
- field = _TYPE_ID_TO_OBJ[typeid]._create_from_ptr(ptr)
- field._field_class = field_class
+def _create_field_from_ptr(ptr, owner_ptr, owner_get_ref, owner_put_ref):
+ field_class_ptr = native_bt.field_borrow_class_const(ptr)
+ utils._handle_ptr(field_class_ptr, "cannot get field object's class")
+ typeid = native_bt.field_class_get_type(field_class_ptr)
+ field = _TYPE_ID_TO_OBJ[typeid]._create_from_ptr_and_get_ref(
+ ptr, owner_ptr, owner_get_ref, owner_put_ref)
return field
return self
-class _RealField(_NumericField, numbers.Real):
+class _IntegerField(_IntegralField, _Field):
pass
-class _IntegerField(_IntegralField):
- _NAME = 'Integer'
-
+class _UnsignedIntegerField(_IntegerField, _Field):
+ _NAME = 'Unsigned integer'
def _value_to_int(self, value):
if not isinstance(value, numbers.Real):
raise TypeError('expecting a real number object')
value = int(value)
-
- if self.field_class.is_signed:
- utils._check_int64(value)
- else:
- utils._check_uint64(value)
+ utils._check_uint64(value)
return value
@property
def _value(self):
- if self.field_class.is_signed:
- ret, value = native_bt.field_signed_integer_get_value(self._ptr)
- else:
- ret, value = native_bt.field_unsigned_integer_get_value(self._ptr)
+ return native_bt.field_unsigned_integer_get_value(self._ptr)
- if ret < 0:
- if not self.is_set:
- return
+ def _set_value(self, value):
+ value = self._value_to_int(value)
+ native_bt.field_unsigned_integer_set_value(self._ptr, value)
+
+ value = property(fset=_set_value)
+
+
+class _SignedIntegerField(_IntegerField, _Field):
+ _NAME = 'Signed integer'
+ def _value_to_int(self, value):
+ if not isinstance(value, numbers.Real):
+ raise TypeError('expecting a real number object')
- utils._handle_ret(ret, "cannot get integer field's value")
+ value = int(value)
+ utils._check_int64(value)
return value
+ @property
+ def _value(self):
+ return native_bt.field_signed_integer_get_value(self._ptr)
+
def _set_value(self, value):
value = self._value_to_int(value)
-
- if self.field_class.is_signed:
- ret = native_bt.field_signed_integer_set_value(self._ptr, value)
- else:
- ret = native_bt.field_unsigned_integer_set_value(self._ptr, value)
-
- utils._handle_ret(ret, "cannot set integer field object's value")
+ native_bt.field_signed_integer_set_value(self._ptr, value)
value = property(fset=_set_value)
-class _FloatingPointNumberField(_RealField):
- _NAME = 'Floating point number'
+class _RealField(_NumericField, numbers.Real):
+ _NAME = 'Real'
def _value_to_float(self, value):
if not isinstance(value, numbers.Real):
@property
def _value(self):
- ret, value = native_bt.field_floating_point_get_value(self._ptr)
-
- if ret < 0:
- if not self.is_set:
- return
-
- utils._handle_ret(ret, "cannot get floating point number field's value")
-
- return value
+ return native_bt.field_real_get_value(self._ptr)
def _set_value(self, value):
value = self._value_to_float(value)
- ret = native_bt.field_floating_point_set_value(self._ptr, value)
- utils._handle_ret(ret, "cannot set floating point number field object's value")
+ native_bt.field_real_set_value(self._ptr, value)
value = property(fset=_set_value)
def __getitem__(self, key):
utils._check_str(key)
- ptr = native_bt.field_structure_get_field_by_name(self._ptr, key)
+ field_ptr = native_bt.field_structure_borrow_member_field_by_name(self._ptr, key)
- if ptr is None:
+ if field_ptr is None:
raise KeyError(key)
- return _create_from_ptr(ptr)
+ return _create_field_from_ptr(field_ptr, self._owner_ptr,
+ self._owner_get_ref,
+ self._owner_put_ref)
def __setitem__(self, key, value):
# raises if key is somehow invalid
_TYPE_ID_TO_OBJ = {
+ native_bt.FIELD_CLASS_TYPE_UNSIGNED_INTEGER: _UnsignedIntegerField,
+ native_bt.FIELD_CLASS_TYPE_SIGNED_INTEGER: _SignedIntegerField,
+ native_bt.FIELD_CLASS_TYPE_REAL: _RealField,
+ native_bt.FIELD_CLASS_TYPE_STRING: _StringField,
+ native_bt.FIELD_CLASS_TYPE_STRUCTURE: _StructureField,
}
utils._handle_ret(ret, "cannot set integer field class object's mapped clock class")
+class _UnsignedIntegerFieldClass(_IntegerFieldClass):
+ pass
+
+
class _SignedIntegerFieldClass(_IntegerFieldClass):
pass
+
+class UnsignedIntegerFieldClass(_UnsignedIntegerFieldClass):
+ _NAME = 'UnsignedInteger'
+
+
class SignedIntegerFieldClass(_SignedIntegerFieldClass):
_NAME = 'SignedInteger'
-class FloatingPointNumberFieldClass(_FieldClass, _AlignmentProp, _ByteOrderProp):
- _NAME = 'Floating point number'
+
+class RealFieldClass(_FieldClass):
+ _NAME = 'Real'
def __init__(self, alignment=None, byte_order=None, exponent_size=None,
mantissa_size=None):
assert(is_canceled >= 0)
return is_canceled > 0
+ def create_output_port_message_iterator(self, output_port):
+ utils._check_type(output_port, bt2.port._OutputPort)
+ msg_iter_ptr = native_bt.port_output_message_iterator_create(self._ptr, output_port._ptr)
+
+ if msg_iter_ptr is None:
+ raise bt2.CreationError('cannot create output port message iterator')
+
+ return bt2.message_iterator._OutputPortMessageIterator(msg_iter_ptr)
+
def __eq__(self, other):
if type(other) is not type(self):
return False
class _Message(object._SharedObject):
- pass
+ _get_ref = staticmethod(native_bt.message_get_ref)
+ _put_ref = staticmethod(native_bt.message_put_ref)
class _CopyableMessage(_Message):
return cpy
-class EventMessage(_CopyableMessage):
+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)
+ @property
+ def event(self):
+ event_ptr = native_bt.message_event_borrow_event(self._ptr)
+ assert event_ptr is not None
+ return bt2.event._Event._create_from_ptr_and_get_ref(
+ event_ptr, self._ptr, self._get_ref, self._put_ref)
- if ptr is None:
- raise bt2.CreationError('cannot create event message object')
+ @property
+ def default_clock_snapshot(self):
+ if self.event.event_class.stream_class.default_clock_class is None:
+ return None
- super().__init__(ptr)
+ snapshot_ptr = native_bt.message_event_borrow_default_clock_snapshot_const(self._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)
+ return bt2.clock_snapshot._ClockSnapshot._create_from_ptr_and_get_ref(
+ snapshot_ptr, self._ptr, self._get_ref, self._put_ref)
@property
def clock_class_priority_map(self):
return EventMessage(self.event, self.clock_class_priority_map)
-class PacketBeginningMessage(_CopyableMessage):
+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)
return PacketEndMessage(self.packet)
-class StreamBeginningMessage(_CopyableMessage):
+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)
_MESSAGE_TYPE_TO_CLS = {
- native_bt.MESSAGE_TYPE_EVENT: EventMessage,
- native_bt.MESSAGE_TYPE_PACKET_BEGINNING: PacketBeginningMessage,
+ 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_BEGINNING: _StreamBeginningMessage,
native_bt.MESSAGE_TYPE_STREAM_END: StreamEndMessage,
native_bt.MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY: InactivityMessage,
native_bt.MESSAGE_TYPE_DISCARDED_PACKETS: _DiscardedPacketsMessage,
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:
+ if 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)
class _GenericMessageIterator(object._SharedObject, _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 __init__(self, ptr):
+ self._current_msgs = []
+ self._at = 0
+ super().__init__(ptr)
def __next__(self):
- self._next()
- return self._get_msg()
+ if len(self._current_msgs) == self._at:
+ status, msgs = self._get_msg_range(self._ptr)
+ self._handle_status(status,
+ 'unexpected error: cannot advance the message iterator')
+ self._current_msgs = msgs
+ self._at = 0
+
+ msg_ptr = self._current_msgs[self._at]
+ self._at += 1
+
+ return bt2.message._create_from_ptr(msg_ptr)
class _PrivateConnectionMessageIterator(_GenericMessageIterator):
class _OutputPortMessageIterator(_GenericMessageIterator):
- pass
+ _get_msg_range = staticmethod(native_bt.py3_port_output_get_msg_range)
+ _get_ref = staticmethod(native_bt.port_output_message_iterator_get_ref)
+ _put_ref = staticmethod(native_bt.port_output_message_iterator_put_ref)
class _UserMessageIterator(_MessageIterator):
utils._check_type(msg, bt2.message._Message)
- # take a new reference for the native part
- msg._get()
- return int(msg._ptr)
+ # Release the reference to the native part.
+ ptr = msg._release()
+ return int(ptr)
+
+ # Validate that the presence or lack of presence of a
+ # `default_clock_snapshot` value is valid in the context of `stream_class`.
+ @staticmethod
+ def _validate_default_clock_snapshot(stream_class, default_clock_snapshot):
+ stream_class_has_default_clock_class = stream_class.default_clock_class is not None
+
+ if stream_class_has_default_clock_class and default_clock_snapshot is None:
+ raise bt2.Error(
+ 'stream class has a default clock class, default_clock_snapshot should not be None')
+
+ if not stream_class_has_default_clock_class and default_clock_snapshot is not None:
+ raise bt2.Error(
+ 'stream class has no default clock class, default_clock_snapshot should be None')
+
+ def _create_event_message(self, event_class, packet, default_clock_snapshot=None):
+ utils._check_type(event_class, bt2.event_class.EventClass)
+ utils._check_type(packet, bt2.packet._Packet)
+ self._validate_default_clock_snapshot(packet.stream.stream_class, default_clock_snapshot)
+
+ if default_clock_snapshot is not None:
+ utils._check_uint64(default_clock_snapshot)
+ ptr = native_bt.message_event_create_with_default_clock_snapshot(
+ self._ptr, event_class._ptr, packet._ptr, default_clock_snapshot)
+ else:
+ ptr = native_bt.message_event_create(
+ self._ptr, event_class._ptr, packet._ptr)
+
+ if ptr is None:
+ raise bt2.CreationError('cannot create event message object')
+
+ return bt2.message._EventMessage(ptr)
+
+ def _create_stream_beginning_message(self, stream):
+ utils._check_type(stream, bt2.stream._Stream)
+
+ ptr = native_bt.message_stream_beginning_create(self._ptr, stream._ptr)
+ if ptr is None:
+ raise bt2.CreationError('cannot create stream beginning message object')
+
+ return bt2.message._StreamBeginningMessage(ptr)
+
+ def _create_packet_beginning_message(self, packet, default_clock_snapshot=None):
+ utils._check_type(packet, bt2.packet._Packet)
+
+ if packet.stream.stream_class.packets_have_default_beginning_clock_snapshot:
+ if default_clock_snapshot is None:
+ raise bt2.CreationError("packet beginning messages in this stream must have a default clock snapshots")
+
+ utils._check_uint64(default_clock_snapshot)
+ ptr = native_bt.message_packet_beginning_create_with_default_clock_snapshot(
+ self._ptr, packet._ptr, default_clock_snapshot)
+ else:
+ if default_clock_snapshot is not None:
+ raise bt2.CreationError("packet beginning messages in this stream must not have a default clock snapshots")
+
+ ptr = native_bt.message_packet_beginning_create(self._ptr, packet._ptr)
+
+ if ptr is None:
+ raise bt2.CreationError('cannot create packet beginning message object')
+
+ return bt2.message._PacketBeginningMessage(ptr)
def _create_from_ptr_and_get_ref(cls, ptr, owner_ptr,
owner_get_ref, owner_put_ref):
obj = cls.__new__(cls)
+
obj._ptr = ptr
obj._owner_ptr = owner_ptr
+ obj._owner_get_ref = owner_get_ref
obj._owner_put_ref = owner_put_ref
- owner_get_ref(obj._owner_ptr)
+
+ obj._owner_get_ref(obj._owner_ptr)
+
return obj
def __del__(self):
class _Packet(object._SharedObject):
+ _get_ref = staticmethod(native_bt.packet_get_ref)
+ _put_ref = staticmethod(native_bt.packet_put_ref)
+
@property
def stream(self):
- stream_ptr = native_bt.packet_get_stream(self._ptr)
- assert(stream_ptr)
- return bt2.stream._Stream._create_from_ptr(stream_ptr)
+ stream_ptr = native_bt.packet_borrow_stream(self._ptr)
+ assert stream_ptr is not None
+ return bt2.stream._Stream._create_from_ptr_and_get_ref(stream_ptr)
@property
def header_field(self):
@property
def context_field(self):
- field_ptr = native_bt.packet_get_context(self._ptr)
+ field_ptr = native_bt.packet_borrow_context_field(self._ptr)
if field_ptr is None:
return
- return bt2.field._create_from_ptr(field_ptr)
+ return bt2.field._create_field_from_ptr(field_ptr, self._ptr,
+ self._get_ref,
+ self._put_ref)
+
@context_field.setter
def context_field(self, context_field):
bt2.field_class.SignedIntegerFieldClass,
'signed integer', range, display_base)
+ def create_unsigned_integer_field_class(self, range=None, display_base=None):
+ return self._create_integer_field_class(native_bt.field_class_unsigned_integer_create,
+ bt2.field_class.UnsignedIntegerFieldClass,
+ 'unsigned integer', range, display_base)
+
+ def create_real_field_class(self, is_single_precision=False):
+ field_class_ptr = native_bt.field_class_real_create(self._ptr)
+ self._check_create_status(field_class_ptr, 'real')
+
+ field_class = bt2.field_class.RealFieldClass._create_from_ptr(field_class_ptr)
+
+ field_class._single_precision = is_single_precision
+
+ return field_class
+
def create_structure_field_class(self):
field_class_ptr = native_bt.field_class_structure_create(self._ptr)
self._check_create_status(field_class_ptr, 'structure')
from collections import OrderedDict
-from bt2 import value
import unittest
-import copy
import bt2
-@unittest.skip("this is broken")
class EventTestCase(unittest.TestCase):
- def setUp(self):
- self._ec = self._create_ec()
-
- def tearDown(self):
- del self._ec
-
- def _create_ec(self, with_eh=True, with_sec=True, with_ec=True, with_ep=True):
- # event header
- if with_eh:
- eh = bt2.StructureFieldClass()
- eh += OrderedDict((
- ('id', bt2.IntegerFieldClass(8)),
- ('ts', bt2.IntegerFieldClass(32)),
- ))
- else:
- eh = None
-
- # stream event context
- if with_sec:
- sec = bt2.StructureFieldClass()
- sec += OrderedDict((
- ('cpu_id', bt2.IntegerFieldClass(8)),
- ('stuff', bt2.FloatingPointNumberFieldClass()),
- ))
- else:
- sec = None
-
- # packet context
- pc = bt2.StructureFieldClass()
- pc += OrderedDict((
- ('something', bt2.IntegerFieldClass(8)),
- ('something_else', bt2.FloatingPointNumberFieldClass()),
- ))
-
- # stream class
- sc = bt2.StreamClass()
- sc.packet_context_field_class = pc
- sc.event_header_field_class = eh
- sc.event_context_field_class = sec
-
- # event context
- if with_ec:
- ec = bt2.StructureFieldClass()
- ec += OrderedDict((
- ('ant', bt2.IntegerFieldClass(16, is_signed=True)),
- ('msg', bt2.StringFieldClass()),
- ))
- else:
- ec = None
-
- # event payload
- if with_ep:
- ep = bt2.StructureFieldClass()
- ep += OrderedDict((
- ('giraffe', bt2.IntegerFieldClass(32)),
- ('gnu', bt2.IntegerFieldClass(8)),
- ('mosquito', bt2.IntegerFieldClass(8)),
- ))
- else:
- ep = None
-
- # event class
- event_class = bt2.EventClass('ec')
- event_class.context_field_class = ec
- event_class.payload_field_class = ep
- sc.add_event_class(event_class)
- return event_class
+ def _create_test_event_message(self, packet_fields_config=None,
+ event_fields_config=None,
+ with_clockclass=False,
+ with_cc=False, with_sc=False,
+ with_ep=False):
+
+ class MyIter(bt2._UserMessageIterator):
+ def __init__(self):
+ self._at = 0
+
+ def __next__(self):
+ if self._at == 0:
+ msg = self._create_stream_beginning_message(test_obj.stream)
+ elif self._at == 1:
+ assert test_obj.packet
+ msg = self._create_packet_beginning_message(test_obj.packet)
+ elif self._at == 2:
+ default_clock_snapshot = 789 if with_clockclass else None
+ assert test_obj.packet
+ msg = self._create_event_message(test_obj.event_class, test_obj.packet, default_clock_snapshot)
+ if event_fields_config is not None:
+ event_fields_config(msg.event)
+ elif self._at == 3:
+ msg = self._create_packet_end_message(test_obj.packet)
+ elif self._at == 4:
+ msg = self._create_stream_end_message(test_obj.stream)
+ elif self._at >= 5:
+ raise bt2.Stop
+
+ self._at += 1
+ return msg
+
+ class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ def __init__(self, params):
+ self._add_output_port('out')
+ tc = self._create_trace_class()
+
+ clock_class = None
+ if with_clockclass:
+ clock_class = self._create_clock_class(frequency=1000)
+
+ # event common context (stream-class-defined)
+ cc = None
+ if with_cc:
+ cc = tc.create_structure_field_class()
+ cc += OrderedDict((
+ ('cpu_id', tc.create_signed_integer_field_class(8)),
+ ('stuff', tc.create_real_field_class()),
+ ))
+
+ # packet context (stream-class-defined)
+ pc = tc.create_structure_field_class()
+ pc += OrderedDict((
+ ('something', tc.create_unsigned_integer_field_class(8)),
+ ('something_else', tc.create_real_field_class()),
+ ))
+
+ stream_class = tc.create_stream_class(default_clock_class=clock_class,
+ event_common_context_field_class=cc,
+ packet_context_field_class=pc)
+
+ # specific context (event-class-defined)
+ sc = None
+ if with_sc:
+ sc = tc.create_structure_field_class()
+ sc += OrderedDict((
+ ('ant', tc.create_signed_integer_field_class(16)),
+ ('msg', tc.create_string_field_class()),
+ ))
+
+ # event payload
+ ep = None
+ if with_ep:
+ ep = tc.create_structure_field_class()
+ ep += OrderedDict((
+ ('giraffe', tc.create_signed_integer_field_class(32)),
+ ('gnu', tc.create_signed_integer_field_class(8)),
+ ('mosquito', tc.create_signed_integer_field_class(8)),
+ ))
+
+ event_class = stream_class.create_event_class(name='garou',
+ specific_context_field_class=sc,
+ payload_field_class=ep)
+
+ trace = tc()
+ stream = trace.create_stream(stream_class)
+ packet = stream.create_packet()
+
+ if packet_fields_config is not None:
+ packet_fields_config(packet)
+
+ test_obj.packet = packet
+ test_obj.stream = stream
+ test_obj.event_class = event_class
+
+ test_obj = self
+ self._graph = bt2.Graph()
+ self._src_comp = self._graph.add_component(MySrc, 'my_source')
+ self._msg_iter = self._graph.create_output_port_message_iterator(self._src_comp.output_ports['out'])
+
+ for i, msg in enumerate(self._msg_iter):
+ if i == 2:
+ return msg
def test_attr_event_class(self):
- ev = self._ec()
- self.assertEqual(ev.event_class.addr, self._ec.addr)
+ msg = self._create_test_event_message()
+ self.assertEqual(msg.event.event_class.addr, self.event_class.addr)
def test_attr_name(self):
- ev = self._ec()
- self.assertEqual(ev.name, self._ec.name)
+ msg = self._create_test_event_message()
+ self.assertEqual(msg.event.name, self.event_class.name)
def test_attr_id(self):
- ev = self._ec()
- self.assertEqual(ev.id, self._ec.id)
-
- def test_get_event_header_field(self):
- ev = self._ec()
- ev.header_field['id'] = 23
- ev.header_field['ts'] = 1234
- self.assertEqual(ev.header_field['id'], 23)
- self.assertEqual(ev.header_field['ts'], 1234)
-
- def test_set_event_header_field(self):
- eh = self._ec.stream_class.event_header_field_class()
- eh['id'] = 17
- eh['ts'] = 188
- ev = self._ec()
- ev.header_field = eh
- self.assertEqual(ev.header_field['id'], 17)
- self.assertEqual(ev.header_field['ts'], 188)
-
- def test_get_stream_event_context_field(self):
- ev = self._ec()
- ev.stream_event_context_field['cpu_id'] = 1
- ev.stream_event_context_field['stuff'] = 13.194
- self.assertEqual(ev.stream_event_context_field['cpu_id'], 1)
- self.assertEqual(ev.stream_event_context_field['stuff'], 13.194)
-
- def test_set_stream_event_context_field(self):
- sec = self._ec.stream_class.event_context_field_class()
- sec['cpu_id'] = 2
- sec['stuff'] = 19.19
- ev = self._ec()
- ev.stream_event_context_field = sec
- self.assertEqual(ev.stream_event_context_field['cpu_id'], 2)
- self.assertEqual(ev.stream_event_context_field['stuff'], 19.19)
-
- def test_no_stream_event_context(self):
- ec = self._create_ec(with_sec=False)
- ev = ec()
- self.assertIsNone(ev.stream_event_context_field)
-
- def test_get_event_context_field(self):
- ev = self._ec()
- ev.context_field['ant'] = -1
- ev.context_field['msg'] = 'hellooo'
- self.assertEqual(ev.context_field['ant'], -1)
- self.assertEqual(ev.context_field['msg'], 'hellooo')
-
- def test_set_event_context_field(self):
- ec = self._ec.context_field_class()
- ec['ant'] = 2
- ec['msg'] = 'hi there'
- ev = self._ec()
- ev.context_field = ec
- self.assertEqual(ev.context_field['ant'], 2)
- self.assertEqual(ev.context_field['msg'], 'hi there')
-
- def test_no_event_context(self):
- ec = self._create_ec(with_ec=False)
- ev = ec()
- self.assertIsNone(ev.context_field)
+ msg = self._create_test_event_message()
+ self.assertEqual(msg.event.id, self.event_class.id)
+
+ def test_get_common_context_field(self):
+ def event_fields_config(event):
+ event.common_context_field['cpu_id'] = 1
+ event.common_context_field['stuff'] = 13.194
+
+ msg = self._create_test_event_message(event_fields_config=event_fields_config, with_cc=True)
+
+ self.assertEqual(msg.event.common_context_field['cpu_id'], 1)
+ self.assertEqual(msg.event.common_context_field['stuff'], 13.194)
+
+ def test_no_common_context_field(self):
+ msg = self._create_test_event_message(with_cc=False)
+ self.assertIsNone(msg.event.common_context_field)
+
+ def test_get_specific_context_field(self):
+ def event_fields_config(event):
+ event.specific_context_field['ant'] = -1
+ event.specific_context_field['msg'] = 'hellooo'
+
+ msg = self._create_test_event_message(event_fields_config=event_fields_config, with_sc=True)
+
+ self.assertEqual(msg.event.specific_context_field['ant'], -1)
+ self.assertEqual(msg.event.specific_context_field['msg'], 'hellooo')
+
+ def test_no_specific_context_field(self):
+ msg = self._create_test_event_message(with_sc=False)
+ self.assertIsNone(msg.event.specific_context_field)
def test_get_event_payload_field(self):
- ev = self._ec()
- ev.payload_field['giraffe'] = 1
- ev.payload_field['gnu'] = 23
- ev.payload_field['mosquito'] = 42
- self.assertEqual(ev.payload_field['giraffe'], 1)
- self.assertEqual(ev.payload_field['gnu'], 23)
- self.assertEqual(ev.payload_field['mosquito'], 42)
-
- def test_set_event_payload_field(self):
- ep = self._ec.payload_field_class()
- ep['giraffe'] = 2
- ep['gnu'] = 124
- ep['mosquito'] = 17
- ev = self._ec()
- ev.payload_field = ep
- self.assertEqual(ev.payload_field['giraffe'], 2)
- self.assertEqual(ev.payload_field['gnu'], 124)
- self.assertEqual(ev.payload_field['mosquito'], 17)
-
- def test_clock_snapshot(self):
- tc = bt2.Trace()
- tc.add_stream_class(self._ec.stream_class)
- cc = bt2.ClockClass('hi', 1000)
- tc.add_clock_class(cc)
- ev = self._ec()
- ev.clock_snapshots.add(cc(177))
- self.assertEqual(ev.clock_snapshots[cc].cycles, 177)
-
- def test_no_clock_snapshot(self):
- tc = bt2.Trace()
- tc.add_stream_class(self._ec.stream_class)
- cc = bt2.ClockClass('hi', 1000)
- tc.add_clock_class(cc)
- ev = self._ec()
- self.assertIsNone(ev.clock_snapshots[cc])
-
- def test_no_packet(self):
- ev = self._ec()
- self.assertIsNone(ev.packet)
-
- def test_packet(self):
- tc = bt2.Trace()
- tc.packet_header_field_class = bt2.StructureFieldClass()
- tc.packet_header_field_class.append_field('magic', bt2.IntegerFieldClass(32))
- tc.packet_header_field_class.append_field('stream_id', bt2.IntegerFieldClass(16))
- tc.add_stream_class(self._ec.stream_class)
- ev = self._ec()
- self._fill_ev(ev)
- stream = self._ec.stream_class()
- packet = stream.create_packet()
- packet.header_field['magic'] = 0xc1fc1fc1
- packet.header_field['stream_id'] = 0
- packet.context_field['something'] = 154
- packet.context_field['something_else'] = 17.2
- ev.packet = packet
- self.assertEqual(ev.packet.addr, packet.addr)
-
- def test_no_stream(self):
- ev = self._ec()
- self.assertIsNone(ev.stream)
+ def event_fields_config(event):
+ event.payload_field['giraffe'] = 1
+ event.payload_field['gnu'] = 23
+ event.payload_field['mosquito'] = 42
- def test_stream(self):
- tc = bt2.Trace()
- tc.packet_header_field_class = bt2.StructureFieldClass()
- tc.packet_header_field_class.append_field('magic', bt2.IntegerFieldClass(32))
- tc.packet_header_field_class.append_field('stream_id', bt2.IntegerFieldClass(16))
- tc.add_stream_class(self._ec.stream_class)
- ev = self._ec()
- self._fill_ev(ev)
- stream = self._ec.stream_class()
- packet = stream.create_packet()
- packet.header_field['magic'] = 0xc1fc1fc1
- packet.header_field['stream_id'] = 0
- packet.context_field['something'] = 154
- packet.context_field['something_else'] = 17.2
- ev.packet = packet
- self.assertEqual(ev.stream.addr, stream.addr)
-
- def _fill_ev(self, ev):
- ev.header_field['id'] = 23
- ev.header_field['ts'] = 1234
- ev.stream_event_context_field['cpu_id'] = 1
- ev.stream_event_context_field['stuff'] = 13.194
- ev.context_field['ant'] = -1
- ev.context_field['msg'] = 'hellooo'
- ev.payload_field['giraffe'] = 1
- ev.payload_field['gnu'] = 23
- ev.payload_field['mosquito'] = 42
-
- def _get_full_ev(self):
- tc = bt2.Trace()
- tc.add_stream_class(self._ec.stream_class)
- cc = bt2.ClockClass('hi', 1000)
- tc.add_clock_class(cc)
- ev = self._ec()
- self._fill_ev(ev)
- ev.clock_snapshots.add(cc(234))
- return ev
+ msg = self._create_test_event_message(event_fields_config=event_fields_config, with_ep=True)
- def test_getitem(self):
- tc = bt2.Trace()
- tc.packet_header_field_class = bt2.StructureFieldClass()
- tc.packet_header_field_class.append_field('magic', bt2.IntegerFieldClass(32))
- tc.packet_header_field_class.append_field('stream_id', bt2.IntegerFieldClass(16))
- tc.add_stream_class(self._ec.stream_class)
- ev = self._ec()
- self._fill_ev(ev)
- stream = self._ec.stream_class()
- packet = stream.create_packet()
- packet.header_field['magic'] = 0xc1fc1fc1
- packet.header_field['stream_id'] = 0
- packet.context_field['something'] = 154
- packet.context_field['something_else'] = 17.2
+ self.assertEqual(msg.event.payload_field['giraffe'], 1)
+ self.assertEqual(msg.event.payload_field['gnu'], 23)
+ self.assertEqual(msg.event.payload_field['mosquito'], 42)
- with self.assertRaises(KeyError):
- ev['magic']
+ def test_no_payload_field(self):
+ msg = self._create_test_event_message(with_ep=False)
+ self.assertIsNone(msg.event.payload_field)
- ev.packet = packet
- self.assertEqual(ev['mosquito'], 42)
- self.assertEqual(ev['gnu'], 23)
+ def test_clock_value(self):
+ msg = self._create_test_event_message(with_clockclass=True)
+ self.assertEqual(msg.default_clock_snapshot.value, 789)
+
+ def test_no_clock_value(self):
+ msg = self._create_test_event_message(with_clockclass=False)
+ self.assertIsNone(msg.default_clock_snapshot)
+
+ def test_stream(self):
+ msg = self._create_test_event_message()
+ self.assertEqual(msg.event.stream.addr, self.stream.addr)
+
+ def test_getitem(self):
+ def event_fields_config(event):
+ event.payload_field['giraffe'] = 1
+ event.payload_field['gnu'] = 23
+ event.payload_field['mosquito'] = 42
+ event.specific_context_field['ant'] = -1
+ event.specific_context_field['msg'] = 'hellooo'
+ event.common_context_field['cpu_id'] = 1
+ event.common_context_field['stuff'] = 13.194
+
+ def packet_fields_config(packet):
+ packet.context_field['something'] = 154
+ packet.context_field['something_else'] = 17.2
+
+ msg = self._create_test_event_message(packet_fields_config=packet_fields_config,
+ event_fields_config=event_fields_config,
+ with_cc=True, with_sc=True, with_ep=True)
+ ev = msg.event
+
+ # Test event fields
self.assertEqual(ev['giraffe'], 1)
- self.assertEqual(ev['msg'], 'hellooo')
+ self.assertEqual(ev['gnu'], 23)
+ self.assertEqual(ev['mosquito'], 42)
self.assertEqual(ev['ant'], -1)
- self.assertEqual(ev['stuff'], 13.194)
+ self.assertEqual(ev['msg'], 'hellooo')
self.assertEqual(ev['cpu_id'], 1)
- self.assertEqual(ev['ts'], 1234)
- self.assertEqual(ev['id'], 23)
- self.assertEqual(ev['something_else'], 17.2)
+ self.assertEqual(ev['stuff'], 13.194)
+
+ # Test packet fields
self.assertEqual(ev['something'], 154)
- self.assertEqual(ev['stream_id'], 0)
- self.assertEqual(ev['magic'], 0xc1fc1fc1)
+ self.assertEqual(ev['something_else'], 17.2)
with self.assertRaises(KeyError):
ev['yes']
- def test_eq(self):
- tc = bt2.Trace()
- tc.add_stream_class(self._ec.stream_class)
- cc = bt2.ClockClass('hi', 1000)
- tc.add_clock_class(cc)
- ev1 = self._ec()
- self._fill_ev(ev1)
- ev1.clock_snapshots.add(cc(234))
- ev2 = self._ec()
- self._fill_ev(ev2)
- ev2.clock_snapshots.add(cc(234))
- self.assertEqual(ev1, ev2)
-
- def test_ne_header_field(self):
- tc = bt2.Trace()
- tc.add_stream_class(self._ec.stream_class)
- cc = bt2.ClockClass('hi', 1000)
- tc.add_clock_class(cc)
- ev1 = self._ec()
- self._fill_ev(ev1)
- ev1.header_field['id'] = 19
- ev1.clock_snapshots.add(cc(234))
- ev2 = self._ec()
- self._fill_ev(ev2)
- ev2.clock_snapshots.add(cc(234))
- self.assertNotEqual(ev1, ev2)
-
- def test_ne_stream_event_context_field(self):
- tc = bt2.Trace()
- tc.add_stream_class(self._ec.stream_class)
- cc = bt2.ClockClass('hi', 1000)
- tc.add_clock_class(cc)
- ev1 = self._ec()
- self._fill_ev(ev1)
- ev1.stream_event_context_field['cpu_id'] = 3
- ev1.clock_snapshots.add(cc(234))
- ev2 = self._ec()
- self._fill_ev(ev2)
- ev2.clock_snapshots.add(cc(234))
- self.assertNotEqual(ev1, ev2)
-
- def test_ne_context_field(self):
- tc = bt2.Trace()
- tc.add_stream_class(self._ec.stream_class)
- cc = bt2.ClockClass('hi', 1000)
- tc.add_clock_class(cc)
- ev1 = self._ec()
- self._fill_ev(ev1)
- ev1.context_field['ant'] = -3
- ev1.clock_snapshots.add(cc(234))
- ev2 = self._ec()
- self._fill_ev(ev2)
- ev2.clock_snapshots.add(cc(234))
- self.assertNotEqual(ev1, ev2)
-
- def test_ne_payload_field(self):
- tc = bt2.Trace()
- tc.add_stream_class(self._ec.stream_class)
- cc = bt2.ClockClass('hi', 1000)
- tc.add_clock_class(cc)
- ev1 = self._ec()
- self._fill_ev(ev1)
- ev1.payload_field['mosquito'] = 98
- ev1.clock_snapshots.add(cc(234))
- ev2 = self._ec()
- self._fill_ev(ev2)
- ev2.clock_snapshots.add(cc(234))
- self.assertNotEqual(ev1, ev2)
-
- def test_eq_invalid(self):
- ev = self._ec()
- self.assertFalse(ev == 23)
-
- def _test_copy(self, func):
- tc = bt2.Trace()
- tc.add_stream_class(self._ec.stream_class)
- cc = bt2.ClockClass('hi', 1000)
- tc.add_clock_class(cc)
- ev = self._ec()
- self._fill_ev(ev)
- ev.clock_snapshots.add(cc(234))
- cpy = func(ev)
- self.assertIsNot(ev, cpy)
- self.assertNotEqual(ev.addr, cpy.addr)
- self.assertEqual(ev, cpy)
-
- def test_copy(self):
- self._test_copy(copy.copy)
-
- def test_deepcopy(self):
- self._test_copy(copy.deepcopy)
+
+if __name__ == "__main__":
+ unittest.main()
projects / babeltrace.git / commitdiff
