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