X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_clock_class.py;h=b9d25e54b0f970e1edb10172c6ffd8b0b73d6c53;hb=d14ddbbab48ce07804d4252228fceee692d5dea4;hp=1f3959127208b274f23b0a3615ee05ac9f436631;hpb=d24d56638469189904fb6ddbb3c725817b3e9417;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_clock_class.py b/tests/bindings/python/bt2/test_clock_class.py index 1f395912..b9d25e54 100644 --- a/tests/bindings/python/bt2/test_clock_class.py +++ b/tests/bindings/python/bt2/test_clock_class.py @@ -18,9 +18,11 @@ import unittest import uuid -import copy import bt2 -from utils import run_in_component_init +import utils +from utils import run_in_component_init, TestOutputPortMessageIterator +from bt2 import value as bt2_value +from bt2 import clock_class as bt2_clock_class class ClockClassOffsetTestCase(unittest.TestCase): @@ -88,6 +90,7 @@ class ClockClassTestCase(unittest.TestCase): self.assertEqual(cc.offset, bt2.ClockClassOffset()) self.assertTrue(cc.origin_is_unix_epoch) self.assertIsNone(cc.uuid) + self.assertEqual(len(cc.user_attributes), 0) def test_create_name(self): def f(comp_self): @@ -169,7 +172,9 @@ class ClockClassTestCase(unittest.TestCase): 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) + 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) @@ -179,12 +184,14 @@ class ClockClassTestCase(unittest.TestCase): return comp_self._create_clock_class(frequency=1000) cc = run_in_component_init(f) - with self.assertRaises(bt2.OverflowError): - cc.cycles_to_ns_from_origin(2**63) + with self.assertRaises(bt2._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')) + return comp_self._create_clock_class( + uuid=uuid.UUID('b43372c32ef0be28444dfc1c5cdafd33') + ) cc = run_in_component_init(f) self.assertEqual(cc.uuid, uuid.UUID('b43372c32ef0be28444dfc1c5cdafd33')) @@ -195,25 +202,46 @@ class ClockClassTestCase(unittest.TestCase): self.assertRaisesInComponentInit(TypeError, f) + def test_create_user_attributes(self): + def f(comp_self): + return comp_self._create_clock_class(user_attributes={'salut': 23}) + + cc = run_in_component_init(f) + self.assertEqual(cc.user_attributes, {'salut': 23}) + self.assertIs(type(cc.user_attributes), bt2_value.MapValue) + + def test_create_invalid_user_attributes(self): + def f(comp_self): + return comp_self._create_clock_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_clock_class(user_attributes=23) + + self.assertRaisesInComponentInit(TypeError, f) + + def test_const_user_attributes(self): + cc = utils.get_const_event_message().default_clock_snapshot.clock_class + self.assertIs(type(cc.user_attributes), bt2_value._MapValueConst) + class ClockSnapshotTestCase(unittest.TestCase): def setUp(self): def f(comp_self): - cc = comp_self._create_clock_class(1000, 'my_cc', - offset=bt2.ClockClassOffset(45, 354)) + 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_beginning_default_clock_snapshot=True, - packets_have_end_default_clock_snapshot=True) + _sc = _tc.create_stream_class(default_clock_class=_cc) _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 @@ -226,14 +254,10 @@ class ClockSnapshotTestCase(unittest.TestCase): if self._at == 0: notif = self._create_stream_beginning_message(_stream) elif self._at == 1: - notif = self._create_packet_beginning_message(_packet, 100) + notif = self._create_event_message(_ec, _stream, 123) elif self._at == 2: - notif = self._create_event_message(_ec, _packet, 123) + notif = self._create_event_message(_ec, _stream, 2 ** 63) 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 @@ -242,18 +266,19 @@ class ClockSnapshotTestCase(unittest.TestCase): return notif class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter): - def __init__(self, params): + def __init__(self, params, obj): 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']) + self._msg_iter = TestOutputPortMessageIterator( + self._graph, self._src_comp.output_ports['out'] + ) for i, msg in enumerate(self._msg_iter): - if i == 2: + if i == 1: self._msg = msg - elif i == 3: + elif i == 2: self._msg_clock_overflow = msg break @@ -263,21 +288,24 @@ class ClockSnapshotTestCase(unittest.TestCase): def test_create_default(self): self.assertEqual( - self._msg.default_clock_snapshot.clock_class.addr, self._cc.addr) + self._msg.default_clock_snapshot.clock_class.addr, self._cc.addr + ) self.assertEqual(self._msg.default_clock_snapshot.value, 123) def test_clock_class(self): - self.assertEqual( - self._msg.default_clock_snapshot.clock_class.addr, self._cc.addr) + cc = self._msg.default_clock_snapshot.clock_class + self.assertEqual(cc.addr, self._cc.addr) + self.assertIs(type(cc), bt2_clock_class._ClockClassConst) 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) + self._msg.default_clock_snapshot.ns_from_origin, ns_from_origin + ) def test_ns_from_origin_overflow(self): - with self.assertRaises(bt2.OverflowError): + with self.assertRaises(bt2._OverflowError): self._msg_clock_overflow.default_clock_snapshot.ns_from_origin def test_eq_int(self): @@ -298,3 +326,7 @@ class ClockSnapshotTestCase(unittest.TestCase): self.assertTrue(self._msg.default_clock_snapshot <= 123) self.assertFalse(self._msg.default_clock_snapshot <= 100) + + +if __name__ == '__main__': + unittest.main()