[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.
#

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_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, config, 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, config, 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_double_precision_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_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 += [
                        ('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):
        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):
        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):
        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)

        # 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_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()
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	import unittest
9cf643d1	20	import bt2
f0a42b33	21	import utils
6c373cc9	22	from utils import TestOutputPortMessageIterator
9cf643d1	23
f0a42b33 FD	24	from bt2 import field as bt2_field
	25	from bt2 import stream as bt2_stream
	26	from bt2 import event_class as bt2_event_class
	27	from bt2 import clock_snapshot as bt2_clock_snapshot
	28
9cf643d1 PP	29
9cf643d1 PP	30	class EventTestCase(unittest.TestCase):
f0a42b33	31	def _create_test_const_event_message(
cfbd7cf3 FD	32	self,
	33	packet_fields_config=None,
	34	event_fields_config=None,
	35	with_clockclass=False,
	36	with_cc=False,
	37	with_sc=False,
	38	with_ep=False,
	39	with_packet=False,
	40	):
2ae9f48c	41	class MyIter(bt2._UserMessageIterator):
8d8b141d	42	def __init__(self, config, self_output_port):
2ae9f48c	43	self._at = 0
cfbd7cf3	44	self._msgs = [self._create_stream_beginning_message(test_obj.stream)]
2ae9f48c	45
26fc5aed	46	if with_packet:
2ae9f48c	47	assert test_obj.packet
cfbd7cf3 FD	48	self._msgs.append(
	49	self._create_packet_beginning_message(test_obj.packet)
	50	)
26fc5aed PP	51
	52	default_clock_snapshot = 789 if with_clockclass else None
	53
	54	if with_packet:
2ae9f48c	55	assert test_obj.packet
26fc5aed PP	56	ev_parent = test_obj.packet
	57	else:
	58	assert test_obj.stream
	59	ev_parent = test_obj.stream
	60
cfbd7cf3 FD	61	msg = self._create_event_message(
	62	test_obj.event_class, ev_parent, default_clock_snapshot
	63	)
26fc5aed PP	64
	65	if event_fields_config is not None:
	66	event_fields_config(msg.event)
	67
	68	self._msgs.append(msg)
	69
	70	if with_packet:
	71	self._msgs.append(self._create_packet_end_message(test_obj.packet))
	72
	73	self._msgs.append(self._create_stream_end_message(test_obj.stream))
	74
	75	def __next__(self):
	76	if self._at == len(self._msgs):
2ae9f48c SM	77	raise bt2.Stop
2ae9f48c SM	78
26fc5aed	79	msg = self._msgs[self._at]
2ae9f48c SM	80	self._at += 1
	81	return msg
	82
	83	class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
59225a3e	84	def __init__(self, config, params, obj):
2ae9f48c SM	85	self._add_output_port('out')
	86	tc = self._create_trace_class()
	87
	88	clock_class = None
	89	if with_clockclass:
	90	clock_class = self._create_clock_class(frequency=1000)
	91
	92	# event common context (stream-class-defined)
	93	cc = None
	94	if with_cc:
	95	cc = tc.create_structure_field_class()
45c51519	96	cc += [
2ae9f48c	97	('cpu_id', tc.create_signed_integer_field_class(8)),
fe4df857	98	('stuff', tc.create_double_precision_real_field_class()),
45c51519	99	]
2ae9f48c SM	100
2ae9f48c SM	101	# packet context (stream-class-defined)
26fc5aed PP	102	pc = None
	103
	104	if with_packet:
	105	pc = tc.create_structure_field_class()
45c51519	106	pc += [
26fc5aed	107	('something', tc.create_unsigned_integer_field_class(8)),
fe4df857 FD	108	(
	109	'something_else',
	110	tc.create_double_precision_real_field_class(),
	111	),
45c51519	112	]
2ae9f48c	113
cfbd7cf3 FD	114	stream_class = tc.create_stream_class(
	115	default_clock_class=clock_class,
	116	event_common_context_field_class=cc,
	117	packet_context_field_class=pc,
	118	supports_packets=with_packet,
	119	)
