X-Git-Url: http://drtracing.org/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_message.py;h=5a3c2c1d671c6a27ca06cdc9828dcec322bd08b8;hb=84eba0d9b419444e95b2f560c660ea57bb2142d8;hp=5b513361ded29b500cf1591e625380e1aa536170;hpb=61d96b89a9c8b3b523558a3c007c2aaedaba2a55;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_message.py b/tests/bindings/python/bt2/test_message.py index 5b513361..5a3c2c1d 100644 --- a/tests/bindings/python/bt2/test_message.py +++ b/tests/bindings/python/bt2/test_message.py @@ -16,9 +16,9 @@ # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. # -import collections import unittest import bt2 +from utils import TestOutputPortMessageIterator class AllMessagesTestCase(unittest.TestCase): @@ -102,7 +102,7 @@ class AllMessagesTestCase(unittest.TestCase): return msg class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter): - def __init__(self, params): + def __init__(self, params, obj): self._add_output_port('out', params) with_cc = bool(params['with_cc']) @@ -148,32 +148,30 @@ class AllMessagesTestCase(unittest.TestCase): def test_all_msg_with_cc(self): params = {'with_cc': True} self._src_comp = self._graph.add_component(self._src, 'my_source', params) - 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 == 0: - self.assertIsInstance(msg, bt2.message._StreamBeginningMessage) + self.assertIsInstance(msg, bt2._StreamBeginningMessage) self.assertEqual(msg.stream.addr, self._stream.addr) self.assertIsInstance( - msg.default_clock_snapshot, bt2.clock_snapshot._UnknownClockSnapshot + msg.default_clock_snapshot, bt2._UnknownClockSnapshot ) elif i == 1: - self.assertIsInstance(msg, bt2.message._PacketBeginningMessage) + self.assertIsInstance(msg, bt2._PacketBeginningMessage) self.assertEqual(msg.packet.addr, self._packet.addr) self.assertEqual(msg.default_clock_snapshot.value, i) elif i == 2: - self.assertIsInstance(msg, bt2.message._EventMessage) + self.assertIsInstance(msg, bt2._EventMessage) self.assertEqual(msg.event.cls.addr, self._event_class.addr) self.assertEqual(msg.default_clock_snapshot.value, i) elif i == 3: - self.assertIsInstance( - msg, bt2.message._MessageIteratorInactivityMessage - ) + self.assertIsInstance(msg, bt2._MessageIteratorInactivityMessage) self.assertEqual(msg.default_clock_snapshot.value, i) elif i == 4: - self.assertIsInstance(msg, bt2.message._DiscardedEventsMessage) + self.assertIsInstance(msg, bt2._DiscardedEventsMessage) self.assertEqual(msg.stream.addr, self._stream.addr) self.assertEqual(msg.count, 890) self.assertEqual( @@ -182,11 +180,11 @@ class AllMessagesTestCase(unittest.TestCase): self.assertEqual(msg.beginning_default_clock_snapshot.value, i) self.assertEqual(msg.end_default_clock_snapshot.value, i) elif i == 5: - self.assertIsInstance(msg, bt2.message._PacketEndMessage) + self.assertIsInstance(msg, bt2._PacketEndMessage) self.assertEqual(msg.packet.addr, self._packet.addr) self.assertEqual(msg.default_clock_snapshot.value, i) elif i == 6: - self.assertIsInstance(msg, bt2.message._DiscardedPacketsMessage) + self.assertIsInstance(msg, bt2._DiscardedPacketsMessage) self.assertEqual(msg.stream.addr, self._stream.addr) self.assertEqual(msg.count, 678) self.assertEqual( @@ -195,10 +193,10 @@ class AllMessagesTestCase(unittest.TestCase): self.assertEqual(msg.beginning_default_clock_snapshot.value, i) self.assertEqual(msg.end_default_clock_snapshot.value, i) elif i == 7: - self.assertIsInstance(msg, bt2.message._StreamEndMessage) + self.assertIsInstance(msg, bt2._StreamEndMessage) self.assertEqual(msg.stream.addr, self._stream.addr) self.assertIsInstance( - msg.default_clock_snapshot, bt2.clock_snapshot._UnknownClockSnapshot + msg.default_clock_snapshot, bt2._UnknownClockSnapshot ) else: raise Exception @@ -206,49 +204,67 @@ class AllMessagesTestCase(unittest.TestCase): def test_all_msg_without_cc(self): params = {'with_cc': False} self._src_comp = self._graph.add_component(self._src, 'my_source', params) - 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 == 0: - self.assertIsInstance(msg, bt2.message._StreamBeginningMessage) + self.assertIsInstance(msg, bt2._StreamBeginningMessage) self.assertEqual(msg.stream.addr, self._stream.addr) - with self.assertRaises(bt2.NonexistentClockSnapshot): + with self.assertRaisesRegex( + ValueError, 'stream class has no default clock class' + ): msg.default_clock_snapshot elif i == 1: - self.assertIsInstance(msg, bt2.message._PacketBeginningMessage) + self.assertIsInstance(msg, bt2._PacketBeginningMessage) self.assertEqual(msg.packet.addr, self._packet.addr) elif i == 2: - self.assertIsInstance(msg, bt2.message._EventMessage) + self.assertIsInstance(msg, bt2._EventMessage) self.assertEqual(msg.event.cls.addr, self._event_class.addr) - with self.assertRaises(bt2.NonexistentClockSnapshot): + with self.assertRaisesRegex( + ValueError, 'stream class has no default clock class' + ): msg.default_clock_snapshot elif i == 3: - self.assertIsInstance(msg, bt2.message._DiscardedEventsMessage) + self.assertIsInstance(msg, bt2._DiscardedEventsMessage) self.assertEqual(msg.stream.addr, self._stream.addr) self.assertEqual(msg.count, 890) self.assertIsNone(msg.stream.cls.default_clock_class) - with self.assertRaises(bt2.NonexistentClockSnapshot): + with self.assertRaisesRegex( + ValueError, + 'such a message has no clock snapshots for this stream class', + ): msg.beginning_default_clock_snapshot - with self.assertRaises(bt2.NonexistentClockSnapshot): + with self.assertRaisesRegex( + ValueError, + 'such a message has no clock snapshots for this stream class', + ): msg.end_default_clock_snapshot elif i == 4: - self.assertIsInstance(msg, bt2.message._PacketEndMessage) + self.assertIsInstance(msg, bt2._PacketEndMessage) self.assertEqual(msg.packet.addr, self._packet.addr) elif i == 5: - self.assertIsInstance(msg, bt2.message._DiscardedPacketsMessage) + self.assertIsInstance(msg, bt2._DiscardedPacketsMessage) self.assertEqual(msg.stream.addr, self._stream.addr) self.assertEqual(msg.count, 678) self.assertIsNone(msg.stream.cls.default_clock_class) - with self.assertRaises(bt2.NonexistentClockSnapshot): + with self.assertRaisesRegex( + ValueError, + 'such a message has no clock snapshots for this stream class', + ): msg.beginning_default_clock_snapshot - with self.assertRaises(bt2.NonexistentClockSnapshot): + with self.assertRaisesRegex( + ValueError, + 'such a message has no clock snapshots for this stream class', + ): msg.end_default_clock_snapshot elif i == 6: - self.assertIsInstance(msg, bt2.message._StreamEndMessage) + self.assertIsInstance(msg, bt2._StreamEndMessage) self.assertEqual(msg.stream.addr, self._stream.addr) - with self.assertRaises(bt2.NonexistentClockSnapshot): + with self.assertRaisesRegex( + ValueError, 'stream class has no default clock class' + ): msg.default_clock_snapshot else: raise Exception @@ -257,15 +273,15 @@ class AllMessagesTestCase(unittest.TestCase): params = {'with_cc': True, 'with_stream_msgs_clock_snapshots': True} self._src_comp = self._graph.add_component(self._src, 'my_source', params) - 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'] ) msgs = list(self._msg_iter) msg_stream_beg = msgs[0] - self.assertIsInstance(msg_stream_beg, bt2.message._StreamBeginningMessage) + self.assertIsInstance(msg_stream_beg, bt2._StreamBeginningMessage) self.assertEqual(msg_stream_beg.default_clock_snapshot.value, 0) msg_stream_end = msgs[7] - self.assertIsInstance(msg_stream_end, bt2.message._StreamEndMessage) + self.assertIsInstance(msg_stream_end, bt2._StreamEndMessage) self.assertEqual(msg_stream_end.default_clock_snapshot.value, 7)