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(_BT_CTF_TRACES_PATH,
  28                                              'intersection',
  29                                              '3eventsintersect')
  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['paths'], ['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 def _count_msgs_by_type(msgs):
  59     res = {}
  60
  61     for msg in msgs:
  62         t = type(msg)
  63         n = res.get(t, 0)
  64         res[t] = n + 1
  65
  66     return res
  67
  68
  69 class TraceCollectionMessageIteratorTestCase(unittest.TestCase):
  70     def test_create_wrong_stream_intersection_mode_type(self):
  71         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
  72
  73         with self.assertRaises(TypeError):
  74             bt2.TraceCollectionMessageIterator(specs, stream_intersection_mode=23)
  75
  76     def test_create_wrong_begin_type(self):
  77         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
  78
  79         with self.assertRaises(TypeError):
  80             bt2.TraceCollectionMessageIterator(specs, begin='hi')
  81
  82     def test_create_wrong_end_type(self):
  83         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
  84
  85         with self.assertRaises(TypeError):
  86             bt2.TraceCollectionMessageIterator(specs, begin='lel')
  87
  88     def test_create_no_such_plugin(self):
  89         specs = [bt2.ComponentSpec('77', '101', _3EVENTS_INTERSECT_TRACE_PATH)]
  90
  91         with self.assertRaises(bt2.Error):
  92             bt2.TraceCollectionMessageIterator(specs)
  93
  94     def test_create_begin_s(self):
  95         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
  96         bt2.TraceCollectionMessageIterator(specs, begin=19457.918232)
  97
  98     def test_create_end_s(self):
  99         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
 100         bt2.TraceCollectionMessageIterator(specs, end=123.12312)
 101
 102     def test_create_begin_datetime(self):
 103         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
 104         bt2.TraceCollectionMessageIterator(specs, begin=datetime.datetime.now())
 105
 106     def test_create_end_datetime(self):
 107         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
 108         bt2.TraceCollectionMessageIterator(specs, end=datetime.datetime.now())
 109
 110     def test_iter_no_intersection(self):
 111         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
 112         msg_iter = bt2.TraceCollectionMessageIterator(specs)
 113         msgs = list(msg_iter)
 114         self.assertEqual(len(msgs), 32)
 115         hist = _count_msgs_by_type(msgs)
 116         self.assertEqual(hist[bt2.message._EventMessage], 8)
 117
 118     # Same as the above, but we pass a single spec instead of a spec list.
 119     def test_iter_specs_not_list(self):
 120         spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
 121         msg_iter = bt2.TraceCollectionMessageIterator(spec)
 122         msgs = list(msg_iter)
 123         self.assertEqual(len(msgs), 32)
 124         hist = _count_msgs_by_type(msgs)
 125         self.assertEqual(hist[bt2.message._EventMessage], 8)
 126
 127     def test_iter_custom_filter(self):
 128         src_spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
 129         flt_spec = bt2.ComponentSpec('utils', 'trimmer', {
 130             'end': '13515309.000000075',
 131         })
 132         msg_iter = bt2.TraceCollectionMessageIterator(src_spec, flt_spec)
 133         hist = _count_msgs_by_type(msg_iter)
 134         self.assertEqual(hist[bt2.message._EventMessage], 5)
 135
 136     def test_iter_intersection(self):
 137         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
 138         msg_iter = bt2.TraceCollectionMessageIterator(specs, stream_intersection_mode=True)
 139         msgs = list(msg_iter)
 140         self.assertEqual(len(msgs), 19)
 141         hist = _count_msgs_by_type(msgs)
 142         self.assertEqual(hist[bt2.message._EventMessage], 3)
 143
 144     def test_iter_intersection_no_path_param(self):
 145         specs = [bt2.ComponentSpec('text', 'dmesg', {'read-from-stdin': True})]
 146
 147         with self.assertRaises(bt2.Error):
 148             bt2.TraceCollectionMessageIterator(specs, stream_intersection_mode=True)
 149
 150     def test_iter_no_intersection_two_traces(self):
 151         spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
 152         specs = [spec, spec]
 153         msg_iter = bt2.TraceCollectionMessageIterator(specs)
 154         msgs = list(msg_iter)
 155         self.assertEqual(len(msgs), 64)
 156         hist = _count_msgs_by_type(msgs)
 157         self.assertEqual(hist[bt2.message._EventMessage], 16)
 158
 159     def test_iter_no_intersection_begin(self):
 160         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
 161         msg_iter = bt2.TraceCollectionMessageIterator(specs, begin=13515309.000000023)
 162         hist = _count_msgs_by_type(msg_iter)
 163         self.assertEqual(hist[bt2.message._EventMessage], 6)
 164
 165     def test_iter_no_intersection_end(self):
 166         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
 167         msg_iter = bt2.TraceCollectionMessageIterator(specs, end=13515309.000000075)
 168         hist = _count_msgs_by_type(msg_iter)
 169         self.assertEqual(hist[bt2.message._EventMessage], 5)