tests/bindings/python/bt2/test_event.py

   1 from collections import OrderedDict
   2 from bt2 import value
   3 import unittest
   4 import copy
   5 import bt2
   6
   7
   8 @unittest.skip("this is broken")
   9 class EventTestCase(unittest.TestCase):
  10     def setUp(self):
  11         self._ec = self._create_ec()
  12
  13     def tearDown(self):
  14         del self._ec
  15
  16     def _create_ec(self, with_eh=True, with_sec=True, with_ec=True, with_ep=True):
  17         # event header
  18         if with_eh:
  19             eh = bt2.StructureFieldClass()
  20             eh += OrderedDict((
  21                 ('id', bt2.IntegerFieldClass(8)),
  22                 ('ts', bt2.IntegerFieldClass(32)),
  23             ))
  24         else:
  25             eh = None
  26
  27         # stream event context
  28         if with_sec:
  29             sec = bt2.StructureFieldClass()
  30             sec += OrderedDict((
  31                 ('cpu_id', bt2.IntegerFieldClass(8)),
  32                 ('stuff', bt2.FloatingPointNumberFieldClass()),
  33             ))
  34         else:
  35             sec = None
  36
  37         # packet context
  38         pc = bt2.StructureFieldClass()
  39         pc += OrderedDict((
  40             ('something', bt2.IntegerFieldClass(8)),
  41             ('something_else', bt2.FloatingPointNumberFieldClass()),
  42         ))
  43
  44         # stream class
  45         sc = bt2.StreamClass()
  46         sc.packet_context_field_class = pc
  47         sc.event_header_field_class = eh
  48         sc.event_context_field_class = sec
  49
  50         # event context
  51         if with_ec:
  52             ec = bt2.StructureFieldClass()
  53             ec += OrderedDict((
  54                 ('ant', bt2.IntegerFieldClass(16, is_signed=True)),
  55                 ('msg', bt2.StringFieldClass()),
  56             ))
  57         else:
  58             ec = None
  59
  60         # event payload
  61         if with_ep:
  62             ep = bt2.StructureFieldClass()
  63             ep += OrderedDict((
  64                 ('giraffe', bt2.IntegerFieldClass(32)),
  65                 ('gnu', bt2.IntegerFieldClass(8)),
  66                 ('mosquito', bt2.IntegerFieldClass(8)),
  67             ))
  68         else:
  69             ep = None
  70
  71         # event class
  72         event_class = bt2.EventClass('ec')
  73         event_class.context_field_class = ec
  74         event_class.payload_field_class = ep
  75         sc.add_event_class(event_class)
  76         return event_class
  77
  78     def test_attr_event_class(self):
  79         ev = self._ec()
  80         self.assertEqual(ev.event_class.addr, self._ec.addr)
  81
  82     def test_attr_name(self):
  83         ev = self._ec()
  84         self.assertEqual(ev.name, self._ec.name)
  85
  86     def test_attr_id(self):
  87         ev = self._ec()
  88         self.assertEqual(ev.id, self._ec.id)
  89
  90     def test_get_event_header_field(self):
  91         ev = self._ec()
  92         ev.header_field['id'] = 23
  93         ev.header_field['ts'] = 1234
  94         self.assertEqual(ev.header_field['id'], 23)
  95         self.assertEqual(ev.header_field['ts'], 1234)
  96
  97     def test_set_event_header_field(self):
  98         eh = self._ec.stream_class.event_header_field_class()
  99         eh['id'] = 17
 100         eh['ts'] = 188
 101         ev = self._ec()
 102         ev.header_field = eh
 103         self.assertEqual(ev.header_field['id'], 17)
 104         self.assertEqual(ev.header_field['ts'], 188)
 105
 106     def test_get_stream_event_context_field(self):
 107         ev = self._ec()
 108         ev.stream_event_context_field['cpu_id'] = 1
 109         ev.stream_event_context_field['stuff'] = 13.194
 110         self.assertEqual(ev.stream_event_context_field['cpu_id'], 1)
 111         self.assertEqual(ev.stream_event_context_field['stuff'], 13.194)
 112
 113     def test_set_stream_event_context_field(self):
 114         sec = self._ec.stream_class.event_context_field_class()
 115         sec['cpu_id'] = 2
 116         sec['stuff'] = 19.19
 117         ev = self._ec()
 118         ev.stream_event_context_field = sec
 119         self.assertEqual(ev.stream_event_context_field['cpu_id'], 2)
 120         self.assertEqual(ev.stream_event_context_field['stuff'], 19.19)
 121
 122     def test_no_stream_event_context(self):
 123         ec = self._create_ec(with_sec=False)
 124         ev = ec()
 125         self.assertIsNone(ev.stream_event_context_field)
 126
 127     def test_get_event_context_field(self):
 128         ev = self._ec()
 129         ev.context_field['ant'] = -1
 130         ev.context_field['msg'] = 'hellooo'
 131         self.assertEqual(ev.context_field['ant'], -1)
 132         self.assertEqual(ev.context_field['msg'], 'hellooo')
 133
 134     def test_set_event_context_field(self):
 135         ec = self._ec.context_field_class()
 136         ec['ant'] = 2
 137         ec['msg'] = 'hi there'
 138         ev = self._ec()
 139         ev.context_field = ec
 140         self.assertEqual(ev.context_field['ant'], 2)
 141         self.assertEqual(ev.context_field['msg'], 'hi there')
 142
 143     def test_no_event_context(self):
 144         ec = self._create_ec(with_ec=False)
 145         ev = ec()
 146         self.assertIsNone(ev.context_field)
 147
 148     def test_get_event_payload_field(self):
 149         ev = self._ec()
 150         ev.payload_field['giraffe'] = 1
 151         ev.payload_field['gnu'] = 23
 152         ev.payload_field['mosquito'] = 42
 153         self.assertEqual(ev.payload_field['giraffe'], 1)
 154         self.assertEqual(ev.payload_field['gnu'], 23)
 155         self.assertEqual(ev.payload_field['mosquito'], 42)
 156
 157     def test_set_event_payload_field(self):
 158         ep = self._ec.payload_field_class()
 159         ep['giraffe'] = 2
 160         ep['gnu'] = 124
 161         ep['mosquito'] = 17
 162         ev = self._ec()
 163         ev.payload_field = ep
 164         self.assertEqual(ev.payload_field['giraffe'], 2)
 165         self.assertEqual(ev.payload_field['gnu'], 124)
 166         self.assertEqual(ev.payload_field['mosquito'], 17)
 167
 168     def test_clock_snapshot(self):
 169         tc = bt2.Trace()
 170         tc.add_stream_class(self._ec.stream_class)
 171         cc = bt2.ClockClass('hi', 1000)
 172         tc.add_clock_class(cc)
 173         ev = self._ec()
 174         ev.clock_snapshots.add(cc(177))
 175         self.assertEqual(ev.clock_snapshots[cc].cycles, 177)
 176
 177     def test_no_clock_snapshot(self):
 178         tc = bt2.Trace()
 179         tc.add_stream_class(self._ec.stream_class)
 180         cc = bt2.ClockClass('hi', 1000)
 181         tc.add_clock_class(cc)
 182         ev = self._ec()
 183         self.assertIsNone(ev.clock_snapshots[cc])
 184
 185     def test_no_packet(self):
 186         ev = self._ec()
 187         self.assertIsNone(ev.packet)
 188
 189     def test_packet(self):
 190         tc = bt2.Trace()
 191         tc.packet_header_field_class = bt2.StructureFieldClass()
 192         tc.packet_header_field_class.append_field('magic', bt2.IntegerFieldClass(32))
 193         tc.packet_header_field_class.append_field('stream_id', bt2.IntegerFieldClass(16))
 194         tc.add_stream_class(self._ec.stream_class)
 195         ev = self._ec()
 196         self._fill_ev(ev)
 197         stream = self._ec.stream_class()
 198         packet = stream.create_packet()
 199         packet.header_field['magic'] = 0xc1fc1fc1
 200         packet.header_field['stream_id'] = 0
 201         packet.context_field['something'] = 154
 202         packet.context_field['something_else'] = 17.2
 203         ev.packet = packet
 204         self.assertEqual(ev.packet.addr, packet.addr)
 205
 206     def test_no_stream(self):
 207         ev = self._ec()
 208         self.assertIsNone(ev.stream)
 209
 210     def test_stream(self):
 211         tc = bt2.Trace()
 212         tc.packet_header_field_class = bt2.StructureFieldClass()
 213         tc.packet_header_field_class.append_field('magic', bt2.IntegerFieldClass(32))
 214         tc.packet_header_field_class.append_field('stream_id', bt2.IntegerFieldClass(16))
 215         tc.add_stream_class(self._ec.stream_class)
 216         ev = self._ec()
 217         self._fill_ev(ev)
 218         stream = self._ec.stream_class()
 219         packet = stream.create_packet()
 220         packet.header_field['magic'] = 0xc1fc1fc1
 221         packet.header_field['stream_id'] = 0
 222         packet.context_field['something'] = 154
 223         packet.context_field['something_else'] = 17.2
 224         ev.packet = packet
 225         self.assertEqual(ev.stream.addr, stream.addr)
 226
 227     def _fill_ev(self, ev):
 228         ev.header_field['id'] = 23
 229         ev.header_field['ts'] = 1234
 230         ev.stream_event_context_field['cpu_id'] = 1
 231         ev.stream_event_context_field['stuff'] = 13.194
 232         ev.context_field['ant'] = -1
 233         ev.context_field['msg'] = 'hellooo'
 234         ev.payload_field['giraffe'] = 1
 235         ev.payload_field['gnu'] = 23
 236         ev.payload_field['mosquito'] = 42
 237
 238     def _get_full_ev(self):
 239         tc = bt2.Trace()
 240         tc.add_stream_class(self._ec.stream_class)
 241         cc = bt2.ClockClass('hi', 1000)
 242         tc.add_clock_class(cc)
 243         ev = self._ec()
 244         self._fill_ev(ev)
 245         ev.clock_snapshots.add(cc(234))
 246         return ev
 247
 248     def test_getitem(self):
 249         tc = bt2.Trace()
 250         tc.packet_header_field_class = bt2.StructureFieldClass()
 251         tc.packet_header_field_class.append_field('magic', bt2.IntegerFieldClass(32))
 252         tc.packet_header_field_class.append_field('stream_id', bt2.IntegerFieldClass(16))
 253         tc.add_stream_class(self._ec.stream_class)
 254         ev = self._ec()
 255         self._fill_ev(ev)
 256         stream = self._ec.stream_class()
 257         packet = stream.create_packet()
 258         packet.header_field['magic'] = 0xc1fc1fc1
 259         packet.header_field['stream_id'] = 0
 260         packet.context_field['something'] = 154
 261         packet.context_field['something_else'] = 17.2
 262
 263         with self.assertRaises(KeyError):
 264             ev['magic']
 265
 266         ev.packet = packet
 267         self.assertEqual(ev['mosquito'], 42)
 268         self.assertEqual(ev['gnu'], 23)
 269         self.assertEqual(ev['giraffe'], 1)
 270         self.assertEqual(ev['msg'], 'hellooo')
 271         self.assertEqual(ev['ant'], -1)
 272         self.assertEqual(ev['stuff'], 13.194)
 273         self.assertEqual(ev['cpu_id'], 1)
 274         self.assertEqual(ev['ts'], 1234)
 275         self.assertEqual(ev['id'], 23)
 276         self.assertEqual(ev['something_else'], 17.2)
 277         self.assertEqual(ev['something'], 154)
 278         self.assertEqual(ev['stream_id'], 0)
 279         self.assertEqual(ev['magic'], 0xc1fc1fc1)
 280
 281         with self.assertRaises(KeyError):
 282             ev['yes']
 283
 284     def test_eq(self):
 285         tc = bt2.Trace()
 286         tc.add_stream_class(self._ec.stream_class)
 287         cc = bt2.ClockClass('hi', 1000)
 288         tc.add_clock_class(cc)
 289         ev1 = self._ec()
 290         self._fill_ev(ev1)
 291         ev1.clock_snapshots.add(cc(234))
 292         ev2 = self._ec()
 293         self._fill_ev(ev2)
 294         ev2.clock_snapshots.add(cc(234))
 295         self.assertEqual(ev1, ev2)
 296
 297     def test_ne_header_field(self):
 298         tc = bt2.Trace()
 299         tc.add_stream_class(self._ec.stream_class)
 300         cc = bt2.ClockClass('hi', 1000)
 301         tc.add_clock_class(cc)
 302         ev1 = self._ec()
 303         self._fill_ev(ev1)
 304         ev1.header_field['id'] = 19
 305         ev1.clock_snapshots.add(cc(234))
 306         ev2 = self._ec()
 307         self._fill_ev(ev2)
 308         ev2.clock_snapshots.add(cc(234))
 309         self.assertNotEqual(ev1, ev2)
 310
 311     def test_ne_stream_event_context_field(self):
 312         tc = bt2.Trace()
 313         tc.add_stream_class(self._ec.stream_class)
 314         cc = bt2.ClockClass('hi', 1000)
 315         tc.add_clock_class(cc)
 316         ev1 = self._ec()
 317         self._fill_ev(ev1)
 318         ev1.stream_event_context_field['cpu_id'] = 3
 319         ev1.clock_snapshots.add(cc(234))
 320         ev2 = self._ec()
 321         self._fill_ev(ev2)
 322         ev2.clock_snapshots.add(cc(234))
 323         self.assertNotEqual(ev1, ev2)
 324
 325     def test_ne_context_field(self):
 326         tc = bt2.Trace()
 327         tc.add_stream_class(self._ec.stream_class)
 328         cc = bt2.ClockClass('hi', 1000)
 329         tc.add_clock_class(cc)
 330         ev1 = self._ec()
 331         self._fill_ev(ev1)
 332         ev1.context_field['ant'] = -3
 333         ev1.clock_snapshots.add(cc(234))
 334         ev2 = self._ec()
 335         self._fill_ev(ev2)
 336         ev2.clock_snapshots.add(cc(234))
 337         self.assertNotEqual(ev1, ev2)
 338
 339     def test_ne_payload_field(self):
 340         tc = bt2.Trace()
 341         tc.add_stream_class(self._ec.stream_class)
 342         cc = bt2.ClockClass('hi', 1000)
 343         tc.add_clock_class(cc)
 344         ev1 = self._ec()
 345         self._fill_ev(ev1)
 346         ev1.payload_field['mosquito'] = 98
 347         ev1.clock_snapshots.add(cc(234))
 348         ev2 = self._ec()
 349         self._fill_ev(ev2)
 350         ev2.clock_snapshots.add(cc(234))
 351         self.assertNotEqual(ev1, ev2)
 352
 353     def test_eq_invalid(self):
 354         ev = self._ec()
 355         self.assertFalse(ev == 23)
 356
 357     def _test_copy(self, func):
 358         tc = bt2.Trace()
 359         tc.add_stream_class(self._ec.stream_class)
 360         cc = bt2.ClockClass('hi', 1000)
 361         tc.add_clock_class(cc)
 362         ev = self._ec()
 363         self._fill_ev(ev)
 364         ev.clock_snapshots.add(cc(234))
 365         cpy = func(ev)
 366         self.assertIsNot(ev, cpy)
 367         self.assertNotEqual(ev.addr, cpy.addr)
 368         self.assertEqual(ev, cpy)
 369
 370     def test_copy(self):
 371         self._test_copy(copy.copy)
 372
 373     def test_deepcopy(self):
 374         self._test_copy(copy.deepcopy)