X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_event.py;h=eedc9688b306ae7ec799866dc83b28c6a483495d;hb=26fc5aedf;hp=6cef201ee071aadf98b0dd8f1f6c857b7a9c0c3b;hpb=976c241df475c9d99161a1100517c3c308074d7f;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_event.py b/tests/bindings/python/bt2/test_event.py index 6cef201e..eedc9688 100644 --- a/tests/bindings/python/bt2/test_event.py +++ b/tests/bindings/python/bt2/test_event.py @@ -1,374 +1,259 @@ +# +# 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 -from bt2 import values 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.StructureFieldType() - eh += OrderedDict(( - ('id', bt2.IntegerFieldType(8)), - ('ts', bt2.IntegerFieldType(32)), - )) - else: - eh = None - - # stream event context - if with_sec: - sec = bt2.StructureFieldType() - sec += OrderedDict(( - ('cpu_id', bt2.IntegerFieldType(8)), - ('stuff', bt2.FloatingPointNumberFieldType()), - )) - else: - sec = None - - # packet context - pc = bt2.StructureFieldType() - pc += OrderedDict(( - ('something', bt2.IntegerFieldType(8)), - ('something_else', bt2.FloatingPointNumberFieldType()), - )) - - # stream class - sc = bt2.StreamClass() - sc.packet_context_field_type = pc - sc.event_header_field_type = eh - sc.event_context_field_type = sec - - # event context - if with_ec: - ec = bt2.StructureFieldType() - ec += OrderedDict(( - ('ant', bt2.IntegerFieldType(16, is_signed=True)), - ('msg', bt2.StringFieldType()), - )) - else: - ec = None - - # event payload - if with_ep: - ep = bt2.StructureFieldType() - ep += OrderedDict(( - ('giraffe', bt2.IntegerFieldType(32)), - ('gnu', bt2.IntegerFieldType(8)), - ('mosquito', bt2.IntegerFieldType(8)), - )) - else: - ep = None - - # event class - event_class = bt2.EventClass('ec') - event_class.context_field_type = ec - event_class.payload_field_type = 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, 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), + ] + + if with_packet: + assert test_obj.packet + 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 + 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): + 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 = None + + if with_packet: + 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, + supports_packets=with_packet) + + # 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) + + if with_packet: + packet = stream.create_packet() + + if packet_fields_config is not None: + assert packet + packet_fields_config(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']) + + for msg in self._msg_iter: + if type(msg) is bt2._EventMessage: + 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.cls.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_type() - 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_type() - 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_type() - 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_type() - 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 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) + + self.assertEqual(msg.event.payload_field['giraffe'], 1) + self.assertEqual(msg.event.payload_field['gnu'], 23) + self.assertEqual(msg.event.payload_field['mosquito'], 42) + + def test_no_payload_field(self): + msg = self._create_test_event_message(with_ep=False) + self.assertIsNone(msg.event.payload_field) def test_clock_value(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_values.add(cc(177)) - self.assertEqual(ev.clock_values[cc].cycles, 177) + msg = self._create_test_event_message(with_clockclass=True) + self.assertEqual(msg.default_clock_snapshot.value, 789) def test_no_clock_value(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_values[cc]) - - def test_no_packet(self): - ev = self._ec() - self.assertIsNone(ev.packet) - - def test_packet(self): - tc = bt2.Trace() - tc.packet_header_field_type = bt2.StructureFieldType() - tc.packet_header_field_type.append_field('magic', bt2.IntegerFieldType(32)) - tc.packet_header_field_type.append_field('stream_id', bt2.IntegerFieldType(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) + msg = self._create_test_event_message(with_clockclass=False) + with self.assertRaises(bt2.NonexistentClockSnapshot): + msg.default_clock_snapshot def test_stream(self): - tc = bt2.Trace() - tc.packet_header_field_type = bt2.StructureFieldType() - tc.packet_header_field_type.append_field('magic', bt2.IntegerFieldType(32)) - tc.packet_header_field_type.append_field('stream_id', bt2.IntegerFieldType(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_values.add(cc(234)) - return ev + msg = self._create_test_event_message() + self.assertEqual(msg.event.stream.addr, self.stream.addr) def test_getitem(self): - tc = bt2.Trace() - tc.packet_header_field_type = bt2.StructureFieldType() - tc.packet_header_field_type.append_field('magic', bt2.IntegerFieldType(32)) - tc.packet_header_field_type.append_field('stream_id', bt2.IntegerFieldType(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 - - with self.assertRaises(KeyError): - ev['magic'] - - ev.packet = packet - self.assertEqual(ev['mosquito'], 42) - self.assertEqual(ev['gnu'], 23) + 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, + with_packet=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_values.add(cc(234)) - ev2 = self._ec() - self._fill_ev(ev2) - ev2.clock_values.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_values.add(cc(234)) - ev2 = self._ec() - self._fill_ev(ev2) - ev2.clock_values.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_values.add(cc(234)) - ev2 = self._ec() - self._fill_ev(ev2) - ev2.clock_values.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_values.add(cc(234)) - ev2 = self._ec() - self._fill_ev(ev2) - ev2.clock_values.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_values.add(cc(234)) - ev2 = self._ec() - self._fill_ev(ev2) - ev2.clock_values.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_values.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()