diff --git a/src/pipecat/tests/utils.py b/src/pipecat/tests/utils.py index 55bda9cea..7fcecc3ff 100644 --- a/src/pipecat/tests/utils.py +++ b/src/pipecat/tests/utils.py @@ -26,7 +26,8 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor class SleepFrame(SystemFrame): """This frame is used by test framework to introduce some sleep time before the next frame is pushed. This is useful to control system frames vs data or - control frames.""" + control frames. + """ sleep: float = 0.1 diff --git a/test-requirements.txt b/test-requirements.txt index 7566de351..6a92557e0 100644 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -29,3 +29,4 @@ soxr~=0.5.0 together~=1.2.7 transformers~=4.48.0 websockets~=13.1 +Markdown~=3.7 \ No newline at end of file diff --git a/tests/test_markdown_text_filter.py b/tests/test_markdown_text_filter.py new file mode 100644 index 000000000..c8cae97cc --- /dev/null +++ b/tests/test_markdown_text_filter.py @@ -0,0 +1,230 @@ +# +# Copyright (c) 2024-2025 Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +import unittest + +from pipecat.utils.text.markdown_text_filter import MarkdownTextFilter + + +class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.filter = MarkdownTextFilter() + + async def test_basic_markdown_removal(self): + """Test removal of basic Markdown formatting while preserving content.""" + input_text = """ + **Bold text** and *italic text* + 1. Numbered list item + - Bullet point + Some `inline code` here + """ + + expected_text = """ + Bold text and italic text + 1. Numbered list item + - Bullet point + Some inline code here + """ + + result = self.filter.filter(input_text) + self.assertEqual(result.strip(), expected_text.strip()) + + async def test_space_preservation(self): + """Test preservation of leading and trailing spaces (for + word-by-word streaming in bot-tts-text). + """ + input_text = [ + " Leading spaces", + "Trailing spaces ", + " Both ends ", + " Multiple spaces between words ", + ] + + for text in input_text: + result = self.filter.filter(text) + self.assertEqual( + len(result), len(text), f"Space preservation failed for: '{text}'\nGot: '{result}'" + ) + # Check if spaces are in the same positions + for i, char in enumerate(text): + if char == " ": + self.assertEqual( + result[i], " ", f"Space at position {i} was not preserved in: '{text}'" + ) + + async def test_repeated_character_removal(self): + """Test removal of repeated character sequences (5 or more).""" + test_cases = { + "Hello!!!!!World": "HelloWorld", # 5 exclamations removed + "Test####ing": "Test####ing", # 4 hashes preserved + "Normal text": "Normal text", # No repeated chars + "!!!!!": "", # All repeated chars removed + "Mixed!!!!!...../////": "Mixed", # Multiple repeated sequences + "Text^^^^test": "Text^^^^test", # 4 carets preserved + "Text^^^^^test": "Texttest", # 5 carets removed + "Dots....here": "Dots....here", # 4 dots preserved + "Dots.....here": "Dotshere", # 5 dots removed + } + + for input_text, expected in test_cases.items(): + result = self.filter.filter(input_text) + self.assertEqual( + result, + expected, + f"Failed to handle repeated characters in: '{input_text}'\nExpected: '{expected}'\nGot: '{result}'", + ) + + async def test_numbered_list_preservation(self): + """Test that numbered lists are preserved correctly.""" + input_text = """1. First item + 2. Second item + 3. Third item with **bold**""" + + expected = """1. First item + 2. Second item + 3. Third item with bold""" + + result = self.filter.filter(input_text) + self.assertEqual( + result.strip(), + expected.strip(), + f"Numbered list preservation failed.\nExpected:\n{expected}\nGot:\n{result}", + ) + + async def test_html_entity_conversion(self): + """Test conversion of HTML entities to their plain text equivalents.""" + test_cases = { + "This & that": "This & that", + "1 < 2": "1 < 2", + "2 > 1": "2 > 1", + "Line break": "Line break", + "Mixed & <entities>": "Mixed & ", + } + + for input_text, expected in test_cases.items(): + result = self.filter.filter(input_text) + self.assertEqual(result, expected, f"HTML entity conversion failed for: '{input_text}'") + + async def test_asterisk_removal(self): + """Test removal of Markdown asterisk formatting.""" + test_cases = { + "**bold text**": "bold text", # Double asterisks + "*italic text*": "italic text", # Single asterisks + "**bold** and *italic*": "bold and italic", # Mixed + "multiple**bold**words": "multipleboldwords", # No spaces + "edge**cases***here*": "edgecaseshere", # Adjacent asterisks + } + + for input_text, expected in test_cases.items(): + result = self.filter.filter(input_text) + self.assertEqual(result, expected, f"Asterisk removal failed for: '{input_text}'") + + async def test_newline_handling(self): + """Test handling of empty and whitespace-only lines.""" + test_cases = { + "Line 1\n\nLine 2": "Line 1\n Line 2", # Empty line becomes space + "Line 1\n \nLine 2": "Line 1\n Line 2", # Whitespace line becomes single space + "Text\n\n\nMore": "Text\n More", # Multiple empty lines become spaces + } + + for input_text, expected in test_cases.items(): + result = self.filter.filter(input_text) + self.assertEqual( + result, expected, f"Newline handling failed for:\n{input_text}\nGot:\n{result}" + ) + + async def test_numbered_list_marker_handling(self): + """Test handling of numbered lists with the special §NUM§ marker.""" + test_cases = { + "1. First\n2. Second": "1. First\n2. Second", # Basic numbered list + " 1. Indented": " 1. Indented", # Indented numbered list + "1. Item\nText\n2. Item": "1. Item\nText\n2. Item", # Text between items + "1.No space": "1.No space", # Not a list item (no space) + "12. Large number": "12. Large number", # Multi-digit numbers + } + + for input_text, expected in test_cases.items(): + result = self.filter.filter(input_text) + self.assertEqual( + result, + expected, + f"Numbered list handling failed for:\n{input_text}\nGot:\n{result}", + ) + + async def test_inline_code_handling(self): + """Test handling of inline code with backticks.""" + test_cases = { + "`code`": "code", # Basic inline code + "Text `code` more": "Text code more", # Inline code within text + "``nested`code``": "nested`code", # Nested backticks + "`code1` and `code2`": "code1 and code2", # Multiple inline codes + "No``space``between": "Nospacebetween", # No spaces around backticks + } + + for input_text, expected in test_cases.items(): + result = self.filter.filter(input_text) + self.assertEqual(result, expected, f"Inline code handling failed for: '{input_text}'") + + async def test_simple_table_removal(self): + """Test removal of a simple markdown table.""" + filter = MarkdownTextFilter(params=MarkdownTextFilter.InputParams(filter_tables=True)) + + input_text = "| Column 1 | Column 2 |\n|----------|----------|\n| Cell 1 | Cell 2 |" + + expected = "" + + result = filter.filter(input_text) + self.assertEqual( + result.strip(), + expected.strip(), + f"Simple table removal failed.\nExpected:\n{expected}\nGot:\n{result}", + ) + + async def test_feature_toggles(self): + """Test enabling/disabling specific filter features.""" + # Create a filter with all features disabled + filter = MarkdownTextFilter( + params=MarkdownTextFilter.InputParams( + enable_text_filter=False, + filter_code=False, + filter_tables=False, + ) + ) + + # Test with text filtering disabled + text_with_markdown = "**bold** and *italic* with `code`" + self.assertEqual( + filter.filter(text_with_markdown), + text_with_markdown, + "Disabled filter should not modify text", + ) + + # Enable just text filtering + filter.update_settings({"enable_text_filter": True}) + self.assertEqual( + filter.filter(text_with_markdown), + "bold and italic with code", + "Enabled filter should remove markdown", + ) + + async def test_settings_update(self): + """Test that filter settings can be updated at runtime.""" + filter = MarkdownTextFilter() + + # Initial state - formatting should be removed + input_text = "**bold** and *italic*" + self.assertEqual(filter.filter(input_text), "bold and italic") + + # Disable text filtering + filter.update_settings({"enable_text_filter": False}) + self.assertEqual(filter.filter(input_text), input_text, "Text filtering should be disabled") + + # Re-enable text filtering + filter.update_settings({"enable_text_filter": True}) + self.assertEqual( + filter.filter(input_text), "bold and italic", "Text filtering should be re-enabled" + )