[babeltrace.git] / tests / bindings / python / bt2 / test_event.py

#
# 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):

        class MyIter(bt2._UserMessageIterator):
            def __init__(self, self_output_port):
                self._at = 0

            def __next__(self):
                if self._at == 0:
                    msg = self._create_stream_beginning_message(test_obj.stream)
                elif self._at == 1:
                    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
                    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:
                    raise bt2.Stop

                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 = 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)

                # 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)
                packet = stream.create_packet()

                if packet_fields_config is not None:
                    packet_fields_config(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 i, msg in enumerate(self._msg_iter):
            if i == 2:
                return msg

    def test_attr_event_class(self):
        msg = self._create_test_event_message()
        self.assertEqual(msg.event.cls.addr, self.event_class.addr)

    def test_attr_name(self):
        msg = self._create_test_event_message()
        self.assertEqual(msg.event.name, self.event_class.name)

    def test_attr_id(self):
        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):
        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):
        msg = self._create_test_event_message(with_clockclass=True)
        self.assertEqual(msg.default_clock_snapshot.value, 789)

    def test_no_clock_value(self):
        msg = self._create_test_event_message(with_clockclass=False)
        with self.assertRaises(bt2.NonexistentClockSnapshot):
            msg.default_clock_snapshot

    def test_stream(self):
        msg = self._create_test_event_message()
        self.assertEqual(msg.event.stream.addr, self.stream.addr)

    def test_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_event_message(packet_fields_config=packet_fields_config,
                                              event_fields_config=event_fields_config,
                                              with_cc=True, with_sc=True, with_ep=True)
        ev = msg.event

        # Test event fields
        self.assertEqual(ev['giraffe'], 1)
        self.assertEqual(ev['gnu'], 23)
        self.assertEqual(ev['mosquito'], 42)
        self.assertEqual(ev['ant'], -1)
        self.assertEqual(ev['msg'], 'hellooo')
        self.assertEqual(ev['cpu_id'], 1)
        self.assertEqual(ev['stuff'], 13.194)

        # Test packet fields
        self.assertEqual(ev['something'], 154)
        self.assertEqual(ev['something_else'], 17.2)

        with self.assertRaises(KeyError):
            ev['yes']


if __name__ == "__main__":
    unittest.main()
Commit	Line	Data
d2d857a8 MJ	1	#
	2	# Copyright (C) 2019 EfficiOS Inc.
	3	#
	4	# This program is free software; you can redistribute it and/or
	5	# modify it under the terms of the GNU General Public License
	6	# as published by the Free Software Foundation; only version 2
	7	# of the License.
	8	#
	9	# This program is distributed in the hope that it will be useful,
	10	# but WITHOUT ANY WARRANTY; without even the implied warranty of
	11	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	12	# GNU General Public License for more details.
	13	#
	14	# You should have received a copy of the GNU General Public License
	15	# along with this program; if not, write to the Free Software
	16	# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
	17	#
	18
9cf643d1	19	from collections import OrderedDict
9cf643d1	20	import unittest
9cf643d1 PP	21	import bt2
	22
	23
	24	class EventTestCase(unittest.TestCase):
2ae9f48c SM	25	def _create_test_event_message(self, packet_fields_config=None,
	26	event_fields_config=None,
	27	with_clockclass=False,
	28	with_cc=False, with_sc=False,
	29	with_ep=False):
	30
	31	class MyIter(bt2._UserMessageIterator):
