tests/lib/ctf-writer/test_auto_populate.py

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