X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_packet.py;h=aa1c2d6a619cf989b382a93316921291180dcb92;hb=f0a42b33ac3951cd5cb2ee0f66ac04437a681621;hp=65c268e87059ff580907fea6d3c2fb40824173d7;hpb=976c241df475c9d99161a1100517c3c308074d7f;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_packet.py b/tests/bindings/python/bt2/test_packet.py index 65c268e8..aa1c2d6a 100644 --- a/tests/bindings/python/bt2/test_packet.py +++ b/tests/bindings/python/bt2/test_packet.py @@ -1,150 +1,110 @@ -from collections import OrderedDict -from bt2 import values +# +# 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 copy -import bt2 +import utils +from utils import run_in_component_init +from bt2 import stream as bt2_stream +from bt2 import field as bt2_field -@unittest.skip("this is broken") class PacketTestCase(unittest.TestCase): - def setUp(self): - self._packet = self._create_packet() - - def tearDown(self): - del self._packet + @staticmethod + def _create_packet(with_pc): + def create_tc_cc(comp_self): + cc = comp_self._create_clock_class(frequency=1000, name='my_cc') + tc = comp_self._create_trace_class() + return cc, tc - def _create_packet(self, with_ph=True, with_pc=True): - # event header - eh = bt2.StructureFieldType() - eh += OrderedDict(( - ('id', bt2.IntegerFieldType(8)), - ('ts', bt2.IntegerFieldType(32)), - )) + clock_class, tc = run_in_component_init(create_tc_cc) # stream event context - sec = bt2.StructureFieldType() - sec += OrderedDict(( - ('cpu_id', bt2.IntegerFieldType(8)), - ('stuff', bt2.FloatingPointNumberFieldType()), - )) + sec = tc.create_structure_field_class() + sec += [ + ('cpu_id', tc.create_signed_integer_field_class(8)), + ('stuff', tc.create_real_field_class()), + ] # packet context + pc = None if with_pc: - pc = bt2.StructureFieldType() - pc += OrderedDict(( - ('something', bt2.IntegerFieldType(8)), - ('something_else', bt2.FloatingPointNumberFieldType()), - )) - else: - pc = None + pc = tc.create_structure_field_class() + pc += [ + ('something', tc.create_signed_integer_field_class(8)), + ('something_else', tc.create_real_field_class()), + ('events_discarded', tc.create_unsigned_integer_field_class(64)), + ('packet_seq_num', tc.create_unsigned_integer_field_class(64)), + ] # stream class - sc = bt2.StreamClass() - sc.packet_context_field_type = pc - sc.event_header_field_type = eh - sc.event_context_field_type = sec + sc = tc.create_stream_class( + default_clock_class=clock_class, + event_common_context_field_class=sec, + packet_context_field_class=pc, + supports_packets=True, + ) # event context - ec = bt2.StructureFieldType() - ec += OrderedDict(( - ('ant', bt2.IntegerFieldType(16, is_signed=True)), - ('msg', bt2.StringFieldType()), - )) + ec = tc.create_structure_field_class() + ec += [ + ('ant', tc.create_signed_integer_field_class(16)), + ('msg', tc.create_string_field_class()), + ] # event payload - ep = bt2.StructureFieldType() - ep += OrderedDict(( - ('giraffe', bt2.IntegerFieldType(32)), - ('gnu', bt2.IntegerFieldType(8)), - ('mosquito', bt2.IntegerFieldType(8)), - )) + 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)), + ] # event class - event_class = bt2.EventClass('ec') - event_class.context_field_type = ec - event_class.payload_field_type = ep - sc.add_event_class(event_class) - - # packet header - if with_ph: - ph = bt2.StructureFieldType() - ph += OrderedDict(( - ('magic', bt2.IntegerFieldType(32)), - ('stream_id', bt2.IntegerFieldType(16)), - )) - else: - ph = None - - # trace c;ass - tc = bt2.Trace() - tc.packet_header_field_type = ph - tc.add_stream_class(sc) + event_class = sc.create_event_class(name='ec', payload_field_class=ep) + event_class.common_context_field_class = ec + + # trace + trace = tc() # stream - stream = sc() + stream = trace.create_stream(sc) # packet - return stream.create_packet() + return stream.create_packet(), stream, pc def test_attr_stream(self): - self.assertIsNotNone(self._packet.stream) + packet, stream, _ = self._create_packet(with_pc=True) + self.assertEqual(packet.stream.addr, stream.addr) + self.assertIs(type(packet.stream), bt2_stream._Stream) - def test_get_header_field(self): - self.assertIsNotNone(self._packet.header_field) + def test_const_attr_stream(self): + packet = utils.get_const_packet_beginning_message().packet + self.assertIs(type(packet.stream), bt2_stream._StreamConst) - def test_no_header_field(self): - packet = self._create_packet(with_ph=False) - self.assertIsNone(packet.header_field) + def test_context_field(self): + packet, stream, pc_fc = self._create_packet(with_pc=True) + self.assertEqual(packet.context_field.cls.addr, pc_fc.addr) + self.assertIs(type(packet.context_field), bt2_field._StructureField) - def test_get_context_field(self): - self.assertIsNotNone(self._packet.context_field) + def test_const_context_field(self): + packet = utils.get_const_packet_beginning_message().packet + self.assertIs(type(packet.context_field), bt2_field._StructureFieldConst) def test_no_context_field(self): - packet = self._create_packet(with_pc=False) + packet, _, _ = self._create_packet(with_pc=False) self.assertIsNone(packet.context_field) - - def _fill_packet(self, packet): - packet.header_field['magic'] = 0xc1fc1fc1 - packet.header_field['stream_id'] = 23 - packet.context_field['something'] = 17 - packet.context_field['something_else'] = 188.88 - - def test_eq(self): - packet1 = self._create_packet() - self._fill_packet(packet1) - packet2 = self._create_packet() - self._fill_packet(packet2) - self.assertEqual(packet1, packet2) - - def test_ne_header_field(self): - packet1 = self._create_packet() - self._fill_packet(packet1) - packet2 = self._create_packet() - self._fill_packet(packet2) - packet2.header_field['stream_id'] = 18 - self.assertNotEqual(packet1, packet2) - - def test_ne_context_field(self): - packet1 = self._create_packet() - self._fill_packet(packet1) - packet2 = self._create_packet() - self._fill_packet(packet2) - packet2.context_field['something_else'] = 1948.11 - self.assertNotEqual(packet1, packet2) - - def test_eq_invalid(self): - self.assertFalse(self._packet == 23) - - def _test_copy(self, func): - packet = self._create_packet() - self._fill_packet(packet) - cpy = func(packet) - self.assertIsNot(packet, cpy) - self.assertNotEqual(packet.addr, cpy.addr) - self.assertEqual(packet, cpy) - - def test_copy(self): - self._test_copy(copy.copy) - - def test_deepcopy(self): - self._test_copy(copy.deepcopy)