2ae9f48c SM	120
	121	# specific context (event-class-defined)
	122	sc = None
	123	if with_sc:
	124	sc = tc.create_structure_field_class()
45c51519	125	sc += [
2ae9f48c SM	126	('ant', tc.create_signed_integer_field_class(16)),
2ae9f48c SM	127	('msg', tc.create_string_field_class()),
45c51519	128	]
2ae9f48c SM	129
	130	# event payload
	131	ep = None
	132	if with_ep:
	133	ep = tc.create_structure_field_class()
45c51519	134	ep += [
2ae9f48c SM	135	('giraffe', tc.create_signed_integer_field_class(32)),
	136	('gnu', tc.create_signed_integer_field_class(8)),
	137	('mosquito', tc.create_signed_integer_field_class(8)),
45c51519	138	]
2ae9f48c	139
cfbd7cf3 FD	140	event_class = stream_class.create_event_class(
	141	name='garou',
	142	specific_context_field_class=sc,
	143	payload_field_class=ep,
	144	)
2ae9f48c SM	145
	146	trace = tc()
	147	stream = trace.create_stream(stream_class)
26fc5aed PP	148
	149	if with_packet:
	150	packet = stream.create_packet()
2ae9f48c SM	151
2ae9f48c SM	152	if packet_fields_config is not None:
26fc5aed	153	assert packet
2ae9f48c SM	154	packet_fields_config(packet)
2ae9f48c SM	155
26fc5aed PP	156	if with_packet:
	157	test_obj.packet = packet
	158
2ae9f48c SM	159	test_obj.stream = stream
	160	test_obj.event_class = event_class
	161
	162	test_obj = self
	163	self._graph = bt2.Graph()
	164	self._src_comp = self._graph.add_component(MySrc, 'my_source')
6c373cc9 PP	165	self._msg_iter = TestOutputPortMessageIterator(
6c373cc9 PP	166	self._graph, self._src_comp.output_ports['out']
cfbd7cf3	167	)
2ae9f48c	168
26fc5aed	169	for msg in self._msg_iter:
f0a42b33 FD	170	if type(msg) is bt2._EventMessageConst:
f0a42b33 FD	171	self._event_msg = msg
2ae9f48c	172	return msg
9cf643d1	173
f0a42b33 FD	174	def test_const_attr_event_class(self):
f0a42b33 FD	175	msg = self._create_test_const_event_message()
e8ac1aae	176	self.assertEqual(msg.event.cls.addr, self.event_class.addr)
f0a42b33 FD	177	self.assertIs(type(msg.event.cls), bt2_event_class._EventClassConst)
	178
	179	def test_attr_event_class(self):
	180	msg = utils.get_event_message()
	181	self.assertIs(type(msg.event.cls), bt2_event_class._EventClass)
9cf643d1	182
f0a42b33 FD	183	def test_const_attr_name(self):
f0a42b33 FD	184	msg = self._create_test_const_event_message()
2ae9f48c	185	self.assertEqual(msg.event.name, self.event_class.name)
9cf643d1	186
f0a42b33 FD	187	def test_const_attr_id(self):
f0a42b33 FD	188	msg = self._create_test_const_event_message()
2ae9f48c SM	189	self.assertEqual(msg.event.id, self.event_class.id)
2ae9f48c SM	190
f0a42b33	191	def test_const_get_common_context_field(self):
2ae9f48c SM	192	def event_fields_config(event):
	193	event.common_context_field['cpu_id'] = 1
	194	event.common_context_field['stuff'] = 13.194
	195
f0a42b33	196	msg = self._create_test_const_event_message(
cfbd7cf3 FD	197	event_fields_config=event_fields_config, with_cc=True
cfbd7cf3 FD	198	)
2ae9f48c SM	199
	200	self.assertEqual(msg.event.common_context_field['cpu_id'], 1)
	201	self.assertEqual(msg.event.common_context_field['stuff'], 13.194)
