tests/bindings/python/bt2/test_trace_collection_message_iterator.py

   1 import unittest
   2 import datetime
   3 import bt2
   4 import os
   5 import os.path
   6
   7
   8 _TEST_CTF_TRACES_PATH = os.environ['TEST_CTF_TRACES_PATH']
   9 _3EVENTS_INTERSECT_TRACE_PATH = os.path.join(_TEST_CTF_TRACES_PATH,
  10                                              'intersection',
  11                                              '3eventsintersect')
  12
  13
  14 class ComponentSpecTestCase(unittest.TestCase):
  15     def test_create_good_no_params(self):
  16         bt2.ComponentSpec('plugin', 'compcls')
  17
  18     def test_create_good_with_params(self):
  19         bt2.ComponentSpec('plugin', 'compcls', {'salut': 23})
  20
  21     def test_create_good_with_path_params(self):
  22         spec = bt2.ComponentSpec('plugin', 'compcls', 'a path')
  23         self.assertEqual(spec.params['paths'], ['a path'])
  24
  25     def test_create_wrong_plugin_name_type(self):
  26         with self.assertRaises(TypeError):
  27             bt2.ComponentSpec(23, 'compcls')
  28
  29     def test_create_wrong_component_class_name_type(self):
  30         with self.assertRaises(TypeError):
  31             bt2.ComponentSpec('plugin', 190)
  32
  33     def test_create_wrong_params_type(self):
  34         with self.assertRaises(TypeError):
  35             bt2.ComponentSpec('dwdw', 'compcls', datetime.datetime.now())
  36
  37
  38 # Return a map, msg type -> number of messages of this type.
  39
  40 def _count_msgs_by_type(msgs):
  41     res = {}
  42
  43     for msg in msgs:
  44         t = type(msg)
  45         n = res.get(t, 0)
  46         res[t] = n + 1
  47
  48     return res
  49
  50
  51 class TraceCollectionMessageIteratorTestCase(unittest.TestCase):
  52     def test_create_wrong_stream_intersection_mode_type(self):
  53         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
  54
  55         with self.assertRaises(TypeError):
  56             bt2.TraceCollectionMessageIterator(specs, stream_intersection_mode=23)
  57
  58     def test_create_wrong_begin_type(self):
  59         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
  60
  61         with self.assertRaises(TypeError):
  62             bt2.TraceCollectionMessageIterator(specs, begin='hi')
  63
  64     def test_create_wrong_end_type(self):
  65         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
  66
  67         with self.assertRaises(TypeError):
  68             bt2.TraceCollectionMessageIterator(specs, begin='lel')
  69
  70     def test_create_no_such_plugin(self):
  71         specs = [bt2.ComponentSpec('77', '101', _3EVENTS_INTERSECT_TRACE_PATH)]
  72
  73         with self.assertRaises(bt2.Error):
  74             bt2.TraceCollectionMessageIterator(specs)
  75
  76     def test_create_begin_s(self):
  77         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
  78         bt2.TraceCollectionMessageIterator(specs, begin=19457.918232)
  79
  80     def test_create_end_s(self):
  81         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
  82         bt2.TraceCollectionMessageIterator(specs, end=123.12312)
  83
  84     def test_create_begin_datetime(self):
  85         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
  86         bt2.TraceCollectionMessageIterator(specs, begin=datetime.datetime.now())
  87
  88     def test_create_end_datetime(self):
  89         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
  90         bt2.TraceCollectionMessageIterator(specs, end=datetime.datetime.now())
  91
  92     def test_iter_no_intersection(self):
  93         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
  94         msg_iter = bt2.TraceCollectionMessageIterator(specs)
  95         msgs = list(msg_iter)
  96         self.assertEqual(len(msgs), 32)
  97         hist = _count_msgs_by_type(msgs)
  98         self.assertEqual(hist[bt2.message._EventMessage], 8)
  99
 100     # Same as the above, but we pass a single spec instead of a spec list.
 101     def test_iter_specs_not_list(self):
 102         spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
 103         msg_iter = bt2.TraceCollectionMessageIterator(spec)
 104         msgs = list(msg_iter)
 105         self.assertEqual(len(msgs), 32)
 106         hist = _count_msgs_by_type(msgs)
 107         self.assertEqual(hist[bt2.message._EventMessage], 8)
 108
 109     def test_iter_custom_filter(self):
 110         src_spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
 111         flt_spec = bt2.ComponentSpec('utils', 'trimmer', {
 112             'end': '13515309.000000075',
 113         })
 114         msg_iter = bt2.TraceCollectionMessageIterator(src_spec, flt_spec)
 115         hist = _count_msgs_by_type(msg_iter)
 116         self.assertEqual(hist[bt2.message._EventMessage], 5)
 117
 118     def test_iter_intersection(self):
 119         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
 120         msg_iter = bt2.TraceCollectionMessageIterator(specs, stream_intersection_mode=True)
 121         msgs = list(msg_iter)
 122         self.assertEqual(len(msgs), 19)
 123         hist = _count_msgs_by_type(msgs)
 124         self.assertEqual(hist[bt2.message._EventMessage], 3)
 125
 126     def test_iter_intersection_no_path_param(self):
 127         specs = [bt2.ComponentSpec('text', 'dmesg', {'read-from-stdin': True})]
 128
 129         with self.assertRaises(bt2.Error):
 130             bt2.TraceCollectionMessageIterator(specs, stream_intersection_mode=True)
 131
 132     def test_iter_no_intersection_two_traces(self):
 133         spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
 134         specs = [spec, spec]
 135         msg_iter = bt2.TraceCollectionMessageIterator(specs)
 136         msgs = list(msg_iter)
 137         self.assertEqual(len(msgs), 64)
 138         hist = _count_msgs_by_type(msgs)
 139         self.assertEqual(hist[bt2.message._EventMessage], 16)
 140
 141     def test_iter_no_intersection_begin(self):
 142         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
 143         msg_iter = bt2.TraceCollectionMessageIterator(specs, begin=13515309.000000023)
 144         hist = _count_msgs_by_type(msg_iter)
 145         self.assertEqual(hist[bt2.message._EventMessage], 6)
 146
 147     def test_iter_no_intersection_end(self):
 148         specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
 149         msg_iter = bt2.TraceCollectionMessageIterator(specs, end=13515309.000000075)
 150         hist = _count_msgs_by_type(msg_iter)
 151         self.assertEqual(hist[bt2.message._EventMessage], 5)