tests/bindings/python/bt2/test_trace.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 uuid
  20 import unittest
  21 import utils
  22 from utils import get_default_trace_class
  23 from bt2 import trace_class as bt2_trace_class
  24 from bt2 import value as bt2_value
  25 from bt2 import trace as bt2_trace
  26 from bt2 import stream as bt2_stream
  27 from bt2 import utils as bt2_utils
  28
  29
  30 class TraceTestCase(unittest.TestCase):
  31     def setUp(self):
  32         self._tc = get_default_trace_class()
  33
  34     def test_create_default(self):
  35         trace = self._tc()
  36         self.assertIsNone(trace.name)
  37         self.assertIsNone(trace.uuid)
  38         self.assertEqual(len(trace.environment), 0)
  39         self.assertEqual(len(trace.user_attributes), 0)
  40
  41     def test_create_invalid_name(self):
  42         with self.assertRaises(TypeError):
  43             self._tc(name=17)
  44
  45     def test_create_user_attributes(self):
  46         trace = self._tc(user_attributes={'salut': 23})
  47         self.assertEqual(trace.user_attributes, {'salut': 23})
  48         self.assertIs(type(trace.user_attributes), bt2_value.MapValue)
  49
  50     def test_create_invalid_user_attributes(self):
  51         with self.assertRaises(TypeError):
  52             self._tc(user_attributes=object())
  53
  54     def test_create_invalid_user_attributes_value_type(self):
  55         with self.assertRaises(TypeError):
  56             self._tc(user_attributes=23)
  57
  58     def test_attr_trace_class(self):
  59         trace = self._tc()
  60         self.assertEqual(trace.cls.addr, self._tc.addr)
  61         self.assertIs(type(trace.cls), bt2_trace_class._TraceClass)
  62
  63     def test_const_attr_trace_class(self):
  64         trace = utils.get_const_stream_beginning_message().stream.trace
  65         self.assertIs(type(trace.cls), bt2_trace_class._TraceClassConst)
  66
  67     def test_attr_name(self):
  68         trace = self._tc(name='mein trace')
  69         self.assertEqual(trace.name, 'mein trace')
  70
  71     def test_attr_uuid(self):
  72         trace = self._tc(uuid=uuid.UUID('da7d6b6f-3108-4706-89bd-ab554732611b'))
  73         self.assertEqual(trace.uuid, uuid.UUID('da7d6b6f-3108-4706-89bd-ab554732611b'))
  74
  75     def test_env_get(self):
  76         trace = self._tc(environment={'hello': 'you', 'foo': -5})
  77         self.assertIs(type(trace.environment), bt2_trace._TraceEnvironment)
  78         self.assertIs(type(trace.environment['foo']), bt2_value.SignedIntegerValue)
  79         self.assertEqual(trace.environment['hello'], 'you')
  80         self.assertEqual(trace.environment['foo'], -5)
  81
  82     def test_env_iter(self):
  83         trace = self._tc(environment={'hello': 'you', 'foo': -5})
  84         values = set(trace.environment)
  85         self.assertEqual(values, {'hello', 'foo'})
  86
  87     def test_const_env_get(self):
  88         trace = utils.get_const_stream_beginning_message().stream.trace
  89         self.assertIs(type(trace.environment), bt2_trace._TraceEnvironmentConst)
  90         self.assertIs(
  91             type(trace.environment['patate']), bt2_value._SignedIntegerValueConst
  92         )
  93
  94     def test_const_env_iter(self):
  95         trace = utils.get_const_stream_beginning_message().stream.trace
  96         values = set(trace.environment)
  97         self.assertEqual(values, {'patate'})
  98
  99     def test_const_env_set(self):
 100         trace = utils.get_const_stream_beginning_message().stream.trace
 101         with self.assertRaises(TypeError):
 102             trace.environment['patate'] = 33
 103
 104     def test_env_get_non_existent(self):
 105         trace = self._tc(environment={'hello': 'you', 'foo': -5})
 106
 107         with self.assertRaises(KeyError):
 108             trace.environment['lel']
 109
 110     def test_len(self):
 111         trace = self._tc()
 112         sc = self._tc.create_stream_class()
 113         self.assertEqual(len(trace), 0)
 114
 115         trace.create_stream(sc)
 116         self.assertEqual(len(trace), 1)
 117
 118     def _create_trace_with_some_streams(self):
 119         sc = self._tc.create_stream_class(assigns_automatic_stream_id=False)
 120         trace = self._tc()
 121         trace.create_stream(sc, id=12)
 122         trace.create_stream(sc, id=15)
 123         trace.create_stream(sc, id=17)
 124
 125         return trace
 126
 127     def test_iter(self):
 128         trace = self._create_trace_with_some_streams()
 129         stream_ids = set(trace)
 130         self.assertEqual(stream_ids, {12, 15, 17})
 131
 132     def test_getitem(self):
 133         trace = self._create_trace_with_some_streams()
 134         self.assertEqual(trace[12].id, 12)
 135         self.assertIs(type(trace[12]), bt2_stream._Stream)
 136
 137     def test_const_getitem(self):
 138         trace = utils.get_const_stream_beginning_message().stream.trace
 139         self.assertIs(type(trace[0]), bt2_stream._StreamConst)
 140
 141     def test_getitem_invalid_key(self):
 142         trace = self._create_trace_with_some_streams()
 143         with self.assertRaises(KeyError):
 144             trace[18]
 145
 146     def test_destruction_listener(self):
 147         def on_trace_class_destruction(trace_class):
 148             nonlocal num_trace_class_destroyed_calls
 149             num_trace_class_destroyed_calls += 1
 150
 151         def on_trace_destruction(trace):
 152             nonlocal num_trace_destroyed_calls
 153             num_trace_destroyed_calls += 1
 154
 155         num_trace_class_destroyed_calls = 0
 156         num_trace_destroyed_calls = 0
 157
 158         trace_class = get_default_trace_class()
 159         stream_class = trace_class.create_stream_class()
 160         trace = trace_class()
 161         stream = trace.create_stream(stream_class)
 162
 163         trace_class.add_destruction_listener(on_trace_class_destruction)
 164         td_handle1 = trace.add_destruction_listener(on_trace_destruction)
 165         td_handle2 = trace.add_destruction_listener(on_trace_destruction)
 166
 167         self.assertIs(type(td_handle1), bt2_utils._ListenerHandle)
 168
 169         trace.remove_destruction_listener(td_handle2)
 170
 171         self.assertEqual(num_trace_class_destroyed_calls, 0)
 172         self.assertEqual(num_trace_destroyed_calls, 0)
 173
 174         del trace
 175
 176         self.assertEqual(num_trace_class_destroyed_calls, 0)
 177         self.assertEqual(num_trace_destroyed_calls, 0)
 178
 179         del stream
 180
 181         self.assertEqual(num_trace_class_destroyed_calls, 0)
 182         self.assertEqual(num_trace_destroyed_calls, 1)
 183
 184         del trace_class
 185
 186         self.assertEqual(num_trace_class_destroyed_calls, 0)
 187         self.assertEqual(num_trace_destroyed_calls, 1)
 188
 189         del stream_class
 190
 191         self.assertEqual(num_trace_class_destroyed_calls, 1)
 192         self.assertEqual(num_trace_destroyed_calls, 1)
 193
 194     def test_remove_destruction_listener_wrong_type(self):
 195         trace_class = get_default_trace_class()
 196         trace = trace_class()
 197
 198         with self.assertRaisesRegex(
 199             TypeError, r"'int' is not a '<class 'bt2.utils._ListenerHandle'>' object"
 200         ):
 201             trace.remove_destruction_listener(123)
 202
 203     def test_remove_destruction_listener_wrong_object(self):
 204         def on_trace_destruction(trace):
 205             pass
 206
 207         trace_class_1 = get_default_trace_class()
 208         trace1 = trace_class_1()
 209         trace_class_2 = get_default_trace_class()
 210         trace2 = trace_class_2()
 211
 212         handle1 = trace1.add_destruction_listener(on_trace_destruction)
 213
 214         with self.assertRaisesRegex(
 215             ValueError,
 216             r'This trace destruction listener does not match the trace object\.',
 217         ):
 218             trace2.remove_destruction_listener(handle1)
 219
 220     def test_remove_destruction_listener_twice(self):
 221         def on_trace_destruction(trace_class):
 222             pass
 223
 224         trace_class = get_default_trace_class()
 225         trace = trace_class()
 226         handle = trace.add_destruction_listener(on_trace_destruction)
 227
 228         trace.remove_destruction_listener(handle)
 229
 230         with self.assertRaisesRegex(
 231             ValueError, r'This trace destruction listener was already removed\.'
 232         ):
 233             trace.remove_destruction_listener(handle)
 234
 235     def test_raise_in_destruction_listener(self):
 236         def on_trace_destruction(trace):
 237             raise ValueError('it hurts')
 238
 239         trace_class = get_default_trace_class()
 240         trace = trace_class()
 241         trace.add_destruction_listener(on_trace_destruction)
 242
 243         del trace
 244
 245
 246 if __name__ == '__main__':
 247     unittest.main()