wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test fixes

Browse Source
This commit is contained in:
mattie ruth backman
2025-10-21 15:04:34 -04:00
parent fe9aa3383e
commit 5c8635570d
4 changed files with 42 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
examples/foundational/35-pattern-pair-voice-switching.py
View File

@@ -110,6 +110,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
pattern_id="voice_tag",
start_pattern="<voice>",
end_pattern="</voice>",
type="voice",
remove_match=True,
)

10
src/pipecat/extensions/ivr/ivr_navigator.py
View File

@@ -114,15 +114,19 @@ class IVRProcessor(FrameProcessor):
def _setup_xml_patterns(self):
"""Set up XML pattern detection and handlers."""
# Register DTMF pattern
self._aggregator.add_pattern_pair("dtmf", "<dtmf>", "</dtmf>", remove_match=True)
self._aggregator.add_pattern_pair(
"dtmf", "<dtmf>", "</dtmf>", type="dtmf", remove_match=True
)
self._aggregator.on_pattern_match("dtmf", self._handle_dtmf_action)
# Register mode pattern
self._aggregator.add_pattern_pair("mode", "<mode>", "</mode>", remove_match=True)
self._aggregator.add_pattern_pair(
"mode", "<mode>", "</mode>", type="mode", remove_match=True
)
self._aggregator.on_pattern_match("mode", self._handle_mode_action)
# Register IVR pattern
self._aggregator.add_pattern_pair("ivr", "<ivr>", "</ivr>", remove_match=True)
self._aggregator.add_pattern_pair("ivr", "<ivr>", "</ivr>", type="ivr", remove_match=True)
self._aggregator.on_pattern_match("ivr", self._handle_ivr_action)
async def process_frame(self, frame: Frame, direction: FrameDirection):

29
tests/test_pattern_pair_aggregator.py
View File

@@ -20,8 +20,16 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
pattern_id="test_pattern",
start_pattern="<test>",
end_pattern="</test>",
type="test",
remove_match=True,
)
self.aggregator.add_pattern_pair(
pattern_id="code_pattern",
start_pattern="<code>",
end_pattern="</code>",
type="code",
remove_match=False,
)
# Register the mock handler
self.aggregator.on_pattern_match("test_pattern", self.test_handler)
@@ -30,7 +38,8 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
# First part doesn't complete the pattern
result = await self.aggregator.aggregate("Hello <test>pattern")
self.assertIsNone(result)
self.assertEqual(self.aggregator.text, "Hello <test>pattern")
self.assertEqual(self.aggregator.text.text, "Hello <test>pattern")
self.assertEqual(self.aggregator.text.type, "test")
# Second part completes the pattern and includes an exclamation point
result = await self.aggregator.aggregate(" content</test>!")
@@ -45,14 +54,15 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
# The exclamation point should be treated as a sentence boundary,
# so the result should include just text up to and including "!"
self.assertEqual(result, "Hello !")
self.assertEqual(result.text, "Hello !")
self.assertEqual(result.type, "sentence")
# Next sentence should be processed separately
result = await self.aggregator.aggregate(" This is another sentence.")
self.assertEqual(result, " This is another sentence.")
self.assertEqual(result.text, " This is another sentence.")
# Buffer should be empty after returning a complete sentence
self.assertEqual(self.aggregator.text, "")
self.assertEqual(self.aggregator.text.text, "")
async def test_incomplete_pattern(self):
# Add text with incomplete pattern
@@ -65,11 +75,12 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
self.test_handler.assert_not_called()
# Buffer should contain the incomplete text
self.assertEqual(self.aggregator.text, "Hello <test>pattern content")
self.assertEqual(self.aggregator.text.text, "Hello <test>pattern content")
self.assertEqual(self.aggregator.text.type, "test")
# Reset and confirm buffer is cleared
await self.aggregator.reset()
self.assertEqual(self.aggregator.text, "")
self.assertEqual(self.aggregator.text.text, "")
async def test_multiple_patterns(self):
# Set up multiple patterns and handlers
@@ -109,7 +120,7 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
self.assertEqual(result, "Hello I am <em>very</em> excited to meet you!")
# Buffer should be empty
self.assertEqual(self.aggregator.text, "")
self.assertEqual(self.aggregator.text.text, "")
async def test_handle_interruption(self):
# Start with incomplete pattern
@@ -120,7 +131,7 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
await self.aggregator.handle_interruption()
# Buffer should be cleared
self.assertEqual(self.aggregator.text, "")
self.assertEqual(self.aggregator.text.text, "")
# Handler should not have been called
self.test_handler.assert_not_called()
@@ -144,4 +155,4 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
self.assertEqual(result, "Hello Final sentence.")
# Buffer should be empty
self.assertEqual(self.aggregator.text, "")
self.assertEqual(self.aggregator.text.text, "")

24
tests/test_skip_tags_aggregator.py
View File

@@ -18,16 +18,18 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase):
# No tags involved, aggregate at end of sentence.
result = await self.aggregator.aggregate("Hello Pipecat!")
self.assertEqual(result, "Hello Pipecat!")
self.assertEqual(self.aggregator.text, "")
self.assertEqual(result.text, "Hello Pipecat!")
self.assertEqual(result.type, "sentence")
self.assertEqual(self.aggregator.text.text, "")
async def test_basic_tags(self):
await self.aggregator.reset()
# Tags involved, avoid aggregation during tags.
result = await self.aggregator.aggregate("My email is <spell>foo@pipecat.ai</spell>.")
self.assertEqual(result, "My email is <spell>foo@pipecat.ai</spell>.")
self.assertEqual(self.aggregator.text, "")
self.assertEqual(result.text, "My email is <spell>foo@pipecat.ai</spell>.")
self.assertEqual(result.type, "sentence")
self.assertEqual(self.aggregator.text.text, "")
async def test_streaming_tags(self):
await self.aggregator.reset()
@@ -35,20 +37,22 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase):
# Tags involved, stream small chunk of texts.
result = await self.aggregator.aggregate("My email is <sp")
self.assertIsNone(result)
self.assertEqual(self.aggregator.text, "My email is <sp")
self.assertEqual(self.aggregator.text.text, "My email is <sp")
result = await self.aggregator.aggregate("ell>foo.")
self.assertIsNone(result)
self.assertEqual(self.aggregator.text, "My email is <spell>foo.")
self.assertEqual(self.aggregator.text.text, "My email is <spell>foo.")
result = await self.aggregator.aggregate("bar@pipecat.")
self.assertIsNone(result)
self.assertEqual(self.aggregator.text, "My email is <spell>foo.bar@pipecat.")
self.assertEqual(self.aggregator.text.text, "My email is <spell>foo.bar@pipecat.")
result = await self.aggregator.aggregate("ai</spe")
self.assertIsNone(result)
self.assertEqual(self.aggregator.text, "My email is <spell>foo.bar@pipecat.ai</spe")
self.assertEqual(self.aggregator.text.text, "My email is <spell>foo.bar@pipecat.ai</spe")
self.assertEqual(self.aggregator.text.type, "sentence")
result = await self.aggregator.aggregate("ll>.")
self.assertEqual(result, "My email is <spell>foo.bar@pipecat.ai</spell>.")
self.assertEqual(self.aggregator.text, "")
self.assertEqual(result.text, "My email is <spell>foo.bar@pipecat.ai</spell>.")
self.assertEqual(self.aggregator.text.text, "")
self.assertEqual(self.aggregator.text.type, "sentence")