tests/bindings/python/bt2/test_packet.py

   1 from collections import OrderedDict
   2 from bt2 import values
   3 import unittest
   4 import copy
   5 import bt2
   6
   7
   8 class PacketTestCase(unittest.TestCase):
   9     def setUp(self):
  10         self._packet = self._create_packet()
  11
  12     def tearDown(self):
  13         del self._packet
  14
  15     def _create_packet(self, with_ph=True, with_pc=True):
  16         # event header
  17         eh = bt2.StructureFieldType()
  18         eh += OrderedDict((
  19             ('id', bt2.IntegerFieldType(8)),
  20             ('ts', bt2.IntegerFieldType(32)),
  21         ))
  22
  23         # stream event context
  24         sec = bt2.StructureFieldType()
  25         sec += OrderedDict((
  26             ('cpu_id', bt2.IntegerFieldType(8)),
  27             ('stuff', bt2.FloatingPointNumberFieldType()),
  28         ))
  29
  30         # packet context
  31         if with_pc:
  32             pc = bt2.StructureFieldType()
  33             pc += OrderedDict((
  34                 ('something', bt2.IntegerFieldType(8)),
  35                 ('something_else', bt2.FloatingPointNumberFieldType()),
  36             ))
  37         else:
  38             pc = None
  39
  40         # stream class
  41         sc = bt2.StreamClass()
  42         sc.packet_context_field_type = pc
  43         sc.event_header_field_type = eh
  44         sc.event_context_field_type = sec
  45
  46         # event context
  47         ec = bt2.StructureFieldType()
  48         ec += OrderedDict((
  49             ('ant', bt2.IntegerFieldType(16, is_signed=True)),
  50             ('msg', bt2.StringFieldType()),
  51         ))
  52
  53         # event payload
  54         ep = bt2.StructureFieldType()
  55         ep += OrderedDict((
  56             ('giraffe', bt2.IntegerFieldType(32)),
  57             ('gnu', bt2.IntegerFieldType(8)),
  58             ('mosquito', bt2.IntegerFieldType(8)),
  59         ))
  60
  61         # event class
  62         event_class = bt2.EventClass('ec')
  63         event_class.context_field_type = ec
  64         event_class.payload_field_type = ep
  65         sc.add_event_class(event_class)
  66
  67         # packet header
  68         if with_ph:
  69             ph = bt2.StructureFieldType()
  70             ph += OrderedDict((
  71                 ('magic', bt2.IntegerFieldType(32)),
  72                 ('stream_id', bt2.IntegerFieldType(16)),
  73             ))
  74         else:
  75             ph = None
  76
  77         # trace c;ass
  78         tc = bt2.Trace()
  79         tc.packet_header_field_type = ph
  80         tc.add_stream_class(sc)
  81
  82         # stream
  83         stream = sc()
  84
  85         # packet
  86         return stream.create_packet()
  87
  88     def test_attr_stream(self):
  89         self.assertIsNotNone(self._packet.stream)
  90
  91     def test_get_header_field(self):
  92         self.assertIsNotNone(self._packet.header_field)
  93
  94     def test_no_header_field(self):
  95         packet = self._create_packet(with_ph=False)
  96         self.assertIsNone(packet.header_field)
  97
  98     def test_get_context_field(self):
  99         self.assertIsNotNone(self._packet.context_field)
 100
 101     def test_no_context_field(self):
 102         packet = self._create_packet(with_pc=False)
 103         self.assertIsNone(packet.context_field)
 104
 105     def _fill_packet(self, packet):
 106         packet.header_field['magic'] = 0xc1fc1fc1
 107         packet.header_field['stream_id'] = 23
 108         packet.context_field['something'] = 17
 109         packet.context_field['something_else'] = 188.88
 110
 111     def test_eq(self):
 112         packet1 = self._create_packet()
 113         self._fill_packet(packet1)
 114         packet2 = self._create_packet()
 115         self._fill_packet(packet2)
 116         self.assertEqual(packet1, packet2)
 117
 118     def test_ne_header_field(self):
 119         packet1 = self._create_packet()
 120         self._fill_packet(packet1)
 121         packet2 = self._create_packet()
 122         self._fill_packet(packet2)
 123         packet2.header_field['stream_id'] = 18
 124         self.assertNotEqual(packet1, packet2)
 125
 126     def test_ne_context_field(self):
 127         packet1 = self._create_packet()
 128         self._fill_packet(packet1)
 129         packet2 = self._create_packet()
 130         self._fill_packet(packet2)
 131         packet2.context_field['something_else'] = 1948.11
 132         self.assertNotEqual(packet1, packet2)
 133
 134     def test_eq_invalid(self):
 135         self.assertFalse(self._packet == 23)
 136
 137     def _test_copy(self, func):
 138         packet = self._create_packet()
 139         self._fill_packet(packet)
 140         cpy = func(packet)
 141         self.assertIsNot(packet, cpy)
 142         self.assertNotEqual(packet.addr, cpy.addr)
 143         self.assertEqual(packet, cpy)
 144
 145     def test_copy(self):
 146         self._test_copy(copy.copy)
 147
 148     def test_deepcopy(self):
 149         self._test_copy(copy.deepcopy)