From: Simon Marchi Date: Thu, 9 May 2019 19:28:15 +0000 (-0400) Subject: bt2: Adapt test_trace.py and make it pass X-Git-Url: http://git.efficios.com/?p=babeltrace.git;a=commitdiff_plain;h=8c2367b884576ed438bc2e06cfc9205b2436838d bt2: Adapt test_trace.py and make it pass test_trace.py is heavily modified, because most of the responsibilities of Trace have been moved to TraceClass. A number of concepts, such as packet headers, also disappeared. Similarly, trace.py loses a lot of weight. Note that copy and equality operations are removed, since we ruled that they were complex but not particularly useful. A trace implements abc.Mapping, and maps stream ids to stream objects. A trace is created by "calling" a trace class, hence the addition of the __call__ method to TraceClass. The test_trace test creates some streams and requires to control the ids of those streams, so the create_stream_class method and StreamClass type are enhanced to support that. Change-Id: If23089abfbcbe5f0064069f93bd91126f1549b62 Signed-off-by: Simon Marchi Signed-off-by: Francis Deslauriers Reviewed-on: https://review.lttng.org/c/babeltrace/+/1290 Tested-by: jenkins Reviewed-by: Philippe Proulx --- diff --git a/bindings/python/bt2/bt2/stream.py b/bindings/python/bt2/bt2/stream.py index 22ff0dfd..c37e0878 100644 --- a/bindings/python/bt2/bt2/stream.py +++ b/bindings/python/bt2/bt2/stream.py @@ -62,6 +62,9 @@ class _StreamBase(object._SharedObject): class _Stream(_StreamBase): + _get_ref = staticmethod(native_bt.stream_get_ref) + _put_ref = staticmethod(native_bt.stream_put_ref) + def create_packet(self): packet_ptr = native_bt.packet_create(self._ptr) diff --git a/bindings/python/bt2/bt2/stream_class.py b/bindings/python/bt2/bt2/stream_class.py index e94c3b22..ccf3a84f 100644 --- a/bindings/python/bt2/bt2/stream_class.py +++ b/bindings/python/bt2/bt2/stream_class.py @@ -121,6 +121,16 @@ class StreamClass(object._SharedObject, collections.abc.Mapping): ret = native_bt.stream_class_set_name(self._ptr, name) utils._handle_ret(ret, "cannot set stream class object's name") + @property + def assigns_automatic_stream_id(self): + return native_bt.stream_class_assigns_automatic_stream_id(self._ptr) + + def _assigns_automatic_stream_id(self, auto_id): + utils._check_bool(auto_id) + return native_bt.stream_class_set_assigns_automatic_stream_id(self._ptr, auto_id) + + _assigns_automatic_stream_id = property(fset=_assigns_automatic_stream_id) + @property def id(self): id = native_bt.stream_class_get_id(self._ptr) diff --git a/bindings/python/bt2/bt2/trace.py b/bindings/python/bt2/bt2/trace.py index 2f3f24e7..aa2757c0 100644 --- a/bindings/python/bt2/bt2/trace.py +++ b/bindings/python/bt2/bt2/trace.py @@ -25,319 +25,91 @@ import bt2.field_class import collections.abc import bt2.value import bt2.stream -import copy import bt2 +import functools -class _StreamClassIterator(collections.abc.Iterator): - def __init__(self, trace): - self._trace = trace - self._at = 0 +def _trace_destruction_listener_from_native(user_listener, trace_ptr): + trace = bt2.trace.Trace._create_from_ptr_and_get_ref(trace_ptr) + user_listener(trace) - def __next__(self): - if self._at == len(self._trace): - raise StopIteration - sc_ptr = native_bt.trace_get_stream_class_by_index(self._trace._ptr, - self._at) - assert(sc_ptr) - id = native_bt.stream_class_get_id(sc_ptr) - native_bt.put(sc_ptr) - assert(id >= 0) - self._at += 1 - return id - - -class _TraceStreams(collections.abc.Sequence): - def __init__(self, trace): - self._trace = trace +class Trace(object._SharedObject, collections.abc.Mapping): + _get_ref = staticmethod(native_bt.trace_get_ref) + _put_ref = staticmethod(native_bt.trace_put_ref) def __len__(self): - count = native_bt.trace_get_stream_count(self._trace._ptr) - assert(count >= 0) + count = native_bt.trace_get_stream_count(self._ptr) + assert count >= 0 return count - def __getitem__(self, index): - utils._check_uint64(index) - - if index >= len(self): - raise IndexError - - stream_ptr = native_bt.trace_get_stream_by_index(self._trace._ptr, - index) - assert(stream_ptr) - return bt2.stream._create_from_ptr(stream_ptr) - - -class _TraceClockClassesIterator(collections.abc.Iterator): - def __init__(self, trace_clock_classes): - self._trace_clock_classes = trace_clock_classes - self._at = 0 - - def __next__(self): - if self._at == len(self._trace_clock_classes): - raise StopIteration - - trace_ptr = self._trace_clock_classes._trace._ptr - cc_ptr = native_bt.trace_get_clock_class_by_index(trace_ptr, self._at) - assert(cc_ptr) - name = native_bt.clock_class_get_name(cc_ptr) - native_bt.put(cc_ptr) - assert(name is not None) - self._at += 1 - return name + def __getitem__(self, id): + utils._check_uint64(id) + stream_ptr = native_bt.trace_borrow_stream_by_id_const(self._ptr, id) -class _TraceClockClasses(collections.abc.Mapping): - def __init__(self, trace): - self._trace = trace + if stream_ptr is None: + raise KeyError(id) - def __getitem__(self, key): - utils._check_str(key) - cc_ptr = native_bt.trace_get_clock_class_by_name(self._trace._ptr, key) - - if cc_ptr is None: - raise KeyError(key) - - return bt2.ClockClass._create_from_ptr(cc_ptr) - - def __len__(self): - count = native_bt.trace_get_clock_class_count(self._trace._ptr) - assert(count >= 0) - return count + return bt2.stream._Stream._create_from_ptr_and_get_ref(stream_ptr) def __iter__(self): - return _TraceClockClassesIterator(self) - - -class _TraceEnvIterator(collections.abc.Iterator): - def __init__(self, trace_env): - self._trace_env = trace_env - self._at = 0 - - def __next__(self): - if self._at == len(self._trace_env): - raise StopIteration - - trace_ptr = self._trace_env._trace._ptr - entry_name = native_bt.trace_get_environment_field_name_by_index(trace_ptr, - self._at) - assert(entry_name is not None) - self._at += 1 - return entry_name + for idx in range(len(self)): + stream_ptr = native_bt.trace_borrow_stream_by_index_const(self._ptr, idx) + assert stream_ptr is not None + id = native_bt.stream_get_id(stream_ptr) + assert id >= 0 -class _TraceEnv(collections.abc.MutableMapping): - def __init__(self, trace): - self._trace = trace - - def __getitem__(self, key): - utils._check_str(key) - value_ptr = native_bt.trace_get_environment_field_value_by_name(self._trace._ptr, - key) - - if value_ptr is None: - raise KeyError(key) - - return bt2.value._create_from_ptr(value_ptr) - - def __setitem__(self, key, value): - utils._check_str(key) - value = bt2.create_value(value) - ret = native_bt.trace_set_environment_field(self._trace._ptr, - key, value._ptr) - utils._handle_ret(ret, "cannot set trace class object's environment entry") - - def __delitem__(self, key): - raise NotImplementedError - - def __len__(self): - count = native_bt.trace_get_environment_field_count(self._trace._ptr) - assert(count >= 0) - return count - - def __iter__(self): - return _TraceEnvIterator(self) - - -class Trace(object._SharedObject, collections.abc.Mapping): - def __init__(self, name=None, native_byte_order=None, env=None, - packet_header_field_class=None, clock_classes=None, - stream_classes=None): - ptr = native_bt.trace_create() - - if ptr is None: - raise bt2.CreationError('cannot create trace class object') - - super().__init__(ptr) - - if name is not None: - self.name = name - - if native_byte_order is not None: - self.native_byte_order = native_byte_order - - if packet_header_field_class is not None: - self.packet_header_field_class = packet_header_field_class - - if env is not None: - for key, value in env.items(): - self.env[key] = value - - if clock_classes is not None: - for clock_class in clock_classes: - self.add_clock_class(clock_class) - - if stream_classes is not None: - for stream_class in stream_classes: - self.add_stream_class(stream_class) - - def __getitem__(self, key): - utils._check_int64(key) - sc_ptr = native_bt.trace_get_stream_class_by_id(self._ptr, key) - - if sc_ptr is None: - raise KeyError(key) - - return bt2.StreamClass._create_from_ptr(sc_ptr) - - def __len__(self): - count = native_bt.trace_get_stream_class_count(self._ptr) - assert(count >= 0) - return count - - def __iter__(self): - return _StreamClassIterator(self) - - def add_stream_class(self, stream_class): - utils._check_type(stream_class, bt2.StreamClass) - ret = native_bt.trace_add_stream_class(self._ptr, stream_class._ptr) - utils._handle_ret(ret, "cannot add stream class object to trace class object") + yield id @property def name(self): return native_bt.trace_get_name(self._ptr) - @name.setter - def name(self, name): + def _name(self, name): utils._check_str(name) ret = native_bt.trace_set_name(self._ptr, name) utils._handle_ret(ret, "cannot set trace class object's name") - @property - def native_byte_order(self): - bo = native_bt.trace_get_native_byte_order(self._ptr) - assert(bo >= 0) - return bo - - @native_byte_order.setter - def native_byte_order(self, native_byte_order): - utils._check_int(native_byte_order) - ret = native_bt.trace_set_native_byte_order(self._ptr, native_byte_order) - utils._handle_ret(ret, "cannot set trace class object's native byte order") - - @property - def is_static(self): - is_static = native_bt.trace_is_static(self._ptr) - return is_static > 0 - - def set_is_static(self): - ret = native_bt.trace_set_is_static(self._ptr) - utils._handle_ret(ret, "cannot set trace object as static") + _name = property(fset=_name) - @property - def env(self): - return _TraceEnv(self) - - @property - def clock_classes(self): - return _TraceClockClasses(self) + def create_stream(self, stream_class, id=None, name=None): + utils._check_type(stream_class, bt2.stream_class.StreamClass) - def add_clock_class(self, clock_class): - utils._check_type(clock_class, bt2.ClockClass) - ret = native_bt.trace_add_clock_class(self._ptr, clock_class._ptr) - utils._handle_ret(ret, "cannot add clock class object to trace class object") - - @property - def streams(self): - return _TraceStreams(self) - - @property - def packet_header_field_class(self): - fc_ptr = native_bt.trace_get_packet_header_type(self._ptr) + if stream_class.assigns_automatic_stream_id: + if id is not None: + raise bt2.CreationError("id provided, but stream class assigns automatic stream ids") - if fc_ptr is None: - return + stream_ptr = native_bt.stream_create(stream_class._ptr, self._ptr) + else: + if id is None: + raise bt2.CreationError("id not provided, but stream class does not assign automatic stream ids") - return bt2.field_class._create_from_ptr(fc_ptr) + utils._check_uint64(id) + stream_ptr = native_bt.stream_create_with_id(stream_class._ptr, self._ptr, id) - @packet_header_field_class.setter - def packet_header_field_class(self, packet_header_field_class): - packet_header_field_class_ptr = None + if stream_ptr is None: + raise bt2.CreationError('cannot create stream object') - if packet_header_field_class is not None: - utils._check_type(packet_header_field_class, bt2.field_class._FieldClass) - packet_header_field_class_ptr = packet_header_field_class._ptr + stream = bt2.stream._Stream._create_from_ptr(stream_ptr) - ret = native_bt.trace_set_packet_header_type(self._ptr, - packet_header_field_class_ptr) - utils._handle_ret(ret, "cannot set trace class object's packet header field class") - - def __eq__(self, other): - if type(other) is not type(self): - # not comparing apples to apples - return False - - if self.addr == other.addr: - return True - - self_stream_classes = list(self.values()) - self_clock_classes = list(self.clock_classes.values()) - self_env = {key: val for key, val in self.env.items()} - other_stream_classes = list(other.values()) - other_clock_classes = list(other.clock_classes.values()) - other_env = {key: val for key, val in other.env.items()} - self_props = ( - self_stream_classes, - self_clock_classes, - self_env, - self.name, - self.native_byte_order, - self.packet_header_field_class, - ) - other_props = ( - other_stream_classes, - other_clock_classes, - other_env, - other.name, - other.native_byte_order, - other.packet_header_field_class, - ) - return self_props == other_props - - def _copy(self, gen_copy_func, sc_copy_func): - cpy = Trace() - - if self.name is not None: - cpy.name = self.name - - cpy.packet_header_field_class = gen_copy_func(self.packet_header_field_class) - - for key, val in self.env.items(): - cpy.env[key] = gen_copy_func(val) + if name is not None: + stream._name = name - for clock_class in self.clock_classes.values(): - cpy.add_clock_class(gen_copy_func(clock_class)) + return stream - for stream_class in self.values(): - cpy.add_stream_class(sc_copy_func(stream_class)) + def add_destruction_listener(self, listener): + '''Add a listener to be called when the trace is destroyed.''' + if not callable(listener): + raise TypeError("'listener' parameter is not callable") - return cpy + fn = native_bt.py3_trace_add_destruction_listener + listener_from_native = functools.partial(_trace_destruction_listener_from_native, + listener) - def __copy__(self): - return self._copy(lambda obj: obj, copy.copy) + listener_id = fn(self._ptr, listener_from_native) + if listener_id is None: + utils._raise_bt2_error('cannot add destruction listener to trace object') - def __deepcopy__(self, memo): - cpy = self._copy(copy.deepcopy, copy.deepcopy) - memo[id(self)] = cpy - return cpy + return bt2._ListenerHandle(listener_id, self) diff --git a/bindings/python/bt2/bt2/trace_class.py b/bindings/python/bt2/bt2/trace_class.py index 1f473a30..4ee7702f 100644 --- a/bindings/python/bt2/bt2/trace_class.py +++ b/bindings/python/bt2/bt2/trace_class.py @@ -117,6 +117,21 @@ class TraceClass(object._SharedObject, collections.abc.Mapping): _uuid = property(fset=_uuid) + # Instantiate a trace of this class. + + def __call__(self, name=None): + trace_ptr = native_bt.trace_create(self._ptr) + + if trace_ptr is None: + raise bt2.CreationError('cannot create trace class object') + + trace = bt2.trace.Trace._create_from_ptr(trace_ptr) + + if name is not None: + trace._name = name + + return trace + # Number of stream classes in this trace class. def __len__(self): @@ -149,7 +164,8 @@ class TraceClass(object._SharedObject, collections.abc.Mapping): def env(self): return _TraceClassEnv(self) - def create_stream_class(self, id=None): + def create_stream_class(self, id=None, + assigns_automatic_stream_id=True): if self.assigns_automatic_stream_class_id: if id is not None: @@ -163,7 +179,11 @@ class TraceClass(object._SharedObject, collections.abc.Mapping): utils._check_uint64(id) sc_ptr = native_bt.stream_class_create_with_id(self._ptr, id) - return bt2.StreamClass._create_from_ptr(sc_ptr) + sc = bt2.stream_class.StreamClass._create_from_ptr(sc_ptr) + + sc._assigns_automatic_stream_id = assigns_automatic_stream_id + + return sc @property def assigns_automatic_stream_class_id(self): diff --git a/tests/bindings/python/bt2/Makefile.am b/tests/bindings/python/bt2/Makefile.am index b5da0c91..09261c70 100644 --- a/tests/bindings/python/bt2/Makefile.am +++ b/tests/bindings/python/bt2/Makefile.am @@ -21,4 +21,5 @@ EXTRA_DIST = \ test_trace.py \ test_trace_collection_message_iterator.py \ test_value.py \ + utils.py \ .coveragerc diff --git a/tests/bindings/python/bt2/test_trace.py b/tests/bindings/python/bt2/test_trace.py index d4c7de90..69029244 100644 --- a/tests/bindings/python/bt2/test_trace.py +++ b/tests/bindings/python/bt2/test_trace.py @@ -1,343 +1,94 @@ -from bt2 import value import unittest -import copy -import uuid -import bt2 +from utils import get_default_trace_class -@unittest.skip("this is broken") 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_fc = bt2.StructureFieldClass() - packet_context_fc.append_field('menu', bt2.FloatingPointNumberFieldClass()) - packet_context_fc.append_field('sticker', bt2.StringFieldClass()) - event_header_fc = bt2.StructureFieldClass() - event_header_fc.append_field('id', bt2.IntegerFieldClass(19)) - event_context_fc = bt2.StructureFieldClass() - event_context_fc.append_field('msg', bt2.StringFieldClass()) - return bt2.StreamClass(name=name, id=id, - packet_context_field_class=packet_context_fc, - event_header_field_class=event_header_fc, - event_context_field_class=event_context_fc, - event_classes=(ec1, ec2)) - - def _create_event_classes(self): - context_fc = bt2.StructureFieldClass() - context_fc.append_field('allo', bt2.StringFieldClass()) - context_fc.append_field('zola', bt2.IntegerFieldClass(18)) - payload_fc = bt2.StructureFieldClass() - payload_fc.append_field('zoom', bt2.StringFieldClass()) - ec1 = bt2.EventClass('event23', id=23, context_field_class=context_fc, - payload_field_class=payload_fc) - ec2 = bt2.EventClass('event17', id=17, context_field_class=payload_fc, - payload_field_class=context_fc) - 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_fc = bt2.StructureFieldClass() - header_fc.append_field('magic', bt2.IntegerFieldClass(32)) - header_fc.append_field('stream_id', bt2.IntegerFieldClass(32)) - return header_fc + trace = self._tc() + self.assertEqual(trace.name, None) 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_class=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_class, 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 - - def test_assign_static(self): - self._tc.set_is_static() - self.assertTrue(self._tc.is_static) - - 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) + trace = self._tc(name='my name') + self.assertEqual(trace.name, 'my name') - def test_assign_invalid_native_byte_order(self): + def test_create_invalid_name(self): with self.assertRaises(TypeError): - self._tc.native_byte_order = 'lel' - - def test_assign_packet_header_field_class(self): - header_fc = bt2.StructureFieldClass() - header_fc.append_field('magic', bt2.IntegerFieldClass(32)) - self._tc.packet_header_field_class = header_fc - self.assertEqual(self._tc.packet_header_field_class, header_fc) - - def test_assign_no_packet_header_field_class(self): - self._tc.packet_header_field_class = None - self.assertIsNone(self._tc.packet_header_field_class) - - 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)) + self._tc(name=17) - def _pre_copy(self): - self._tc.packet_header_field_class = 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_class.addr, cpy.packet_header_field_class.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_class.addr, cpy.packet_header_field_class.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_len(self): + trace = self._tc() + sc = self._tc.create_stream_class() + self.assertEqual(len(trace), 0) - def test_getitem(self): - self._tc.add_stream_class(self._sc) - self.assertEqual(self._tc[3].addr, self._sc.addr) + trace.create_stream(sc) + self.assertEqual(len(trace), 1) - def test_getitem_wrong_key_type(self): - self._tc.add_stream_class(self._sc) - with self.assertRaises(TypeError): - self._tc['hello'] + 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) - def test_getitem_wrong_key(self): - self._tc.add_stream_class(self._sc) - with self.assertRaises(KeyError): - self._tc[4] - - def test_len(self): - self.assertEqual(len(self._tc), 0) - self._tc.add_stream_class(self._sc) - self.assertEqual(len(self._tc), 1) + return trace def test_iter(self): - self._tc.packet_header_field_class = 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) + trace = self._create_trace_with_some_streams() + stream_ids = set(trace) + self.assertEqual(stream_ids, {12, 15, 17}) - 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_getitem(self): + trace = self._create_trace_with_some_streams() - def test_env_getitem_wrong_key(self): - with self.assertRaises(KeyError): - self._tc.env['lel'] + self.assertEqual(trace[12].id, 12) - def test_clock_classes_getitem_wrong_key(self): + def test_getitem_invalid_key(self): + trace = self._create_trace_with_some_streams() with self.assertRaises(KeyError): - self._tc.clock_classes['lel'] + trace[18] - def test_streams_none(self): - self.assertEqual(len(self._tc.streams), 0) + def test_destruction_listener(self): + def on_trace_class_destruction(trace_class): + nonlocal trace_class_destroyed + trace_class_destroyed = True - 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 on_trace_destruction(trace): + nonlocal trace_destroyed + trace_destroyed = True - 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() + trace_destroyed = False + trace_class_destroyed = False - for stream in self._tc.streams: - sids.add(stream.id) + trace_class = get_default_trace_class() + stream_class = trace_class.create_stream_class() + trace = trace_class() + stream = trace.create_stream(stream_class) - self.assertEqual(len(sids), 3) - self.assertTrue(12 in sids and 15 in sids and 17 in sids) + trace_class.add_destruction_listener(on_trace_class_destruction) + trace.add_destruction_listener(on_trace_destruction) - 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() + self.assertFalse(trace_class_destroyed) + self.assertFalse(trace_destroyed) - def test_eq(self): - cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - self.assertEqual(tc1, tc2) + del trace - def test_ne_name(self): - cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - self.assertNotEqual(tc1, tc2) + self.assertFalse(trace_class_destroyed) + self.assertFalse(trace_destroyed) - def test_ne_packet_header_field_class(self): - cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects() - header_fc.append_field('yes', bt2.StringFieldClass()) - tc2 = bt2.Trace(name='my name', - native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, - env={'the_string': 'value', 'the_int': 23}, - packet_header_field_class=header_fc, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - self.assertNotEqual(tc1, tc2) + del stream - def test_ne_native_byte_order(self): - cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - self.assertNotEqual(tc1, tc2) + self.assertFalse(trace_class_destroyed) + self.assertTrue(trace_destroyed) - def test_ne_env(self): - cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - self.assertNotEqual(tc1, tc2) + del trace_class - def test_ne_clock_classes(self): - cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - self.assertNotEqual(tc1, tc2) + self.assertFalse(trace_class_destroyed) + self.assertTrue(trace_destroyed) - def test_ne_stream_classes(self): - cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc, - clock_classes=(cc1, cc2), - stream_classes=(sc1, sc2)) - self.assertNotEqual(tc1, tc2) + del stream_class - def test_eq_invalid(self): - self.assertFalse(self._tc == 23) + self.assertTrue(trace_class_destroyed) + self.assertTrue(trace_destroyed)