X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_packet.py;h=5448c03c8fa5df893a5670d7924d92009790b767;hb=f5567ea88d172767b34373bc6e402da8bfd85ef8;hp=51391cfb16ca0bda3e3986df2a5508b1f85c2070;hpb=811644b8fe5fb9946972a7ace9df02ed872f448a;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_packet.py b/tests/bindings/python/bt2/test_packet.py index 51391cfb..5448c03c 100644 --- a/tests/bindings/python/bt2/test_packet.py +++ b/tests/bindings/python/bt2/test_packet.py @@ -1,149 +1,101 @@ -from collections import OrderedDict -from bt2 import values +# SPDX-License-Identifier: GPL-2.0-only +# +# Copyright (C) 2019 EfficiOS Inc. +# + 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 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_double_precision_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_double_precision_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) + +if __name__ == "__main__": + unittest.main()