f0a42b33 FD	202	self.assertIs(
	203	type(msg.event.common_context_field), bt2_field._StructureFieldConst
	204	)
	205
	206	def test_attr_common_context_field(self):
	207	msg = utils.get_event_message()
	208	self.assertIs(type(msg.event.common_context_field), bt2_field._StructureField)
2ae9f48c	209
f0a42b33 FD	210	def test_const_no_common_context_field(self):
f0a42b33 FD	211	msg = self._create_test_const_event_message(with_cc=False)
2ae9f48c SM	212	self.assertIsNone(msg.event.common_context_field)
2ae9f48c SM	213
f0a42b33	214	def test_const_get_specific_context_field(self):
2ae9f48c SM	215	def event_fields_config(event):
	216	event.specific_context_field['ant'] = -1
	217	event.specific_context_field['msg'] = 'hellooo'
	218
f0a42b33	219	msg = self._create_test_const_event_message(
cfbd7cf3 FD	220	event_fields_config=event_fields_config, with_sc=True
cfbd7cf3 FD	221	)
2ae9f48c SM	222
	223	self.assertEqual(msg.event.specific_context_field['ant'], -1)
	224	self.assertEqual(msg.event.specific_context_field['msg'], 'hellooo')
f0a42b33 FD	225	self.assertIs(
	226	type(msg.event.specific_context_field), bt2_field._StructureFieldConst
	227	)
	228
	229	def test_attr_specific_context_field(self):
	230	msg = utils.get_event_message()
	231	self.assertIs(type(msg.event.specific_context_field), bt2_field._StructureField)
2ae9f48c	232
f0a42b33 FD	233	def test_const_no_specific_context_field(self):
f0a42b33 FD	234	msg = self._create_test_const_event_message(with_sc=False)
2ae9f48c	235	self.assertIsNone(msg.event.specific_context_field)
9cf643d1	236
f0a42b33	237	def test_const_get_event_payload_field(self):
2ae9f48c SM	238	def event_fields_config(event):
	239	event.payload_field['giraffe'] = 1
	240	event.payload_field['gnu'] = 23
	241	event.payload_field['mosquito'] = 42
9cf643d1	242
f0a42b33	243	msg = self._create_test_const_event_message(
cfbd7cf3 FD	244	event_fields_config=event_fields_config, with_ep=True
cfbd7cf3 FD	245	)
9cf643d1	246
2ae9f48c SM	247	self.assertEqual(msg.event.payload_field['giraffe'], 1)
	248	self.assertEqual(msg.event.payload_field['gnu'], 23)
	249	self.assertEqual(msg.event.payload_field['mosquito'], 42)
f0a42b33 FD	250	self.assertIs(type(msg.event.payload_field), bt2_field._StructureFieldConst)
	251
	252	def test_attr_payload_field(self):
	253	msg = utils.get_event_message()
	254	self.assertIs(type(msg.event.payload_field), bt2_field._StructureField)
9cf643d1	255
f0a42b33 FD	256	def test_const_no_payload_field(self):
f0a42b33 FD	257	msg = self._create_test_const_event_message(with_ep=False)
2ae9f48c	258	self.assertIsNone(msg.event.payload_field)
9cf643d1	259
f0a42b33 FD	260	def test_const_clock_value(self):
	261	msg = self._create_test_const_event_message(with_clockclass=True)
	262	self.assertEqual(msg.default_clock_snapshot.value, 789)
	263	self.assertIs(
	264	type(msg.default_clock_snapshot), bt2_clock_snapshot._ClockSnapshotConst
	265	)
	266
2ae9f48c	267	def test_clock_value(self):
f0a42b33	268	msg = utils.get_event_message()
2ae9f48c	269	self.assertEqual(msg.default_clock_snapshot.value, 789)
f0a42b33 FD	270	self.assertIs(
	271	type(msg.default_clock_snapshot), bt2_clock_snapshot._ClockSnapshotConst
	272	)
