X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_event.py;h=470ae01d086140edf3dd1b532e50236a752b8364;hb=e30908cfe5b1a65d01c52cf21e7134abf9697eaf;hp=a2727130a82c36ad7f83b41f52febd9cb4e1fce9;hpb=32d2d479b457ba07d60c09928362a44f0bc2762c;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_event.py b/tests/bindings/python/bt2/test_event.py index a2727130..470ae01d 100644 --- a/tests/bindings/python/bt2/test_event.py +++ b/tests/bindings/python/bt2/test_event.py @@ -16,46 +16,72 @@ # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. # -from collections import OrderedDict import unittest import bt2 +import utils +from utils import TestOutputPortMessageIterator +from bt2 import field as bt2_field +from bt2 import stream as bt2_stream +from bt2 import event_class as bt2_event_class +from bt2 import clock_snapshot as bt2_clock_snapshot -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): +class EventTestCase(unittest.TestCase): + def _create_test_const_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): + def __init__(self, config, self_output_port): self._at = 0 + self._msgs = [self._create_stream_beginning_message(test_obj.stream)] - def __next__(self): - if self._at == 0: - msg = self._create_stream_beginning_message(test_obj.stream) - elif self._at == 1: + if with_packet: 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 + self._msgs.append( + self._create_packet_beginning_message(test_obj.packet) + ) + + default_clock_snapshot = 789 if with_clockclass else None + + if with_packet: 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: + ev_parent = test_obj.packet + else: + assert test_obj.stream + ev_parent = test_obj.stream + + 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) + + self._msgs.append(msg) + + if with_packet: + self._msgs.append(self._create_packet_end_message(test_obj.packet)) + + self._msgs.append(self._create_stream_end_message(test_obj.stream)) + + def __next__(self): + if self._at == len(self._msgs): raise bt2.Stop + msg = self._msgs[self._at] self._at += 1 return msg class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter): - def __init__(self, params): + def __init__(self, config, params, obj): self._add_output_port('out') tc = self._create_trace_class() @@ -67,135 +93,201 @@ class EventTestCase(unittest.TestCase): cc = None if with_cc: cc = tc.create_structure_field_class() - cc += OrderedDict(( + cc += [ ('cpu_id', tc.create_signed_integer_field_class(8)), - ('stuff', tc.create_real_field_class()), - )) + ('stuff', tc.create_double_precision_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) + pc = None + + if with_packet: + pc = tc.create_structure_field_class() + pc += [ + ('something', tc.create_unsigned_integer_field_class(8)), + ( + 'something_else', + tc.create_double_precision_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, + ) # specific context (event-class-defined) sc = None if with_sc: sc = tc.create_structure_field_class() - sc += OrderedDict(( + sc += [ ('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(( + ep += [ ('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) + 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 with_packet: + packet = stream.create_packet() if packet_fields_config is not None: + assert packet packet_fields_config(packet) - test_obj.packet = packet + if with_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']) + self._msg_iter = TestOutputPortMessageIterator( + self._graph, self._src_comp.output_ports['out'] + ) - for i, msg in enumerate(self._msg_iter): - if i == 2: + for msg in self._msg_iter: + if type(msg) is bt2._EventMessageConst: + self._event_msg = msg return msg - def test_attr_event_class(self): - msg = self._create_test_event_message() + def test_const_attr_event_class(self): + msg = self._create_test_const_event_message() self.assertEqual(msg.event.cls.addr, self.event_class.addr) + self.assertIs(type(msg.event.cls), bt2_event_class._EventClassConst) + + def test_attr_event_class(self): + msg = utils.get_event_message() + self.assertIs(type(msg.event.cls), bt2_event_class._EventClass) - def test_attr_name(self): - msg = self._create_test_event_message() + def test_const_attr_name(self): + msg = self._create_test_const_event_message() self.assertEqual(msg.event.name, self.event_class.name) - def test_attr_id(self): - msg = self._create_test_event_message() + def test_const_attr_id(self): + msg = self._create_test_const_event_message() self.assertEqual(msg.event.id, self.event_class.id) - def test_get_common_context_field(self): + def test_const_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) + msg = self._create_test_const_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) + self.assertIs( + type(msg.event.common_context_field), bt2_field._StructureFieldConst + ) - def test_no_common_context_field(self): - msg = self._create_test_event_message(with_cc=False) + def test_attr_common_context_field(self): + msg = utils.get_event_message() + self.assertIs(type(msg.event.common_context_field), bt2_field._StructureField) + + def test_const_no_common_context_field(self): + msg = self._create_test_const_event_message(with_cc=False) self.assertIsNone(msg.event.common_context_field) - def test_get_specific_context_field(self): + def test_const_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) + msg = self._create_test_const_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') + self.assertIs( + type(msg.event.specific_context_field), bt2_field._StructureFieldConst + ) + + def test_attr_specific_context_field(self): + msg = utils.get_event_message() + self.assertIs(type(msg.event.specific_context_field), bt2_field._StructureField) - def test_no_specific_context_field(self): - msg = self._create_test_event_message(with_sc=False) + def test_const_no_specific_context_field(self): + msg = self._create_test_const_event_message(with_sc=False) self.assertIsNone(msg.event.specific_context_field) - def test_get_event_payload_field(self): + def test_const_get_event_payload_field(self): def event_fields_config(event): event.payload_field['giraffe'] = 1 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_const_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) self.assertEqual(msg.event.payload_field['mosquito'], 42) + self.assertIs(type(msg.event.payload_field), bt2_field._StructureFieldConst) + + def test_attr_payload_field(self): + msg = utils.get_event_message() + self.assertIs(type(msg.event.payload_field), bt2_field._StructureField) - def test_no_payload_field(self): - msg = self._create_test_event_message(with_ep=False) + def test_const_no_payload_field(self): + msg = self._create_test_const_event_message(with_ep=False) self.assertIsNone(msg.event.payload_field) - def test_clock_value(self): - msg = self._create_test_event_message(with_clockclass=True) + def test_const_clock_value(self): + msg = self._create_test_const_event_message(with_clockclass=True) self.assertEqual(msg.default_clock_snapshot.value, 789) + self.assertIs( + type(msg.default_clock_snapshot), bt2_clock_snapshot._ClockSnapshotConst + ) - def test_no_clock_value(self): - msg = self._create_test_event_message(with_clockclass=False) - with self.assertRaises(bt2.NonexistentClockSnapshot): + def test_clock_value(self): + msg = utils.get_event_message() + self.assertEqual(msg.default_clock_snapshot.value, 789) + self.assertIs( + type(msg.default_clock_snapshot), bt2_clock_snapshot._ClockSnapshotConst + ) + + def test_const_no_clock_value(self): + msg = self._create_test_const_event_message(with_clockclass=False) + with self.assertRaisesRegex( + ValueError, 'stream class has no default clock class' + ): msg.default_clock_snapshot - def test_stream(self): - msg = self._create_test_event_message() + def test_const_stream(self): + msg = self._create_test_const_event_message() self.assertEqual(msg.event.stream.addr, self.stream.addr) + self.assertIs(type(msg.event.stream), bt2_stream._StreamConst) - def test_getitem(self): + def test_stream(self): + msg = utils.get_event_message() + self.assertIs(type(msg.event.stream), bt2_stream._Stream) + + def test_const_getitem(self): def event_fields_config(event): event.payload_field['giraffe'] = 1 event.payload_field['gnu'] = 23 @@ -209,27 +301,63 @@ class EventTestCase(unittest.TestCase): 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) + msg = self._create_test_const_event_message( + packet_fields_config=packet_fields_config, + event_fields_config=event_fields_config, + with_cc=True, + with_sc=True, + with_ep=True, + with_packet=True, + ) ev = msg.event # Test event fields self.assertEqual(ev['giraffe'], 1) + self.assertIs(type(ev['giraffe']), bt2_field._SignedIntegerFieldConst) self.assertEqual(ev['gnu'], 23) self.assertEqual(ev['mosquito'], 42) self.assertEqual(ev['ant'], -1) + self.assertIs(type(ev['ant']), bt2_field._SignedIntegerFieldConst) self.assertEqual(ev['msg'], 'hellooo') self.assertEqual(ev['cpu_id'], 1) + self.assertIs(type(ev['cpu_id']), bt2_field._SignedIntegerFieldConst) self.assertEqual(ev['stuff'], 13.194) # Test packet fields self.assertEqual(ev['something'], 154) + self.assertIs(type(ev['something']), bt2_field._UnsignedIntegerFieldConst) self.assertEqual(ev['something_else'], 17.2) with self.assertRaises(KeyError): ev['yes'] + def test_const_getitem_no_packet(self): + def event_fields_config(event): + event.payload_field['giraffe'] = 1 + event.payload_field['gnu'] = 23 + event.payload_field['mosquito'] = 42 + + msg = self._create_test_const_event_message( + event_fields_config=event_fields_config, + with_ep=True, + ) + ev = msg.event + + with self.assertRaises(KeyError): + ev['yes'] + + def test_getitem(self): + msg = utils.get_event_message() + ev = msg.event + self.assertEqual(ev['giraffe'], 1) + self.assertIs(type(ev['giraffe']), bt2_field._SignedIntegerField) + self.assertEqual(ev['ant'], -1) + self.assertIs(type(ev['ant']), bt2_field._SignedIntegerField) + self.assertEqual(ev['cpu_id'], 1) + self.assertIs(type(ev['cpu_id']), bt2_field._SignedIntegerField) + self.assertEqual(ev['something'], 154) + self.assertIs(type(ev['something']), bt2_field._UnsignedIntegerField) + if __name__ == "__main__": unittest.main()