X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_clock_class.py;h=c4b34bd29b14110f7674873177600e4db523f085;hb=8d8b141db4c46135a35be19e4a1c192f6a36d67b;hp=de0c001c6d6b11235812b8518f2021ff51464617;hpb=be7bbff934d18e407853436dd9f7da23c8c20743;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_clock_class.py b/tests/bindings/python/bt2/test_clock_class.py index de0c001c..c4b34bd2 100644 --- a/tests/bindings/python/bt2/test_clock_class.py +++ b/tests/bindings/python/bt2/test_clock_class.py @@ -1,8 +1,28 @@ +# +# 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 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): @@ -70,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): @@ -151,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) @@ -161,12 +184,14 @@ class ClockClassTestCase(unittest.TestCase): 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) + 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')) @@ -177,45 +202,62 @@ 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_default_beginning_clock_snapshot=True, - packets_have_default_end_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 class MyIter(bt2._UserMessageIterator): - def __init__(self): + def __init__(self, config, self_port_output): 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) + 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 @@ -224,18 +266,19 @@ class ClockSnapshotTestCase(unittest.TestCase): return notif class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter): - def __init__(self, params): + def __init__(self, config, 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 @@ -245,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(OverflowError): + with self.assertRaises(bt2._OverflowError): self._msg_clock_overflow.default_clock_snapshot.ns_from_origin def test_eq_int(self): @@ -280,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()