2ae9f48c	273
f0a42b33 FD	274	def test_const_no_clock_value(self):
f0a42b33 FD	275	msg = self._create_test_const_event_message(with_clockclass=False)
1153eccb SM	276	with self.assertRaisesRegex(
	277	ValueError, 'stream class has no default clock class'
	278	):
9ec609ec	279	msg.default_clock_snapshot
2ae9f48c	280
f0a42b33 FD	281	def test_const_stream(self):
f0a42b33 FD	282	msg = self._create_test_const_event_message()
2ae9f48c	283	self.assertEqual(msg.event.stream.addr, self.stream.addr)
f0a42b33	284	self.assertIs(type(msg.event.stream), bt2_stream._StreamConst)
2ae9f48c	285
f0a42b33 FD	286	def test_stream(self):
	287	msg = utils.get_event_message()
	288	self.assertIs(type(msg.event.stream), bt2_stream._Stream)
	289
	290	def test_const_getitem(self):
2ae9f48c SM	291	def event_fields_config(event):
	292	event.payload_field['giraffe'] = 1
	293	event.payload_field['gnu'] = 23
	294	event.payload_field['mosquito'] = 42
	295	event.specific_context_field['ant'] = -1
	296	event.specific_context_field['msg'] = 'hellooo'
	297	event.common_context_field['cpu_id'] = 1
	298	event.common_context_field['stuff'] = 13.194
	299
	300	def packet_fields_config(packet):
	301	packet.context_field['something'] = 154
	302	packet.context_field['something_else'] = 17.2
	303
f0a42b33	304	msg = self._create_test_const_event_message(
cfbd7cf3 FD	305	packet_fields_config=packet_fields_config,
	306	event_fields_config=event_fields_config,
	307	with_cc=True,
	308	with_sc=True,
	309	with_ep=True,
	310	with_packet=True,
	311	)
2ae9f48c SM	312	ev = msg.event
	313
	314	# Test event fields
9cf643d1	315	self.assertEqual(ev['giraffe'], 1)
f0a42b33	316	self.assertIs(type(ev['giraffe']), bt2_field._SignedIntegerFieldConst)
2ae9f48c SM	317	self.assertEqual(ev['gnu'], 23)
2ae9f48c SM	318	self.assertEqual(ev['mosquito'], 42)
9cf643d1	319	self.assertEqual(ev['ant'], -1)
f0a42b33	320	self.assertIs(type(ev['ant']), bt2_field._SignedIntegerFieldConst)
2ae9f48c	321	self.assertEqual(ev['msg'], 'hellooo')
9cf643d1	322	self.assertEqual(ev['cpu_id'], 1)
f0a42b33	323	self.assertIs(type(ev['cpu_id']), bt2_field._SignedIntegerFieldConst)
2ae9f48c SM	324	self.assertEqual(ev['stuff'], 13.194)
	325
	326	# Test packet fields
9cf643d1	327	self.assertEqual(ev['something'], 154)
f0a42b33	328	self.assertIs(type(ev['something']), bt2_field._UnsignedIntegerFieldConst)
2ae9f48c	329	self.assertEqual(ev['something_else'], 17.2)
9cf643d1 PP	330
	331	with self.assertRaises(KeyError):
	332	ev['yes']
	333
f0a42b33 FD	334	def test_getitem(self):
	335	msg = utils.get_event_message()
	336	ev = msg.event
	337	self.assertEqual(ev['giraffe'], 1)
	338	self.assertIs(type(ev['giraffe']), bt2_field._SignedIntegerField)
	339	self.assertEqual(ev['ant'], -1)
	340	self.assertIs(type(ev['ant']), bt2_field._SignedIntegerField)
	341	self.assertEqual(ev['cpu_id'], 1)
	342	self.assertIs(type(ev['cpu_id']), bt2_field._SignedIntegerField)
	343	self.assertEqual(ev['something'], 154)
	344	self.assertIs(type(ev['something']), bt2_field._UnsignedIntegerField)
	345
2ae9f48c SM	346
	347	if __name__ == "__main__":
	348	unittest.main()