X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_event.py;h=74fe7cff1e374d4e044fda296b5748f3dedfdbf7;hb=cfbd7cf3bde05e8a6606478889dcd663604ef7b5;hp=4c21ee7fe144dd204a963ff9da93f057fd8a4d34;hpb=c5f330cd909f5dfbdb519546e875b4427434ba4f;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_event.py b/tests/bindings/python/bt2/test_event.py index 4c21ee7f..74fe7cff 100644 --- a/tests/bindings/python/bt2/test_event.py +++ b/tests/bindings/python/bt2/test_event.py @@ -1,38 +1,76 @@ +# +# Copyright (C) 2019 EfficiOS Inc. +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU General Public License +# as published by the Free Software Foundation; only version 2 +# of the License. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. +# + from collections import OrderedDict import unittest import bt2 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): - + 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)] - 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 @@ -49,68 +87,83 @@ 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()), - )) + ] # 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_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 = self._graph.create_output_port_message_iterator( + 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._EventMessage: return msg def test_attr_event_class(self): msg = self._create_test_event_message() - self.assertEqual(msg.event.event_class.addr, self.event_class.addr) + self.assertEqual(msg.event.cls.addr, self.event_class.addr) def test_attr_name(self): msg = self._create_test_event_message() @@ -125,7 +178,9 @@ class EventTestCase(unittest.TestCase): 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) @@ -139,7 +194,9 @@ class EventTestCase(unittest.TestCase): 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') @@ -154,7 +211,9 @@ class EventTestCase(unittest.TestCase): 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) @@ -170,7 +229,7 @@ class EventTestCase(unittest.TestCase): def test_no_clock_value(self): msg = self._create_test_event_message(with_clockclass=False) - with self.assertRaises(bt2.NoDefaultClockClass): + with self.assertRaises(bt2.NonexistentClockSnapshot): msg.default_clock_snapshot def test_stream(self): @@ -191,9 +250,14 @@ 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_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