diff --git a/src/modules/Makefile.am b/src/modules/Makefile.am index 5cc53d53..e0bc2997 100644 --- a/src/modules/Makefile.am +++ b/src/modules/Makefile.am @@ -30,6 +30,7 @@ if HAVE_PYTHON speechd_python_modules_pythondir = $(pythondir)/speechd_python_modules dist_speechd_python_modules_python_PYTHON = \ speechd_python_modules/__init__.py \ + speechd_python_modules/module_readline.py \ speechd_python_modules/module_utils.py \ speechd_python_modules/speechd_types.py endif diff --git a/src/modules/speechd_python_modules/module_readline.py b/src/modules/speechd_python_modules/module_readline.py new file mode 100644 index 00000000..7681f1fe --- /dev/null +++ b/src/modules/speechd_python_modules/module_readline.py @@ -0,0 +1,131 @@ +# +# module_readline.py - Input buffering for Python output modules. +# +# Copyright (C) 2020 Samuel Thibault +# Copyright (C) 2026 Jean-François David +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions +# are met: +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS ``AS IS'' AND +# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE +# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL +# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS +# OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) +# HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT +# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY +# OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF +# SUCH DAMAGE. +# + +import os +import select +import sys + + +READ_CHUNK = 4096 + +_fd_buffers = {} + + +class _ReadBuffer: + def __init__(self): + self.data = bytearray() + self.no_lf = 0 + +def module_readline(source=None, block=True): + if source is None: + source = sys.stdin + if isinstance(source, int): + return _readline_fd(source, block) + + fd = _source_fd(source) + if fd is not None: + return _readline_fd(fd, block) + + if not block: + return None + + line = source.readline() + return _decode_complete_line(line) + + +def _readline_fd(fd, block): + state = _fd_buffers.get(fd) + + while True: + if state is not None: + newline = state.data.find(b"\n", state.no_lf) + if newline != -1: + line = bytes(state.data[: newline + 1]) + del state.data[: newline + 1] + state.no_lf = 0 + if not state.data: + _fd_buffers.pop(fd, None) + return _decode_bytes(line) + + state.no_lf = len(state.data) + + try: + readable, _, _ = select.select([fd], [], [], None if block else 0) + except (InterruptedError, BlockingIOError): + if not block: + return None + continue + except OSError: + _fd_buffers.pop(fd, None) + return None + + if not readable: + return None + + try: + chunk = os.read(fd, READ_CHUNK) + except (InterruptedError, BlockingIOError): + if not block: + return None + continue + except OSError: + _fd_buffers.pop(fd, None) + return None + + if not chunk: + if state is not None: + _fd_buffers.pop(fd, None) + return None + + if state is None: + state = _ReadBuffer() + _fd_buffers[fd] = state + + state.data.extend(chunk) + + +def _source_fd(source): + try: + return source.fileno() + except (AttributeError, OSError, ValueError): + return None + + +def _decode_complete_line(line): + if not line: + return None + if not line.endswith(b"\n" if isinstance(line, bytes) else "\n"): + return None + if isinstance(line, bytes): + return _decode_bytes(line) + return line + + +def _decode_bytes(data): + return data.decode("utf-8", "surrogateescape") diff --git a/src/modules/speechd_python_modules/module_utils.py b/src/modules/speechd_python_modules/module_utils.py index a4b26d6a..184500e3 100644 --- a/src/modules/speechd_python_modules/module_utils.py +++ b/src/modules/speechd_python_modules/module_utils.py @@ -27,6 +27,58 @@ # +import re + + +log_level = 0 +Debug = 0 +CustomDebugFile = None + + +def module_loglevel_set(cur_item, cur_value): + global log_level + + if cur_item != "log_level": + return -1 + + match = re.match(r"\s*([+-]?[0-9]+)", cur_value) + if match is None: + return -1 + + log_level = int(match.group(1), 10) + return 0 + + +# TODO Add an equivalent of C MSG() that writes to CustomDebugFile. +def module_debug(enable, filename): + global CustomDebugFile, Debug + + if enable: + try: + new_custom_debug_file = open(filename, "w+") + except OSError: + return -1 + + if CustomDebugFile is not None: + CustomDebugFile.close() + CustomDebugFile = new_custom_debug_file + if Debug == 1: + Debug = 3 + else: + Debug = 2 + else: + if Debug == 3: + Debug = 1 + else: + Debug = 0 + + if CustomDebugFile is not None: + CustomDebugFile.close() + CustomDebugFile = None + + return 0 + + def module_strip_ssml(message: str) -> str: out = [] append = out.append diff --git a/src/tests/Makefile.am b/src/tests/Makefile.am index e03091ee..7e38714d 100644 --- a/src/tests/Makefile.am +++ b/src/tests/Makefile.am @@ -43,6 +43,11 @@ AUTOM4TE = autom4te AUTOTEST = $(AUTOM4TE) --language=autotest TESTSUITE_AT = c_api.at python_module.at +PYTHON_TESTS = \ + python/test_imports.py \ + python/test_module_readline.py \ + python/test_module_utils.py \ + python/test_speechd_types.py TESTSUITE = ./testsuite $(TESTSUITE): package.m4 testsuite.at $(TESTSUITE_AT) $(AUTOTEST) -I '$(srcdir)' -o $@.tmp $@.at @@ -79,7 +84,7 @@ run_test_LDADD = $(c_api)/libspeechd.la $(GLIB_LIBS) $(EXTRA_SOCKET_LIBS) EXTRA_DIST= basic.test general.test keys.test priority_progress.test \ pronunciation.test punctuation.test sound_icons.test spelling.test \ ssml.test stop_and_pause.test voices.test yo.wav \ - atlocal.in testsuite.at $(TESTSUITE_AT) sayfortune.sh + atlocal.in testsuite.at $(TESTSUITE_AT) $(PYTHON_TESTS) sayfortune.sh clean-local: test ! -f $(TESTSUITE) || $(SHELL) $(TESTSUITE) --clean diff --git a/src/tests/python/test_imports.py b/src/tests/python/test_imports.py new file mode 100644 index 00000000..1b37e332 --- /dev/null +++ b/src/tests/python/test_imports.py @@ -0,0 +1,42 @@ +# +# test_imports.py - Python module import tests +# +# Copyright (C) 2026 Jean-François David +# +# This 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; either version 2, or (at your option) +# any later version. +# +# This software 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, see . + +import os +import unittest + +from speechd_python_modules import module_readline, module_utils, speechd_types + + +class ImportsTest(unittest.TestCase): + def test_modules_are_imported_from_package_directory(self): + module_root = os.environ.get( + "TEST_PYTHONPATH", + os.path.join(os.path.dirname(__file__), "..", "..", "modules"), + ) + package_dir = os.path.join(module_root, "speechd_python_modules") + expected_dir = os.path.realpath(package_dir) + modules = [module_readline, module_utils, speechd_types] + + for module in modules: + with self.subTest(module=module.__name__): + module_file = os.path.realpath(module.__file__) + self.assertEqual(os.path.dirname(module_file), expected_dir) + + +if __name__ == "__main__": + unittest.main() diff --git a/src/tests/python/test_module_readline.py b/src/tests/python/test_module_readline.py new file mode 100644 index 00000000..326c726a --- /dev/null +++ b/src/tests/python/test_module_readline.py @@ -0,0 +1,105 @@ +# +# test_module_readline.py - Python module_readline unit tests +# +# Copyright (C) 2026 Jean-François David +# +# This 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; either version 2, or (at your option) +# any later version. +# +# This software 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, see . + +import io +import os +import unittest + +import speechd_python_modules.module_readline as module_readline + + +class ModuleReadlineTest(unittest.TestCase): + def setUp(self): + self._fds = [] + + def tearDown(self): + for fd in self._fds: + module_readline._fd_buffers.pop(fd, None) + for fd in reversed(self._fds): + try: + os.close(fd) + except OSError: + pass + + def pipe(self): + read_fd, write_fd = os.pipe() + self._fds.extend((read_fd, write_fd)) + return read_fd, write_fd + + def close_fd(self, fd): + os.close(fd) + self._fds.remove(fd) + module_readline._fd_buffers.pop(fd, None) + + def test_nonblocking_empty_fd_returns_none(self): + read_fd, _write_fd = self.pipe() + + self.assertIsNone(module_readline.module_readline(read_fd, block=False)) + + def test_nonblocking_partial_line_is_buffered(self): + read_fd, write_fd = self.pipe() + + os.write(write_fd, b"partial") + self.assertIsNone(module_readline.module_readline(read_fd, block=False)) + + os.write(write_fd, b"\nnext\n") + self.assertEqual( + module_readline.module_readline(read_fd, block=False), + "partial\n", + ) + self.assertEqual( + module_readline.module_readline(read_fd, block=False), + "next\n", + ) + + def test_eof_with_partial_line_returns_none(self): + read_fd, write_fd = self.pipe() + + os.write(write_fd, b"partial") + self.close_fd(write_fd) + + self.assertIsNone(module_readline.module_readline(read_fd, block=True)) + self.assertNotIn(read_fd, module_readline._fd_buffers) + + def test_invalid_utf8_round_trips_with_surrogateescape(self): + read_fd, write_fd = self.pipe() + + os.write(write_fd, b"bad\xff\n") + + line = module_readline.module_readline(read_fd, block=True) + self.assertEqual(line, "bad\udcff\n") + self.assertEqual(line.encode("utf-8", "surrogateescape"), b"bad\xff\n") + + def test_file_like_source_uses_readline_fallback(self): + source = io.StringIO("hello\n") + + self.assertEqual(module_readline.module_readline(source), "hello\n") + + def test_incomplete_file_like_line_returns_none(self): + source = io.StringIO("partial") + + self.assertIsNone(module_readline.module_readline(source)) + + def test_nonblocking_file_like_source_without_fd_returns_none(self): + source = io.StringIO("hello\n") + + self.assertIsNone(module_readline.module_readline(source, block=False)) + + +if __name__ == "__main__": + unittest.main() diff --git a/src/tests/python/test_module_utils.py b/src/tests/python/test_module_utils.py new file mode 100644 index 00000000..fb5eea63 --- /dev/null +++ b/src/tests/python/test_module_utils.py @@ -0,0 +1,135 @@ +# +# test_module_utils.py - Python module_utils unit tests +# +# Copyright (C) 2026 Jean-François David +# +# This 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; either version 2, or (at your option) +# any later version. +# +# This software 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, see . + +import os +import tempfile +import unittest + +from speechd_python_modules import module_utils + + +class ModuleUtilsTest(unittest.TestCase): + def setUp(self): + self.reset_debug_state() + module_utils.log_level = 0 + + def tearDown(self): + self.reset_debug_state() + module_utils.log_level = 0 + + def reset_debug_state(self): + if module_utils.CustomDebugFile is not None: + module_utils.CustomDebugFile.close() + module_utils.CustomDebugFile = None + module_utils.Debug = 0 + + def test_module_strip_ssml_removes_tags(self): + strip_ssml = module_utils.module_strip_ssml + + self.assertEqual(strip_ssml("Plain text."), "Plain text.") + self.assertEqual( + strip_ssml("Hello world."), + "Hello world.", + ) + self.assertEqual( + strip_ssml('Hello world'), + "Hello world", + ) + self.assertEqual(strip_ssml('text'), "text") + self.assertEqual(strip_ssml("Keep , &, \"quotes\" and 'apostrophes'.", + ) + + def test_module_strip_ssml_keeps_unknown_entities(self): + self.assertEqual( + module_utils.module_strip_ssml("Unknown © entity stays."), + "Unknown © entity stays.", + ) + + def test_module_loglevel_set_accepts_numeric_prefix(self): + self.assertEqual(module_utils.module_loglevel_set("log_level", " -12xyz"), 0) + self.assertEqual(module_utils.log_level, -12) + + def test_module_loglevel_set_rejects_wrong_item(self): + module_utils.log_level = 4 + + self.assertEqual(module_utils.module_loglevel_set("rate", "8"), -1) + self.assertEqual(module_utils.log_level, 4) + + def test_module_loglevel_set_rejects_non_numeric_value(self): + module_utils.log_level = 4 + + self.assertEqual(module_utils.module_loglevel_set("log_level", "abc"), -1) + self.assertEqual(module_utils.log_level, 4) + + def test_module_debug_enable_and_disable_custom_file(self): + with tempfile.TemporaryDirectory() as directory: + filename = os.path.join(directory, "debug.log") + + self.assertEqual(module_utils.module_debug(True, filename), 0) + self.assertEqual(module_utils.Debug, 2) + self.assertIsNotNone(module_utils.CustomDebugFile) + self.assertFalse(module_utils.CustomDebugFile.closed) + + custom_debug_file = module_utils.CustomDebugFile + self.assertEqual(module_utils.module_debug(False, None), 0) + self.assertEqual(module_utils.Debug, 0) + self.assertIsNone(module_utils.CustomDebugFile) + self.assertTrue(custom_debug_file.closed) + + def test_module_debug_preserves_stdout_debug_state(self): + with tempfile.TemporaryDirectory() as directory: + filename = os.path.join(directory, "debug.log") + module_utils.Debug = 1 + + self.assertEqual(module_utils.module_debug(True, filename), 0) + self.assertEqual(module_utils.Debug, 3) + + self.assertEqual(module_utils.module_debug(False, None), 0) + self.assertEqual(module_utils.Debug, 1) + + def test_module_debug_closes_previous_file(self): + with tempfile.TemporaryDirectory() as directory: + first = os.path.join(directory, "first.log") + second = os.path.join(directory, "second.log") + + self.assertEqual(module_utils.module_debug(True, first), 0) + first_file = module_utils.CustomDebugFile + self.assertEqual(module_utils.module_debug(True, second), 0) + + self.assertTrue(first_file.closed) + self.assertFalse(module_utils.CustomDebugFile.closed) + + def test_module_debug_reports_open_failure(self): + with tempfile.TemporaryDirectory() as directory: + filename = os.path.join(directory, "missing", "debug.log") + + self.assertEqual(module_utils.module_debug(True, filename), -1) + self.assertEqual(module_utils.Debug, 0) + self.assertIsNone(module_utils.CustomDebugFile) + + +if __name__ == "__main__": + unittest.main() diff --git a/src/tests/python/test_speechd_types.py b/src/tests/python/test_speechd_types.py new file mode 100644 index 00000000..c9488c4d --- /dev/null +++ b/src/tests/python/test_speechd_types.py @@ -0,0 +1,55 @@ +# +# test_speechd_types.py - Python speechd_types unit tests +# +# Copyright (C) 2026 Jean-François David +# +# This 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; either version 2, or (at your option) +# any later version. +# +# This software 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, see . + +import unittest + +from speechd_python_modules import speechd_types + + +class SpeechdTypesTest(unittest.TestCase): + def test_constants(self): + self.assertEqual(speechd_types.SPD_MSGTYPE_TEXT, 0) + self.assertEqual(speechd_types.SPD_MSGTYPE_SPELL, 99) + self.assertEqual(speechd_types.SPD_AUDIO_LE, 0) + self.assertEqual(speechd_types.SPD_AUDIO_BE, 1) + + def test_spd_voice_fields(self): + voice = speechd_types.SPDVoice(name="name", language="en", variant="male") + + self.assertEqual(voice.name, "name") + self.assertEqual(voice.language, "en") + self.assertEqual(voice.variant, "male") + + def test_audio_track_fields(self): + track = speechd_types.AudioTrack( + bits=16, + num_channels=1, + sample_rate=22050, + num_samples=2, + samples=b"\0\0\1\0", + ) + + self.assertEqual(track.bits, 16) + self.assertEqual(track.num_channels, 1) + self.assertEqual(track.sample_rate, 22050) + self.assertEqual(track.num_samples, 2) + self.assertEqual(track.samples, b"\0\0\1\0") + + +if __name__ == "__main__": + unittest.main() diff --git a/src/tests/python_module.at b/src/tests/python_module.at index 3bc74c3a..dabed2a7 100644 --- a/src/tests/python_module.at +++ b/src/tests/python_module.at @@ -17,122 +17,17 @@ AT_BANNER([Python module]) -AT_SETUP([Python module imports]) -AT_KEYWORDS([python import]) +AT_SETUP([Python module unit tests]) +AT_KEYWORDS([python unit]) AT_SKIP_IF([test "x$TEST_PYTHON_ENABLED" != xyes]) AT_CHECK([[test_pythonpath=${TEST_PYTHONPATH-"$abs_top_srcdir/src/modules"} PYTHONDONTWRITEBYTECODE=1 \ PYTHONNOUSERSITE=1 \ +TEST_PYTHONPATH="$test_pythonpath" \ PYTHONPATH="$test_pythonpath" \ -"$PYTHON" - "$test_pythonpath" <<'PY' -import os -import sys - -from speechd_python_modules import module_utils, speechd_types - -pythonpath_dir = os.path.realpath(sys.argv[1]) -module_dir = os.path.join(pythonpath_dir, "speechd_python_modules") -modules = [module_utils, speechd_types] - - -def check_module_path(module): - module_file = os.path.realpath(module.__file__) - if os.path.dirname(module_file) != module_dir: - raise SystemExit( - "%s imported from %r, expected %r" - % (module.__name__, module_file, module_dir) - ) - - -for module in modules: - check_module_path(module) -PY -]], [0], [], []) - -AT_CLEANUP - -AT_SETUP([module_strip_ssml]) -AT_KEYWORDS([python module_utils ssml]) - -AT_SKIP_IF([test "x$TEST_PYTHON_ENABLED" != xyes]) - -AT_CHECK([[test_pythonpath=${TEST_PYTHONPATH-"$abs_top_srcdir/src/modules"} -PYTHONDONTWRITEBYTECODE=1 \ -PYTHONNOUSERSITE=1 \ -PYTHONPATH="$test_pythonpath" \ -"$PYTHON" - <<'PY' -from speechd_python_modules import module_utils - -strip_ssml = module_utils.module_strip_ssml - - -def check(message, expected): - actual = strip_ssml(message) - if actual != expected: - raise SystemExit( - "strip_ssml(%r) returned %r, expected %r" - % (message, actual, expected) - ) - - -check("Plain text.", "Plain text.") -check("Hello world.", "Hello world.") -check('Hello world', "Hello world") -check('text', "text") -check( - "Use <tag>, &, "quotes" and 'apostrophes'.", - "Use , &, \"quotes\" and 'apostrophes'.", -) -check("Keep &1 +]], [0], [ignore], []) AT_CLEANUP