X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_trace_class.py;h=f9a21a0ffb276c2ac94a8a64bf88280ce34e54c2;hb=f5567ea88d172767b34373bc6e402da8bfd85ef8;hp=265b0de8b40ed8ac0604216688235a870a7a8a95;hpb=fbbe93021d7cd3793003911931bb3abd5e69596a;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_trace_class.py b/tests/bindings/python/bt2/test_trace_class.py index 265b0de8..f9a21a0f 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): @@ -45,12 +86,14 @@ class TraceClassTestCase(unittest.TestCase): tc = run_in_component_init(f) self.assertTrue(tc.assigns_automatic_stream_class_id) - with self.assertRaises(bt2.CreationError): - sc1 = tc.create_stream_class(23) + with self.assertRaises(ValueError): + 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,37 +103,23 @@ 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) # In this mode, it is required to pass an explicit id. - with self.assertRaises(bt2.CreationError): + 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,12 +129,17 @@ 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): - tc['hello'] + tc["hello"] def test_getitem_wrong_key(self): tc, _, _, _ = self._create_trace_class_with_some_stream_classes() @@ -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()