- ev = self._ec()
- self.assertEqual(ev.event_class.addr, self._ec.addr)
-
- def test_attr_name(self):
- ev = self._ec()
- self.assertEqual(ev.name, self._ec.name)
-
- def test_attr_id(self):
- ev = self._ec()
- self.assertEqual(ev.id, self._ec.id)
-
- def test_get_event_header_field(self):
- ev = self._ec()
- ev.header_field['id'] = 23
- ev.header_field['ts'] = 1234
- self.assertEqual(ev.header_field['id'], 23)
- self.assertEqual(ev.header_field['ts'], 1234)
-
- def test_set_event_header_field(self):
- eh = self._ec.stream_class.event_header_field_class()
- eh['id'] = 17
- eh['ts'] = 188
- ev = self._ec()
- ev.header_field = eh
- self.assertEqual(ev.header_field['id'], 17)
- self.assertEqual(ev.header_field['ts'], 188)
-
- def test_get_stream_event_context_field(self):
- ev = self._ec()
- ev.stream_event_context_field['cpu_id'] = 1
- ev.stream_event_context_field['stuff'] = 13.194
- self.assertEqual(ev.stream_event_context_field['cpu_id'], 1)
- self.assertEqual(ev.stream_event_context_field['stuff'], 13.194)
-
- def test_set_stream_event_context_field(self):
- sec = self._ec.stream_class.event_context_field_class()
- sec['cpu_id'] = 2
- sec['stuff'] = 19.19
- ev = self._ec()
- ev.stream_event_context_field = sec
- self.assertEqual(ev.stream_event_context_field['cpu_id'], 2)
- self.assertEqual(ev.stream_event_context_field['stuff'], 19.19)
-
- def test_no_stream_event_context(self):
- ec = self._create_ec(with_sec=False)
- ev = ec()
- self.assertIsNone(ev.stream_event_context_field)
-
- def test_get_event_context_field(self):
- ev = self._ec()
- ev.context_field['ant'] = -1
- ev.context_field['msg'] = 'hellooo'
- self.assertEqual(ev.context_field['ant'], -1)
- self.assertEqual(ev.context_field['msg'], 'hellooo')
-
- def test_set_event_context_field(self):
- ec = self._ec.context_field_class()
- ec['ant'] = 2
- ec['msg'] = 'hi there'
- ev = self._ec()
- ev.context_field = ec
- self.assertEqual(ev.context_field['ant'], 2)
- self.assertEqual(ev.context_field['msg'], 'hi there')
-
- def test_no_event_context(self):
- ec = self._create_ec(with_ec=False)
- ev = ec()
- self.assertIsNone(ev.context_field)
-
- def test_get_event_payload_field(self):
- ev = self._ec()
- ev.payload_field['giraffe'] = 1
- ev.payload_field['gnu'] = 23
- ev.payload_field['mosquito'] = 42
- self.assertEqual(ev.payload_field['giraffe'], 1)
- self.assertEqual(ev.payload_field['gnu'], 23)
- self.assertEqual(ev.payload_field['mosquito'], 42)
-
- def test_set_event_payload_field(self):
- ep = self._ec.payload_field_class()
- ep['giraffe'] = 2
- ep['gnu'] = 124
- ep['mosquito'] = 17
- ev = self._ec()
- ev.payload_field = ep
- self.assertEqual(ev.payload_field['giraffe'], 2)
- self.assertEqual(ev.payload_field['gnu'], 124)
- self.assertEqual(ev.payload_field['mosquito'], 17)
-
- def test_clock_snapshot(self):
- tc = bt2.Trace()
- tc.add_stream_class(self._ec.stream_class)
- cc = bt2.ClockClass('hi', 1000)
- tc.add_clock_class(cc)
- ev = self._ec()
- ev.clock_snapshots.add(cc(177))
- self.assertEqual(ev.clock_snapshots[cc].cycles, 177)
-
- def test_no_clock_snapshot(self):
- tc = bt2.Trace()
- tc.add_stream_class(self._ec.stream_class)
- cc = bt2.ClockClass('hi', 1000)
- tc.add_clock_class(cc)
- ev = self._ec()
- self.assertIsNone(ev.clock_snapshots[cc])
-
- def test_no_packet(self):
- ev = self._ec()
- self.assertIsNone(ev.packet)
-
- def test_packet(self):
- tc = bt2.Trace()
- tc.packet_header_field_class = bt2.StructureFieldClass()
- tc.packet_header_field_class.append_field('magic', bt2.IntegerFieldClass(32))
- tc.packet_header_field_class.append_field('stream_id', bt2.IntegerFieldClass(16))
- tc.add_stream_class(self._ec.stream_class)
- ev = self._ec()
- self._fill_ev(ev)
- stream = self._ec.stream_class()
- packet = stream.create_packet()
- packet.header_field['magic'] = 0xc1fc1fc1
- packet.header_field['stream_id'] = 0
- packet.context_field['something'] = 154
- packet.context_field['something_else'] = 17.2
- ev.packet = packet
- self.assertEqual(ev.packet.addr, packet.addr)
-
- def test_no_stream(self):
- ev = self._ec()
- self.assertIsNone(ev.stream)
+ msg = utils.get_event_message()
+ self.assertIs(type(msg.event.cls), bt2_event_class._EventClass)
+
+ def test_const_attr_name(self):
+ msg = self._create_test_const_event_message()
+ self.assertEqual(msg.event.name, self.event_class.name)
+
+ def test_const_attr_id(self):
+ msg = self._create_test_const_event_message()
+ self.assertEqual(msg.event.id, self.event_class.id)
+
+ def test_const_get_common_context_field(self):
+ def event_fields_config(event):
+ event.common_context_field['cpu_id'] = 1
+ event.common_context_field['stuff'] = 13.194
+
+ msg = self._create_test_const_event_message(
+ event_fields_config=event_fields_config, with_cc=True
+ )
+
+ self.assertEqual(msg.event.common_context_field['cpu_id'], 1)
+ self.assertEqual(msg.event.common_context_field['stuff'], 13.194)
+ self.assertIs(
+ type(msg.event.common_context_field), bt2_field._StructureFieldConst
+ )
+
+ def test_attr_common_context_field(self):
+ msg = utils.get_event_message()
+ self.assertIs(type(msg.event.common_context_field), bt2_field._StructureField)
+
+ def test_const_no_common_context_field(self):
+ msg = self._create_test_const_event_message(with_cc=False)
+ self.assertIsNone(msg.event.common_context_field)
+
+ def test_const_get_specific_context_field(self):
+ def event_fields_config(event):
+ event.specific_context_field['ant'] = -1
+ event.specific_context_field['msg'] = 'hellooo'
+
+ msg = self._create_test_const_event_message(
+ event_fields_config=event_fields_config, with_sc=True
+ )
+
+ self.assertEqual(msg.event.specific_context_field['ant'], -1)
+ self.assertEqual(msg.event.specific_context_field['msg'], 'hellooo')
+ self.assertIs(
+ type(msg.event.specific_context_field), bt2_field._StructureFieldConst
+ )
+
+ def test_attr_specific_context_field(self):
+ msg = utils.get_event_message()
+ self.assertIs(type(msg.event.specific_context_field), bt2_field._StructureField)
+
+ def test_const_no_specific_context_field(self):
+ msg = self._create_test_const_event_message(with_sc=False)
+ self.assertIsNone(msg.event.specific_context_field)
+
+ def test_const_get_event_payload_field(self):
+ def event_fields_config(event):
+ event.payload_field['giraffe'] = 1
+ event.payload_field['gnu'] = 23
+ event.payload_field['mosquito'] = 42
+
+ msg = self._create_test_const_event_message(
+ event_fields_config=event_fields_config, with_ep=True
+ )
+
+ self.assertEqual(msg.event.payload_field['giraffe'], 1)
+ self.assertEqual(msg.event.payload_field['gnu'], 23)
+ self.assertEqual(msg.event.payload_field['mosquito'], 42)
+ self.assertIs(type(msg.event.payload_field), bt2_field._StructureFieldConst)
+
+ def test_attr_payload_field(self):
+ msg = utils.get_event_message()
+ self.assertIs(type(msg.event.payload_field), bt2_field._StructureField)
+
+ def test_const_no_payload_field(self):
+ msg = self._create_test_const_event_message(with_ep=False)
+ self.assertIsNone(msg.event.payload_field)
+
+ def test_const_clock_value(self):
+ msg = self._create_test_const_event_message(with_clockclass=True)
+ self.assertEqual(msg.default_clock_snapshot.value, 789)
+ self.assertIs(
+ type(msg.default_clock_snapshot), bt2_clock_snapshot._ClockSnapshotConst
+ )
+
+ def test_clock_value(self):
+ msg = utils.get_event_message()
+ self.assertEqual(msg.default_clock_snapshot.value, 789)
+ self.assertIs(
+ type(msg.default_clock_snapshot), bt2_clock_snapshot._ClockSnapshotConst
+ )
+
+ def test_const_no_clock_value(self):
+ msg = self._create_test_const_event_message(with_clockclass=False)
+ with self.assertRaisesRegex(
+ ValueError, 'stream class has no default clock class'
+ ):
+ msg.default_clock_snapshot
+
+ def test_const_stream(self):
+ msg = self._create_test_const_event_message()
+ self.assertEqual(msg.event.stream.addr, self.stream.addr)
+ self.assertIs(type(msg.event.stream), bt2_stream._StreamConst)