tests/plugins/src.ctf.fs/query/test_query_trace_info.py

   1 # Copyright (C) 2019 Simon Marchi <simon.marchi@efficios.com>
   2 #
   3 # This program is free software; you can redistribute it and/or
   4 # modify it under the terms of the GNU General Public License
   5 # as published by the Free Software Foundation; only version 2
   6 # of the License.
   7 #
   8 # This program is distributed in the hope that it will be useful,
   9 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11 # GNU General Public License for more details.
  12 #
  13 # You should have received a copy of the GNU General Public License
  14 # along with this program; if not, write to the Free Software
  15 # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
  16
  17 import unittest
  18 import bt2
  19 import os
  20 import re
  21
  22
  23 test_ctf_traces_path = os.environ['BT_CTF_TRACES_PATH']
  24
  25
  26 # Key to sort streams in a predictable order.
  27 def sort_predictably(stream):
  28     if 'range-ns' in stream:
  29         return stream['range-ns']['begin']
  30     else:
  31         return stream['paths'][0]
  32
  33
  34 class QueryTraceInfoClockOffsetTestCase(unittest.TestCase):
  35     def setUp(self):
  36         ctf = bt2.find_plugin('ctf')
  37         self._fs = ctf.source_component_classes['fs']
  38
  39         self._paths = [
  40             os.path.join(test_ctf_traces_path, 'intersection', '3eventsintersect')
  41         ]
  42         self._executor = bt2.QueryExecutor()
  43
  44     def _check(self, trace, offset):
  45         self.assertEqual(trace['range-ns']['begin'], 13515309000000000 + offset)
  46         self.assertEqual(trace['range-ns']['end'], 13515309000000120 + offset)
  47         self.assertEqual(
  48             trace['intersection-range-ns']['begin'], 13515309000000070 + offset
  49         )
  50         self.assertEqual(
  51             trace['intersection-range-ns']['end'], 13515309000000100 + offset
  52         )
  53
  54         streams = sorted(trace['streams'], key=sort_predictably)
  55         self.assertEqual(streams[0]['range-ns']['begin'], 13515309000000000 + offset)
  56         self.assertEqual(streams[0]['range-ns']['end'], 13515309000000100 + offset)
  57         self.assertEqual(streams[1]['range-ns']['begin'], 13515309000000070 + offset)
  58         self.assertEqual(streams[1]['range-ns']['end'], 13515309000000120 + offset)
  59
  60     # Test various cominations of the clock-class-offset-s and
  61     # clock-class-offset-ns parameters to trace-info queries.
  62
  63     # Without clock class offset
  64
  65     def test_no_clock_class_offset(self):
  66         res = self._executor.query(self._fs, 'trace-info', {'paths': self._paths})
  67         trace = res[0]
  68         self._check(trace, 0)
  69
  70     # With clock-class-offset-s
  71
  72     def test_clock_class_offset_s(self):
  73         res = self._executor.query(
  74             self._fs, 'trace-info', {'paths': self._paths, 'clock-class-offset-s': 2}
  75         )
  76         trace = res[0]
  77         self._check(trace, 2000000000)
  78
  79     # With clock-class-offset-ns
  80
  81     def test_clock_class_offset_ns(self):
  82         res = self._executor.query(
  83             self._fs, 'trace-info', {'paths': self._paths, 'clock-class-offset-ns': 2}
  84         )
  85         trace = res[0]
  86         self._check(trace, 2)
  87
  88     # With both, negative
  89
  90     def test_clock_class_offset_both(self):
  91         res = self._executor.query(
  92             self._fs,
  93             'trace-info',
  94             {
  95                 'paths': self._paths,
  96                 'clock-class-offset-s': -2,
  97                 'clock-class-offset-ns': -2,
  98             },
  99         )
 100         trace = res[0]
 101         self._check(trace, -2000000002)
 102
 103     def test_clock_class_offset_s_wrong_type(self):
 104         with self.assertRaises(bt2.InvalidParams):
 105             self._executor.query(
 106                 self._fs,
 107                 'trace-info',
 108                 {'paths': self._paths, 'clock-class-offset-s': "2"},
 109             )
 110
 111     def test_clock_class_offset_s_wrong_type_none(self):
 112         with self.assertRaises(bt2.InvalidParams):
 113             self._executor.query(
 114                 self._fs,
 115                 'trace-info',
 116                 {'paths': self._paths, 'clock-class-offset-s': None},
 117             )
 118
 119     def test_clock_class_offset_ns_wrong_type(self):
 120         with self.assertRaises(bt2.InvalidParams):
 121             self._executor.query(
 122                 self._fs,
 123                 'trace-info',
 124                 {'paths': self._paths, 'clock-class-offset-ns': "2"},
 125             )
 126
 127     def test_clock_class_offset_ns_wrong_type_none(self):
 128         with self.assertRaises(bt2.InvalidParams):
 129             self._executor.query(
 130                 self._fs,
 131                 'trace-info',
 132                 {'paths': self._paths, 'clock-class-offset-ns': None},
 133             )
 134
 135
 136 class QueryTraceInfoPortNameTestCase(unittest.TestCase):
 137     def setUp(self):
 138         ctf = bt2.find_plugin("ctf")
 139         self._fs = ctf.source_component_classes["fs"]
 140
 141         self._executor = bt2.QueryExecutor()
 142
 143     def test_trace_uuid_stream_class_id_no_stream_id(self):
 144         res = self._executor.query(
 145             self._fs,
 146             "trace-info",
 147             {
 148                 "paths": [
 149                     os.path.join(
 150                         test_ctf_traces_path, "intersection", "3eventsintersect"
 151                     )
 152                 ]
 153             },
 154         )
 155         self.assertEqual(len(res), 1)
 156         trace = res[0]
 157         streams = sorted(trace["streams"], key=sort_predictably)
 158         self.assertEqual(len(streams), 2)
 159         self.assertRegexpMatches(
 160             str(streams[0]["port-name"]),
 161             r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*/tests/data/ctf-traces/intersection/3eventsintersect/test_stream_0$",
 162         )
 163         self.assertRegexpMatches(
 164             str(streams[1]["port-name"]),
 165             r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*/tests/data/ctf-traces/intersection/3eventsintersect/test_stream_1$",
 166         )
 167
 168     def test_trace_uuid_no_stream_class_id_no_stream_id(self):
 169         res = self._executor.query(
 170             self._fs,
 171             "trace-info",
 172             {"paths": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
 173         )
 174         self.assertEqual(len(res), 1)
 175         trace = res[0]
 176         streams = sorted(trace["streams"], key=sort_predictably)
 177         self.assertEqual(len(streams), 1)
 178         self.assertRegexpMatches(
 179             str(streams[0]["port-name"]),
 180             r"^2a6422d0-6cee-11e0-8c08-cb07d7b3a564 \| .*/tests/data/ctf-traces/succeed/succeed1/dummystream$",
 181         )
 182
 183
 184 class QueryTraceInfoRangeTestCase(unittest.TestCase):
 185     def setUp(self):
 186         ctf = bt2.find_plugin("ctf")
 187         self._fs = ctf.source_component_classes["fs"]
 188
 189         self._executor = bt2.QueryExecutor()
 190
 191     def test_trace_no_range(self):
 192         # This trace has no `timestamp_begin` and `timestamp_end` in its packet
 193         # context. The `trace-info` query should omit the `range-ns` fields in
 194         # the `trace` and `stream` data structures.
 195
 196         res = self._executor.query(
 197             self._fs,
 198             "trace-info",
 199             {"paths": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
 200         )
 201
 202         self.assertEqual(len(res), 1)
 203         trace = res[0]
 204         streams = trace["streams"]
 205         self.assertEqual(len(streams), 1)
 206
 207         self.assertRaises(KeyError, lambda: trace['range-ns'])
 208         self.assertRaises(KeyError, lambda: streams[0]['range-ns'])
 209
 210
 211 if __name__ == '__main__':
 212     unittest.main()