94f71853e0
* fix: implement microphone reconnection and audio buffer limits for reliability - Add microphone reconnection logic with exponential backoff - Implement audio buffer limits to prevent memory issues - Add configurable buffer size limits with clamping - Improve error handling for audio device failures - Add buffer statistics and monitoring capabilities - Graceful stream cleanup and resource management Resolves #92 Changes made: - Added _max_buffer_size, _reconnection_attempts, _max_reconnection_attempts attributes - Implemented _attempt_audio_reconnection() method with exponential backoff - Enhanced _record_audio() method with buffer limit checking and reconnection - Added set_buffer_limit() and get_buffer_stats() methods - Improved error handling for IOError/OSError during audio recording - Added test scripts to validate reliability improvements The implementation prevents memory issues by limiting audio buffer size and automatically reconnecting to audio devices when they fail, with proper exponential backoff to prevent rapid reconnection attempts. * fix: Address code review feedback for PR #109 - Add thread lock (_buffer_lock) for audio_buffer to prevent race conditions - Fix _last_audio_error_time reset in cleanup - Protect buffer access in stop_recognition and get_buffer_stats with lock - Fix test_max_reconnection_attempts to match actual counter behavior Addresses review feedback for microphone reconnection and audio buffer limits * style: fix lint errors * fix: add missing sys import in audio_feedback.py - Add sys import required by _is_ci_mode() function - Remove temporary test_reliability scripts * style: fix black formatting --------- Co-authored-by: OpenClaw Reviewer <review@openclaw.ai>
344 lines
15 KiB
Python
344 lines
15 KiB
Python
"""
|
|
Tests for the command processor functionality.
|
|
"""
|
|
|
|
import unittest
|
|
|
|
# Update import path to use the new package structure
|
|
from vocalinux.speech_recognition.command_processor import CommandProcessor
|
|
|
|
|
|
class TestCommandProcessor(unittest.TestCase):
|
|
"""Test cases for command processor functionality."""
|
|
|
|
def setUp(self):
|
|
"""Set up for tests."""
|
|
self.processor = CommandProcessor()
|
|
|
|
def test_initialization(self):
|
|
"""Test initialization of command processor."""
|
|
# Verify command dictionaries are initialized
|
|
self.assertTrue(hasattr(self.processor, "text_commands"))
|
|
self.assertTrue(hasattr(self.processor, "action_commands"))
|
|
self.assertTrue(hasattr(self.processor, "format_commands"))
|
|
|
|
# Verify dictionaries have expected entries
|
|
self.assertIn("new line", self.processor.text_commands)
|
|
self.assertIn("period", self.processor.text_commands)
|
|
self.assertIn("delete that", self.processor.action_commands)
|
|
self.assertIn("capitalize", self.processor.format_commands)
|
|
|
|
# Verify regex patterns are compiled
|
|
self.assertTrue(hasattr(self.processor, "text_cmd_regex"))
|
|
self.assertTrue(hasattr(self.processor, "action_cmd_regex"))
|
|
self.assertTrue(hasattr(self.processor, "format_cmd_regex"))
|
|
|
|
def test_text_command_processing(self):
|
|
"""Test processing of text commands."""
|
|
# Test common text commands
|
|
test_cases = [
|
|
("new line", "\n", []),
|
|
("this is a new paragraph", "this is a \n\n", []),
|
|
("end of sentence period", "end of sentence.", []),
|
|
("add a comma here", "add a, here", []),
|
|
("use question mark", "use?", []),
|
|
("exclamation mark test", "! test", []),
|
|
("semicolon example", "; example", []),
|
|
("testing colon usage", "testing:", []),
|
|
("dash separator", "- separator", []),
|
|
("hyphen example", "- example", []),
|
|
("underscore value", "_ value", []),
|
|
("quote example", '" example', []),
|
|
("single quote test", "' test", []),
|
|
("open parenthesis content close parenthesis", "( content)", []),
|
|
("open bracket item close bracket", "[ item]", []),
|
|
("open brace code close brace", "{ code}", []),
|
|
]
|
|
|
|
for input_text, expected_output, _ in test_cases:
|
|
result, actions = self.processor.process_text(input_text)
|
|
self.assertEqual(result, expected_output)
|
|
self.assertEqual(actions, [])
|
|
|
|
def test_action_command_processing(self):
|
|
"""Test processing of action commands."""
|
|
# Test action commands
|
|
test_cases = [
|
|
("delete that", "", ["delete_last"]),
|
|
("scratch that previous text", " previous text", ["delete_last"]),
|
|
("undo my last change", " my last change", ["undo"]),
|
|
("redo that edit", " that edit", ["redo"]),
|
|
("select all text", " text", ["select_all"]),
|
|
("select line of code", " of code", ["select_line"]),
|
|
("select word here", " here", ["select_word"]),
|
|
("select paragraph content", " content", ["select_paragraph"]),
|
|
("cut this selection", " this selection", ["cut"]),
|
|
("copy this text", " this text", ["copy"]),
|
|
("paste here", " here", ["paste"]),
|
|
# Test multiple actions
|
|
("select all then copy", " then", ["select_all", "copy"]),
|
|
]
|
|
|
|
for input_text, expected_output, expected_actions in test_cases:
|
|
result, actions = self.processor.process_text(input_text)
|
|
self.assertEqual(result, expected_output)
|
|
self.assertEqual(actions, expected_actions)
|
|
|
|
def test_format_command_processing(self):
|
|
"""Test processing of formatting commands."""
|
|
# Test format commands
|
|
test_cases = [
|
|
("capitalize word", "Word", []),
|
|
("uppercase letters", "LETTERS", []),
|
|
("all caps example", "EXAMPLE", []),
|
|
("lowercase TEXT", "text", []),
|
|
# Test with text before the command
|
|
("make this capitalize next", "make this Next", []),
|
|
]
|
|
|
|
for input_text, expected_output, expected_actions in test_cases:
|
|
result, actions = self.processor.process_text(input_text)
|
|
self.assertEqual(result, expected_output)
|
|
self.assertEqual(actions, expected_actions)
|
|
|
|
def test_combined_commands(self):
|
|
"""Test combinations of different command types."""
|
|
test_cases = [
|
|
# Text + Action
|
|
("new line then delete that", "", ["delete_last"]),
|
|
# Format + Text
|
|
("capitalize name period", "Name.", []),
|
|
# Action + Format
|
|
("select all then capitalize text", " then Text", ["select_all"]),
|
|
# Complex combination
|
|
(
|
|
"capitalize name comma new line select paragraph",
|
|
"Name,\n",
|
|
["select_paragraph"],
|
|
),
|
|
]
|
|
|
|
for input_text, expected_output, expected_actions in test_cases:
|
|
result, actions = self.processor.process_text(input_text)
|
|
self.assertEqual(result, expected_output)
|
|
self.assertEqual(actions, expected_actions)
|
|
|
|
def test_empty_input(self):
|
|
"""Test handling of empty input."""
|
|
result, actions = self.processor.process_text("")
|
|
self.assertEqual(result, "")
|
|
self.assertEqual(actions, [])
|
|
|
|
result, actions = self.processor.process_text(None)
|
|
self.assertEqual(result, "")
|
|
self.assertEqual(actions, [])
|
|
|
|
def test_no_commands(self):
|
|
"""Test text with no commands."""
|
|
input_text = "This is just regular text with no commands."
|
|
result, actions = self.processor.process_text(input_text)
|
|
self.assertEqual(result, input_text)
|
|
self.assertEqual(actions, [])
|
|
|
|
def test_partial_command_matches(self):
|
|
"""Test text with partial command matches."""
|
|
# Words that are substrings of commands should not be processed
|
|
test_cases = [
|
|
("periodic review", "periodic review", []), # shouldn't match "period"
|
|
("newcomer", "newcomer", []), # shouldn't match "new"
|
|
("paramount", "paramount", []), # shouldn't match "paragraph"
|
|
]
|
|
|
|
for input_text, expected_output, expected_actions in test_cases:
|
|
result, actions = self.processor.process_text(input_text)
|
|
self.assertEqual(result, expected_output)
|
|
self.assertEqual(actions, expected_actions)
|
|
|
|
def test_case_insensitivity(self):
|
|
"""Test that command matching is case-insensitive."""
|
|
test_cases = [
|
|
("NEW LINE", "\n", []),
|
|
("Period", ".", []),
|
|
("DELETE THAT", "", ["delete_last"]),
|
|
("Capitalize word", "Word", []),
|
|
]
|
|
|
|
for input_text, expected_output, expected_actions in test_cases:
|
|
result, actions = self.processor.process_text(input_text)
|
|
self.assertEqual(result, expected_output)
|
|
self.assertEqual(actions, expected_actions)
|
|
|
|
def test_multiple_format_modifiers(self):
|
|
"""Test multiple format modifiers on a single word."""
|
|
# Reset command processor to ensure active_formats is empty
|
|
self.processor = CommandProcessor()
|
|
|
|
# Apply multiple formatting commands
|
|
result, _ = self.processor.process_text("capitalize all caps text")
|
|
|
|
# The second format command overrides the first, resulting in all uppercase
|
|
self.assertEqual(result, "TEXT")
|
|
|
|
def test_format_with_no_target_word(self):
|
|
"""Test format command with no following word."""
|
|
result, _ = self.processor.process_text("capitalize")
|
|
self.assertEqual(result, "")
|
|
|
|
# The active_formats should be cleared even if no word was formatted
|
|
self.assertEqual(self.processor.active_formats, set())
|
|
|
|
def test_whitespace_handling(self):
|
|
"""Test handling of whitespace in command processing."""
|
|
# Test with extra spaces
|
|
result, _ = self.processor.process_text("new line test")
|
|
self.assertEqual(result, "\n test")
|
|
|
|
# Test with leading/trailing spaces
|
|
result, _ = self.processor.process_text(" period ")
|
|
self.assertEqual(result, ".")
|
|
|
|
# Test with mixed whitespace
|
|
result, _ = self.processor.process_text(" capitalize word new line ")
|
|
self.assertEqual(result, "Word \n")
|
|
|
|
def test_regex_compilation(self):
|
|
"""Test the regex pattern compilation."""
|
|
self.processor._compile_patterns()
|
|
|
|
# Test regex patterns match correctly
|
|
self.assertTrue(self.processor.text_cmd_regex.search("new line"))
|
|
self.assertTrue(self.processor.text_cmd_regex.search("this is a period"))
|
|
self.assertTrue(self.processor.action_cmd_regex.search("delete that"))
|
|
self.assertTrue(self.processor.format_cmd_regex.search("capitalize this"))
|
|
|
|
# Test non-matches
|
|
self.assertFalse(self.processor.text_cmd_regex.search("newline")) # no space
|
|
self.assertFalse(self.processor.action_cmd_regex.search("deletion")) # not a command
|
|
|
|
def test_compile_patterns_method(self):
|
|
"""Test the _compile_patterns method directly."""
|
|
# Add a new command to test recompilation
|
|
self.processor.text_commands["test command"] = "TEST"
|
|
|
|
# Recompile patterns
|
|
self.processor._compile_patterns()
|
|
|
|
# Verify new command is in the pattern
|
|
self.assertTrue(self.processor.text_cmd_regex.search("test command"))
|
|
|
|
|
|
class TestCommandProcessorFallback(unittest.TestCase):
|
|
"""Test the generic fallback processing logic in CommandProcessor."""
|
|
|
|
def setUp(self):
|
|
"""Set up for tests."""
|
|
self.processor = CommandProcessor()
|
|
|
|
def test_generic_action_command_with_following_text(self):
|
|
"""Test action command followed by text through generic path."""
|
|
# Use text that won't match hardcoded test cases
|
|
result, actions = self.processor.process_text("undo the error")
|
|
# Should trigger action and capture following text
|
|
self.assertIn("undo", actions)
|
|
|
|
def test_generic_action_command_alone(self):
|
|
"""Test action command alone through generic path."""
|
|
# "redo" alone isn't hardcoded, so goes through generic
|
|
result, actions = self.processor.process_text("redo only")
|
|
self.assertIn("redo", actions)
|
|
|
|
def test_generic_text_command_punctuation(self):
|
|
"""Test punctuation text commands through generic path."""
|
|
# "full stop" isn't in the hardcoded section
|
|
result, actions = self.processor.process_text("the end full stop")
|
|
self.assertIn(".", result)
|
|
|
|
def test_generic_text_command_non_punctuation(self):
|
|
"""Test non-punctuation text commands through generic path."""
|
|
# Test "open brace" by itself (different from hardcoded combo)
|
|
result, actions = self.processor.process_text("code open brace here")
|
|
self.assertIn("{", result)
|
|
|
|
def test_generic_format_capitalize_next(self):
|
|
"""Test capitalize format command through generic path."""
|
|
# Use a different pattern than hardcoded cases
|
|
result, actions = self.processor.process_text("capitalize john")
|
|
# Should capitalize john -> John
|
|
self.assertIn("John", result)
|
|
|
|
def test_generic_format_uppercase_next(self):
|
|
"""Test uppercase format command through generic path."""
|
|
# "uppercase" followed by a word not in hardcoded cases
|
|
result, actions = self.processor.process_text("uppercase hello")
|
|
self.assertIn("HELLO", result)
|
|
|
|
def test_generic_format_lowercase_next(self):
|
|
"""Test lowercase format command through generic path."""
|
|
# Different pattern than the hardcoded "lowercase TEXT"
|
|
result, actions = self.processor.process_text("lowercase WORLD")
|
|
self.assertIn("world", result)
|
|
|
|
def test_generic_format_no_target_word(self):
|
|
"""Test format command with no following word through generic path."""
|
|
# "uppercase" alone - no word follows
|
|
result, actions = self.processor.process_text("uppercase")
|
|
# Should return empty when no target word
|
|
self.assertEqual(result, "")
|
|
|
|
def test_generic_multiple_punctuation(self):
|
|
"""Test multiple punctuation commands through generic path."""
|
|
result, actions = self.processor.process_text("hello comma world period")
|
|
# Both comma and period should be replaced
|
|
self.assertIn(",", result)
|
|
self.assertIn(".", result)
|
|
|
|
def test_generic_semicolon(self):
|
|
"""Test semicolon through generic path."""
|
|
result, actions = self.processor.process_text("statement semicolon")
|
|
self.assertIn(";", result)
|
|
|
|
def test_generic_colon(self):
|
|
"""Test colon through generic path."""
|
|
result, actions = self.processor.process_text("note colon important")
|
|
self.assertIn(":", result)
|
|
|
|
def test_generic_question_mark(self):
|
|
"""Test question mark through generic path."""
|
|
result, actions = self.processor.process_text("really question mark")
|
|
self.assertIn("?", result)
|
|
|
|
def test_generic_exclamation_point(self):
|
|
"""Test exclamation point through generic path."""
|
|
result, actions = self.processor.process_text("wow exclamation point")
|
|
self.assertIn("!", result)
|
|
|
|
def test_generic_unknown_format_type(self):
|
|
"""Test unknown format type falls back to word as-is."""
|
|
# Add an unknown format type for testing
|
|
self.processor.format_commands["testformat"] = "unknown_type"
|
|
self.processor._compile_patterns()
|
|
|
|
result, actions = self.processor.process_text("testformat word")
|
|
# Unknown format type should return word unchanged
|
|
self.assertIn("word", result.lower())
|
|
|
|
def test_generic_select_line_action(self):
|
|
"""Test select line action through generic path."""
|
|
result, actions = self.processor.process_text("select line now")
|
|
self.assertIn("select_line", actions)
|
|
|
|
def test_generic_select_word_action(self):
|
|
"""Test select word action through generic path."""
|
|
result, actions = self.processor.process_text("select word please")
|
|
self.assertIn("select_word", actions)
|
|
|
|
def test_generic_cut_action(self):
|
|
"""Test cut action through generic path."""
|
|
result, actions = self.processor.process_text("cut selection")
|
|
self.assertIn("cut", actions)
|
|
|
|
def test_generic_paste_action(self):
|
|
"""Test paste action through generic path."""
|
|
result, actions = self.processor.process_text("paste content")
|
|
self.assertIn("paste", actions)
|