tests/bindings/python/bt2/test_packet.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # Copyright (C) 2019 EfficiOS Inc.
   4 #
   5
   6 import unittest
   7
   8 import utils
   9 from bt2 import field as bt2_field
  10 from bt2 import stream as bt2_stream
  11 from utils import run_in_component_init
  12
  13
  14 class PacketTestCase(unittest.TestCase):
  15     @staticmethod
  16     def _create_packet(with_pc):
  17         def create_tc_cc(comp_self):
  18             cc = comp_self._create_clock_class(frequency=1000, name="my_cc")
  19             tc = comp_self._create_trace_class()
  20             return cc, tc
  21
  22         clock_class, tc = run_in_component_init(create_tc_cc)
  23
  24         # stream event context
  25         sec = tc.create_structure_field_class()
  26         sec += [
  27             ("cpu_id", tc.create_signed_integer_field_class(8)),
  28             ("stuff", tc.create_double_precision_real_field_class()),
  29         ]
  30
  31         # packet context
  32         pc = None
  33         if with_pc:
  34             pc = tc.create_structure_field_class()
  35             pc += [
  36                 ("something", tc.create_signed_integer_field_class(8)),
  37                 ("something_else", tc.create_double_precision_real_field_class()),
  38                 ("events_discarded", tc.create_unsigned_integer_field_class(64)),
  39                 ("packet_seq_num", tc.create_unsigned_integer_field_class(64)),
  40             ]
  41
  42         # stream class
  43         sc = tc.create_stream_class(
  44             default_clock_class=clock_class,
  45             event_common_context_field_class=sec,
  46             packet_context_field_class=pc,
  47             supports_packets=True,
  48         )
  49
  50         # event context
  51         ec = tc.create_structure_field_class()
  52         ec += [
  53             ("ant", tc.create_signed_integer_field_class(16)),
  54             ("msg", tc.create_string_field_class()),
  55         ]
  56
  57         # event payload
  58         ep = tc.create_structure_field_class()
  59         ep += [
  60             ("giraffe", tc.create_signed_integer_field_class(32)),
  61             ("gnu", tc.create_signed_integer_field_class(8)),
  62             ("mosquito", tc.create_signed_integer_field_class(8)),
  63         ]
  64
  65         # event class
  66         event_class = sc.create_event_class(name="ec", payload_field_class=ep)
  67         event_class.common_context_field_class = ec
  68
  69         # trace
  70         trace = tc()
  71
  72         # stream
  73         stream = trace.create_stream(sc)
  74
  75         # packet
  76         return stream.create_packet(), stream, pc
  77
  78     def test_attr_stream(self):
  79         packet, stream, _ = self._create_packet(with_pc=True)
  80         self.assertEqual(packet.stream.addr, stream.addr)
  81         self.assertIs(type(packet.stream), bt2_stream._Stream)
  82
  83     def test_const_attr_stream(self):
  84         packet = utils.get_const_packet_beginning_message().packet
  85         self.assertIs(type(packet.stream), bt2_stream._StreamConst)
  86
  87     def test_context_field(self):
  88         packet, stream, pc_fc = self._create_packet(with_pc=True)
  89         self.assertEqual(packet.context_field.cls.addr, pc_fc.addr)
  90         self.assertIs(type(packet.context_field), bt2_field._StructureField)
  91
  92     def test_const_context_field(self):
  93         packet = utils.get_const_packet_beginning_message().packet
  94         self.assertIs(type(packet.context_field), bt2_field._StructureFieldConst)
  95
  96     def test_no_context_field(self):
  97         packet, _, _ = self._create_packet(with_pc=False)
  98         self.assertIsNone(packet.context_field)
  99
 100
 101 if __name__ == "__main__":
 102     unittest.main()