c5f330cd	32	def __init__(self, self_output_port):
2ae9f48c SM	33	self._at = 0
	34
	35	def __next__(self):
	36	if self._at == 0:
	37	msg = self._create_stream_beginning_message(test_obj.stream)
	38	elif self._at == 1:
	39	assert test_obj.packet
	40	msg = self._create_packet_beginning_message(test_obj.packet)
	41	elif self._at == 2:
	42	default_clock_snapshot = 789 if with_clockclass else None
	43	assert test_obj.packet
	44	msg = self._create_event_message(test_obj.event_class, test_obj.packet, default_clock_snapshot)
	45	if event_fields_config is not None:
	46	event_fields_config(msg.event)
	47	elif self._at == 3:
	48	msg = self._create_packet_end_message(test_obj.packet)
	49	elif self._at == 4:
	50	msg = self._create_stream_end_message(test_obj.stream)
	51	elif self._at >= 5:
	52	raise bt2.Stop
	53
	54	self._at += 1
	55	return msg
	56
	57	class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
	58	def __init__(self, params):
	59	self._add_output_port('out')
	60	tc = self._create_trace_class()
	61
	62	clock_class = None
	63	if with_clockclass:
	64	clock_class = self._create_clock_class(frequency=1000)
	65
	66	# event common context (stream-class-defined)
	67	cc = None
	68	if with_cc:
	69	cc = tc.create_structure_field_class()
	70	cc += OrderedDict((
	71	('cpu_id', tc.create_signed_integer_field_class(8)),
	72	('stuff', tc.create_real_field_class()),
	73	))
	74
	75	# packet context (stream-class-defined)
	76	pc = tc.create_structure_field_class()
	77	pc += OrderedDict((
	78	('something', tc.create_unsigned_integer_field_class(8)),
	79	('something_else', tc.create_real_field_class()),
	80	))
	81
	82	stream_class = tc.create_stream_class(default_clock_class=clock_class,
	83	event_common_context_field_class=cc,
	84	packet_context_field_class=pc)
	85
	86	# specific context (event-class-defined)
	87	sc = None
	88	if with_sc:
	89	sc = tc.create_structure_field_class()
	90	sc += OrderedDict((
	91	('ant', tc.create_signed_integer_field_class(16)),
	92	('msg', tc.create_string_field_class()),
	93	))
	94
	95	# event payload
	96	ep = None
97	if with_ep:
98	ep = tc.create_structure_field_class()
99	ep += OrderedDict((
100	('giraffe', tc.create_signed_integer_field_class(32)),
101	('gnu', tc.create_signed_integer_field_class(8)),
102	('mosquito', tc.create_signed_integer_field_class(8)),
103	))
104
105	event_class = stream_class.create_event_class(name='garou',
106	specific_context_field_class=sc,
107	payload_field_class=ep)
108
109	trace = tc()
110	stream = trace.create_stream(stream_class)
111	packet = stream.create_packet()
112
113	if packet_fields_config is not None:
114	packet_fields_config(packet)
115
116	test_obj.packet = packet
117	test_obj.stream = stream
118	test_obj.event_class = event_class
119
120	test_obj = self
121	self._graph = bt2.Graph()
122	self._src_comp = self._graph.add_component(MySrc, 'my_source')
123	self._msg_iter = self._graph.create_output_port_message_iterator(self._src_comp.output_ports['out'])
124
125	for i, msg in enumerate(self._msg_iter):
126	if i == 2:
127	return msg
9cf643d1 PP	128
9cf643d1 PP	129	def test_attr_event_class(self):
2ae9f48c	130	msg = self._create_test_event_message()
e8ac1aae	131	self.assertEqual(msg.event.cls.addr, self.event_class.addr)
9cf643d1 PP	132
9cf643d1 PP	133	def test_attr_name(self):
2ae9f48c SM	134	msg = self._create_test_event_message()
2ae9f48c SM	135	self.assertEqual(msg.event.name, self.event_class.name)
9cf643d1 PP	136
9cf643d1 PP	137	def test_attr_id(self):
2ae9f48c SM	138	msg = self._create_test_event_message()
	139	self.assertEqual(msg.event.id, self.event_class.id)
	140
	141	def test_get_common_context_field(self):
	142	def event_fields_config(event):
	143	event.common_context_field['cpu_id'] = 1
	144	event.common_context_field['stuff'] = 13.194
	145
	146	msg = self._create_test_event_message(event_fields_config=event_fields_config, with_cc=True)
	147
	148	self.assertEqual(msg.event.common_context_field['cpu_id'], 1)
	149	self.assertEqual(msg.event.common_context_field['stuff'], 13.194)
	150
	151	def test_no_common_context_field(self):
	152	msg = self._create_test_event_message(with_cc=False)
	153	self.assertIsNone(msg.event.common_context_field)
	154
	155	def test_get_specific_context_field(self):
	156	def event_fields_config(event):
	157	event.specific_context_field['ant'] = -1
	158	event.specific_context_field['msg'] = 'hellooo'
	159
	160	msg = self._create_test_event_message(event_fields_config=event_fields_config, with_sc=True)
	161
	162	self.assertEqual(msg.event.specific_context_field['ant'], -1)
	163	self.assertEqual(msg.event.specific_context_field['msg'], 'hellooo')
	164
	165	def test_no_specific_context_field(self):
	166	msg = self._create_test_event_message(with_sc=False)
	167	self.assertIsNone(msg.event.specific_context_field)
