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