tests/bindings/python/bt2/test_event.py

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