X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_trace.py;h=7fa8d8a56a28fc5d6fb46b96f99324400136b977;hb=f0a42b33ac3951cd5cb2ee0f66ac04437a681621;hp=b2b9cdad103d98462628882bb62c502f3ad74774;hpb=811644b8fe5fb9946972a7ace9df02ed872f448a;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_trace.py b/tests/bindings/python/bt2/test_trace.py index b2b9cdad..7fa8d8a5 100644 --- a/tests/bindings/python/bt2/test_trace.py +++ b/tests/bindings/python/bt2/test_trace.py @@ -1,342 +1,186 @@ -from bt2 import values -import unittest -import copy +# +# Copyright (C) 2019 EfficiOS Inc. +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU General Public License +# as published by the Free Software Foundation; only version 2 +# of the License. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. +# + import uuid -import bt2 +import unittest +import utils +from utils import get_default_trace_class +from bt2 import trace_class as bt2_trace_class +from bt2 import value as bt2_value +from bt2 import trace as bt2_trace +from bt2 import stream as bt2_stream class TraceTestCase(unittest.TestCase): def setUp(self): - self._sc = self._create_stream_class('sc1', 3) - self._tc = bt2.Trace() - - def tearDown(self): - del self._sc - del self._tc - - def _create_stream_class(self, name, id): - ec1, ec2 = self._create_event_classes() - packet_context_ft = bt2.StructureFieldType() - packet_context_ft.append_field('menu', bt2.FloatingPointNumberFieldType()) - packet_context_ft.append_field('sticker', bt2.StringFieldType()) - event_header_ft = bt2.StructureFieldType() - event_header_ft.append_field('id', bt2.IntegerFieldType(19)) - event_context_ft = bt2.StructureFieldType() - event_context_ft.append_field('msg', bt2.StringFieldType()) - return bt2.StreamClass(name=name, id=id, - packet_context_field_type=packet_context_ft, - event_header_field_type=event_header_ft, - event_context_field_type=event_context_ft, - event_classes=(ec1, ec2)) - - def _create_event_classes(self): - context_ft = bt2.StructureFieldType() - context_ft.append_field('allo', bt2.StringFieldType()) - context_ft.append_field('zola', bt2.IntegerFieldType(18)) - payload_ft = bt2.StructureFieldType() - payload_ft.append_field('zoom', bt2.StringFieldType()) - ec1 = bt2.EventClass('event23', id=23, context_field_type=context_ft, - payload_field_type=payload_ft) - ec2 = bt2.EventClass('event17', id=17, context_field_type=payload_ft, - payload_field_type=context_ft) - return ec1, ec2 + self._tc = get_default_trace_class() def test_create_default(self): - self.assertEqual(len(self._tc), 0) - - def _get_std_header(self): - header_ft = bt2.StructureFieldType() - header_ft.append_field('magic', bt2.IntegerFieldType(32)) - header_ft.append_field('stream_id', bt2.IntegerFieldType(32)) - return header_ft - - def test_create_full(self): - clock_classes = bt2.ClockClass('cc1', 1000), bt2.ClockClass('cc2', 30) - sc = self._create_stream_class('sc1', 3) - tc = bt2.Trace(name='my name', - native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, - env={'the_string': 'value', 'the_int': 23}, - packet_header_field_type=self._get_std_header(), - clock_classes=clock_classes, - stream_classes=(sc,)) - self.assertEqual(tc.name, 'my name') - self.assertEqual(tc.native_byte_order, bt2.ByteOrder.LITTLE_ENDIAN) - self.assertEqual(tc.env['the_string'], 'value') - self.assertEqual(tc.env['the_int'], 23) - self.assertEqual(tc.packet_header_field_type, self._get_std_header()) - self.assertEqual(tc.clock_classes['cc1'], clock_classes[0]) - self.assertEqual(tc.clock_classes['cc2'], clock_classes[1]) - self.assertEqual(tc[3], sc) - - def test_assign_name(self): - self._tc.name = 'lel' - self.assertEqual(self._tc.name, 'lel') - - def test_assign_invalid_name(self): - with self.assertRaises(TypeError): - self._tc.name = 17 + trace = self._tc() + self.assertIsNone(trace.name) + self.assertIsNone(trace.uuid) + self.assertEqual(len(trace.environment), 0) + self.assertEqual(len(trace.user_attributes), 0) - def test_assign_static(self): - self._tc.set_is_static() - self.assertTrue(self._tc.is_static) + def test_create_invalid_name(self): + with self.assertRaises(TypeError): + self._tc(name=17) - def test_assign_native_byte_order(self): - self._tc.native_byte_order = bt2.ByteOrder.BIG_ENDIAN - self.assertEqual(self._tc.native_byte_order, bt2.ByteOrder.BIG_ENDIAN) + def test_create_user_attributes(self): + trace = self._tc(user_attributes={'salut': 23}) + self.assertEqual(trace.user_attributes, {'salut': 23}) + self.assertIs(type(trace.user_attributes), bt2_value.MapValue) - def test_assign_invalid_native_byte_order(self): + def test_create_invalid_user_attributes(self): with self.assertRaises(TypeError): - self._tc.native_byte_order = 'lel' - - def test_assign_packet_header_field_type(self): - header_ft = bt2.StructureFieldType() - header_ft.append_field('magic', bt2.IntegerFieldType(32)) - self._tc.packet_header_field_type = header_ft - self.assertEqual(self._tc.packet_header_field_type, header_ft) - - def test_assign_no_packet_header_field_type(self): - self._tc.packet_header_field_type = None - self.assertIsNone(self._tc.packet_header_field_type) - - def _test_copy(self, cpy): - self.assertIsNot(cpy, self._tc) - self.assertNotEqual(cpy.addr, self._tc.addr) - self.assertEqual(cpy, self._tc) - self.assertEqual(len(self._tc), len(cpy)) - - def _pre_copy(self): - self._tc.packet_header_field_type = self._get_std_header() - self._tc.name = 'the trace class' - sc1 = self._create_stream_class('sc1', 3) - sc2 = self._create_stream_class('sc2', 9) - sc3 = self._create_stream_class('sc3', 17) - self._tc.add_clock_class(bt2.ClockClass('cc1', 1000)) - self._tc.add_clock_class(bt2.ClockClass('cc2', 30)) - self._tc.env['allo'] = 'bateau' - self._tc.env['bateau'] = 'cart' - self._tc.add_stream_class(sc1) - self._tc.add_stream_class(sc2) - self._tc.add_stream_class(sc3) - - def test_copy(self): - self._pre_copy() - cpy = copy.copy(self._tc) - self._test_copy(cpy) - self.assertEqual(self._tc.packet_header_field_type.addr, cpy.packet_header_field_type.addr) - self.assertEqual(self._tc.clock_classes['cc1'].addr, cpy.clock_classes['cc1'].addr) - self.assertEqual(self._tc.clock_classes['cc2'].addr, cpy.clock_classes['cc2'].addr) - self.assertEqual(self._tc.env['allo'].addr, cpy.env['allo'].addr) - self.assertEqual(self._tc.env['bateau'].addr, cpy.env['bateau'].addr) - - def test_deepcopy(self): - self._pre_copy() - cpy = copy.deepcopy(self._tc) - self._test_copy(cpy) - self.assertNotEqual(self._tc.packet_header_field_type.addr, cpy.packet_header_field_type.addr) - self.assertNotEqual(self._tc.clock_classes['cc1'].addr, cpy.clock_classes['cc1'].addr) - self.assertNotEqual(self._tc.clock_classes['cc2'].addr, cpy.clock_classes['cc2'].addr) - self.assertNotEqual(self._tc.env['allo'].addr, cpy.env['allo'].addr) - self.assertNotEqual(self._tc.env['bateau'].addr, cpy.env['bateau'].addr) - - def test_getitem(self): - self._tc.add_stream_class(self._sc) - self.assertEqual(self._tc[3].addr, self._sc.addr) + self._tc(user_attributes=object()) - def test_getitem_wrong_key_type(self): - self._tc.add_stream_class(self._sc) + def test_create_invalid_user_attributes_value_type(self): + with self.assertRaises(TypeError): + self._tc(user_attributes=23) + + def test_attr_trace_class(self): + trace = self._tc() + self.assertEqual(trace.cls.addr, self._tc.addr) + self.assertIs(type(trace.cls), bt2_trace_class._TraceClass) + + def test_const_attr_trace_class(self): + trace = utils.get_const_stream_beginning_message().stream.trace + self.assertIs(type(trace.cls), bt2_trace_class._TraceClassConst) + + def test_attr_name(self): + trace = self._tc(name='mein trace') + self.assertEqual(trace.name, 'mein trace') + + def test_attr_uuid(self): + trace = self._tc(uuid=uuid.UUID('da7d6b6f-3108-4706-89bd-ab554732611b')) + self.assertEqual(trace.uuid, uuid.UUID('da7d6b6f-3108-4706-89bd-ab554732611b')) + + def test_env_get(self): + trace = self._tc(environment={'hello': 'you', 'foo': -5}) + self.assertIs(type(trace.environment), bt2_trace._TraceEnvironment) + self.assertIs(type(trace.environment['foo']), bt2_value.SignedIntegerValue) + self.assertEqual(trace.environment['hello'], 'you') + self.assertEqual(trace.environment['foo'], -5) + + def test_env_iter(self): + trace = self._tc(environment={'hello': 'you', 'foo': -5}) + values = set(trace.environment) + self.assertEqual(values, {'hello', 'foo'}) + + def test_const_env_get(self): + trace = utils.get_const_stream_beginning_message().stream.trace + self.assertIs(type(trace.environment), bt2_trace._TraceEnvironmentConst) + self.assertIs( + type(trace.environment['patate']), bt2_value._SignedIntegerValueConst + ) + + def test_env_iter(self): + trace = utils.get_const_stream_beginning_message().stream.trace + values = set(trace.environment) + self.assertEqual(values, {'patate'}) + + def test_const_env_set(self): + trace = utils.get_const_stream_beginning_message().stream.trace with self.assertRaises(TypeError): - self._tc['hello'] + trace.environment['patate'] = 33 + + def test_env_get_non_existent(self): + trace = self._tc(environment={'hello': 'you', 'foo': -5}) - def test_getitem_wrong_key(self): - self._tc.add_stream_class(self._sc) with self.assertRaises(KeyError): - self._tc[4] + trace.environment['lel'] def test_len(self): - self.assertEqual(len(self._tc), 0) - self._tc.add_stream_class(self._sc) - self.assertEqual(len(self._tc), 1) + trace = self._tc() + sc = self._tc.create_stream_class() + self.assertEqual(len(trace), 0) + + trace.create_stream(sc) + self.assertEqual(len(trace), 1) + + def _create_trace_with_some_streams(self): + sc = self._tc.create_stream_class(assigns_automatic_stream_id=False) + trace = self._tc() + trace.create_stream(sc, id=12) + trace.create_stream(sc, id=15) + trace.create_stream(sc, id=17) + + return trace def test_iter(self): - self._tc.packet_header_field_type = self._get_std_header() - sc1 = self._create_stream_class('sc1', 3) - sc2 = self._create_stream_class('sc2', 9) - sc3 = self._create_stream_class('sc3', 17) - self._tc.add_stream_class(sc1) - self._tc.add_stream_class(sc2) - self._tc.add_stream_class(sc3) - - for sid, stream_class in self._tc.items(): - self.assertIsInstance(stream_class, bt2.StreamClass) - - if sid == 3: - self.assertEqual(stream_class.addr, sc1.addr) - elif sid == 9: - self.assertEqual(stream_class.addr, sc2.addr) - elif sid == 17: - self.assertEqual(stream_class.addr, sc3.addr) - - def test_env_getitem_wrong_key(self): - with self.assertRaises(KeyError): - self._tc.env['lel'] + trace = self._create_trace_with_some_streams() + stream_ids = set(trace) + self.assertEqual(stream_ids, {12, 15, 17}) + + def test_getitem(self): + trace = self._create_trace_with_some_streams() + self.assertEqual(trace[12].id, 12) + self.assertIs(type(trace[12]), bt2_stream._Stream) - def test_clock_classes_getitem_wrong_key(self): + def test_const_getitem(self): + trace = utils.get_const_stream_beginning_message().stream.trace + self.assertIs(type(trace[0]), bt2_stream._StreamConst) + + def test_getitem_invalid_key(self): + trace = self._create_trace_with_some_streams() with self.assertRaises(KeyError): - self._tc.clock_classes['lel'] - - def test_streams_none(self): - self.assertEqual(len(self._tc.streams), 0) - - def test_streams_len(self): - self._tc.add_stream_class(self._create_stream_class('sc1', 3)) - stream0 = self._tc[3]() - stream1 = self._tc[3]() - stream2 = self._tc[3]() - self.assertEqual(len(self._tc.streams), 3) - - def test_streams_iter(self): - self._tc.add_stream_class(self._create_stream_class('sc1', 3)) - stream0 = self._tc[3](id=12) - stream1 = self._tc[3](id=15) - stream2 = self._tc[3](id=17) - sids = set() - - for stream in self._tc.streams: - sids.add(stream.id) - - self.assertEqual(len(sids), 3) - self.assertTrue(12 in sids and 15 in sids and 17 in sids) - - def _test_eq_create_objects(self): - cc1_uuid = uuid.UUID('bc7f2f2d-2ee4-4e03-ab1f-2e0e1304e94f') - cc1 = bt2.ClockClass('cc1', 1000, uuid=cc1_uuid) - cc2_uuid = uuid.UUID('da7d6b6f-3108-4706-89bd-ab554732611b') - cc2 = bt2.ClockClass('cc2', 30, uuid=cc2_uuid) - sc1 = self._create_stream_class('sc1', 3) - sc2 = self._create_stream_class('sc2', 9) - return cc1, cc2, sc1, sc2, self._get_std_header() - - def test_eq(self): - cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() - tc1 = bt2.Trace(name='my name', - native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, - env={'the_string': 'value', 'the_int': 23}, - packet_header_field_type=header_ft, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() - tc2 = bt2.Trace(name='my name', - native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, - env={'the_string': 'value', 'the_int': 23}, - packet_header_field_type=header_ft, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - self.assertEqual(tc1, tc2) - - def test_ne_name(self): - cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() - tc1 = bt2.Trace(name='my name2', - native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, - env={'the_string': 'value', 'the_int': 23}, - packet_header_field_type=header_ft, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() - tc2 = bt2.Trace(name='my name', - native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, - env={'the_string': 'value', 'the_int': 23}, - packet_header_field_type=header_ft, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - self.assertNotEqual(tc1, tc2) - - def test_ne_packet_header_field_type(self): - cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() - tc1 = bt2.Trace(name='my name', - native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, - env={'the_string': 'value', 'the_int': 23}, - packet_header_field_type=header_ft, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() - header_ft.append_field('yes', bt2.StringFieldType()) - tc2 = bt2.Trace(name='my name', - native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, - env={'the_string': 'value', 'the_int': 23}, - packet_header_field_type=header_ft, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - self.assertNotEqual(tc1, tc2) - - def test_ne_native_byte_order(self): - cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() - tc1 = bt2.Trace(name='my name', - native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, - env={'the_string': 'value', 'the_int': 23}, - packet_header_field_type=header_ft, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() - tc2 = bt2.Trace(name='my name', - native_byte_order=bt2.ByteOrder.BIG_ENDIAN, - env={'the_string': 'value', 'the_int': 23}, - packet_header_field_type=header_ft, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - self.assertNotEqual(tc1, tc2) - - def test_ne_env(self): - cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() - tc1 = bt2.Trace(name='my name', - native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, - env={'the_string': 'value', 'the_int2': 23}, - packet_header_field_type=header_ft, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() - tc2 = bt2.Trace(name='my name', - native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, - env={'the_string': 'value', 'the_int': 23}, - packet_header_field_type=header_ft, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - self.assertNotEqual(tc1, tc2) - - def test_ne_clock_classes(self): - cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() - tc1 = bt2.Trace(name='my name', - native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, - env={'the_string': 'value', 'the_int': 23}, - packet_header_field_type=header_ft, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() - cc2.frequency = 1234 - tc2 = bt2.Trace(name='my name', - native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, - env={'the_string': 'value', 'the_int': 23}, - packet_header_field_type=header_ft, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - self.assertNotEqual(tc1, tc2) - - def test_ne_stream_classes(self): - cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() - tc1 = bt2.Trace(name='my name', - native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, - env={'the_string': 'value', 'the_int': 23}, - packet_header_field_type=header_ft, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() - sc2.id = 72632 - tc2 = bt2.Trace(name='my name', - native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, - env={'the_string': 'value', 'the_int': 23}, - packet_header_field_type=header_ft, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - self.assertNotEqual(tc1, tc2) - - def test_eq_invalid(self): - self.assertFalse(self._tc == 23) + trace[18] + + def test_destruction_listener(self): + def on_trace_class_destruction(trace_class): + nonlocal trace_class_destroyed + trace_class_destroyed = True + + def on_trace_destruction(trace): + nonlocal trace_destroyed + trace_destroyed = True + + trace_destroyed = False + trace_class_destroyed = False + + trace_class = get_default_trace_class() + stream_class = trace_class.create_stream_class() + trace = trace_class() + stream = trace.create_stream(stream_class) + + trace_class.add_destruction_listener(on_trace_class_destruction) + trace.add_destruction_listener(on_trace_destruction) + + self.assertFalse(trace_class_destroyed) + self.assertFalse(trace_destroyed) + + del trace + + self.assertFalse(trace_class_destroyed) + self.assertFalse(trace_destroyed) + + del stream + + self.assertFalse(trace_class_destroyed) + self.assertTrue(trace_destroyed) + + del trace_class + + self.assertFalse(trace_class_destroyed) + self.assertTrue(trace_destroyed) + + del stream_class + + self.assertTrue(trace_class_destroyed) + self.assertTrue(trace_destroyed)