X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_event.py;h=970b7c2210470f879a2780ed1b64d77c239290e3;hb=f5567ea88d172767b34373bc6e402da8bfd85ef8;hp=a770c8f5d58c4316b6f5de9346e509783415b7c6;hpb=fe4df857056b4a03898f1031f136359ce733b0f5;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_event.py b/tests/bindings/python/bt2/test_event.py index a770c8f5..970b7c22 100644 --- a/tests/bindings/python/bt2/test_event.py +++ b/tests/bindings/python/bt2/test_event.py @@ -1,20 +1,7 @@ +# SPDX-License-Identifier: GPL-2.0-only # # 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 bt2 @@ -82,7 +69,7 @@ class EventTestCase(unittest.TestCase): class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter): def __init__(self, config, params, obj): - self._add_output_port('out') + self._add_output_port("out") tc = self._create_trace_class() clock_class = None @@ -94,8 +81,9 @@ class EventTestCase(unittest.TestCase): if with_cc: cc = tc.create_structure_field_class() cc += [ - ('cpu_id', tc.create_signed_integer_field_class(8)), - ('stuff', tc.create_double_precision_real_field_class()), + ("cpu_id", tc.create_signed_integer_field_class(8)), + ("stuff", tc.create_double_precision_real_field_class()), + ("gnu", tc.create_string_field_class()), ] # packet context (stream-class-defined) @@ -104,9 +92,9 @@ class EventTestCase(unittest.TestCase): if with_packet: pc = tc.create_structure_field_class() pc += [ - ('something', tc.create_unsigned_integer_field_class(8)), + ("something", tc.create_unsigned_integer_field_class(8)), ( - 'something_else', + "something_else", tc.create_double_precision_real_field_class(), ), ] @@ -123,8 +111,8 @@ class EventTestCase(unittest.TestCase): if with_sc: sc = tc.create_structure_field_class() sc += [ - ('ant', tc.create_signed_integer_field_class(16)), - ('msg', tc.create_string_field_class()), + ("ant", tc.create_signed_integer_field_class(16)), + ("msg", tc.create_string_field_class()), ] # event payload @@ -132,13 +120,13 @@ class EventTestCase(unittest.TestCase): if with_ep: ep = tc.create_structure_field_class() ep += [ - ('giraffe', tc.create_signed_integer_field_class(32)), - ('gnu', tc.create_signed_integer_field_class(8)), - ('mosquito', tc.create_signed_integer_field_class(8)), + ("giraffe", tc.create_signed_integer_field_class(32)), + ("gnu", tc.create_signed_integer_field_class(8)), + ("mosquito", tc.create_signed_integer_field_class(8)), ] event_class = stream_class.create_event_class( - name='garou', + name="garou", specific_context_field_class=sc, payload_field_class=ep, ) @@ -161,9 +149,9 @@ class EventTestCase(unittest.TestCase): test_obj = self self._graph = bt2.Graph() - self._src_comp = self._graph.add_component(MySrc, 'my_source') + self._src_comp = self._graph.add_component(MySrc, "my_source") self._msg_iter = TestOutputPortMessageIterator( - self._graph, self._src_comp.output_ports['out'] + self._graph, self._src_comp.output_ports["out"] ) for msg in self._msg_iter: @@ -190,15 +178,17 @@ class EventTestCase(unittest.TestCase): def test_const_get_common_context_field(self): def event_fields_config(event): - event.common_context_field['cpu_id'] = 1 - event.common_context_field['stuff'] = 13.194 + event.common_context_field["cpu_id"] = 1 + event.common_context_field["stuff"] = 13.194 + event.common_context_field["gnu"] = "salut" msg = self._create_test_const_event_message( event_fields_config=event_fields_config, with_cc=True ) - self.assertEqual(msg.event.common_context_field['cpu_id'], 1) - self.assertEqual(msg.event.common_context_field['stuff'], 13.194) + self.assertEqual(msg.event.common_context_field["cpu_id"], 1) + self.assertEqual(msg.event.common_context_field["stuff"], 13.194) + self.assertEqual(msg.event.common_context_field["gnu"], "salut") self.assertIs( type(msg.event.common_context_field), bt2_field._StructureFieldConst ) @@ -213,15 +203,15 @@ class EventTestCase(unittest.TestCase): def test_const_get_specific_context_field(self): def event_fields_config(event): - event.specific_context_field['ant'] = -1 - event.specific_context_field['msg'] = 'hellooo' + event.specific_context_field["ant"] = -1 + event.specific_context_field["msg"] = "hellooo" msg = self._create_test_const_event_message( event_fields_config=event_fields_config, with_sc=True ) - self.assertEqual(msg.event.specific_context_field['ant'], -1) - self.assertEqual(msg.event.specific_context_field['msg'], 'hellooo') + self.assertEqual(msg.event.specific_context_field["ant"], -1) + self.assertEqual(msg.event.specific_context_field["msg"], "hellooo") self.assertIs( type(msg.event.specific_context_field), bt2_field._StructureFieldConst ) @@ -236,17 +226,17 @@ class EventTestCase(unittest.TestCase): def test_const_get_event_payload_field(self): def event_fields_config(event): - event.payload_field['giraffe'] = 1 - event.payload_field['gnu'] = 23 - event.payload_field['mosquito'] = 42 + event.payload_field["giraffe"] = 1 + event.payload_field["gnu"] = 23 + event.payload_field["mosquito"] = 42 msg = self._create_test_const_event_message( event_fields_config=event_fields_config, with_ep=True ) - self.assertEqual(msg.event.payload_field['giraffe'], 1) - self.assertEqual(msg.event.payload_field['gnu'], 23) - self.assertEqual(msg.event.payload_field['mosquito'], 42) + self.assertEqual(msg.event.payload_field["giraffe"], 1) + self.assertEqual(msg.event.payload_field["gnu"], 23) + self.assertEqual(msg.event.payload_field["mosquito"], 42) self.assertIs(type(msg.event.payload_field), bt2_field._StructureFieldConst) def test_attr_payload_field(self): @@ -274,7 +264,7 @@ class EventTestCase(unittest.TestCase): def test_const_no_clock_value(self): msg = self._create_test_const_event_message(with_clockclass=False) with self.assertRaisesRegex( - ValueError, 'stream class has no default clock class' + ValueError, "stream class has no default clock class" ): msg.default_clock_snapshot @@ -287,23 +277,30 @@ class EventTestCase(unittest.TestCase): msg = utils.get_event_message() self.assertIs(type(msg.event.stream), bt2_stream._Stream) - def test_const_getitem(self): - def event_fields_config(event): - event.payload_field['giraffe'] = 1 - event.payload_field['gnu'] = 23 - event.payload_field['mosquito'] = 42 - event.specific_context_field['ant'] = -1 - event.specific_context_field['msg'] = 'hellooo' - event.common_context_field['cpu_id'] = 1 - event.common_context_field['stuff'] = 13.194 - - def packet_fields_config(packet): - packet.context_field['something'] = 154 - packet.context_field['something_else'] = 17.2 + @staticmethod + def _event_payload_fields_config(event): + event.payload_field["giraffe"] = 1 + event.payload_field["gnu"] = 23 + event.payload_field["mosquito"] = 42 + + @staticmethod + def _event_fields_config(event): + EventTestCase._event_payload_fields_config(event) + event.specific_context_field["ant"] = -1 + event.specific_context_field["msg"] = "hellooo" + event.common_context_field["cpu_id"] = 1 + event.common_context_field["stuff"] = 13.194 + event.common_context_field["gnu"] = "salut" + + @staticmethod + def _packet_fields_config(packet): + packet.context_field["something"] = 154 + packet.context_field["something_else"] = 17.2 + def test_const_getitem(self): msg = self._create_test_const_event_message( - packet_fields_config=packet_fields_config, - event_fields_config=event_fields_config, + packet_fields_config=self._packet_fields_config, + event_fields_config=self._event_fields_config, with_cc=True, with_sc=True, with_ep=True, @@ -312,36 +309,151 @@ class EventTestCase(unittest.TestCase): ev = msg.event # Test event fields - self.assertEqual(ev['giraffe'], 1) - self.assertIs(type(ev['giraffe']), bt2_field._SignedIntegerFieldConst) - self.assertEqual(ev['gnu'], 23) - self.assertEqual(ev['mosquito'], 42) - self.assertEqual(ev['ant'], -1) - self.assertIs(type(ev['ant']), bt2_field._SignedIntegerFieldConst) - self.assertEqual(ev['msg'], 'hellooo') - self.assertEqual(ev['cpu_id'], 1) - self.assertIs(type(ev['cpu_id']), bt2_field._SignedIntegerFieldConst) - self.assertEqual(ev['stuff'], 13.194) + self.assertEqual(ev["giraffe"], 1) + self.assertIs(type(ev["giraffe"]), bt2_field._SignedIntegerFieldConst) + self.assertEqual(ev["gnu"], 23) + self.assertEqual(ev["mosquito"], 42) + self.assertEqual(ev["ant"], -1) + self.assertIs(type(ev["ant"]), bt2_field._SignedIntegerFieldConst) + self.assertEqual(ev["msg"], "hellooo") + self.assertEqual(ev["cpu_id"], 1) + self.assertIs(type(ev["cpu_id"]), bt2_field._SignedIntegerFieldConst) + self.assertEqual(ev["stuff"], 13.194) # Test packet fields - self.assertEqual(ev['something'], 154) - self.assertIs(type(ev['something']), bt2_field._UnsignedIntegerFieldConst) - self.assertEqual(ev['something_else'], 17.2) + self.assertEqual(ev["something"], 154) + self.assertIs(type(ev["something"]), bt2_field._UnsignedIntegerFieldConst) + self.assertEqual(ev["something_else"], 17.2) + + with self.assertRaises(KeyError): + ev["yes"] + + def test_const_getitem_no_packet(self): + msg = self._create_test_const_event_message( + event_fields_config=self._event_payload_fields_config, + with_ep=True, + ) + ev = msg.event with self.assertRaises(KeyError): - ev['yes'] + ev["yes"] def test_getitem(self): msg = utils.get_event_message() ev = msg.event - self.assertEqual(ev['giraffe'], 1) - self.assertIs(type(ev['giraffe']), bt2_field._SignedIntegerField) - self.assertEqual(ev['ant'], -1) - self.assertIs(type(ev['ant']), bt2_field._SignedIntegerField) - self.assertEqual(ev['cpu_id'], 1) - self.assertIs(type(ev['cpu_id']), bt2_field._SignedIntegerField) - self.assertEqual(ev['something'], 154) - self.assertIs(type(ev['something']), bt2_field._UnsignedIntegerField) + self.assertEqual(ev["giraffe"], 1) + self.assertIs(type(ev["giraffe"]), bt2_field._SignedIntegerField) + self.assertEqual(ev["ant"], -1) + self.assertIs(type(ev["ant"]), bt2_field._SignedIntegerField) + self.assertEqual(ev["cpu_id"], 1) + self.assertIs(type(ev["cpu_id"]), bt2_field._SignedIntegerField) + self.assertEqual(ev["something"], 154) + self.assertIs(type(ev["something"]), bt2_field._UnsignedIntegerField) + + def test_iter_full(self): + msg = self._create_test_const_event_message( + packet_fields_config=self._packet_fields_config, + event_fields_config=self._event_fields_config, + with_cc=True, + with_sc=True, + with_ep=True, + with_packet=True, + ) + expected_field_names = [ + # payload + "giraffe", + "gnu", + "mosquito", + # specific context + "ant", + "msg", + # common context + "cpu_id", + "stuff", + # packet context + "something", + "something_else", + ] + self.assertEqual(list(msg.event), expected_field_names) + + def test_iter_payload_only(self): + msg = self._create_test_const_event_message( + event_fields_config=self._event_payload_fields_config, + with_ep=True, + ) + expected_field_names = [ + # payload + "giraffe", + "gnu", + "mosquito", + ] + self.assertEqual(list(msg.event), expected_field_names) + + def test_len_full(self): + msg = self._create_test_const_event_message( + packet_fields_config=self._packet_fields_config, + event_fields_config=self._event_fields_config, + with_cc=True, + with_sc=True, + with_ep=True, + with_packet=True, + ) + self.assertEqual(len(msg.event), 9) + + def test_len_payload_only(self): + msg = self._create_test_const_event_message( + packet_fields_config=None, + event_fields_config=self._event_payload_fields_config, + with_ep=True, + ) + self.assertEqual(len(msg.event), 3) + + def test_in_full(self): + msg = self._create_test_const_event_message( + packet_fields_config=self._packet_fields_config, + event_fields_config=self._event_fields_config, + with_cc=True, + with_sc=True, + with_ep=True, + with_packet=True, + ) + field_names = [ + # payload + "giraffe", + "gnu", + "mosquito", + # specific context + "ant", + "msg", + # common context + "cpu_id", + "stuff", + # packet context + "something", + "something_else", + ] + + for field_name in field_names: + self.assertTrue(field_name in msg.event) + + self.assertFalse("lol" in msg.event) + + def test_in_payload_only(self): + msg = self._create_test_const_event_message( + packet_fields_config=None, + event_fields_config=self._event_payload_fields_config, + with_ep=True, + ) + field_names = [ + "giraffe", + "gnu", + "mosquito", + ] + + for field_name in field_names: + self.assertTrue(field_name in msg.event) + + self.assertFalse("lol" in msg.event) if __name__ == "__main__":