X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_event.py;h=72490603475478aae10e2736ce85fd3d6835cf8f;hb=HEAD;hp=21329b5b2e84cf04fd3ea99cb3960d2b6e797cb5;hpb=4b552f8b1184f7178130a8026b2d62f69104aab1;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_event.py b/tests/bindings/python/bt2/test_event.py index 21329b5b..72490603 100644 --- a/tests/bindings/python/bt2/test_event.py +++ b/tests/bindings/python/bt2/test_event.py @@ -1,374 +1,460 @@ -from collections import OrderedDict -from bt2 import values +# SPDX-License-Identifier: GPL-2.0-only +# +# Copyright (C) 2019 EfficiOS Inc. +# + import unittest -import copy + import bt2 +import utils +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 +from utils import TestOutputPortMessageIterator -@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.StructureFieldClass() - eh += OrderedDict(( - ('id', bt2.IntegerFieldClass(8)), - ('ts', bt2.IntegerFieldClass(32)), - )) - else: - eh = None - - # stream event context - if with_sec: - sec = bt2.StructureFieldClass() - sec += OrderedDict(( - ('cpu_id', bt2.IntegerFieldClass(8)), - ('stuff', bt2.FloatingPointNumberFieldClass()), - )) - else: - sec = None - - # packet context - pc = bt2.StructureFieldClass() - pc += OrderedDict(( - ('something', bt2.IntegerFieldClass(8)), - ('something_else', bt2.FloatingPointNumberFieldClass()), - )) - - # stream class - sc = bt2.StreamClass() - sc.packet_context_field_class = pc - sc.event_header_field_class = eh - sc.event_context_field_class = sec - - # event context - if with_ec: - ec = bt2.StructureFieldClass() - ec += OrderedDict(( - ('ant', bt2.IntegerFieldClass(16, is_signed=True)), - ('msg', bt2.StringFieldClass()), - )) - else: - ec = None - - # event payload - if with_ep: - ep = bt2.StructureFieldClass() - ep += OrderedDict(( - ('giraffe', bt2.IntegerFieldClass(32)), - ('gnu', bt2.IntegerFieldClass(8)), - ('mosquito', bt2.IntegerFieldClass(8)), - )) - else: - ep = None - - # event class - event_class = bt2.EventClass('ec') - event_class.context_field_class = ec - event_class.payload_field_class = 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, 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()), + ("gnu", tc.create_string_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): - 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_class() - 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_class() - 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_class() - 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_class() - 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 test_clock_snapshot(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_snapshots.add(cc(177)) - self.assertEqual(ev.clock_snapshots[cc].cycles, 177) - - def test_no_clock_snapshot(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_snapshots[cc]) - - def test_no_packet(self): - ev = self._ec() - self.assertIsNone(ev.packet) - - def test_packet(self): - tc = bt2.Trace() - tc.packet_header_field_class = bt2.StructureFieldClass() - tc.packet_header_field_class.append_field('magic', bt2.IntegerFieldClass(32)) - tc.packet_header_field_class.append_field('stream_id', bt2.IntegerFieldClass(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.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 + event.common_context_field["gnu"] = "salut" + + 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.assertEqual(msg.event.common_context_field["gnu"], "salut") + 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): - tc = bt2.Trace() - tc.packet_header_field_class = bt2.StructureFieldClass() - tc.packet_header_field_class.append_field('magic', bt2.IntegerFieldClass(32)) - tc.packet_header_field_class.append_field('stream_id', bt2.IntegerFieldClass(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_snapshots.add(cc(234)) - return ev - - def test_getitem(self): - tc = bt2.Trace() - tc.packet_header_field_class = bt2.StructureFieldClass() - tc.packet_header_field_class.append_field('magic', bt2.IntegerFieldClass(32)) - tc.packet_header_field_class.append_field('stream_id', bt2.IntegerFieldClass(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 + msg = utils.get_event_message() + self.assertIs(type(msg.event.stream), bt2_stream._Stream) + + @staticmethod + def _event_payload_fields_config(event): + event.payload_field["giraffe"] = 1 + event.payload_field["gnu"] = 23 + event.payload_field["mosquito"] = 42 + + @staticmethod + def _event_fields_config(event): + EventTestCase._event_payload_fields_config(event) + 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 + event.common_context_field["gnu"] = "salut" + + @staticmethod + def _packet_fields_config(packet): + packet.context_field["something"] = 154 + packet.context_field["something_else"] = 17.2 + + def test_const_getitem(self): + msg = self._create_test_const_event_message( + packet_fields_config=self._packet_fields_config, + event_fields_config=self._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['magic'] - - ev.packet = packet - self.assertEqual(ev['mosquito'], 42) - self.assertEqual(ev['gnu'], 23) - self.assertEqual(ev['giraffe'], 1) - self.assertEqual(ev['msg'], 'hellooo') - self.assertEqual(ev['ant'], -1) - self.assertEqual(ev['stuff'], 13.194) - 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['something'], 154) - self.assertEqual(ev['stream_id'], 0) - self.assertEqual(ev['magic'], 0xc1fc1fc1) + ev["yes"] + + def test_const_getitem_no_packet(self): + msg = self._create_test_const_event_message( + event_fields_config=self._event_payload_fields_config, + with_ep=True, + ) + ev = msg.event 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_snapshots.add(cc(234)) - ev2 = self._ec() - self._fill_ev(ev2) - ev2.clock_snapshots.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_snapshots.add(cc(234)) - ev2 = self._ec() - self._fill_ev(ev2) - ev2.clock_snapshots.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_snapshots.add(cc(234)) - ev2 = self._ec() - self._fill_ev(ev2) - ev2.clock_snapshots.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_snapshots.add(cc(234)) - ev2 = self._ec() - self._fill_ev(ev2) - ev2.clock_snapshots.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_snapshots.add(cc(234)) - ev2 = self._ec() - self._fill_ev(ev2) - ev2.clock_snapshots.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_snapshots.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) + 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) + + def test_iter_full(self): + msg = self._create_test_const_event_message( + packet_fields_config=self._packet_fields_config, + event_fields_config=self._event_fields_config, + with_cc=True, + with_sc=True, + with_ep=True, + with_packet=True, + ) + expected_field_names = [ + # payload + "giraffe", + "gnu", + "mosquito", + # specific context + "ant", + "msg", + # common context + "cpu_id", + "stuff", + # packet context + "something", + "something_else", + ] + self.assertEqual(list(msg.event), expected_field_names) + + def test_iter_payload_only(self): + msg = self._create_test_const_event_message( + event_fields_config=self._event_payload_fields_config, + with_ep=True, + ) + expected_field_names = [ + # payload + "giraffe", + "gnu", + "mosquito", + ] + self.assertEqual(list(msg.event), expected_field_names) + + def test_len_full(self): + msg = self._create_test_const_event_message( + packet_fields_config=self._packet_fields_config, + event_fields_config=self._event_fields_config, + with_cc=True, + with_sc=True, + with_ep=True, + with_packet=True, + ) + self.assertEqual(len(msg.event), 9) + + def test_len_payload_only(self): + msg = self._create_test_const_event_message( + packet_fields_config=None, + event_fields_config=self._event_payload_fields_config, + with_ep=True, + ) + self.assertEqual(len(msg.event), 3) + + def test_in_full(self): + msg = self._create_test_const_event_message( + packet_fields_config=self._packet_fields_config, + event_fields_config=self._event_fields_config, + with_cc=True, + with_sc=True, + with_ep=True, + with_packet=True, + ) + field_names = [ + # payload + "giraffe", + "gnu", + "mosquito", + # specific context + "ant", + "msg", + # common context + "cpu_id", + "stuff", + # packet context + "something", + "something_else", + ] + + for field_name in field_names: + self.assertTrue(field_name in msg.event) + + self.assertFalse("lol" in msg.event) + + def test_in_payload_only(self): + msg = self._create_test_const_event_message( + packet_fields_config=None, + event_fields_config=self._event_payload_fields_config, + with_ep=True, + ) + field_names = [ + "giraffe", + "gnu", + "mosquito", + ] + + for field_name in field_names: + self.assertTrue(field_name in msg.event) + + self.assertFalse("lol" in msg.event) + + +if __name__ == "__main__": + unittest.main()