self._trace = bt2.Trace()
self._sc = bt2.StreamClass()
self._ec = bt2.EventClass('salut')
- self._my_int_ft = bt2.IntegerFieldType(32)
- self._ec.payload_field_type = bt2.StructureFieldType()
- self._ec.payload_field_type += collections.OrderedDict([
- ('my_int', self._my_int_ft),
+ self._my_int_fc = bt2.IntegerFieldClass(32)
+ self._ec.payload_field_class = bt2.StructureFieldClass()
+ self._ec.payload_field_class += collections.OrderedDict([
+ ('my_int', self._my_int_fc),
])
self._sc.add_event_class(self._ec)
self._clock_class = bt2.ClockClass('allo', 1000)
self._trace.add_clock_class(self._clock_class)
- self._trace.packet_header_field_type = bt2.StructureFieldType()
- self._trace.packet_header_field_type += collections.OrderedDict([
- ('hello', self._my_int_ft),
+ self._trace.packet_header_field_class = bt2.StructureFieldClass()
+ self._trace.packet_header_field_class += collections.OrderedDict([
+ ('hello', self._my_int_fc),
])
self._trace.add_stream_class(self._sc)
self._cc_prio_map = bt2.ClockClassPriorityMap()
self._packet = self._stream.create_packet()
self._packet.header_field['hello'] = 19487
self._event = self._ec()
- self._event.clock_values.add(self._clock_class(1772))
+ self._event.clock_snapshots.add(self._clock_class(1772))
self._event.payload_field['my_int'] = 23
self._event.packet = self._packet
del self._trace
del self._sc
del self._ec
- del self._my_int_ft
+ del self._my_int_fc
del self._clock_class
del self._cc_prio_map
del self._stream
def test_create_with_cc_prio_map(self):
msg = bt2.InactivityMessage(self._cc_prio_map)
- msg.clock_values.add(self._cc1(123))
- msg.clock_values.add(self._cc2(19487))
+ msg.clock_snapshots.add(self._cc1(123))
+ msg.clock_snapshots.add(self._cc2(19487))
self.assertEqual(len(msg.clock_class_priority_map), 2)
self.assertEqual(msg.clock_class_priority_map, self._cc_prio_map)
- self.assertEqual(msg.clock_values[self._cc1], 123)
- self.assertEqual(msg.clock_values[self._cc2], 19487)
+ self.assertEqual(msg.clock_snapshots[self._cc1], 123)
+ self.assertEqual(msg.clock_snapshots[self._cc2], 19487)
def test_eq(self):
msg = bt2.InactivityMessage(self._cc_prio_map)
- msg.clock_values.add(self._cc1(123))
- msg.clock_values.add(self._cc2(19487))
+ msg.clock_snapshots.add(self._cc1(123))
+ msg.clock_snapshots.add(self._cc2(19487))
cc_prio_map_copy = copy.copy(self._cc_prio_map)
msg2 = bt2.InactivityMessage(cc_prio_map_copy)
- msg2.clock_values.add(self._cc1(123))
- msg2.clock_values.add(self._cc2(19487))
+ msg2.clock_snapshots.add(self._cc1(123))
+ msg2.clock_snapshots.add(self._cc2(19487))
self.assertEqual(msg, msg2)
def test_ne_cc_prio_map(self):
msg = bt2.InactivityMessage(self._cc_prio_map)
- msg.clock_values.add(self._cc1(123))
- msg.clock_values.add(self._cc2(19487))
+ msg.clock_snapshots.add(self._cc1(123))
+ msg.clock_snapshots.add(self._cc2(19487))
cc_prio_map_copy = copy.copy(self._cc_prio_map)
cc_prio_map_copy[self._cc2] = 23
msg2 = bt2.InactivityMessage(cc_prio_map_copy)
self.assertNotEqual(msg, msg2)
- def test_ne_clock_value(self):
+ def test_ne_clock_snapshot(self):
msg = bt2.InactivityMessage(self._cc_prio_map)
- msg.clock_values.add(self._cc1(123))
- msg.clock_values.add(self._cc2(19487))
+ msg.clock_snapshots.add(self._cc1(123))
+ msg.clock_snapshots.add(self._cc2(19487))
msg2 = bt2.InactivityMessage(self._cc_prio_map)
- msg.clock_values.add(self._cc1(123))
- msg.clock_values.add(self._cc2(1847))
+ msg.clock_snapshots.add(self._cc1(123))
+ msg.clock_snapshots.add(self._cc2(1847))
self.assertNotEqual(msg, msg2)
def test_eq_invalid(self):
def test_copy(self):
msg = bt2.InactivityMessage(self._cc_prio_map)
- msg.clock_values.add(self._cc1(123))
- msg.clock_values.add(self._cc2(19487))
+ msg.clock_snapshots.add(self._cc1(123))
+ msg.clock_snapshots.add(self._cc2(19487))
msg_copy = copy.copy(msg)
self.assertEqual(msg, msg_copy)
self.assertNotEqual(msg.addr, msg_copy.addr)
self.assertEqual(msg.clock_class_priority_map.addr,
msg_copy.clock_class_priority_map.addr)
- self.assertEqual(msg_copy.clock_values[self._cc1], 123)
- self.assertEqual(msg_copy.clock_values[self._cc2], 19487)
+ self.assertEqual(msg_copy.clock_snapshots[self._cc1], 123)
+ self.assertEqual(msg_copy.clock_snapshots[self._cc2], 19487)
def test_deepcopy(self):
msg = bt2.InactivityMessage(self._cc_prio_map)
- msg.clock_values.add(self._cc1(123))
- msg.clock_values.add(self._cc2(19487))
+ msg.clock_snapshots.add(self._cc1(123))
+ msg.clock_snapshots.add(self._cc2(19487))
msg_copy = copy.deepcopy(msg)
self.assertEqual(msg, msg_copy)
self.assertNotEqual(msg.addr, msg_copy.addr)
msg_copy.clock_class_priority_map)
self.assertNotEqual(list(msg.clock_class_priority_map)[0].addr,
list(msg_copy.clock_class_priority_map)[0].addr)
- self.assertIsNone(msg_copy.clock_values[self._cc1])
- self.assertIsNone(msg_copy.clock_values[self._cc2])
- self.assertEqual(msg_copy.clock_values[list(msg_copy.clock_class_priority_map)[0]], 123)
- self.assertEqual(msg_copy.clock_values[list(msg_copy.clock_class_priority_map)[1]], 19487)
+ self.assertIsNone(msg_copy.clock_snapshots[self._cc1])
+ self.assertIsNone(msg_copy.clock_snapshots[self._cc2])
+ self.assertEqual(msg_copy.clock_snapshots[list(msg_copy.clock_class_priority_map)[0]], 123)
+ self.assertEqual(msg_copy.clock_snapshots[list(msg_copy.clock_class_priority_map)[1]], 19487)
@unittest.skip("this is broken")
self._sc = bt2.StreamClass()
self._ec = bt2.EventClass('salut')
self._clock_class = bt2.ClockClass('yo', 1000)
- self._uint64_int_ft = bt2.IntegerFieldType(64, mapped_clock_class=self._clock_class)
- self._my_int_ft = bt2.IntegerFieldType(32)
- self._ec.payload_field_type = bt2.StructureFieldType()
- self._ec.payload_field_type += collections.OrderedDict([
- ('my_int', self._my_int_ft),
+ self._uint64_int_fc = bt2.IntegerFieldClass(64, mapped_clock_class=self._clock_class)
+ self._my_int_fc = bt2.IntegerFieldClass(32)
+ self._ec.payload_field_class = bt2.StructureFieldClass()
+ self._ec.payload_field_class += collections.OrderedDict([
+ ('my_int', self._my_int_fc),
])
self._sc.add_event_class(self._ec)
- self._sc.packet_context_field_type = bt2.StructureFieldType()
- self._sc.packet_context_field_type += collections.OrderedDict([
- ('packet_seq_num', self._my_int_ft),
- ('timestamp_begin', self._uint64_int_ft),
- ('timestamp_end', self._uint64_int_ft),
+ self._sc.packet_context_field_class = bt2.StructureFieldClass()
+ self._sc.packet_context_field_class += collections.OrderedDict([
+ ('packet_seq_num', self._my_int_fc),
+ ('timestamp_begin', self._uint64_int_fc),
+ ('timestamp_end', self._uint64_int_fc),
])
self._trace.add_clock_class(self._clock_class)
self._trace.add_stream_class(self._sc)
del self._sc
del self._ec
del self._clock_class
- del self._uint64_int_ft
- del self._my_int_ft
+ del self._uint64_int_fc
+ del self._my_int_fc
del self._stream
def _create_event(self, packet):
def test_stream(self):
self.assertEqual(self._get_msg().stream.addr, self._stream.addr)
- def test_beginning_clock_value(self):
+ def test_beginning_clock_snapshot(self):
msg = self._get_msg()
- beginning_clock_value = msg.beginning_clock_value
- self.assertEqual(beginning_clock_value.clock_class, self._clock_class)
- self.assertEqual(beginning_clock_value, 6)
+ beginning_clock_snapshot = msg.beginning_clock_snapshot
+ self.assertEqual(beginning_clock_snapshot.clock_class, self._clock_class)
+ self.assertEqual(beginning_clock_snapshot, 6)
- def test_end_clock_value(self):
+ def test_end_clock_snapshot(self):
msg = self._get_msg()
- end_clock_value = msg.end_clock_value
- self.assertEqual(end_clock_value.clock_class, self._clock_class)
- self.assertEqual(end_clock_value, 7)
+ end_clock_snapshot = msg.end_clock_snapshot
+ self.assertEqual(end_clock_snapshot.clock_class, self._clock_class)
+ self.assertEqual(end_clock_snapshot, 7)
def test_eq(self):
msg1 = self._get_msg()
self._sc = bt2.StreamClass()
self._ec = bt2.EventClass('salut')
self._clock_class = bt2.ClockClass('yo', 1000)
- self._uint64_int_ft = bt2.IntegerFieldType(64, mapped_clock_class=self._clock_class)
- self._my_int_ft = bt2.IntegerFieldType(32)
- self._ec.payload_field_type = bt2.StructureFieldType()
- self._ec.payload_field_type += collections.OrderedDict([
- ('my_int', self._my_int_ft),
+ self._uint64_int_fc = bt2.IntegerFieldClass(64, mapped_clock_class=self._clock_class)
+ self._my_int_fc = bt2.IntegerFieldClass(32)
+ self._ec.payload_field_class = bt2.StructureFieldClass()
+ self._ec.payload_field_class += collections.OrderedDict([
+ ('my_int', self._my_int_fc),
])
self._sc.add_event_class(self._ec)
- self._sc.packet_context_field_type = bt2.StructureFieldType()
- self._sc.packet_context_field_type += collections.OrderedDict([
- ('events_discarded', self._my_int_ft),
- ('timestamp_begin', self._uint64_int_ft),
- ('timestamp_end', self._uint64_int_ft),
+ self._sc.packet_context_field_class = bt2.StructureFieldClass()
+ self._sc.packet_context_field_class += collections.OrderedDict([
+ ('events_discarded', self._my_int_fc),
+ ('timestamp_begin', self._uint64_int_fc),
+ ('timestamp_end', self._uint64_int_fc),
])
self._trace.add_clock_class(self._clock_class)
self._trace.add_stream_class(self._sc)
del self._sc
del self._ec
del self._clock_class
- del self._uint64_int_ft
- del self._my_int_ft
+ del self._uint64_int_fc
+ del self._my_int_fc
del self._stream
def _create_event(self, packet):
def test_stream(self):
self.assertEqual(self._get_msg().stream.addr, self._stream.addr)
- def test_beginning_clock_value(self):
+ def test_beginning_clock_snapshot(self):
msg = self._get_msg()
- beginning_clock_value = msg.beginning_clock_value
- self.assertEqual(beginning_clock_value.clock_class, self._clock_class)
- self.assertEqual(beginning_clock_value, 6)
+ beginning_clock_snapshot = msg.beginning_clock_snapshot
+ self.assertEqual(beginning_clock_snapshot.clock_class, self._clock_class)
+ self.assertEqual(beginning_clock_snapshot, 6)
- def test_end_clock_value(self):
+ def test_end_clock_snapshot(self):
msg = self._get_msg()
- end_clock_value = msg.end_clock_value
- self.assertEqual(end_clock_value.clock_class, self._clock_class)
- self.assertEqual(end_clock_value, 10)
+ end_clock_snapshot = msg.end_clock_snapshot
+ self.assertEqual(end_clock_snapshot.clock_class, self._clock_class)
+ self.assertEqual(end_clock_snapshot, 10)
def test_eq(self):
msg1 = self._get_msg()