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