tests/bindings/python/bt2/test_packet.py

   1 from collections import OrderedDict
   2 import unittest
   3 from utils import run_in_component_init
   4
   5
   6 class PacketTestCase(unittest.TestCase):
   7     @staticmethod
   8     def _create_packet(with_pc):
   9         def create_tc_cc(comp_self):
  10             cc = comp_self._create_clock_class(frequency=1000, name='my_cc')
  11             tc = comp_self._create_trace_class()
  12             return cc, tc
  13
  14         clock_class, tc = run_in_component_init(create_tc_cc)
  15
  16         # stream event context
  17         sec = tc.create_structure_field_class()
  18         sec += OrderedDict((
  19             ('cpu_id', tc.create_signed_integer_field_class(8)),
  20             ('stuff', tc.create_real_field_class()),
  21         ))
  22
  23         # packet context
  24         pc = None
  25         if with_pc:
  26             pc = tc.create_structure_field_class()
  27             pc += OrderedDict((
  28                 ('something', tc.create_signed_integer_field_class(8)),
  29                 ('something_else', tc.create_real_field_class()),
  30                 ('events_discarded', tc.create_unsigned_integer_field_class(64)),
  31                 ('packet_seq_num', tc.create_unsigned_integer_field_class(64)),
  32             ))
  33
  34         # stream class
  35         sc = tc.create_stream_class(default_clock_class=clock_class,
  36                                     event_common_context_field_class=sec,
  37                                     packet_context_field_class=pc)
  38
  39         # event context
  40         ec = tc.create_structure_field_class()
  41         ec += OrderedDict((
  42             ('ant', tc.create_signed_integer_field_class(16)),
  43             ('msg', tc.create_string_field_class()),
  44         ))
  45
  46         # event payload
  47         ep = tc.create_structure_field_class()
  48         ep += OrderedDict((
  49             ('giraffe', tc.create_signed_integer_field_class(32)),
  50             ('gnu', tc.create_signed_integer_field_class(8)),
  51             ('mosquito', tc.create_signed_integer_field_class(8)),
  52         ))
  53
  54         # event class
  55         event_class = sc.create_event_class(name='ec', payload_field_class=ep)
  56         event_class.common_context_field_class = ec
  57
  58         # trace
  59         trace = tc()
  60
  61         # stream
  62         stream = trace.create_stream(sc)
  63
  64         # packet
  65         return stream.create_packet(), stream, pc
  66
  67     def test_attr_stream(self):
  68         packet, stream, _ = self._create_packet(with_pc=True)
  69         self.assertEqual(packet.stream.addr, stream.addr)
  70
  71     def test_context_field(self):
  72         packet, stream, pc_fc = self._create_packet(with_pc=True)
  73         self.assertEqual(packet.context_field.field_class.addr, pc_fc.addr)
  74
  75     def test_no_context_field(self):
  76         packet, _, _ = self._create_packet(with_pc=False)
  77         self.assertIsNone(packet.context_field)