9cf643d1 PP	168
9cf643d1 PP	169	def test_get_event_payload_field(self):
2ae9f48c SM	170	def event_fields_config(event):
	171	event.payload_field['giraffe'] = 1
	172	event.payload_field['gnu'] = 23
	173	event.payload_field['mosquito'] = 42
9cf643d1	174
2ae9f48c	175	msg = self._create_test_event_message(event_fields_config=event_fields_config, with_ep=True)
9cf643d1	176
2ae9f48c SM	177	self.assertEqual(msg.event.payload_field['giraffe'], 1)
	178	self.assertEqual(msg.event.payload_field['gnu'], 23)
	179	self.assertEqual(msg.event.payload_field['mosquito'], 42)
9cf643d1	180
2ae9f48c SM	181	def test_no_payload_field(self):
	182	msg = self._create_test_event_message(with_ep=False)
	183	self.assertIsNone(msg.event.payload_field)
9cf643d1	184
2ae9f48c SM	185	def test_clock_value(self):
	186	msg = self._create_test_event_message(with_clockclass=True)
	187	self.assertEqual(msg.default_clock_snapshot.value, 789)
	188
	189	def test_no_clock_value(self):
	190	msg = self._create_test_event_message(with_clockclass=False)
2e90378a	191	with self.assertRaises(bt2.NonexistentClockSnapshot):
9ec609ec	192	msg.default_clock_snapshot
2ae9f48c SM	193
	194	def test_stream(self):
	195	msg = self._create_test_event_message()
	196	self.assertEqual(msg.event.stream.addr, self.stream.addr)
	197
	198	def test_getitem(self):
	199	def event_fields_config(event):
	200	event.payload_field['giraffe'] = 1
	201	event.payload_field['gnu'] = 23
	202	event.payload_field['mosquito'] = 42
	203	event.specific_context_field['ant'] = -1
	204	event.specific_context_field['msg'] = 'hellooo'
	205	event.common_context_field['cpu_id'] = 1
	206	event.common_context_field['stuff'] = 13.194
	207
	208	def packet_fields_config(packet):
	209	packet.context_field['something'] = 154
	210	packet.context_field['something_else'] = 17.2
	211
	212	msg = self._create_test_event_message(packet_fields_config=packet_fields_config,
	213	event_fields_config=event_fields_config,
	214	with_cc=True, with_sc=True, with_ep=True)
	215	ev = msg.event
	216
	217	# Test event fields
9cf643d1	218	self.assertEqual(ev['giraffe'], 1)
2ae9f48c SM	219	self.assertEqual(ev['gnu'], 23)
2ae9f48c SM	220	self.assertEqual(ev['mosquito'], 42)
9cf643d1	221	self.assertEqual(ev['ant'], -1)
2ae9f48c	222	self.assertEqual(ev['msg'], 'hellooo')
9cf643d1	223	self.assertEqual(ev['cpu_id'], 1)
2ae9f48c SM	224	self.assertEqual(ev['stuff'], 13.194)
	225
	226	# Test packet fields
9cf643d1	227	self.assertEqual(ev['something'], 154)
2ae9f48c	228	self.assertEqual(ev['something_else'], 17.2)
9cf643d1 PP	229
	230	with self.assertRaises(KeyError):
	231	ev['yes']
	232
2ae9f48c SM	233
	234	if __name__ == "__main__":
	235	unittest.main()