tests/lib/ctf-writer/test_auto_populate.py

   1 import tempfile
   2 import unittest
   3 import bt2
   4 import shutil
   5
   6
   7 class AutoPopulatePacketContextTimestampsTestCase(unittest.TestCase):
   8     def setUp(self):
   9         self._trace_path = tempfile.mkdtemp()
  10         self._writer = bt2.CtfWriter(self._trace_path)
  11         self._cc = bt2.ClockClass('default', int(1e9))
  12         self._writer.trace.add_clock_class(self._cc)
  13         self._sc = bt2.StreamClass()
  14         pcft = bt2.StructureFieldType()
  15         pcft.append_field('packet_size', bt2.IntegerFieldType(32))
  16         pcft.append_field('content_size', bt2.IntegerFieldType(32))
  17         pcft.append_field('timestamp_begin', bt2.IntegerFieldType(64, mapped_clock_class=self._cc))
  18         pcft.append_field('timestamp_end', bt2.IntegerFieldType(64, mapped_clock_class=self._cc))
  19         self._sc.packet_context_field_type = pcft
  20
  21     def tearDown(self):
  22         shutil.rmtree(self._trace_path)
  23
  24     def _get_trace_notifs(self):
  25         del self._cc
  26         del self._sc
  27         del self._ec
  28         del self._stream
  29         del self._writer
  30         specs = [bt2.ComponentSpec('ctf', 'fs', self._trace_path)]
  31         notif_iter = bt2.TraceCollectionNotificationIterator(specs)
  32         return list(notif_iter)
  33
  34     def _complete_sc(self):
  35         self._sc.add_event_class(self._ec)
  36         self._writer.trace.add_stream_class(self._sc)
  37         self._stream = self._sc()
  38
  39     def _complete_sc_event_simple_ts(self):
  40         ehft = bt2.StructureFieldType()
  41         ehft.append_field('id', bt2.IntegerFieldType(32))
  42         ehft.append_field('timestamp', bt2.IntegerFieldType(16, mapped_clock_class=self._cc))
  43         self._sc.event_header_field_type = ehft
  44         self._ec = bt2.EventClass('evt')
  45         payloadft = bt2.StructureFieldType()
  46         payloadft.append_field('value', bt2.IntegerFieldType(32))
  47         self._ec.payload_field_type = payloadft
  48         self._complete_sc()
  49
  50     def _complete_sc_event_ts_in_payload(self):
  51         ehft = bt2.StructureFieldType()
  52         ehft.append_field('id', bt2.IntegerFieldType(32))
  53         ehft.append_field('timestamp', bt2.IntegerFieldType(16, mapped_clock_class=self._cc))
  54         self._sc.event_header_field_type = ehft
  55         self._ec = bt2.EventClass('evt')
  56         payloadft = bt2.StructureFieldType()
  57         payloadft.append_field('value', bt2.IntegerFieldType(32))
  58         payloadft.append_field('a_ts', bt2.IntegerFieldType(8, mapped_clock_class=self._cc))
  59         self._ec.payload_field_type = payloadft
  60         self._complete_sc()
  61
  62     def _append_event_simple_ts(self, ts):
  63         evt = self._ec()
  64         evt.header_field['timestamp'] = ts
  65         evt.payload_field['value'] = 23
  66         self._stream.append_event(evt)
  67
  68     def _append_event_ts_in_payload(self, header_ts, payload_ts):
  69         evt = self._ec()
  70         evt.header_field['timestamp'] = header_ts
  71         evt.payload_field['value'] = 23
  72         evt.payload_field['a_ts'] = payload_ts
  73         self._stream.append_event(evt)
  74
  75     def _set_ts_begin_end(self, ts_begin=None, ts_end=None):
  76         if ts_begin is not None:
  77             self._stream.packet_context_field['timestamp_begin'] = ts_begin
  78
  79         if ts_end is not None:
  80             self._stream.packet_context_field['timestamp_end'] = ts_end
  81
  82     def _get_packet_ts_begin_end(self, notifs):
  83         ts_begin_end_list = []
  84
  85         for notif in notifs:
  86             if type(notif) is bt2.PacketBeginningNotification:
  87                 ts_begin_end_list.append((
  88                     int(notif.packet.context_field['timestamp_begin']),
  89                     int(notif.packet.context_field['timestamp_end']),
  90                 ))
  91
  92         return ts_begin_end_list
  93
  94     def test_ts_inc(self):
  95         self._complete_sc_event_simple_ts()
  96         self._append_event_simple_ts(12)
  97         self._append_event_simple_ts(144)
  98         self._stream.flush()
  99         ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
 100         self.assertEqual(ts_begin_end_list[0][0], 0)
 101         self.assertEqual(ts_begin_end_list[0][1], 144)
 102
 103     def test_ts_equal(self):
 104         self._complete_sc_event_simple_ts()
 105         self._append_event_simple_ts(12)
 106         self._append_event_simple_ts(144)
 107         self._append_event_simple_ts(144)
 108         self._stream.flush()
 109         ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
 110         self.assertEqual(ts_begin_end_list[0][0], 0)
 111         self.assertEqual(ts_begin_end_list[0][1], 144)
 112
 113     def test_ts_wraps(self):
 114         self._complete_sc_event_simple_ts()
 115         self._append_event_simple_ts(12)
 116         self._append_event_simple_ts(144)
 117         self._append_event_simple_ts(11)
 118         self._stream.flush()
 119         ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
 120         self.assertEqual(ts_begin_end_list[0][0], 0)
 121         self.assertEqual(ts_begin_end_list[0][1], 65547)
 122
 123     def test_ts_begin_from_0(self):
 124         self._complete_sc_event_simple_ts()
 125         self._append_event_simple_ts(12)
 126         self._append_event_simple_ts(144)
 127         self._stream.flush()
 128         ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
 129         self.assertEqual(ts_begin_end_list[0][0], 0)
 130         self.assertEqual(ts_begin_end_list[0][1], 144)
 131
 132     def test_ts_begin_from_last_ts_end(self):
 133         self._complete_sc_event_simple_ts()
 134         self._append_event_simple_ts(12)
 135         self._append_event_simple_ts(144)
 136         self._stream.flush()
 137         self._append_event_simple_ts(2001)
 138         self._append_event_simple_ts(3500)
 139         self._stream.flush()
 140         ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
 141         self.assertEqual(ts_begin_end_list[0][0], 0)
 142         self.assertEqual(ts_begin_end_list[0][1], 144)
 143         self.assertEqual(ts_begin_end_list[1][0], 144)
 144         self.assertEqual(ts_begin_end_list[1][1], 3500)
 145
 146     def test_ts_begin_from_provided(self):
 147         self._complete_sc_event_simple_ts()
 148         self._set_ts_begin_end(ts_begin=17)
 149         self._append_event_simple_ts(11)
 150         self._append_event_simple_ts(15)
 151         self._stream.flush()
 152         ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
 153         self.assertEqual(ts_begin_end_list[0][0], 17)
 154         self.assertEqual(ts_begin_end_list[0][1], 65551)
 155
 156     def test_ts_end_from_provided(self):
 157         self._complete_sc_event_simple_ts()
 158         self._set_ts_begin_end(ts_end=1001)
 159         self._append_event_simple_ts(11)
 160         self._append_event_simple_ts(15)
 161         self._stream.flush()
 162         ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
 163         self.assertEqual(ts_begin_end_list[0][0], 0)
 164         self.assertEqual(ts_begin_end_list[0][1], 1001)
 165
 166     def test_ts_end_from_provided_equal(self):
 167         self._complete_sc_event_simple_ts()
 168         self._set_ts_begin_end(ts_end=1001)
 169         self._append_event_simple_ts(11)
 170         self._append_event_simple_ts(1001)
 171         self._stream.flush()
 172         ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
 173         self.assertEqual(ts_begin_end_list[0][0], 0)
 174         self.assertEqual(ts_begin_end_list[0][1], 1001)
 175
 176     def test_ts_end_from_provided_too_small(self):
 177         self._complete_sc_event_simple_ts()
 178         self._set_ts_begin_end(ts_end=1001)
 179         self._append_event_simple_ts(11)
 180         self._append_event_simple_ts(1002)
 181
 182         with self.assertRaises(bt2.Error):
 183             self._stream.flush()
 184
 185     def test_ts_begin_provided_equal_last_ts_end_two_packets(self):
 186         self._complete_sc_event_simple_ts()
 187         self._append_event_simple_ts(12)
 188         self._append_event_simple_ts(1001)
 189         self._stream.flush()
 190         self._set_ts_begin_end(ts_begin=1001)
 191         self._append_event_simple_ts(2001)
 192         self._append_event_simple_ts(3500)
 193         self._stream.flush()
 194         ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
 195         self.assertEqual(ts_begin_end_list[0][0], 0)
 196         self.assertEqual(ts_begin_end_list[0][1], 1001)
 197         self.assertEqual(ts_begin_end_list[1][0], 1001)
 198         self.assertEqual(ts_begin_end_list[1][1], 3500)
 199
 200     def test_ts_begin_provided_less_than_last_ts_end_two_packets(self):
 201         self._complete_sc_event_simple_ts()
 202         self._append_event_simple_ts(12)
 203         self._append_event_simple_ts(1001)
 204         self._stream.flush()
 205         self._set_ts_begin_end(ts_begin=1000)
 206         self._append_event_simple_ts(2001)
 207         self._append_event_simple_ts(3500)
 208
 209         with self.assertRaises(bt2.Error):
 210             self._stream.flush()
 211
 212     def test_ts_end_from_value_event_payload(self):
 213         self._complete_sc_event_ts_in_payload()
 214         self._append_event_ts_in_payload(11, 15)
 215         self._append_event_ts_in_payload(5, 10)
 216         self._append_event_ts_in_payload(18, 10)
 217         self._stream.flush()
 218         ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
 219         self.assertEqual(ts_begin_end_list[0][0], 0)
 220         self.assertEqual(ts_begin_end_list[0][1], 65802)
 221
 222     def test_empty_packet_auto_ts_begin_end(self):
 223         self._complete_sc_event_simple_ts()
 224         self._stream.flush()
 225         ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
 226         self.assertEqual(ts_begin_end_list[0][0], 0)
 227         self.assertEqual(ts_begin_end_list[0][1], 0)
 228
 229     def test_empty_packet_provided_ts_begin(self):
 230         self._complete_sc_event_simple_ts()
 231         self._set_ts_begin_end(ts_begin=1001)
 232         self._stream.flush()
 233         ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
 234         self.assertEqual(ts_begin_end_list[0][0], 1001)
 235         self.assertEqual(ts_begin_end_list[0][1], 1001)
 236
 237     def test_empty_packet_provided_ts_end(self):
 238         self._complete_sc_event_simple_ts()
 239         self._set_ts_begin_end(ts_end=1001)
 240         self._stream.flush()
 241         ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
 242         self.assertEqual(ts_begin_end_list[0][0], 0)
 243         self.assertEqual(ts_begin_end_list[0][1], 1001)
 244
 245     def test_empty_packet_provided_ts_begin_end(self):
 246         self._complete_sc_event_simple_ts()
 247         self._set_ts_begin_end(1001, 3003)
 248         self._stream.flush()
 249         ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
 250         self.assertEqual(ts_begin_end_list[0][0], 1001)
 251         self.assertEqual(ts_begin_end_list[0][1], 3003)
 252
 253     def test_empty_packet_ts_begin_from_last_ts_end(self):
 254         self._complete_sc_event_simple_ts()
 255         self._set_ts_begin_end(1001, 3003)
 256         self._stream.flush()
 257         self._set_ts_begin_end(ts_end=6000)
 258         self._stream.flush()
 259         ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
 260         self.assertEqual(ts_begin_end_list[0][0], 1001)
 261         self.assertEqual(ts_begin_end_list[0][1], 3003)
 262         self.assertEqual(ts_begin_end_list[1][0], 3003)
 263         self.assertEqual(ts_begin_end_list[1][1], 6000)