From: Simon Marchi Date: Mon, 3 Jun 2019 22:41:45 +0000 (-0400) Subject: bt2: Adapt test_clock_class.py and make it pass X-Git-Url: http://git.efficios.com/?p=babeltrace.git;a=commitdiff_plain;h=be7bbff934d18e407853436dd9f7da23c8c20743 bt2: Adapt test_clock_class.py and make it pass This patch adapts test_clock_class to the current BT API and changes what's needed to make it pass. One change in test_clock_class is that clock classes need to be created from self components now, so it requires a bit more boilerplate. Everthing related to equality, copy and deepcopy is removed from ClockClass. However, it remains possible to test a _ClockSnapshot for equality against an integer. I thoough it would be useful to support other relational operations (<, <=, >, >=) for clock snapshots, so I added support for them using functools.total_ordering. The constructors for both _ClockSnapshot and ClockClass are removed, as the user never directly creates those objects anymore. Clock snapshots are obtained from messages, while clock classes are created using the _create_clock_class method of a component. As in previous patches, the public setters are removed, as we only support setting properies when creating an object. Change-Id: I7228b32530f98811cb512243469ae7d0d61a9da1 Signed-off-by: Simon Marchi Signed-off-by: Francis Deslauriers Reviewed-on: https://review.lttng.org/c/babeltrace/+/1299 Reviewed-by: Philippe Proulx Tested-by: jenkins --- diff --git a/bindings/python/bt2/bt2/clock_class.py b/bindings/python/bt2/bt2/clock_class.py index 8eb25385..59a93cd7 100644 --- a/bindings/python/bt2/bt2/clock_class.py +++ b/bindings/python/bt2/bt2/clock_class.py @@ -21,10 +21,8 @@ # THE SOFTWARE. from bt2 import native_bt, object, utils -import uuid as uuidp -import numbers import bt2 -import bt2.clock_snapshot as clock_snapshot +import uuid as uuidp class ClockClassOffset: @@ -42,9 +40,6 @@ class ClockClassOffset: def cycles(self): return self._cycles - def __hash__(self): - return hash((self.seconds, self.cycles)) - def __eq__(self, other): if not isinstance(other, self.__class__): # not comparing apples to apples @@ -53,159 +48,73 @@ class ClockClassOffset: return (self.seconds, self.cycles) == (other.seconds, other.cycles) -class ClockClass(object._SharedObject): +class _ClockClass(object._SharedObject): _get_ref = staticmethod(native_bt.clock_class_get_ref) _put_ref = staticmethod(native_bt.clock_class_put_ref) - def __init__(self, name, frequency, description=None, precision=None, - offset=None, is_absolute=None, uuid=None): - utils._check_str(name) - utils._check_uint64(frequency) - ptr = native_bt.clock_class_create(name, frequency) - - if ptr is None: - raise bt2.CreationError('cannot create clock class object') - - super().__init__(ptr) - - if description is not None: - self.description = description - - if frequency is not None: - self.frequency = frequency - - if precision is not None: - self.precision = precision - - if offset is not None: - self.offset = offset - - if is_absolute is not None: - self.is_absolute = is_absolute - - if uuid is not None: - self.uuid = uuid - - def __eq__(self, other): - if type(self) is not type(other): - # not comparing apples to apples - return False - - self_props = ( - self.name, - self.description, - self.frequency, - self.precision, - self.offset, - self.is_absolute, - self.uuid - ) - other_props = ( - other.name, - other.description, - other.frequency, - other.precision, - other.offset, - other.is_absolute, - other.uuid - ) - return self_props == other_props - - def __copy__(self): - return ClockClass(name=self.name, description=self.description, - frequency=self.frequency, precision=self.precision, - offset=self.offset, is_absolute=self.is_absolute, - uuid=self.uuid) - - def __deepcopy__(self, memo): - cpy = self.__copy__() - memo[id(self)] = cpy - return cpy - - def __hash__(self): - return hash(( - self.name, - self.description, - self.frequency, - self.precision, - self.offset.seconds, - self.offset.cycles, - self.is_absolute, - self.uuid)) - @property def name(self): - name = native_bt.clock_class_get_name(self._ptr) - assert(name is not None) - return name + return native_bt.clock_class_get_name(self._ptr) - @name.setter - def name(self, name): + def _name(self, name): utils._check_str(name) ret = native_bt.clock_class_set_name(self._ptr, name) utils._handle_ret(ret, "cannot set clock class object's name") + _name = property(fset=_name) + @property def description(self): return native_bt.clock_class_get_description(self._ptr) - @description.setter - def description(self, description): + def _description(self, description): utils._check_str(description) ret = native_bt.clock_class_set_description(self._ptr, description) utils._handle_ret(ret, "cannot set clock class object's description") + _description = property(fset=_description) + @property def frequency(self): - frequency = native_bt.clock_class_get_frequency(self._ptr) - assert(frequency >= 1) - return frequency + return native_bt.clock_class_get_frequency(self._ptr) - @frequency.setter - def frequency(self, frequency): + def _frequency(self, frequency): utils._check_uint64(frequency) - ret = native_bt.clock_class_set_frequency(self._ptr, frequency) - utils._handle_ret(ret, "cannot set clock class object's frequency") + native_bt.clock_class_set_frequency(self._ptr, frequency) + + _frequency = property(fset=_frequency) @property def precision(self): precision = native_bt.clock_class_get_precision(self._ptr) - assert(precision >= 0) return precision - @precision.setter - def precision(self, precision): + def _precision(self, precision): utils._check_uint64(precision) - ret = native_bt.clock_class_set_precision(self._ptr, precision) - utils._handle_ret(ret, "cannot set clock class object's precision") + native_bt.clock_class_set_precision(self._ptr, precision) + + _precision = property(fset=_precision) @property def offset(self): - ret, offset_s = native_bt.clock_class_get_offset_s(self._ptr) - assert(ret == 0) - ret, offset_cycles = native_bt.clock_class_get_offset_cycles(self._ptr) - assert(ret == 0) + offset_s, offset_cycles = native_bt.clock_class_get_offset(self._ptr) return ClockClassOffset(offset_s, offset_cycles) - @offset.setter - def offset(self, offset): + def _offset(self, offset): utils._check_type(offset, ClockClassOffset) - ret = native_bt.clock_class_set_offset_s(self._ptr, offset.seconds) - utils._handle_ret(ret, "cannot set clock class object's offset (seconds)") - ret = native_bt.clock_class_set_offset_cycles(self._ptr, offset.cycles) - utils._handle_ret(ret, "cannot set clock class object's offset (cycles)") + native_bt.clock_class_set_offset(self._ptr, offset.seconds, offset.cycles) + + _offset = property(fset=_offset) @property - def is_absolute(self): - is_absolute = native_bt.clock_class_is_absolute(self._ptr) - assert(is_absolute >= 0) - return is_absolute > 0 + def origin_is_unix_epoch(self): + return native_bt.clock_class_origin_is_unix_epoch(self._ptr) - @is_absolute.setter - def is_absolute(self, is_absolute): - utils._check_bool(is_absolute) - ret = native_bt.clock_class_set_is_absolute(self._ptr, int(is_absolute)) - utils._handle_ret(ret, "cannot set clock class object's absoluteness") + def _origin_is_unix_epoch(self, origin_is_unix_epoch): + utils._check_bool(origin_is_unix_epoch) + native_bt.clock_class_set_origin_is_unix_epoch(self._ptr, int(origin_is_unix_epoch)) + + _origin_is_unix_epoch = property(fset=_origin_is_unix_epoch) @property def uuid(self): @@ -216,11 +125,20 @@ class ClockClass(object._SharedObject): return uuidp.UUID(bytes=uuid_bytes) - @uuid.setter - def uuid(self, uuid): + def _uuid(self, uuid): utils._check_type(uuid, uuidp.UUID) - ret = native_bt.clock_class_set_uuid(self._ptr, uuid.bytes) - utils._handle_ret(ret, "cannot set clock class object's UUID") + native_bt.clock_class_set_uuid(self._ptr, uuid.bytes) + + _uuid = property(fset=_uuid) + + def cycles_to_ns_from_origin(self, cycles): + utils._check_uint64(cycles) + ret, ns = native_bt.clock_class_cycles_to_ns_from_origin(self._ptr, cycles) + + error_msg = "cannot convert clock value to nanoseconds from origin for given clock class" + + if ret == native_bt.CLOCK_CLASS_STATUS_OVERFLOW: + raise OverflowError(error_msg) - def __call__(self, cycles): - return clock_snapshot._ClockSnapshot(self._ptr, cycles) + utils._handle_ret(ret, error_msg) + return ns diff --git a/bindings/python/bt2/bt2/clock_snapshot.py b/bindings/python/bt2/bt2/clock_snapshot.py index 4be39e80..85bda824 100644 --- a/bindings/python/bt2/bt2/clock_snapshot.py +++ b/bindings/python/bt2/bt2/clock_snapshot.py @@ -21,61 +21,40 @@ # THE SOFTWARE. from bt2 import native_bt, object, utils -import uuid as uuidp import numbers import bt2 +import functools -def _create_clock_snapshot_from_ptr(ptr): - clock_snapshot = _ClockSnapshot._create_from_ptr(ptr) - return clock_snapshot - - +@functools.total_ordering class _ClockSnapshot(object._UniqueObject): - def __init__(self, clock_class_ptr, cycles): - utils._check_uint64(cycles) - ptr = native_bt.clock_snapshot_create(clock_class_ptr, cycles) - - if ptr is None: - raise bt2.CreationError('cannot create clock value object') - - super().__init__(ptr) - @property def clock_class(self): - ptr = native_bt.clock_snapshot_get_class(self._ptr) - assert(ptr) - return bt2.ClockClass._create_from_ptr(ptr) + cc_ptr = native_bt.clock_snapshot_borrow_clock_class_const(self._ptr) + assert cc_ptr is not None + return bt2.clock_class._ClockClass._create_from_ptr_and_get_ref(cc_ptr) @property def value(self): return native_bt.clock_snapshot_get_value(self._ptr) @property - def ns_from_epoch(self): - ret, ns = native_bt.clock_snapshot_get_value_ns_from_epoch(self._ptr) - utils._handle_ret(ret, "cannot get clock value object's nanoseconds from Epoch") - return ns + def ns_from_origin(self): + ret, ns = native_bt.clock_snapshot_get_ns_from_origin(self._ptr) - def __eq__(self, other): - if isinstance(other, numbers.Integral): - return int(other) == self.cycles + if ret == native_bt.CLOCK_SNAPSHOT_STATUS_OVERFLOW: + raise OverflowError("cannot get clock snapshot's nanoseconds from origin") - if not isinstance(other, self.__class__): - # not comparing apples to apples - return False + return ns - if self.addr == other.addr: - return True + def __eq__(self, other): + if not isinstance(other, numbers.Integral): + return NotImplemented - self_props = self.clock_class, self.cycles - other_props = other.clock_class, other.cycles - return self_props == other_props + return self.value == int(other) - def __copy__(self): - return self.clock_class(self.cycles) + def __lt__(self, other): + if not isinstance(other, numbers.Integral): + return NotImplemented - def __deepcopy__(self, memo): - cpy = self.__copy__() - memo[id(self)] = cpy - return cpy + return self.value < int(other) diff --git a/bindings/python/bt2/bt2/component.py b/bindings/python/bt2/bt2/component.py index fb7f68f0..e500c982 100644 --- a/bindings/python/bt2/bt2/component.py +++ b/bindings/python/bt2/bt2/component.py @@ -641,18 +641,37 @@ class _UserComponent(metaclass=_UserComponentType): return tc - def _create_clock_class(self, frequency=None): + def _create_clock_class(self, frequency=None, name=None, description=None, + precision=None, offset=None, origin_is_unix_epoch=True, + uuid=None): ptr = self._as_self_component_ptr(self._ptr) cc_ptr = native_bt.clock_class_create(ptr) if cc_ptr is None: raise bt2.CreationError('could not create clock class') - cc = bt2.ClockClass._create_from_ptr(cc_ptr) + cc = bt2.clock_class._ClockClass._create_from_ptr(cc_ptr) if frequency is not None: cc._frequency = frequency + if name is not None: + cc._name = name + + if description is not None: + cc._description = description + + if precision is not None: + cc._precision = precision + + if offset is not None: + cc._offset = offset + + cc._origin_is_unix_epoch = origin_is_unix_epoch + + if uuid is not None: + cc._uuid = uuid + return cc diff --git a/bindings/python/bt2/bt2/stream_class.py b/bindings/python/bt2/bt2/stream_class.py index c599dc2a..f2869cb9 100644 --- a/bindings/python/bt2/bt2/stream_class.py +++ b/bindings/python/bt2/bt2/stream_class.py @@ -224,10 +224,10 @@ class StreamClass(object._SharedObject, collections.abc.Mapping): if cc_ptr is None: return - return bt2.clock_class.ClockClass._create_from_ptr_and_get_ref(cc_ptr) + return bt2.clock_class._ClockClass._create_from_ptr_and_get_ref(cc_ptr) def _default_clock_class(self, clock_class): - utils._check_type(clock_class, bt2.clock_class.ClockClass) + utils._check_type(clock_class, bt2.clock_class._ClockClass) native_bt.stream_class_set_default_clock_class( self._ptr, clock_class._ptr) diff --git a/tests/bindings/python/bt2/test_clock_class.py b/tests/bindings/python/bt2/test_clock_class.py index 7498be44..de0c001c 100644 --- a/tests/bindings/python/bt2/test_clock_class.py +++ b/tests/bindings/python/bt2/test_clock_class.py @@ -2,9 +2,9 @@ import unittest import uuid import copy import bt2 +from utils import run_in_component_init -@unittest.skip("this is broken") class ClockClassOffsetTestCase(unittest.TestCase): def test_create_default(self): cco = bt2.ClockClassOffset() @@ -48,265 +48,235 @@ class ClockClassOffsetTestCase(unittest.TestCase): self.assertFalse(bt2.ClockClassOffset() == 23) -@unittest.skip("this is broken") class ClockClassTestCase(unittest.TestCase): - def setUp(self): - self._cc = bt2.ClockClass('salut', 1000000) + def assertRaisesInComponentInit(self, expected_exc_type, user_code): + def f(comp_self): + try: + user_code(comp_self) + except Exception as exc: + return type(exc) - def tearDown(self): - del self._cc + exc_type = run_in_component_init(f) + self.assertIsNotNone(exc_type) + self.assertEqual(exc_type, expected_exc_type) def test_create_default(self): - self.assertEqual(self._cc.name, 'salut') + cc = run_in_component_init(lambda comp_self: comp_self._create_clock_class()) - def test_create_invalid_no_name(self): - with self.assertRaises(TypeError): - bt2.ClockClass() - - def test_create_full(self): - my_uuid = uuid.uuid1() - cc = bt2.ClockClass(name='name', description='some description', - frequency=1001, precision=176, - offset=bt2.ClockClassOffset(45, 3003), - is_absolute=True, uuid=my_uuid) - self.assertEqual(cc.name, 'name') - self.assertEqual(cc.description, 'some description') - self.assertEqual(cc.frequency, 1001) - self.assertEqual(cc.precision, 176) - self.assertEqual(cc.offset, bt2.ClockClassOffset(45, 3003)) - self.assertEqual(cc.is_absolute, True) - self.assertEqual(cc.uuid, copy.deepcopy(my_uuid)) - - def test_assign_name(self): - self._cc.name = 'the_clock' - self.assertEqual(self._cc.name, 'the_clock') - - def test_assign_invalid_name(self): - with self.assertRaises(TypeError): - self._cc.name = 23 + self.assertIsNone(cc.name) + self.assertEqual(cc.frequency, 1000000000) + self.assertIsNone(cc.description) + self.assertEqual(cc.precision, 0) + self.assertEqual(cc.offset, bt2.ClockClassOffset()) + self.assertTrue(cc.origin_is_unix_epoch) + self.assertIsNone(cc.uuid) - def test_assign_description(self): - self._cc.description = 'hi people' - self.assertEqual(self._cc.description, 'hi people') + def test_create_name(self): + def f(comp_self): + return comp_self._create_clock_class(name='the_clock') - def test_assign_invalid_description(self): - with self.assertRaises(TypeError): - self._cc.description = 23 + cc = run_in_component_init(f) + self.assertEqual(cc.name, 'the_clock') - def test_assign_frequency(self): - self._cc.frequency = 987654321 - self.assertEqual(self._cc.frequency, 987654321) + def test_create_invalid_name(self): + def f(comp_self): + comp_self._create_clock_class(name=23) - def test_assign_invalid_frequency(self): - with self.assertRaises(TypeError): - self._cc.frequency = 'lel' + self.assertRaisesInComponentInit(TypeError, f) - def test_assign_precision(self): - self._cc.precision = 12 - self.assertEqual(self._cc.precision, 12) + def test_create_description(self): + def f(comp_self): + return comp_self._create_clock_class(description='hi people') - def test_assign_invalid_precision(self): - with self.assertRaises(TypeError): - self._cc.precision = 'lel' + cc = run_in_component_init(f) + self.assertEqual(cc.description, 'hi people') - def test_assign_offset(self): - self._cc.offset = bt2.ClockClassOffset(12, 56) - self.assertEqual(self._cc.offset, bt2.ClockClassOffset(12, 56)) + def test_create_invalid_description(self): + def f(comp_self): + return comp_self._create_clock_class(description=23) - def test_assign_invalid_offset(self): - with self.assertRaises(TypeError): - self._cc.offset = object() + self.assertRaisesInComponentInit(TypeError, f) - def test_assign_absolute(self): - self._cc.is_absolute = True - self.assertTrue(self._cc.is_absolute) + def test_create_frequency(self): + def f(comp_self): + return comp_self._create_clock_class(frequency=987654321) - def test_assign_invalid_absolute(self): - with self.assertRaises(TypeError): - self._cc.is_absolute = 23 + cc = run_in_component_init(f) + self.assertEqual(cc.frequency, 987654321) - def test_assign_uuid(self): - the_uuid = uuid.uuid1() - self._cc.uuid = the_uuid - self.assertEqual(self._cc.uuid, the_uuid) + def test_create_invalid_frequency(self): + def f(comp_self): + return comp_self._create_clock_class(frequency='lel') - def test_assign_invalid_uuid(self): - with self.assertRaises(TypeError): - self._cc.uuid = object() + self.assertRaisesInComponentInit(TypeError, f) - def test_create_clock_snapshot(self): - cs = self._cc(756) - self.assertEqual(cs.clock_class.addr, self._cc.addr) + def test_create_precision(self): + def f(comp_self): + return comp_self._create_clock_class(precision=12) - def _test_copy(self, cpy): - self.assertIsNot(cpy, self._cc) - self.assertNotEqual(cpy.addr, self._cc.addr) - self.assertEqual(cpy, self._cc) + cc = run_in_component_init(f) + self.assertEqual(cc.precision, 12) - def test_copy(self): - cpy = copy.copy(self._cc) - self._test_copy(cpy) + def test_create_invalid_precision(self): + def f(comp_self): + return comp_self._create_clock_class(precision='lel') - def test_deepcopy(self): - cpy = copy.deepcopy(self._cc) - self._test_copy(cpy) + self.assertRaisesInComponentInit(TypeError, f) - def test_eq(self): - my_uuid = uuid.uuid1() - cc1 = bt2.ClockClass(name='name', description='some description', - frequency=1001, precision=176, - offset=bt2.ClockClassOffset(45, 3003), - is_absolute=True, uuid=my_uuid) - cc2 = bt2.ClockClass(name='name', description='some description', - frequency=1001, precision=176, - offset=bt2.ClockClassOffset(45, 3003), - is_absolute=True, uuid=my_uuid) - self.assertEqual(cc1, cc2) - - def test_ne_name(self): - my_uuid = uuid.uuid1() - cc1 = bt2.ClockClass(name='mane', description='some description', - frequency=1001, precision=176, - offset=bt2.ClockClassOffset(45, 3003), - is_absolute=True, uuid=my_uuid) - cc2 = bt2.ClockClass(name='name', description='some description', - frequency=1001, precision=176, - offset=bt2.ClockClassOffset(45, 3003), - is_absolute=True, uuid=my_uuid) - self.assertNotEqual(cc1, cc2) - - def test_ne_description(self): - my_uuid = uuid.uuid1() - cc1 = bt2.ClockClass(name='name', description='some descripti2', - frequency=1001, precision=176, - offset=bt2.ClockClassOffset(45, 3003), - is_absolute=True, uuid=my_uuid) - cc2 = bt2.ClockClass(name='name', description='some description', - frequency=1001, precision=176, - offset=bt2.ClockClassOffset(45, 3003), - is_absolute=True, uuid=my_uuid) - self.assertNotEqual(cc1, cc2) - - def test_ne_frequency(self): - my_uuid = uuid.uuid1() - cc1 = bt2.ClockClass(name='name', description='some description', - frequency=1003, precision=176, - offset=bt2.ClockClassOffset(45, 3003), - is_absolute=True, uuid=my_uuid) - cc2 = bt2.ClockClass(name='name', description='some description', - frequency=1001, precision=176, - offset=bt2.ClockClassOffset(45, 3003), - is_absolute=True, uuid=my_uuid) - self.assertNotEqual(cc1, cc2) - - def test_ne_precision(self): - my_uuid = uuid.uuid1() - cc1 = bt2.ClockClass(name='name', description='some description', - frequency=1001, precision=171, - offset=bt2.ClockClassOffset(45, 3003), - is_absolute=True, uuid=my_uuid) - cc2 = bt2.ClockClass(name='name', description='some description', - frequency=1001, precision=176, - offset=bt2.ClockClassOffset(45, 3003), - is_absolute=True, uuid=my_uuid) - self.assertNotEqual(cc1, cc2) - - def test_ne_offset(self): - my_uuid = uuid.uuid1() - cc1 = bt2.ClockClass(name='name', description='some description', - frequency=1001, precision=176, - offset=bt2.ClockClassOffset(45, 3001), - is_absolute=True, uuid=my_uuid) - cc2 = bt2.ClockClass(name='name', description='some description', - frequency=1001, precision=176, - offset=bt2.ClockClassOffset(45, 3003), - is_absolute=True, uuid=my_uuid) - self.assertNotEqual(cc1, cc2) - - def test_ne_absolute(self): - my_uuid = uuid.uuid1() - cc1 = bt2.ClockClass(name='name', description='some description', - frequency=1001, precision=176, - offset=bt2.ClockClassOffset(45, 3003), - is_absolute=True, uuid=my_uuid) - cc2 = bt2.ClockClass(name='name', description='some description', - frequency=1001, precision=176, - offset=bt2.ClockClassOffset(45, 3003), - is_absolute=False, uuid=my_uuid) - self.assertNotEqual(cc1, cc2) - - def test_ne_uuid(self): - cc1 = bt2.ClockClass(name='name', description='some description', - frequency=1001, precision=176, - offset=bt2.ClockClassOffset(45, 3003), - is_absolute=True, uuid=uuid.uuid1()) - cc2 = bt2.ClockClass(name='name', description='some description', - frequency=1001, precision=176, - offset=bt2.ClockClassOffset(45, 3003), - is_absolute=True, uuid=uuid.uuid1()) - self.assertNotEqual(cc1, cc2) + def test_create_offset(self): + def f(comp_self): + return comp_self._create_clock_class(offset=bt2.ClockClassOffset(12, 56)) - def test_eq_invalid(self): - self.assertFalse(self._cc == 23) + cc = run_in_component_init(f) + self.assertEqual(cc.offset, bt2.ClockClassOffset(12, 56)) + + def test_create_invalid_offset(self): + def f(comp_self): + return comp_self._create_clock_class(offset=object()) + + self.assertRaisesInComponentInit(TypeError, f) + + def test_create_origin_is_unix_epoch(self): + def f(comp_self): + return comp_self._create_clock_class(origin_is_unix_epoch=False) + + cc = run_in_component_init(f) + self.assertEqual(cc.origin_is_unix_epoch, False) + + def test_create_invalid_origin_is_unix_epoch(self): + def f(comp_self): + return comp_self._create_clock_class(origin_is_unix_epoch=23) + + self.assertRaisesInComponentInit(TypeError, f) + + def test_cycles_to_ns_from_origin(self): + def f(comp_self): + return comp_self._create_clock_class(frequency=10**8, origin_is_unix_epoch=True) + + cc = run_in_component_init(f) + self.assertEqual(cc.cycles_to_ns_from_origin(112), 1120) + + def test_cycles_to_ns_from_origin_overflow(self): + def f(comp_self): + return comp_self._create_clock_class(frequency=1000) + + cc = run_in_component_init(f) + with self.assertRaises(OverflowError): + cc.cycles_to_ns_from_origin(2**63) + + def test_create_uuid(self): + def f(comp_self): + return comp_self._create_clock_class(uuid=uuid.UUID('b43372c32ef0be28444dfc1c5cdafd33')) + + cc = run_in_component_init(f) + self.assertEqual(cc.uuid, uuid.UUID('b43372c32ef0be28444dfc1c5cdafd33')) + + def test_create_invalid_uuid(self): + def f(comp_self): + return comp_self._create_clock_class(uuid=23) + + self.assertRaisesInComponentInit(TypeError, f) -@unittest.skip("this is broken") class ClockSnapshotTestCase(unittest.TestCase): def setUp(self): - self._cc = bt2.ClockClass('salut', 1000, - offset=bt2.ClockClassOffset(45, 354)) - self._cs = self._cc(123) + def f(comp_self): + cc = comp_self._create_clock_class(1000, 'my_cc', + offset=bt2.ClockClassOffset(45, 354)) + tc = comp_self._create_trace_class() + + return (cc, tc) + + _cc, _tc = run_in_component_init(f) + _trace = _tc() + _sc = _tc.create_stream_class(default_clock_class=_cc, + packets_have_default_beginning_clock_snapshot=True, + packets_have_default_end_clock_snapshot=True) + _ec = _sc.create_event_class(name='salut') + _stream = _trace.create_stream(_sc) + _packet = _stream.create_packet() + self._packet = _packet + self._stream = _stream + self._ec = _ec + self._cc = _cc + + class MyIter(bt2._UserMessageIterator): + def __init__(self): + self._at = 0 + + def __next__(self): + if self._at == 0: + notif = self._create_stream_beginning_message(_stream) + elif self._at == 1: + notif = self._create_packet_beginning_message(_packet, 100) + elif self._at == 2: + notif = self._create_event_message(_ec, _packet, 123) + elif self._at == 3: + notif = self._create_event_message(_ec, _packet, 2**63) + elif self._at == 4: + notif = self._create_packet_end_message(_packet) + elif self._at == 5: + notif = self._create_stream_end_message(_stream) + else: + raise bt2.Stop + + self._at += 1 + return notif + + class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter): + def __init__(self, params): + self._add_output_port('out') + + self._graph = bt2.Graph() + self._src_comp = self._graph.add_component(MySrc, 'my_source') + self._msg_iter = self._graph.create_output_port_message_iterator( + self._src_comp.output_ports['out']) + + for i, msg in enumerate(self._msg_iter): + if i == 2: + self._msg = msg + elif i == 3: + self._msg_clock_overflow = msg + break def tearDown(self): del self._cc - del self._cs + del self._msg def test_create_default(self): - self.assertEqual(self._cs.clock_class.addr, self._cc.addr) - self.assertEqual(self._cs.cycles, 123) + self.assertEqual( + self._msg.default_clock_snapshot.clock_class.addr, self._cc.addr) + self.assertEqual(self._msg.default_clock_snapshot.value, 123) - def test_create_invalid_cycles_type(self): - with self.assertRaises(TypeError): - self._cc('yes') + def test_clock_class(self): + self.assertEqual( + self._msg.default_clock_snapshot.clock_class.addr, self._cc.addr) - def test_ns_from_epoch(self): - s_from_epoch = 45 + ((354 + 123) / 1000) - ns_from_epoch = int(s_from_epoch * 1e9) - self.assertEqual(self._cs.ns_from_epoch, ns_from_epoch) + def test_ns_from_origin(self): + s_from_origin = 45 + ((354 + 123) / 1000) + ns_from_origin = int(s_from_origin * 1e9) + self.assertEqual( + self._msg.default_clock_snapshot.ns_from_origin, ns_from_origin) - def test_eq(self): - cs1 = self._cc(123) - cs2 = self._cc(123) - self.assertEqual(cs1, cs2) + def test_ns_from_origin_overflow(self): + with self.assertRaises(OverflowError): + self._msg_clock_overflow.default_clock_snapshot.ns_from_origin def test_eq_int(self): - cs1 = self._cc(123) - self.assertEqual(cs1, 123) - - def test_ne_clock_class(self): - cc1 = bt2.ClockClass('yes', 1500) - cc2 = bt2.ClockClass('yes', 1501) - cs1 = cc1(123) - cs2 = cc2(123) - self.assertNotEqual(cs1, cs2) - - def test_ne_cycles(self): - cs1 = self._cc(123) - cs2 = self._cc(125) - self.assertNotEqual(cs1, cs2) + self.assertEqual(self._msg.default_clock_snapshot, 123) def test_eq_invalid(self): - self.assertFalse(self._cs == 23) + self.assertFalse(self._msg.default_clock_snapshot == 23) + + def test_comparison(self): + self.assertTrue(self._msg.default_clock_snapshot > 100) + self.assertFalse(self._msg.default_clock_snapshot > 200) - def _test_copy(self, cpy): - self.assertIsNot(cpy, self._cs) - self.assertNotEqual(cpy.addr, self._cs.addr) - self.assertEqual(cpy, self._cs) + self.assertTrue(self._msg.default_clock_snapshot >= 123) + self.assertFalse(self._msg.default_clock_snapshot >= 200) - def test_copy(self): - cpy = copy.copy(self._cs) - self._test_copy(cpy) + self.assertTrue(self._msg.default_clock_snapshot < 200) + self.assertFalse(self._msg.default_clock_snapshot < 100) - def test_deepcopy(self): - cpy = copy.deepcopy(self._cs) - self._test_copy(cpy) + self.assertTrue(self._msg.default_clock_snapshot <= 123) + self.assertFalse(self._msg.default_clock_snapshot <= 100)