X-Git-Url: http://git.efficios.com/?p=babeltrace.git;a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_trace_class.py;h=67f6456b7f894c87e4291b021c0e2125894d03c4;hp=b36dafa1a1023d77aa73beec8657f5f330600e22;hb=0235b0db7de5bcacdb3650c92461f2ce5eb2143d;hpb=2c6f8520ad68e1ca0c9554d705c39790e7d3ef5f diff --git a/tests/bindings/python/bt2/test_trace_class.py b/tests/bindings/python/bt2/test_trace_class.py index b36dafa1..67f6456b 100644 --- a/tests/bindings/python/bt2/test_trace_class.py +++ b/tests/bindings/python/bt2/test_trace_class.py @@ -1,10 +1,30 @@ -import uuid +# SPDX-License-Identifier: GPL-2.0-only +# +# Copyright (C) 2019 EfficiOS Inc. +# + import unittest -import bt2 -from utils import run_in_component_init, get_default_trace_class +from utils import ( + run_in_component_init, + get_default_trace_class, + get_const_stream_beginning_message, +) +from bt2 import stream_class as bt2_stream_class +from bt2 import trace_class as bt2_trace_class +from bt2 import utils as bt2_utils class TraceClassTestCase(unittest.TestCase): + def assertRaisesInComponentInit(self, expected_exc_type, user_code): + def f(comp_self): + try: + user_code(comp_self) + except Exception as exc: + return type(exc) + + exc_type = run_in_component_init(f) + self.assertIsNotNone(exc_type) + self.assertEqual(exc_type, expected_exc_type) def test_create_default(self): def f(comp_self): @@ -13,17 +33,36 @@ class TraceClassTestCase(unittest.TestCase): tc = run_in_component_init(f) self.assertEqual(len(tc), 0) - self.assertEqual(len(tc.env), 0) - self.assertIsNone(tc.uuid) + self.assertIs(type(tc), bt2_trace_class._TraceClass) self.assertTrue(tc.assigns_automatic_stream_class_id) + self.assertEqual(len(tc.user_attributes), 0) - def test_uuid(self): + def test_create_user_attributes(self): def f(comp_self): - return comp_self._create_trace_class(uuid=uuid.UUID('da7d6b6f-3108-4706-89bd-ab554732611b')) + return comp_self._create_trace_class(user_attributes={'salut': 23}) tc = run_in_component_init(f) + self.assertEqual(tc.user_attributes, {'salut': 23}) - self.assertEqual(tc.uuid, uuid.UUID('da7d6b6f-3108-4706-89bd-ab554732611b')) + def test_create_invalid_user_attributes(self): + def f(comp_self): + return comp_self._create_trace_class(user_attributes=object()) + + self.assertRaisesInComponentInit(TypeError, f) + + def test_create_invalid_user_attributes_value_type(self): + def f(comp_self): + return comp_self._create_trace_class(user_attributes=23) + + self.assertRaisesInComponentInit(TypeError, f) + + def test_create_invalid_automatic_stream_class_id_type(self): + def f(comp_self): + return comp_self._create_trace_class( + assigns_automatic_stream_class_id='perchaude' + ) + + self.assertRaisesInComponentInit(TypeError, f) def test_automatic_stream_class_id(self): def f(comp_self): @@ -36,6 +75,8 @@ class TraceClassTestCase(unittest.TestCase): sc1 = tc.create_stream_class() sc2 = tc.create_stream_class() + self.assertIs(type(sc1), bt2_stream_class._StreamClass) + self.assertIs(type(sc2), bt2_stream_class._StreamClass) self.assertNotEqual(sc1.id, sc2.id) def test_automatic_stream_class_id_raises(self): @@ -46,11 +87,13 @@ class TraceClassTestCase(unittest.TestCase): self.assertTrue(tc.assigns_automatic_stream_class_id) with self.assertRaises(ValueError): - sc1 = tc.create_stream_class(23) + tc.create_stream_class(23) def test_no_assigns_automatic_stream_class_id(self): def f(comp_self): - return comp_self._create_trace_class(assigns_automatic_stream_class_id=False) + return comp_self._create_trace_class( + assigns_automatic_stream_class_id=False + ) tc = run_in_component_init(f) self.assertFalse(tc.assigns_automatic_stream_class_id) @@ -60,7 +103,9 @@ class TraceClassTestCase(unittest.TestCase): def test_no_assigns_automatic_stream_class_id_raises(self): def f(comp_self): - return comp_self._create_trace_class(assigns_automatic_stream_class_id=False) + return comp_self._create_trace_class( + assigns_automatic_stream_class_id=False + ) tc = run_in_component_init(f) self.assertFalse(tc.assigns_automatic_stream_class_id) @@ -69,28 +114,12 @@ class TraceClassTestCase(unittest.TestCase): with self.assertRaises(ValueError): tc.create_stream_class() - def test_env_get(self): - def f(comp_self): - return comp_self._create_trace_class(env={'hello': 'you', 'foo': -5}) - - tc = run_in_component_init(f) - - self.assertEqual(tc.env['hello'], 'you') - self.assertEqual(tc.env['foo'], -5) - - def test_env_get_non_existent(self): - def f(comp_self): - return comp_self._create_trace_class(env={'hello': 'you', 'foo': -5}) - - tc = run_in_component_init(f) - - with self.assertRaises(KeyError): - tc.env['lel'] - @staticmethod def _create_trace_class_with_some_stream_classes(): def f(comp_self): - return comp_self._create_trace_class(assigns_automatic_stream_class_id=False) + return comp_self._create_trace_class( + assigns_automatic_stream_class_id=False + ) tc = run_in_component_init(f) sc1 = tc.create_stream_class(id=12) @@ -100,8 +129,13 @@ class TraceClassTestCase(unittest.TestCase): def test_getitem(self): tc, _, _, sc3 = self._create_trace_class_with_some_stream_classes() + self.assertIs(type(tc[2018]), bt2_stream_class._StreamClass) self.assertEqual(tc[2018].addr, sc3.addr) + def test_const_getitem(self): + const_tc = get_const_stream_beginning_message().stream.trace.cls + self.assertIs(type(const_tc[0]), bt2_stream_class._StreamClassConst) + def test_getitem_wrong_key_type(self): tc, _, _, _ = self._create_trace_class_with_some_stream_classes() with self.assertRaises(TypeError): @@ -122,27 +156,87 @@ class TraceClassTestCase(unittest.TestCase): tc, sc1, sc2, sc3 = self._create_trace_class_with_some_stream_classes() for sc_id, stream_class in tc.items(): - self.assertIsInstance(stream_class, bt2.stream_class._StreamClass) - if sc_id == 12: + self.assertIs(type(stream_class), bt2_stream_class._StreamClass) self.assertEqual(stream_class.addr, sc1.addr) elif sc_id == 54: self.assertEqual(stream_class.addr, sc2.addr) elif sc_id == 2018: self.assertEqual(stream_class.addr, sc3.addr) + def test_const_iter(self): + const_tc = get_const_stream_beginning_message().stream.trace.cls + const_sc = list(const_tc.values())[0] + self.assertIs(type(const_sc), bt2_stream_class._StreamClassConst) + def test_destruction_listener(self): def on_trace_class_destruction(trace_class): - nonlocal trace_class_destroyed - trace_class_destroyed = True + nonlocal num_destruct_calls + num_destruct_calls += 1 - trace_class_destroyed = False + num_destruct_calls = 0 trace_class = get_default_trace_class() - trace_class.add_destruction_listener(on_trace_class_destruction) - self.assertFalse(trace_class_destroyed) + handle1 = trace_class.add_destruction_listener(on_trace_class_destruction) + self.assertIs(type(handle1), bt2_utils._ListenerHandle) + + handle2 = trace_class.add_destruction_listener(on_trace_class_destruction) + + trace_class.remove_destruction_listener(handle2) + + self.assertEqual(num_destruct_calls, 0) + + del trace_class + + self.assertEqual(num_destruct_calls, 1) + + def test_remove_destruction_listener_wrong_type(self): + trace_class = get_default_trace_class() + + with self.assertRaisesRegex( + TypeError, r"'int' is not a '' object" + ): + trace_class.remove_destruction_listener(123) + + def test_remove_destruction_listener_wrong_object(self): + def on_trace_class_destruction(trace_class): + pass + + trace_class_1 = get_default_trace_class() + trace_class_2 = get_default_trace_class() + + handle1 = trace_class_1.add_destruction_listener(on_trace_class_destruction) + + with self.assertRaisesRegex( + ValueError, + r'This trace class destruction listener does not match the trace class object\.', + ): + trace_class_2.remove_destruction_listener(handle1) + + def test_remove_destruction_listener_twice(self): + def on_trace_class_destruction(trace_class): + pass + + trace_class = get_default_trace_class() + handle = trace_class.add_destruction_listener(on_trace_class_destruction) + + trace_class.remove_destruction_listener(handle) + + with self.assertRaisesRegex( + ValueError, r'This trace class destruction listener was already removed\.' + ): + trace_class.remove_destruction_listener(handle) + + def test_raise_in_destruction_listener(self): + def on_trace_class_destruction(trace_class): + raise ValueError('it hurts') + + trace_class = get_default_trace_class() + trace_class.add_destruction_listener(on_trace_class_destruction) del trace_class - self.assertTrue(trace_class_destroyed) + +if __name__ == '__main__': + unittest.main()