-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 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
-@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
+ 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
- with self.assertRaises(KeyError):
- ev['magic']
-
- ev.packet = packet
- self.assertEqual(ev['mosquito'], 42)
- self.assertEqual(ev['gnu'], 23)
+ 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.assertEqual(ev['msg'], 'hellooo')
+ self.assertIs(type(ev['giraffe']), bt2_field._SignedIntegerFieldConst)
+ self.assertEqual(ev['gnu'], 23)
+ self.assertEqual(ev['mosquito'], 42)
self.assertEqual(ev['ant'], -1)
- self.assertEqual(ev['stuff'], 13.194)
+ self.assertIs(type(ev['ant']), bt2_field._SignedIntegerFieldConst)
+ 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.assertIs(type(ev['cpu_id']), bt2_field._SignedIntegerFieldConst)
+ 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.assertIs(type(ev['something']), bt2_field._UnsignedIntegerFieldConst)
+ 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_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)
+ 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_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()
projects / babeltrace.git / blobdiff