tests/bindings/python/bt2/test_packet.py

   1 #
   2 # Copyright (C) 2019 EfficiOS Inc.
   3 #
   4 # This program is free software; you can redistribute it and/or
   5 # modify it under the terms of the GNU General Public License
   6 # as published by the Free Software Foundation; only version 2
   7 # of the License.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU General Public License
  15 # along with this program; if not, write to the Free Software
  16 # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
  17 #
  18
  19 import unittest
  20 import utils
  21 from utils import run_in_component_init
  22 from bt2 import stream as bt2_stream
  23 from bt2 import field as bt2_field
  24
  25
  26 class PacketTestCase(unittest.TestCase):
  27     @staticmethod
  28     def _create_packet(with_pc):
  29         def create_tc_cc(comp_self):
  30             cc = comp_self._create_clock_class(frequency=1000, name='my_cc')
  31             tc = comp_self._create_trace_class()
  32             return cc, tc
  33
  34         clock_class, tc = run_in_component_init(create_tc_cc)
  35
  36         # stream event context
  37         sec = tc.create_structure_field_class()
  38         sec += [
  39             ('cpu_id', tc.create_signed_integer_field_class(8)),
  40             ('stuff', tc.create_double_precision_real_field_class()),
  41         ]
  42
  43         # packet context
  44         pc = None
  45         if with_pc:
  46             pc = tc.create_structure_field_class()
  47             pc += [
  48                 ('something', tc.create_signed_integer_field_class(8)),
  49                 ('something_else', tc.create_double_precision_real_field_class()),
  50                 ('events_discarded', tc.create_unsigned_integer_field_class(64)),
  51                 ('packet_seq_num', tc.create_unsigned_integer_field_class(64)),
  52             ]
  53
  54         # stream class
  55         sc = tc.create_stream_class(
  56             default_clock_class=clock_class,
  57             event_common_context_field_class=sec,
  58             packet_context_field_class=pc,
  59             supports_packets=True,
  60         )
  61
  62         # event context
  63         ec = tc.create_structure_field_class()
  64         ec += [
  65             ('ant', tc.create_signed_integer_field_class(16)),
  66             ('msg', tc.create_string_field_class()),
  67         ]
  68
  69         # event payload
  70         ep = tc.create_structure_field_class()
  71         ep += [
  72             ('giraffe', tc.create_signed_integer_field_class(32)),
  73             ('gnu', tc.create_signed_integer_field_class(8)),
  74             ('mosquito', tc.create_signed_integer_field_class(8)),
  75         ]
  76
  77         # event class
  78         event_class = sc.create_event_class(name='ec', payload_field_class=ep)
  79         event_class.common_context_field_class = ec
  80
  81         # trace
  82         trace = tc()
  83
  84         # stream
  85         stream = trace.create_stream(sc)
  86
  87         # packet
  88         return stream.create_packet(), stream, pc
  89
  90     def test_attr_stream(self):
  91         packet, stream, _ = self._create_packet(with_pc=True)
  92         self.assertEqual(packet.stream.addr, stream.addr)
  93         self.assertIs(type(packet.stream), bt2_stream._Stream)
  94
  95     def test_const_attr_stream(self):
  96         packet = utils.get_const_packet_beginning_message().packet
  97         self.assertIs(type(packet.stream), bt2_stream._StreamConst)
  98
  99     def test_context_field(self):
 100         packet, stream, pc_fc = self._create_packet(with_pc=True)
 101         self.assertEqual(packet.context_field.cls.addr, pc_fc.addr)
 102         self.assertIs(type(packet.context_field), bt2_field._StructureField)
 103
 104     def test_const_context_field(self):
 105         packet = utils.get_const_packet_beginning_message().packet
 106         self.assertIs(type(packet.context_field), bt2_field._StructureFieldConst)
 107
 108     def test_no_context_field(self):
 109         packet, _, _ = self._create_packet(with_pc=False)
 110         self.assertIsNone(packet.context_field)
 111
 112
 113 if __name__ == '__main__':
 114     unittest.main()