test fixes

Author: mattieruth

examples/foundational/35-pattern-pair-voice-switching.py

@@ -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,
     )
 

src/pipecat/extensions/ivr/ivr_navigator.py

@@ -114,15 +114,19 @@ def _get_conversation_history(self) -> List[dict]:
     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):

tests/test_pattern_pair_aggregator.py

@@ -20,8 +20,16 @@ def setUp(self):
             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 @@ async def test_pattern_match_and_removal(self):
         # 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 @@ async def test_pattern_match_and_removal(self):
 
         # 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 @@ async def test_incomplete_pattern(self):
         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 @@ async def test_multiple_patterns(self):
         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 @@ async def test_handle_interruption(self):
         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 @@ async def test_pattern_across_sentences(self):
         self.assertEqual(result, "Hello  Final sentence.")
 
         # Buffer should be empty
-        self.assertEqual(self.aggregator.text, "")
+        self.assertEqual(self.aggregator.text.text, "")

tests/test_skip_tags_aggregator.py

@@ -18,37 +18,41 @@ async def test_no_tags(self):
 
         # 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>[email protected]</spell>.")
-        self.assertEqual(result, "My email is <spell>[email protected]</spell>.")
-        self.assertEqual(self.aggregator.text, "")
+        self.assertEqual(result.text, "My email is <spell>[email protected]</spell>.")
+        self.assertEqual(result.type, "sentence")
+        self.assertEqual(self.aggregator.text.text, "")
 
     async def test_streaming_tags(self):
         await self.aggregator.reset()
 
         # 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>[email protected]</spe")
+        self.assertEqual(self.aggregator.text.text, "My email is <spell>[email protected]</spe")
+        self.assertEqual(self.aggregator.text.type, "sentence")
 
         result = await self.aggregator.aggregate("ll>.")
-        self.assertEqual(result, "My email is <spell>[email protected]</spell>.")
-        self.assertEqual(self.aggregator.text, "")
+        self.assertEqual(result.text, "My email is <spell>[email protected]</spell>.")
+        self.assertEqual(self.aggregator.text.text, "")
+        self.assertEqual(self.aggregator.text.type, "sentence")