tests/bindings/python/bt2/test_trace_collection_message_iterator.py

   1 #
   2 # Copyright (C) 2019 EfficiOS Inc.
   3 #
   4 # This program is free software; you can redistribute it and/or
   5 # modify it under the terms of the GNU General Public License
   6 # as published by the Free Software Foundation; only version 2
   7 # of the License.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU General Public License
  15 # along with this program; if not, write to the Free Software
  16 # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
  17 #
  18
  19 import unittest
  20 import datetime
  21 import bt2
  22 import os
  23 import os.path
  24
  25
  26 _BT_CTF_TRACES_PATH = os.environ['BT_CTF_TRACES_PATH']
  27 _3EVENTS_INTERSECT_TRACE_PATH = os.path.join(
  28     _BT_CTF_TRACES_PATH, 'intersection', '3eventsintersect'
  29 )
  30
  31
  32 class ComponentSpecTestCase(unittest.TestCase):
  33     def test_create_good_no_params(self):
  34         bt2.ComponentSpec('plugin', 'compcls')
  35
  36     def test_create_good_with_params(self):
  37         bt2.ComponentSpec('plugin', 'compcls', {'salut': 23})
  38
  39     def test_create_good_with_path_params(self):
  40         spec = bt2.ComponentSpec('plugin', 'compcls', 'a path')
  41         self.assertEqual(spec.params['inputs'], ['a path'])
  42
  43     def test_create_wrong_plugin_name_type(self):
  44         with self.assertRaises(TypeError):
  45             bt2.ComponentSpec(23, 'compcls')
  46
  47     def test_create_wrong_component_class_name_type(self):
  48         with self.assertRaises(TypeError):
  49             bt2.ComponentSpec('plugin', 190)
  50
  51     def test_create_wrong_params_type(self):
  52         with self.assertRaises(TypeError):
  53             bt2.ComponentSpec('dwdw', 'compcls', datetime.datetime.now())
  54
  55
  56 # Return a map, msg type -> number of messages of this type.
  57
  58
  59 def _count_msgs_by_type(msgs):
  60     res = {}
  61
  62     for msg in msgs:
  63         t = type(msg)
  64         n = res.get(t, 0)
  65         res[t] = n + 1
  66
  67     return res
  68
  69
  70 class TraceCollectionMessageIteratorTestCase(unittest.TestCase):
  71     def test_create_wrong_stream_intersection_mode_type(self):
  72         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
  73
  74         with self.assertRaises(TypeError):
  75             bt2.TraceCollectionMessageIterator(specs, stream_intersection_mode=23)
  76
  77     def test_create_wrong_begin_type(self):
  78         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
  79
  80         with self.assertRaises(TypeError):
  81             bt2.TraceCollectionMessageIterator(specs, begin='hi')
  82
  83     def test_create_wrong_end_type(self):
  84         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
  85
  86         with self.assertRaises(TypeError):
  87             bt2.TraceCollectionMessageIterator(specs, begin='lel')
  88
  89     def test_create_no_such_plugin(self):
  90         specs = [bt2.ComponentSpec('77', '101', _3EVENTS_INTERSECT_TRACE_PATH)]
  91
  92         with self.assertRaises(bt2.Error):
  93             bt2.TraceCollectionMessageIterator(specs)
  94
  95     def test_create_begin_s(self):
  96         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
  97         bt2.TraceCollectionMessageIterator(specs, begin=19457.918232)
  98
  99     def test_create_end_s(self):
 100         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
 101         bt2.TraceCollectionMessageIterator(specs, end=123.12312)
 102
 103     def test_create_begin_datetime(self):
 104         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
 105         bt2.TraceCollectionMessageIterator(specs, begin=datetime.datetime.now())
 106
 107     def test_create_end_datetime(self):
 108         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
 109         bt2.TraceCollectionMessageIterator(specs, end=datetime.datetime.now())
 110
 111     def test_iter_no_intersection(self):
 112         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
 113         msg_iter = bt2.TraceCollectionMessageIterator(specs)
 114         msgs = list(msg_iter)
 115         self.assertEqual(len(msgs), 28)
 116         hist = _count_msgs_by_type(msgs)
 117         self.assertEqual(hist[bt2.message._EventMessage], 8)
 118
 119     # Same as the above, but we pass a single spec instead of a spec list.
 120     def test_iter_specs_not_list(self):
 121         spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
 122         msg_iter = bt2.TraceCollectionMessageIterator(spec)
 123         msgs = list(msg_iter)
 124         self.assertEqual(len(msgs), 28)
 125         hist = _count_msgs_by_type(msgs)
 126         self.assertEqual(hist[bt2.message._EventMessage], 8)
 127
 128     def test_iter_custom_filter(self):
 129         src_spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
 130         flt_spec = bt2.ComponentSpec('utils', 'trimmer', {'end': '13515309.000000075'})
 131         msg_iter = bt2.TraceCollectionMessageIterator(src_spec, flt_spec)
 132         hist = _count_msgs_by_type(msg_iter)
 133         self.assertEqual(hist[bt2.message._EventMessage], 5)
 134
 135     def test_iter_intersection(self):
 136         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
 137         msg_iter = bt2.TraceCollectionMessageIterator(
 138             specs, stream_intersection_mode=True
 139         )
 140         msgs = list(msg_iter)
 141         self.assertEqual(len(msgs), 15)
 142         hist = _count_msgs_by_type(msgs)
 143         self.assertEqual(hist[bt2.message._EventMessage], 3)
 144
 145     def test_iter_intersection_no_inputs_param(self):
 146         specs = [bt2.ComponentSpec('text', 'dmesg', {'read-from-stdin': True})]
 147
 148         with self.assertRaises(bt2.Error):
 149             bt2.TraceCollectionMessageIterator(specs, stream_intersection_mode=True)
 150
 151     def test_iter_no_intersection_two_traces(self):
 152         spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
 153         specs = [spec, spec]
 154         msg_iter = bt2.TraceCollectionMessageIterator(specs)
 155         msgs = list(msg_iter)
 156         self.assertEqual(len(msgs), 56)
 157         hist = _count_msgs_by_type(msgs)
 158         self.assertEqual(hist[bt2.message._EventMessage], 16)
 159
 160     def test_iter_no_intersection_begin(self):
 161         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
 162         msg_iter = bt2.TraceCollectionMessageIterator(specs, begin=13515309.000000023)
 163         hist = _count_msgs_by_type(msg_iter)
 164         self.assertEqual(hist[bt2.message._EventMessage], 6)
 165
 166     def test_iter_no_intersection_end(self):
 167         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
 168         msg_iter = bt2.TraceCollectionMessageIterator(specs, end=13515309.000000075)
 169         hist = _count_msgs_by_type(msg_iter)
 170         self.assertEqual(hist[bt2.message._EventMessage], 5)