X-Git-Url: https://git.efficios.com/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_clock_class.py;h=3805c0d56168f87a18d94e935fe1f855f54c2f2b;hb=6c373cc9;hp=ddded862d5a6ac540342d703366db40b585e88d1;hpb=9b24b6aa4258079f2b59fa156dd3e070969f6f2a;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_clock_class.py b/tests/bindings/python/bt2/test_clock_class.py index ddded862..3805c0d5 100644 --- a/tests/bindings/python/bt2/test_clock_class.py +++ b/tests/bindings/python/bt2/test_clock_class.py @@ -1,8 +1,26 @@ +# +# 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 +from utils import run_in_component_init, TestOutputPortMessageIterator class ClockClassOffsetTestCase(unittest.TestCase): @@ -151,7 +169,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 +181,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')) @@ -181,21 +203,18 @@ class ClockClassTestCase(unittest.TestCase): 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 @@ -208,14 +227,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 @@ -224,18 +239,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 @@ -245,21 +261,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) + self._msg.default_clock_snapshot.clock_class.addr, self._cc.addr + ) 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):