tests/bindings/python/babeltrace/test_reader_trace_intersection.py

   1 #!/usr/bin/env python3
   2 #
   3 # The MIT License (MIT)
   4 #
   5 # Copyright (C) 2016 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
   6 #
   7 # Permission is hereby granted, free of charge, to any person obtaining a copy
   8 # of this software and associated documentation files (the "Software"), to deal
   9 # in the Software without restriction, including without limitation the rights
  10 # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  11 # copies of the Software, and to permit persons to whom the Software is
  12 # furnished to do so, subject to the following conditions:
  13 #
  14 # The above copyright notice and this permission notice shall be included in
  15 # all copies or substantial portions of the Software.
  16 #
  17 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  18 # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  19 # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  20 # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  21 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  22 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  23 # SOFTWARE.
  24
  25 import tempfile
  26 import babeltrace.writer as btw
  27 import babeltrace.reader as btr
  28 import uuid
  29 import shutil
  30 import unittest
  31
  32 class Entry(object):
  33     def __init__(self, stream_id, timestamp=None, end_of_packet=False):
  34         self.stream_id = stream_id
  35         self.timestamp = timestamp
  36         self.end_of_packet = end_of_packet
  37
  38
  39 class Packet(object):
  40     def __init__(self, timestamps):
  41         self.timestamps = timestamps
  42
  43
  44 class TraceIntersectionTestCase(unittest.TestCase):
  45     def _create_trace(self, stream_descriptions):
  46         trace_path = tempfile.mkdtemp()
  47         trace = btw.Writer(trace_path)
  48         clock = btw.Clock('test_clock')
  49         clock.uuid = self._clock_uuid
  50         trace.add_clock(clock)
  51
  52         integer_field_type = btw.IntegerFieldDeclaration(32)
  53
  54         event_class = btw.EventClass('simple_event')
  55         event_class.add_field(integer_field_type, 'int_field')
  56
  57         stream_class = btw.StreamClass('test_stream')
  58         stream_class.add_event_class(event_class)
  59         stream_class.clock = clock
  60
  61         streams = []
  62         stream_entries = []
  63         for stream_id, stream_packets in enumerate(stream_descriptions):
  64             stream = trace.create_stream(stream_class)
  65             streams.append(stream)
  66
  67             for packet in stream_packets:
  68                 for timestamp in packet.timestamps:
  69                     stream_entries.append(Entry(stream_id, timestamp))
  70                 # Mark the last inserted entry as the end of packet
  71                 stream_entries[len(stream_entries) - 1].end_of_packet = True
  72
  73         # Sort stream entries which will provide us with a time-ordered list of
  74         # events to insert in the streams.
  75         for entry in sorted(stream_entries, key=lambda entry: entry.timestamp):
  76             clock.time = entry.timestamp
  77             event = btw.Event(event_class)
  78             event.payload('int_field').value = entry.stream_id
  79             streams[entry.stream_id].append_event(event)
  80             if entry.end_of_packet is True:
  81                 streams[entry.stream_id].flush()
  82
  83         return trace_path
  84
  85     def setUp(self):
  86         self._clock_uuid = uuid.uuid4()
  87         self._trace_path_early = self._create_trace(
  88             [
  89                 [Packet(range(1, 7)), Packet(range(11, 18))],
  90                 [Packet(range(8, 15)), Packet(range(22, 24)), Packet(range(30, 60))],
  91                 [Packet(range(11, 14))]
  92             ]
  93         )
  94         self._trace_path_late = self._create_trace(
  95             [
  96                 [Packet(range(100, 105)), Packet(range(109, 120))],
  97                 [Packet(range(88, 95)), Packet(range(96, 110)), Packet(range(112, 140))],
  98                 [Packet(range(99, 105))]
  99             ]
 100         )
 101
 102         self._expected_timestamps_early = []
 103         for ts in range(11, 14):
 104             for stream in range(3):
 105                 self._expected_timestamps_early.append(ts)
 106
 107         self._expected_timestamps_late = []
 108         for ts in range(100, 105):
 109             for stream in range(3):
 110                 self._expected_timestamps_late.append(ts)
 111
 112         self._expected_timestamps_union = (self._expected_timestamps_early +
 113                                            self._expected_timestamps_late)
 114
 115     def tearDown(self):
 116         shutil.rmtree(self._trace_path_early)
 117         shutil.rmtree(self._trace_path_late)
 118         pass
 119
 120     @staticmethod
 121     def _check_trace_expected_timestamps(trace_paths, expected_timestamps):
 122         traces = btr.TraceCollection(intersect_mode=True)
 123         for trace_path in trace_paths:
 124             trace_handle = traces.add_trace(trace_path, 'ctf')
 125             if trace_handle is None:
 126                 print('# Failed to open trace at {}'.format(trace_path))
 127                 return False
 128         for event in traces.events:
 129             expected_timestamp = expected_timestamps.pop(0)
 130             if event.timestamp != expected_timestamp:
 131                 print('# Unexpected timestamp ({}), expected {}'.format(
 132                     event.timestamp, expected_timestamp))
 133                 return False
 134         return True
 135
 136     def test_trace_early(self):
 137         self._check_trace_expected_timestamps([self._trace_path_early],
 138                                               self._expected_timestamps_early)
 139
 140     def test_trace_late(self):
 141         self._check_trace_expected_timestamps([self._trace_path_late],
 142                                               self._expected_timestamps_late)
 143
 144     def test_trace_intersection(self):
 145         self._check_trace_expected_timestamps([self._trace_path_early,
 146                                                self._trace_path_late],
 147                                               self._expected_timestamps_union)