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

Add unit test exercising synchronous usage of PipelineTask.add_observer() right after initializing the PipelineTask (before anything else is done with it)

Browse Source
This commit is contained in:
Paul Kompfner
2025-06-09 11:27:39 -04:00
parent 1cd96f94ff
commit 513ce26200
1 changed files with 25 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
tests/test_pipeline.py
View File

@@ -117,7 +117,8 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
async def test_task_add_observer(self): async def test_task_add_observer(self):
frame_received = False frame_received = False
frame_add_count = 0 frame_count_1 = 0
frame_count_2 = 0
class CustomObserver(BaseObserver): class CustomObserver(BaseObserver):
async def on_push_frame(self, data: FramePushed): async def on_push_frame(self, data: FramePushed):
@@ -126,28 +127,41 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
if isinstance(data.frame, TextFrame): if isinstance(data.frame, TextFrame):
frame_received = True frame_received = True
class CustomAddObserver(BaseObserver): class CustomAddObserver1(BaseObserver):
async def on_push_frame(self, data: FramePushed): async def on_push_frame(self, data: FramePushed):
nonlocal frame_add_count nonlocal frame_count_1
if isinstance(data.source, IdentityFilter) and isinstance(data.frame, TextFrame): if isinstance(data.source, IdentityFilter) and isinstance(data.frame, TextFrame):
frame_add_count += 1 frame_count_1 += 1
class CustomAddObserver2(BaseObserver):
async def on_push_frame(self, data: FramePushed):
nonlocal frame_count_2
if isinstance(data.source, IdentityFilter) and isinstance(data.frame, TextFrame):
frame_count_2 += 1
identity = IdentityFilter() identity = IdentityFilter()
pipeline = Pipeline([identity]) pipeline = Pipeline([identity])
task = PipelineTask(pipeline, observers=[CustomObserver()]) task = PipelineTask(pipeline, observers=[CustomObserver()])
# Add a new observer right away, before doing anything else with the task.
observer1 = CustomAddObserver1()
task.add_observer(observer1)
task.set_event_loop(asyncio.get_event_loop()) task.set_event_loop(asyncio.get_event_loop())
async def delayed_add_observer(): async def delayed_add_observer():
observer = CustomAddObserver() observer2 = CustomAddObserver2()
# Wait after the pipeline is started and add an observer. # Wait after the pipeline is started and add another observer.
await asyncio.sleep(0.1) await asyncio.sleep(0.1)
task.add_observer(observer) task.add_observer(observer2)
# Push a TextFrame and wait for the observer to pick it up. # Push a TextFrame and wait for the observer to pick it up.
await task.queue_frame(TextFrame(text="Hello Downstream!")) await task.queue_frame(TextFrame(text="Hello Downstream!"))
await asyncio.sleep(0.1) await asyncio.sleep(0.1)
# Remove the observer # Remove both observers.
await task.remove_observer(observer) await task.remove_observer(observer1)
await task.remove_observer(observer2)
# Push another TextFrame. This time the counter should not # Push another TextFrame. This time the counter should not
# increments since we have removed the observer. # increments since we have removed the observer.
await task.queue_frame(TextFrame(text="Hello Downstream!")) await task.queue_frame(TextFrame(text="Hello Downstream!"))
@@ -158,7 +172,8 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
await asyncio.gather(task.run(), delayed_add_observer()) await asyncio.gather(task.run(), delayed_add_observer())
assert frame_received assert frame_received
assert frame_add_count == 1 assert frame_count_1 == 1
assert frame_count_2 == 1
async def test_task_started_ended_event_handler(self): async def test_task_started_ended_event_handler(self):
start_received = False start_received = False