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['TEST_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
  36     def setUp(self):
  37         ctf = bt2.find_plugin('ctf')
  38         self._fs = ctf.source_component_classes['fs']
  39
  40         self._paths = [os.path.join(test_ctf_traces_path, 'intersection', '3eventsintersect')]
  41         self._executor = bt2.QueryExecutor()
  42
  43     def _check(self, trace, offset):
  44         self.assertEqual(trace['range-ns']['begin'], 13515309000000000 + offset)
  45         self.assertEqual(trace['range-ns']['end'], 13515309000000120 + offset)
  46         self.assertEqual(trace['intersection-range-ns']['begin'], 13515309000000070 + offset)
  47         self.assertEqual(trace['intersection-range-ns']['end'], 13515309000000100 + offset)
  48
  49         streams = sorted(trace['streams'], key=sort_predictably)
  50         self.assertEqual(streams[0]['range-ns']['begin'], 13515309000000000 + offset)
  51         self.assertEqual(streams[0]['range-ns']['end'], 13515309000000100 + offset)
  52         self.assertEqual(streams[1]['range-ns']['begin'], 13515309000000070 + offset)
  53         self.assertEqual(streams[1]['range-ns']['end'], 13515309000000120 + offset)
  54
  55     # Test various cominations of the clock-class-offset-s and
  56     # clock-class-offset-ns parameters to trace-info queries.
  57
  58     # Without clock class offset
  59
  60     def test_no_clock_class_offset(self):
  61         res = self._executor.query(self._fs, 'trace-info', {
  62             'paths': self._paths,
  63         })
  64         trace = res[0]
  65         self._check(trace, 0)
  66
  67     # With clock-class-offset-s
  68
  69     def test_clock_class_offset_s(self):
  70         res = self._executor.query(self._fs, 'trace-info', {
  71             'paths': self._paths,
  72             'clock-class-offset-s': 2,
  73         })
  74         trace = res[0]
  75         self._check(trace, 2000000000)
  76
  77     # With clock-class-offset-ns
  78
  79     def test_clock_class_offset_ns(self):
  80         res = self._executor.query(self._fs, 'trace-info', {
  81             'paths': self._paths,
  82             'clock-class-offset-ns': 2,
  83         })
  84         trace = res[0]
  85         self._check(trace, 2)
  86
  87     # With both, negative
  88
  89     def test_clock_class_offset_both(self):
  90         res = self._executor.query(self._fs, 'trace-info', {
  91             'paths': self._paths,
  92             'clock-class-offset-s': -2,
  93             'clock-class-offset-ns': -2,
  94         })
  95         trace = res[0]
  96         self._check(trace, -2000000002)
  97
  98     def test_clock_class_offset_s_wrong_type(self):
  99         with self.assertRaises(bt2.InvalidQueryParams):
 100             self._executor.query(self._fs, 'trace-info', {
 101                 'paths': self._paths,
 102                 'clock-class-offset-s': "2",
 103             })
 104
 105     def test_clock_class_offset_s_wrong_type_none(self):
 106         with self.assertRaises(bt2.InvalidQueryParams):
 107             self._executor.query(self._fs, 'trace-info', {
 108                 'paths': self._paths,
 109                 'clock-class-offset-s': None,
 110             })
 111
 112     def test_clock_class_offset_ns_wrong_type(self):
 113         with self.assertRaises(bt2.InvalidQueryParams):
 114             self._executor.query(self._fs, 'trace-info', {
 115                 'paths': self._paths,
 116                 'clock-class-offset-ns': "2",
 117             })
 118
 119     def test_clock_class_offset_ns_wrong_type_none(self):
 120         with self.assertRaises(bt2.InvalidQueryParams):
 121             self._executor.query(self._fs, 'trace-info', {
 122                 'paths': self._paths,
 123                 'clock-class-offset-ns': None,
 124             })
 125
 126
 127 class QueryTraceInfoPortNameTestCase(unittest.TestCase):
 128     def setUp(self):
 129         ctf = bt2.find_plugin("ctf")
 130         self._fs = ctf.source_component_classes["fs"]
 131
 132         self._executor = bt2.QueryExecutor()
 133
 134     def test_trace_uuid_stream_class_id_no_stream_id(self):
 135         res = self._executor.query(
 136             self._fs,
 137             "trace-info",
 138             {
 139                 "paths": [
 140                     os.path.join(
 141                         test_ctf_traces_path, "intersection", "3eventsintersect"
 142                     )
 143                 ]
 144             },
 145         )
 146         self.assertEqual(len(res), 1)
 147         trace = res[0]
 148         streams = sorted(trace["streams"], key=sort_predictably)
 149         self.assertEqual(len(streams), 2)
 150         self.assertRegexpMatches(
 151             str(streams[0]["port-name"]),
 152             r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*/tests/ctf-traces/intersection/3eventsintersect/test_stream_0$",
 153         )
 154         self.assertRegexpMatches(
 155             str(streams[1]["port-name"]),
 156             r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*/tests/ctf-traces/intersection/3eventsintersect/test_stream_1$",
 157         )
 158
 159     def test_trace_uuid_no_stream_class_id_no_stream_id(self):
 160         res = self._executor.query(
 161             self._fs,
 162             "trace-info",
 163             {"paths": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
 164         )
 165         self.assertEqual(len(res), 1)
 166         trace = res[0]
 167         streams = sorted(trace["streams"], key=sort_predictably)
 168         self.assertEqual(len(streams), 1)
 169         self.assertRegexpMatches(
 170             str(streams[0]["port-name"]),
 171             r"^2a6422d0-6cee-11e0-8c08-cb07d7b3a564 \| .*/tests/ctf-traces/succeed/succeed1/dummystream$",
 172         )
 173
 174
 175 class QueryTraceInfoRangeTestCase(unittest.TestCase):
 176     def setUp(self):
 177         ctf = bt2.find_plugin("ctf")
 178         self._fs = ctf.source_component_classes["fs"]
 179
 180         self._executor = bt2.QueryExecutor()
 181
 182     def test_trace_no_range(self):
 183         # This trace has no `timestamp_begin` and `timestamp_end` in its packet
 184         # context. The `trace-info` query should omit the `range-ns` fields in
 185         # the `trace` and `stream` data structures.
 186
 187         res = self._executor.query(
 188             self._fs,
 189             "trace-info",
 190             {"paths": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
 191         )
 192
 193         self.assertEqual(len(res), 1)
 194         trace = res[0]
 195         streams = trace["streams"]
 196         self.assertEqual(len(streams), 1)
 197
 198         self.assertRaises(KeyError, lambda: trace['range-ns'])
 199         self.assertRaises(KeyError, lambda: streams[0]['range-ns'])
 200
 201
 202 if __name__ == '__main__':
 203     unittest.main()