-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 run_in_component_init
+from bt2 import stream as bt2_stream
+from bt2 import field as bt2_field
class PacketTestCase(unittest.TestCase):
- def setUp(self):
- self._packet = self._create_packet()
+ @staticmethod
+ def _create_packet(with_pc):
+ def create_tc_cc(comp_self):
+ cc = comp_self._create_clock_class(frequency=1000, name="my_cc")
+ tc = comp_self._create_trace_class()
+ return cc, tc
- def _create_packet(self, with_ph=True, with_pc=True):
- # event header
- eh = bt2.StructureFieldType()
- eh += OrderedDict((
- ('id', bt2.IntegerFieldType(8)),
- ('ts', bt2.IntegerFieldType(32)),
- ))
+ clock_class, tc = run_in_component_init(create_tc_cc)
# stream event context
- sec = bt2.StructureFieldType()
- sec += OrderedDict((
- ('cpu_id', bt2.IntegerFieldType(8)),
- ('stuff', bt2.FloatingPointNumberFieldType()),
- ))
+ sec = tc.create_structure_field_class()
+ sec += [
+ ("cpu_id", tc.create_signed_integer_field_class(8)),
+ ("stuff", tc.create_double_precision_real_field_class()),
+ ]
# packet context
+ pc = None
if with_pc:
- pc = bt2.StructureFieldType()
- pc += OrderedDict((
- ('something', bt2.IntegerFieldType(8)),
- ('something_else', bt2.FloatingPointNumberFieldType()),
- ))
- else:
- pc = None
+ pc = tc.create_structure_field_class()
+ pc += [
+ ("something", tc.create_signed_integer_field_class(8)),
+ ("something_else", tc.create_double_precision_real_field_class()),
+ ("events_discarded", tc.create_unsigned_integer_field_class(64)),
+ ("packet_seq_num", tc.create_unsigned_integer_field_class(64)),
+ ]
# stream class
- sc = bt2.StreamClass()
- sc.packet_context_field_type = pc
- sc.event_header_field_type = eh
- sc.event_context_field_type = sec
+ sc = tc.create_stream_class(
+ default_clock_class=clock_class,
+ event_common_context_field_class=sec,
+ packet_context_field_class=pc,
+ supports_packets=True,
+ )
# event context
- ec = bt2.StructureFieldType()
- ec += OrderedDict((
- ('ant', bt2.IntegerFieldType(16, is_signed=True)),
- ('msg', bt2.StringFieldType()),
- ))
+ ec = tc.create_structure_field_class()
+ ec += [
+ ("ant", tc.create_signed_integer_field_class(16)),
+ ("msg", tc.create_string_field_class()),
+ ]
# event payload
- ep = bt2.StructureFieldType()
- ep += OrderedDict((
- ('giraffe', bt2.IntegerFieldType(32)),
- ('gnu', bt2.IntegerFieldType(8)),
- ('mosquito', bt2.IntegerFieldType(8)),
- ))
+ 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
- event_class = bt2.EventClass('ec')
- event_class.context_field_type = ec
- event_class.payload_field_type = ep
- sc.add_event_class(event_class)
-
- # packet header
- if with_ph:
- ph = bt2.StructureFieldType()
- ph += OrderedDict((
- ('magic', bt2.IntegerFieldType(32)),
- ('stream_id', bt2.IntegerFieldType(16)),
- ))
- else:
- ph = None
-
- # trace c;ass
- tc = bt2.Trace()
- tc.packet_header_field_type = ph
- tc.add_stream_class(sc)
+ event_class = sc.create_event_class(name="ec", payload_field_class=ep)
+ event_class.common_context_field_class = ec
+
+ # trace
+ trace = tc()
# stream
- stream = sc()
+ stream = trace.create_stream(sc)
# packet
- return stream.create_packet()
+ return stream.create_packet(), stream, pc
def test_attr_stream(self):
- self.assertIsNotNone(self._packet.stream)
+ packet, stream, _ = self._create_packet(with_pc=True)
+ self.assertEqual(packet.stream.addr, stream.addr)
+ self.assertIs(type(packet.stream), bt2_stream._Stream)
- def test_get_header_field(self):
- self.assertIsNotNone(self._packet.header_field)
+ def test_const_attr_stream(self):
+ packet = utils.get_const_packet_beginning_message().packet
+ self.assertIs(type(packet.stream), bt2_stream._StreamConst)
- def test_no_header_field(self):
- packet = self._create_packet(with_ph=False)
- self.assertIsNone(packet.header_field)
+ def test_context_field(self):
+ packet, stream, pc_fc = self._create_packet(with_pc=True)
+ self.assertEqual(packet.context_field.cls.addr, pc_fc.addr)
+ self.assertIs(type(packet.context_field), bt2_field._StructureField)
- def test_get_context_field(self):
- self.assertIsNotNone(self._packet.context_field)
+ def test_const_context_field(self):
+ packet = utils.get_const_packet_beginning_message().packet
+ self.assertIs(type(packet.context_field), bt2_field._StructureFieldConst)
def test_no_context_field(self):
- packet = self._create_packet(with_pc=False)
+ packet, _, _ = self._create_packet(with_pc=False)
self.assertIsNone(packet.context_field)
- def _fill_packet(self, packet):
- packet.header_field['magic'] = 0xc1fc1fc1
- packet.header_field['stream_id'] = 23
- packet.context_field['something'] = 17
- packet.context_field['something_else'] = 188.88
-
- def test_eq(self):
- packet1 = self._create_packet()
- self._fill_packet(packet1)
- packet2 = self._create_packet()
- self._fill_packet(packet2)
- self.assertEqual(packet1, packet2)
-
- def test_ne_header_field(self):
- packet1 = self._create_packet()
- self._fill_packet(packet1)
- packet2 = self._create_packet()
- self._fill_packet(packet2)
- packet2.header_field['stream_id'] = 18
- self.assertNotEqual(packet1, packet2)
-
- def test_ne_context_field(self):
- packet1 = self._create_packet()
- self._fill_packet(packet1)
- packet2 = self._create_packet()
- self._fill_packet(packet2)
- packet2.context_field['something_else'] = 1948.11
- self.assertNotEqual(packet1, packet2)
-
- def test_eq_invalid(self):
- self.assertFalse(self._packet == 23)
-
- def _test_copy(self, func):
- packet = self._create_packet()
- self._fill_packet(packet)
- cpy = func(packet)
- self.assertIsNot(packet, cpy)
- self.assertNotEqual(packet.addr, cpy.addr)
- self.assertEqual(packet, cpy)
-
- def test_copy(self):
- self._test_copy(copy.copy)
-
- def test_deepcopy(self):
- self._test_copy(copy.deepcopy)
+
+if __name__ == "__main__":
+ unittest.main()
projects / babeltrace.git / blobdiff