From: Jérémie Galarneau Date: Wed, 30 Aug 2017 20:02:54 +0000 (-0400) Subject: Tests: add tests for the babeltrace python package X-Git-Url: http://git.efficios.com/?p=babeltrace.git;a=commitdiff_plain;h=b016b06b258cde220c44ec5f32093996868c382f Tests: add tests for the babeltrace python package Signed-off-by: Jérémie Galarneau --- diff --git a/configure.ac b/configure.ac index c4ea8242..da6a5980 100644 --- a/configure.ac +++ b/configure.ac @@ -679,6 +679,7 @@ AC_CONFIG_FILES([ tests/bindings/Makefile tests/bindings/python/Makefile tests/bindings/python/bt2/Makefile + tests/bindings/python/babeltrace/Makefile tests/plugins/Makefile tests/python-plugin-provider/Makefile extras/Makefile @@ -726,7 +727,10 @@ AC_CONFIG_FILES([tests/plugins/test_dwarf_complete], [chmod +x tests/plugins/tes AC_CONFIG_FILES([tests/plugins/test_bin_info_complete], [chmod +x tests/plugins/test_bin_info_complete]) AS_IF([test "x$enable_python_bindings" = xyes], - [AC_CONFIG_FILES([tests/bindings/python/bt2/test_python_bt2], [chmod +x tests/bindings/python/bt2/test_python_bt2])] + [ + AC_CONFIG_FILES([tests/bindings/python/bt2/test_python_bt2], [chmod +x tests/bindings/python/bt2/test_python_bt2]) + AC_CONFIG_FILES([tests/bindings/python/babeltrace/test_python_babeltrace], [chmod +x tests/bindings/python/babeltrace/test_python_babeltrace]) + ] ) AS_IF([test "x$enable_python_plugins" = "xyes"], diff --git a/tests/bindings/python/Makefile.am b/tests/bindings/python/Makefile.am index f3cec072..fbc28130 100644 --- a/tests/bindings/python/Makefile.am +++ b/tests/bindings/python/Makefile.am @@ -1 +1 @@ -SUBDIRS = bt2 +SUBDIRS = bt2 babeltrace diff --git a/tests/bindings/python/babeltrace/Makefile.am b/tests/bindings/python/babeltrace/Makefile.am new file mode 100644 index 00000000..19900b43 --- /dev/null +++ b/tests/bindings/python/babeltrace/Makefile.am @@ -0,0 +1,12 @@ +EXTRA_DIST = \ + test_reader.py \ + test_reader_event.py \ + test_reader_event_declaration.py \ + test_reader_field_declaration.py + +LOG_DRIVER_FLAGS='--merge' +LOG_DRIVER = env AM_TAP_AWK='$(AWK)' $(SHELL) $(top_srcdir)/config/tap-driver.sh + +if ENABLE_PYTHON_BINDINGS +TESTS = test_python_babeltrace +endif diff --git a/tests/bindings/python/babeltrace/test_python_babeltrace b/tests/bindings/python/babeltrace/test_python_babeltrace new file mode 100755 index 00000000..bcf0df4a --- /dev/null +++ b/tests/bindings/python/babeltrace/test_python_babeltrace @@ -0,0 +1,57 @@ +#!/bin/bash +# +# Copyright (C) 2017 - Philippe Proulx +# 2017 - Jérémie Galarneau +# +# 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. +# + +NO_SH_TAP=1 +. "/home/jgalar/EfficiOS/src/babeltrace/tests/utils/common.sh" + +check_coverage() { + coverage run "$@" +} + +export BABELTRACE_PYTHON_BT2_NO_TRACEBACK=1 +export TEST_PLUGIN_PLUGINS_PATH="${BT_BUILD_PATH}/plugins" +export BABELTRACE_PLUGIN_PATH="${BT_BUILD_PATH}/plugins/ctf:${BT_BUILD_PATH}/plugins/utils:${BT_BUILD_PATH}/plugins/text" +export LD_LIBRARY_PATH="${BT_BUILD_PATH}/lib/.libs" +PYTHON_BUILD_DIR="${BT_BUILD_PATH}/bindings/python/bt2/build/build_lib:${BT_BUILD_PATH}/bindings/python/babeltrace/build/build_lib" +TESTS_UTILS_PYTHON_DIR="${BT_SRC_PATH}/tests/utils/python" +TESTS_CTF_TRACES="${BT_SRC_PATH}/tests/ctf-traces" +TESTRUNNER_PY="${BT_SRC_PATH}/tests/utils/python/testrunner.py" +THIS_DIR="${BT_SRC_PATH}/tests/bindings/python/babeltrace" + +if test "x$TESTALL_COVERAGE" = "x1"; then + EXEC=check_coverage +else + EXEC="/usr/bin/python" + +fi + +PYTHONPATH="$PYTHON_BUILD_DIR:$TESTS_UTILS_PYTHON_DIR" \ + "$EXEC" "$TESTRUNNER_PY" "$THIS_DIR" +res=$? + +if test "x$TESTALL_COVERAGE_REPORT" = "x1"; then + coverage report -m +fi + +if test "x$TESTALL_COVERAGE_HTML" = "x1"; then + coverage html +fi + +exit $res diff --git a/tests/bindings/python/babeltrace/test_python_babeltrace.in b/tests/bindings/python/babeltrace/test_python_babeltrace.in new file mode 100644 index 00000000..fec9e033 --- /dev/null +++ b/tests/bindings/python/babeltrace/test_python_babeltrace.in @@ -0,0 +1,57 @@ +#!/bin/bash +# +# Copyright (C) 2017 - Philippe Proulx +# 2017 - Jérémie Galarneau +# +# 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. +# + +NO_SH_TAP=1 +. "@abs_top_builddir@/tests/utils/common.sh" + +check_coverage() { + coverage run "$@" +} + +export BABELTRACE_PYTHON_BT2_NO_TRACEBACK=1 +export TEST_PLUGIN_PLUGINS_PATH="${BT_BUILD_PATH}/plugins" +export BABELTRACE_PLUGIN_PATH="${BT_BUILD_PATH}/plugins/ctf:${BT_BUILD_PATH}/plugins/utils:${BT_BUILD_PATH}/plugins/text" +export LD_LIBRARY_PATH="${BT_BUILD_PATH}/lib/.libs" +PYTHON_BUILD_DIR="${BT_BUILD_PATH}/bindings/python/bt2/build/build_lib:${BT_BUILD_PATH}/bindings/python/babeltrace/build/build_lib" +TESTS_UTILS_PYTHON_DIR="${BT_SRC_PATH}/tests/utils/python" +TESTS_CTF_TRACES="${BT_SRC_PATH}/tests/ctf-traces" +TESTRUNNER_PY="${BT_SRC_PATH}/tests/utils/python/testrunner.py" +THIS_DIR="${BT_SRC_PATH}/tests/bindings/python/babeltrace" + +if test "x$TESTALL_COVERAGE" = "x1"; then + EXEC=check_coverage +else + EXEC="@PYTHON@" + +fi + +PYTHONPATH="$PYTHON_BUILD_DIR:$TESTS_UTILS_PYTHON_DIR" \ + "$EXEC" "$TESTRUNNER_PY" "$THIS_DIR" +res=$? + +if test "x$TESTALL_COVERAGE_REPORT" = "x1"; then + coverage report -m +fi + +if test "x$TESTALL_COVERAGE_HTML" = "x1"; then + coverage html +fi + +exit $res diff --git a/tests/bindings/python/babeltrace/test_reader.py b/tests/bindings/python/babeltrace/test_reader.py new file mode 100644 index 00000000..cd6adc32 --- /dev/null +++ b/tests/bindings/python/babeltrace/test_reader.py @@ -0,0 +1,2 @@ +import babeltrace +import unittest diff --git a/tests/bindings/python/babeltrace/test_reader_event.py b/tests/bindings/python/babeltrace/test_reader_event.py new file mode 100644 index 00000000..1ea1ce51 --- /dev/null +++ b/tests/bindings/python/babeltrace/test_reader_event.py @@ -0,0 +1,195 @@ +# The MIT License (MIT) +# +# Copyright (c) 2017 Jérémie Galarneau +# Copyright (c) 2017 Philippe Proulx +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. + +import collections +import unittest +import bt2 +import babeltrace +import datetime + + +class EventTestCase(unittest.TestCase): + def setUp(self): + self._values = { + 'ph_field_1' : 42, + 'ph_field_2' : 'bla bla', + 'spc_field' : 'some string', + 'seh_field' : 'another string', + 'sec_field' : 68752, + 'ec_field' : 89, + 'ef_field' : 8476, + } + + self._int_ft = bt2.IntegerFieldType(32) + self._str_ft = bt2.StringFieldType() + + self._trace = bt2.Trace() + self._trace.packet_header_field_type = bt2.StructureFieldType() + self._trace.packet_header_field_type += collections.OrderedDict([ + ('ph_field_1', self._int_ft), + ('ph_field_2', self._str_ft), + ]) + + self._sc = bt2.StreamClass() + self._sc.packet_context_field_type = bt2.StructureFieldType() + self._sc.packet_context_field_type += collections.OrderedDict([ + ('spc_field', self._str_ft), + ]) + + self._sc.event_header_field_type = bt2.StructureFieldType() + self._sc.event_header_field_type += collections.OrderedDict([ + ('seh_field', self._str_ft), + ]) + + self._sc.event_context_field_type = bt2.StructureFieldType() + self._sc.event_context_field_type += collections.OrderedDict([ + ('sec_field', self._int_ft), + ]) + + self._clock_class = bt2.ClockClass('allo', 1000) + self._trace.add_clock_class(self._clock_class) + + self._ec = bt2.EventClass('event_class_name') + self._ec.context_field_type = bt2.StructureFieldType() + self._ec.context_field_type += collections.OrderedDict([ + ('ec_field', self._int_ft), + ]) + self._ec.payload_field_type = bt2.StructureFieldType() + self._ec.payload_field_type += collections.OrderedDict([ + ('ef_field', self._int_ft), + ]) + + self._sc.add_event_class(self._ec) + + self._trace.add_stream_class(self._sc) + self._cc_prio_map = bt2.ClockClassPriorityMap() + self._cc_prio_map[self._clock_class] = 231 + self._stream = self._sc() + self._packet = self._stream.create_packet() + self._packet.header_field['ph_field_1'] = self._values['ph_field_1'] + self._packet.header_field['ph_field_2'] = self._values['ph_field_2'] + self._packet.context_field['spc_field'] = self._values['spc_field'] + + self._event = self._ec() + self._event.add_clock_value(self._clock_class(1772)) + self._event.header_field['seh_field'] = self._values['seh_field'] + self._event.stream_event_context_field['sec_field'] = self._values[ + 'sec_field'] + self._event.context_field['ec_field'] = self._values['ec_field'] + self._event.payload_field['ef_field'] = self._values['ef_field'] + self._event.packet = self._packet + + def tearDown(self): + del self._trace + del self._sc + del self._ec + del self._int_ft + del self._str_ft + del self._clock_class + del self._cc_prio_map + del self._stream + del self._packet + del self._event + + def _get_event(self): + notif = bt2.EventNotification(self._event, self._cc_prio_map) + return babeltrace.reader_event._create_event(notif) + + def test_attr_name(self): + event = self._get_event() + self.assertEqual(event.name, 'event_class_name') + + def test_attr_cycles(self): + event = self._get_event() + self.assertEqual(event.cycles, 1772) + + def test_attr_timestamp(self): + event = self._get_event() + clock_class = self._cc_prio_map.highest_priority_clock_class + self.assertEqual(event.timestamp, 1772 * (1E9 / clock_class.frequency)) + + def test_attr_datetime(self): + event = self._get_event() + clock_class = self._cc_prio_map.highest_priority_clock_class + ns = self._event.clock_value(clock_class).ns_from_epoch + self.assertEqual(datetime.date.fromtimestamp(ns / 1E9), event.datetime) + + def test_getitem(self): + event = self._get_event() + for name, value in self._values.items(): + self.assertEqual(event[name], value) + + with self.assertRaises(KeyError): + field = event['non-existant-key'] + + def test_field_list_with_scope(self): + event = self._get_event() + self.assertEqual( + set(event.field_list_with_scope( + babeltrace.CTFScope.TRACE_PACKET_HEADER)), + set(['ph_field_1', 'ph_field_2'])) + + self.assertEqual( + set(event.field_list_with_scope( + babeltrace.CTFScope.STREAM_PACKET_CONTEXT)), + set(['spc_field'])) + + self.assertEqual( + set(event.field_list_with_scope( + babeltrace.CTFScope.STREAM_EVENT_HEADER)), + set(['seh_field'])) + + self.assertEqual( + set(event.field_list_with_scope( + babeltrace.CTFScope.STREAM_EVENT_CONTEXT)), + set(['sec_field'])) + + self.assertEqual( + set(event.field_list_with_scope( + babeltrace.CTFScope.EVENT_CONTEXT)), + set(['ec_field'])) + + self.assertEqual( + set(event.field_list_with_scope( + babeltrace.CTFScope.EVENT_FIELDS)), + set(['ef_field'])) + + def test_field_with_scope(self): + event = self._get_event() + self.assertEqual(event.field_with_scope( + 'seh_field', babeltrace.CTFScope.STREAM_EVENT_HEADER), + self._values['seh_field']) + + def test_get(self): + event = self._get_event() + self.assertEqual(event.get('spc_field'), self._values['spc_field']) + self.assertEqual(event.get('non-existant field', 'No field'), + 'No field') + + def test_keys(self): + event = self._get_event() + self.assertEqual(set(self._values.keys()), set(event.keys())) + + def test_len(self): + event = self._get_event() + self.assertEqual(len(self._values), len(event)) diff --git a/tests/bindings/python/babeltrace/test_reader_event_declaration.py b/tests/bindings/python/babeltrace/test_reader_event_declaration.py new file mode 100644 index 00000000..2664ee56 --- /dev/null +++ b/tests/bindings/python/babeltrace/test_reader_event_declaration.py @@ -0,0 +1,165 @@ +# The MIT License (MIT) +# +# Copyright (c) 2017 Jérémie Galarneau +# Copyright (c) 2017 Philippe Proulx +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. + +import collections +import unittest +import bt2 +import babeltrace +import babeltrace.reader_event_declaration as event_declaration + +class EventDeclarationTestCase(unittest.TestCase): + def setUp(self): + self._values = { + 'ph_field_1' : 42, + 'ph_field_2' : 'bla bla', + 'spc_field' : 'some string', + 'seh_field' : 'another string', + 'sec_field' : 68752, + 'ec_field' : 89, + 'ef_field' : 8476, + } + + self._int_ft = bt2.IntegerFieldType(32) + self._str_ft = bt2.StringFieldType() + + self._trace = bt2.Trace() + self._trace.packet_header_field_type = bt2.StructureFieldType() + self._trace.packet_header_field_type += collections.OrderedDict([ + ('ph_field_1', self._int_ft), + ('ph_field_2', self._str_ft), + ]) + + self._sc = bt2.StreamClass() + self._sc.packet_context_field_type = bt2.StructureFieldType() + self._sc.packet_context_field_type += collections.OrderedDict([ + ('spc_field', self._str_ft), + ]) + + self._sc.event_header_field_type = bt2.StructureFieldType() + self._sc.event_header_field_type += collections.OrderedDict([ + ('seh_field', self._str_ft), + ]) + + self._sc.event_context_field_type = bt2.StructureFieldType() + self._sc.event_context_field_type += collections.OrderedDict([ + ('sec_field', self._int_ft), + ]) + + self._clock_class = bt2.ClockClass('allo', 1000) + self._trace.add_clock_class(self._clock_class) + + self._ec = bt2.EventClass('event_class_name') + self._ec.id = 42 + self._ec.context_field_type = bt2.StructureFieldType() + self._ec.context_field_type += collections.OrderedDict([ + ('ec_field', self._int_ft), + ]) + self._ec.payload_field_type = bt2.StructureFieldType() + self._ec.payload_field_type += collections.OrderedDict([ + ('ef_field', self._int_ft), + ]) + + self._sc.add_event_class(self._ec) + + self._trace.add_stream_class(self._sc) + self._cc_prio_map = bt2.ClockClassPriorityMap() + self._cc_prio_map[self._clock_class] = 231 + self._stream = self._sc() + self._packet = self._stream.create_packet() + self._packet.header_field['ph_field_1'] = self._values['ph_field_1'] + self._packet.header_field['ph_field_2'] = self._values['ph_field_2'] + self._packet.context_field['spc_field'] = self._values['spc_field'] + + def tearDown(self): + del self._trace + del self._sc + del self._ec + del self._int_ft + del self._str_ft + del self._clock_class + del self._cc_prio_map + del self._stream + del self._packet + + def _get_event_declaration(self): + return event_declaration._create_event_declaration(self._ec) + + def test_name(self): + declaration = self._get_event_declaration() + self.assertEqual(declaration.name, 'event_class_name') + + def test_id(self): + declaration = self._get_event_declaration() + self.assertEqual(declaration.id, 42) + + def test_fields(self): + declaration = self._get_event_declaration() + fields = declaration.fields + self.assertEqual(len(list(fields)), len(self._values)) + + def test_fields_scope(self): + declaration = self._get_event_declaration() + event_fields = list( + declaration.fields_scope(babeltrace.CTFScope.EVENT_FIELDS)) + self.assertEqual(len(event_fields), 1) + self.assertEqual(event_fields[0].name, 'ef_field') + self.assertEqual(event_fields[0].scope, + babeltrace.CTFScope.EVENT_FIELDS) + + event_ctx_fields = list( + declaration.fields_scope(babeltrace.CTFScope.EVENT_CONTEXT)) + self.assertEqual(len(event_ctx_fields), 1) + self.assertEqual(event_ctx_fields[0].name, 'ec_field') + self.assertEqual(event_ctx_fields[0].scope, + babeltrace.CTFScope.EVENT_CONTEXT) + + stream_ectx_fields = list( + declaration.fields_scope(babeltrace.CTFScope.STREAM_EVENT_CONTEXT)) + self.assertEqual(len(stream_ectx_fields), 1) + self.assertEqual(stream_ectx_fields[0].name, 'sec_field') + self.assertEqual(stream_ectx_fields[0].scope, + babeltrace.CTFScope.STREAM_EVENT_CONTEXT) + + stream_eh_fields = list( + declaration.fields_scope(babeltrace.CTFScope.STREAM_EVENT_HEADER)) + self.assertEqual(len(stream_eh_fields), 1) + self.assertEqual(stream_eh_fields[0].name, 'seh_field') + self.assertEqual(stream_eh_fields[0].scope, + babeltrace.CTFScope.STREAM_EVENT_HEADER) + + stream_pctx_fields = list( + declaration.fields_scope(babeltrace.CTFScope.STREAM_PACKET_CONTEXT)) + self.assertEqual(len(stream_pctx_fields), 1) + self.assertEqual(stream_pctx_fields[0].name, 'spc_field') + self.assertEqual(stream_pctx_fields[0].scope, + babeltrace.CTFScope.STREAM_PACKET_CONTEXT) + + stream_ph_fields = list( + declaration.fields_scope(babeltrace.CTFScope.TRACE_PACKET_HEADER)) + self.assertEqual(len(stream_ph_fields), 2) + self.assertEqual(stream_ph_fields[0].name, 'ph_field_1') + self.assertEqual(stream_ph_fields[0].scope, + babeltrace.CTFScope.TRACE_PACKET_HEADER) + self.assertEqual(stream_ph_fields[1].name, 'ph_field_2') + self.assertEqual(stream_ph_fields[1].scope, + babeltrace.CTFScope.TRACE_PACKET_HEADER) diff --git a/tests/bindings/python/babeltrace/test_reader_field_declaration.py b/tests/bindings/python/babeltrace/test_reader_field_declaration.py new file mode 100644 index 00000000..baa15be3 --- /dev/null +++ b/tests/bindings/python/babeltrace/test_reader_field_declaration.py @@ -0,0 +1,172 @@ +# The MIT License (MIT) +# +# Copyright (c) 2017 Jérémie Galarneau +# Copyright (c) 2017 Philippe Proulx +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. + +import collections +import unittest +import copy +import bt2 +import babeltrace +import babeltrace.reader_field_declaration as field_declaration +import datetime +import random + + +class FieldDeclarationTestCase(unittest.TestCase): + def setUp(self): + pass + + def tearDown(self): + pass + + @staticmethod + def _get_declaration(field_type): + return field_declaration._create_field_declaration( + field_type, 'a_name', babeltrace.CTFScope.TRACE_PACKET_HEADER) + + def test_type(self): + int_ft = bt2.IntegerFieldType(32) + + enum_ft = bt2.EnumerationFieldType(int_ft) + enum_ft.append_mapping('corner', 23) + enum_ft.append_mapping('zoom', 17, 20) + enum_ft.append_mapping('mellotron', 1001) + enum_ft.append_mapping('giorgio', 2000, 3000) + + array_ft = bt2.ArrayFieldType(int_ft, 5) + seq_ft = bt2.SequenceFieldType(int_ft, 'the_len_field') + float_ft = bt2.FloatingPointNumberFieldType() + + struct_ft = bt2.StructureFieldType() + struct_ft.append_field('a', int_ft) + struct_ft.append_field('b', int_ft) + struct_ft.append_field('c', int_ft) + + _string_ft = bt2.StringFieldType() + + variant_ft = bt2.VariantFieldType('tag', enum_ft) + variant_ft.append_field('corner', int_ft) + variant_ft.append_field('zoom', array_ft) + variant_ft.append_field('mellotron', float_ft) + variant_ft.append_field('giorgio', struct_ft) + + expected_types = { + babeltrace.CTFTypeId.INTEGER: int_ft, + babeltrace.CTFTypeId.FLOAT: float_ft, + babeltrace.CTFTypeId.ENUM: enum_ft, + babeltrace.CTFTypeId.STRING: _string_ft, + babeltrace.CTFTypeId.STRUCT: struct_ft, + babeltrace.CTFTypeId.VARIANT: variant_ft, + babeltrace.CTFTypeId.ARRAY: array_ft, + babeltrace.CTFTypeId.SEQUENCE: seq_ft, + } + + for type_id, ft in expected_types.items(): + declaration = self._get_declaration(ft) + self.assertIsNotNone(declaration) + self.assertEqual(declaration.type, type_id) + + def test_int_signedness(self): + int_ft = bt2.IntegerFieldType(size=32, is_signed=True) + declaration = self._get_declaration(int_ft) + self.assertEqual(declaration.signedness, 1) + + uint_ft = bt2.IntegerFieldType(size=32, is_signed=False) + declaration = self._get_declaration(uint_ft) + self.assertEqual(declaration.signedness, 0) + + def test_int_base(self): + int_ft = bt2.IntegerFieldType(size=32, base=8) + declaration = self._get_declaration(int_ft) + self.assertEqual(declaration.base, 8) + + int_ft = bt2.IntegerFieldType(size=32, base=16) + declaration = self._get_declaration(int_ft) + self.assertEqual(declaration.base, 16) + + int_ft = bt2.IntegerFieldType(size=32, base=10) + declaration = self._get_declaration(int_ft) + self.assertEqual(declaration.base, 10) + + int_ft = bt2.IntegerFieldType(size=32, base=2) + declaration = self._get_declaration(int_ft) + self.assertEqual(declaration.base, 2) + + def test_int_byte_order(self): + expected_byte_orders = { + bt2.ByteOrder.NATIVE: babeltrace.ByteOrder.BYTE_ORDER_NATIVE, + bt2.ByteOrder.LITTLE_ENDIAN: babeltrace.ByteOrder.BYTE_ORDER_LITTLE_ENDIAN, + bt2.ByteOrder.BIG_ENDIAN: babeltrace.ByteOrder.BYTE_ORDER_BIG_ENDIAN, + bt2.ByteOrder.NETWORK: babeltrace.ByteOrder.BYTE_ORDER_NETWORK, + } + + for bt2_bo, bt_bo in expected_byte_orders.items(): + int_ft = bt2.IntegerFieldType(size=32, byte_order=bt2_bo) + declaration = self._get_declaration(int_ft) + self.assertEqual(declaration.byte_order, bt_bo) + + def test_int_size(self): + int_ft = bt2.IntegerFieldType(size=32, base=8) + declaration = self._get_declaration(int_ft) + self.assertEqual(declaration.size, 32) + self.assertEqual(declaration.length, 32) + + int_ft = bt2.IntegerFieldType(size=12, base=8) + declaration = self._get_declaration(int_ft) + self.assertEqual(declaration.size, 12) + self.assertEqual(declaration.length, 12) + + def test_int_encoding(self): + expected_encodings = { + bt2.Encoding.NONE: babeltrace.CTFStringEncoding.NONE, + bt2.Encoding.ASCII: babeltrace.CTFStringEncoding.ASCII, + bt2.Encoding.UTF8: babeltrace.CTFStringEncoding.UTF8, + } + + for bt2_encoding, bt_encoding in expected_encodings.items(): + int_ft = bt2.IntegerFieldType(size=32, encoding=bt2_encoding) + declaration = self._get_declaration(int_ft) + self.assertEqual(declaration.encoding, bt_encoding) + + def test_array_length(self): + int_ft = bt2.IntegerFieldType(32) + array_ft = bt2.ArrayFieldType(int_ft, 5) + declaration = self._get_declaration(array_ft) + self.assertEqual(declaration.length, 5) + + def test_array_element_declaration(self): + int_ft = bt2.IntegerFieldType(size=32, is_signed=True, base=8) + array_ft = bt2.ArrayFieldType(int_ft, 5) + declaration = self._get_declaration(array_ft) + element_declaration = declaration.element_declaration + self.assertEqual(element_declaration.type, babeltrace.CTFTypeId.INTEGER) + self.assertEqual(element_declaration.size, 32) + self.assertEqual(element_declaration.base, 8) + + def test_sequence_element_declaration(self): + int_ft = bt2.IntegerFieldType(size=32, is_signed=True, base=8) + seq_ft = bt2.SequenceFieldType(int_ft, 'len_field') + declaration = self._get_declaration(seq_ft) + element_declaration = declaration.element_declaration + self.assertEqual(element_declaration.type, babeltrace.CTFTypeId.INTEGER) + self.assertEqual(element_declaration.size, 32) + self.assertEqual(element_declaration.base, 8)