# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
-from collections import OrderedDict
import unittest
import bt2
+from utils import TestOutputPortMessageIterator
class EventTestCase(unittest.TestCase):
- def _create_test_event_message(self, packet_fields_config=None,
- event_fields_config=None,
- with_clockclass=False,
- with_cc=False, with_sc=False,
- with_ep=False, with_packet=False):
-
+ def _create_test_event_message(
+ self,
+ packet_fields_config=None,
+ event_fields_config=None,
+ with_clockclass=False,
+ with_cc=False,
+ with_sc=False,
+ with_ep=False,
+ with_packet=False,
+ ):
class MyIter(bt2._UserMessageIterator):
def __init__(self, self_output_port):
self._at = 0
- self._msgs = [
- self._create_stream_beginning_message(test_obj.stream),
- ]
+ self._msgs = [self._create_stream_beginning_message(test_obj.stream)]
if with_packet:
assert test_obj.packet
- self._msgs.append(self._create_packet_beginning_message(test_obj.packet))
+ self._msgs.append(
+ self._create_packet_beginning_message(test_obj.packet)
+ )
default_clock_snapshot = 789 if with_clockclass else None
assert test_obj.stream
ev_parent = test_obj.stream
- msg = self._create_event_message(test_obj.event_class, ev_parent, default_clock_snapshot)
+ msg = self._create_event_message(
+ test_obj.event_class, ev_parent, default_clock_snapshot
+ )
if event_fields_config is not None:
event_fields_config(msg.event)
return msg
class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
tc = self._create_trace_class()
('something_else', tc.create_real_field_class()),
]
- stream_class = tc.create_stream_class(default_clock_class=clock_class,
- event_common_context_field_class=cc,
- packet_context_field_class=pc,
- supports_packets=with_packet)
+ stream_class = tc.create_stream_class(
+ default_clock_class=clock_class,
+ event_common_context_field_class=cc,
+ packet_context_field_class=pc,
+ supports_packets=with_packet,
+ )
# specific context (event-class-defined)
sc = None
('mosquito', tc.create_signed_integer_field_class(8)),
]
- event_class = stream_class.create_event_class(name='garou',
- specific_context_field_class=sc,
- payload_field_class=ep)
+ event_class = stream_class.create_event_class(
+ name='garou',
+ specific_context_field_class=sc,
+ payload_field_class=ep,
+ )
trace = tc()
stream = trace.create_stream(stream_class)
test_obj = self
self._graph = bt2.Graph()
self._src_comp = self._graph.add_component(MySrc, 'my_source')
- self._msg_iter = self._graph.create_output_port_message_iterator(self._src_comp.output_ports['out'])
+ self._msg_iter = TestOutputPortMessageIterator(
+ self._graph, self._src_comp.output_ports['out']
+ )
for msg in self._msg_iter:
if type(msg) is bt2._EventMessage:
event.common_context_field['cpu_id'] = 1
event.common_context_field['stuff'] = 13.194
- msg = self._create_test_event_message(event_fields_config=event_fields_config, with_cc=True)
+ msg = self._create_test_event_message(
+ event_fields_config=event_fields_config, with_cc=True
+ )
self.assertEqual(msg.event.common_context_field['cpu_id'], 1)
self.assertEqual(msg.event.common_context_field['stuff'], 13.194)
event.specific_context_field['ant'] = -1
event.specific_context_field['msg'] = 'hellooo'
- msg = self._create_test_event_message(event_fields_config=event_fields_config, with_sc=True)
+ msg = self._create_test_event_message(
+ event_fields_config=event_fields_config, with_sc=True
+ )
self.assertEqual(msg.event.specific_context_field['ant'], -1)
self.assertEqual(msg.event.specific_context_field['msg'], 'hellooo')
event.payload_field['gnu'] = 23
event.payload_field['mosquito'] = 42
- msg = self._create_test_event_message(event_fields_config=event_fields_config, with_ep=True)
+ msg = self._create_test_event_message(
+ event_fields_config=event_fields_config, with_ep=True
+ )
self.assertEqual(msg.event.payload_field['giraffe'], 1)
self.assertEqual(msg.event.payload_field['gnu'], 23)
def test_no_clock_value(self):
msg = self._create_test_event_message(with_clockclass=False)
- with self.assertRaises(bt2.NonexistentClockSnapshot):
+ with self.assertRaisesRegex(
+ ValueError, 'stream class has no default clock class'
+ ):
msg.default_clock_snapshot
def test_stream(self):
packet.context_field['something'] = 154
packet.context_field['something_else'] = 17.2
- msg = self._create_test_event_message(packet_fields_config=packet_fields_config,
- event_fields_config=event_fields_config,
- with_cc=True, with_sc=True, with_ep=True,
- with_packet=True)
+ msg = self._create_test_event_message(
+ packet_fields_config=packet_fields_config,
+ event_fields_config=event_fields_config,
+ with_cc=True,
+ with_sc=True,
+ with_ep=True,
+ with_packet=True,
+ )
ev = msg.event
# Test event fields