X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_event.py;h=9c71c0204e182f0ca8c711d8d9810f7a5e5539ca;hb=f0a42b33ac3951cd5cb2ee0f66ac04437a681621;hp=6e7bdc679e906b334789bd9e6e2c08ec779f7f20;hpb=9cf643d1c6524358b231b0f143103aabfa3e5773;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_event.py b/tests/bindings/python/bt2/test_event.py index 6e7bdc67..9c71c020 100644 --- a/tests/bindings/python/bt2/test_event.py +++ b/tests/bindings/python/bt2/test_event.py @@ -1,370 +1,345 @@ -from collections import OrderedDict -from bt2 import values +# +# 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. +# + import unittest -import copy 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 setUp(self): - self._ec = self._create_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_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): + 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, obj): + 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 += [ + ('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 += [ + ('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 += [ + ('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 += [ + ('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 = TestOutputPortMessageIterator( + self._graph, self._src_comp.output_ports['out'] + ) + + for msg in self._msg_iter: + if type(msg) is bt2._EventMessageConst: + self._event_msg = msg + return msg + + 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): - ev = self._ec() - self.assertEqual(ev.event_class.addr, self._ec.addr) - - def test_attr_name(self): - ev = self._ec() - self.assertEqual(ev.name, self._ec.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) - - 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) + msg = utils.get_event_message() + self.assertIs(type(msg.event.cls), bt2_event_class._EventClass) + + def test_const_attr_name(self): + msg = self._create_test_const_event_message() + self.assertEqual(msg.event.name, self.event_class.name) + + def test_const_attr_id(self): + msg = self._create_test_const_event_message() + self.assertEqual(msg.event.id, self.event_class.id) + + 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_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_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_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_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_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_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_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_const_no_payload_field(self): + msg = self._create_test_const_event_message(with_ep=False) + self.assertIsNone(msg.event.payload_field) + + 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_clock_value(self): - tc = bt2.Trace() - tc.add_stream_class(self._ec.stream_class) - cc = bt2.ClockClass('hi') - tc.add_clock_class(cc) - ev = self._ec() - ev.set_clock_value(cc.create_clock_value(177)) - self.assertEqual(ev.get_clock_value(cc).cycles, 177) - - def test_no_clock_value(self): - tc = bt2.Trace() - tc.add_stream_class(self._ec.stream_class) - cc = bt2.ClockClass('hi') - tc.add_clock_class(cc) - ev = self._ec() - self.assertIsNone(ev.get_clock_value(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 = 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_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_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') - tc.add_clock_class(cc) - ev = self._ec() - self._fill_ev(ev) - ev.clock_value(cc, 234) - return ev + 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 + 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_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) - 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 + # 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['magic'] + ev['yes'] - ev.packet = packet - self.assertEqual(ev['mosquito'], 42) - self.assertEqual(ev['gnu'], 23) + def test_getitem(self): + msg = utils.get_event_message() + ev = msg.event self.assertEqual(ev['giraffe'], 1) - self.assertEqual(ev['msg'], 'hellooo') + self.assertIs(type(ev['giraffe']), bt2_field._SignedIntegerField) self.assertEqual(ev['ant'], -1) - self.assertEqual(ev['stuff'], 13.194) + self.assertIs(type(ev['ant']), bt2_field._SignedIntegerField) self.assertEqual(ev['cpu_id'], 1) - self.assertEqual(ev['ts'], 1234) - self.assertEqual(ev['id'], 23) - self.assertEqual(ev['something_else'], 17.2) + self.assertIs(type(ev['cpu_id']), bt2_field._SignedIntegerField) self.assertEqual(ev['something'], 154) - self.assertEqual(ev['stream_id'], 0) - self.assertEqual(ev['magic'], 0xc1fc1fc1) + self.assertIs(type(ev['something']), bt2_field._UnsignedIntegerField) - 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') - tc.add_clock_class(cc) - ev1 = self._ec() - self._fill_ev(ev1) - ev1.set_clock_value(cc.create_clock_value(234)) - ev2 = self._ec() - self._fill_ev(ev2) - ev2.set_clock_value(cc.create_clock_value(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') - tc.add_clock_class(cc) - ev1 = self._ec() - self._fill_ev(ev1) - ev1.header_field['id'] = 19 - ev1.set_clock_value(cc.create_clock_value(234)) - ev2 = self._ec() - self._fill_ev(ev2) - ev2.set_clock_value(cc.create_clock_value(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') - tc.add_clock_class(cc) - ev1 = self._ec() - self._fill_ev(ev1) - ev1.stream_event_context_field['cpu_id'] = 3 - ev1.set_clock_value(cc.create_clock_value(234)) - ev2 = self._ec() - self._fill_ev(ev2) - ev2.set_clock_value(cc.create_clock_value(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') - tc.add_clock_class(cc) - ev1 = self._ec() - self._fill_ev(ev1) - ev1.context_field['ant'] = -3 - ev1.set_clock_value(cc.create_clock_value(234)) - ev2 = self._ec() - self._fill_ev(ev2) - ev2.set_clock_value(cc.create_clock_value(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') - tc.add_clock_class(cc) - ev1 = self._ec() - self._fill_ev(ev1) - ev1.payload_field['mosquito'] = 98 - ev1.set_clock_value(cc.create_clock_value(234)) - ev2 = self._ec() - self._fill_ev(ev2) - ev2.set_clock_value(cc.create_clock_value(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') - tc.add_clock_class(cc) - ev = self._ec() - self._fill_ev(ev) - ev.set_clock_value(cc.create_clock_value(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()