NeoAnd/adk-python
1
0
Fork 0
mirror of https://github.com/EvolutionAPI/adk-python.git synced 2025-07-13 15:14:50 -06:00
Code Issues Actions Packages Projects Releases Wiki Activity

Copybara import of the project:

Browse Source 
--
73826d258b136f92a8da8171f7dc14d5f07de8dd by Calvin Giles <calvin.giles@trademe.co.nz>:

fix: Enable InMemoryRunner to be used in async tests
COPYBARA_INTEGRATE_REVIEW=https://github.com/google/adk-python/pull/868 from calvingiles:enable-test-runner-in-async fb9033ed6f350a3114859715cae51798f864ecf6
PiperOrigin-RevId: 763233472
This commit is contained in:
Calvin Giles 2025-05-25 18:49:00 -07:00 committed by Copybara-Service
parent a380d70ac7
commit 1a43cf0f60
1 changed files with 14 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
tests/unittests/testing_utils.py
View File

@ -177,17 +177,13 @@ class InMemoryRunner:
@property
def session(self) -> Session:
if not self.session_id:
session = asyncio.run(
self.runner.session_service.create_session(
app_name='test_app', user_id='test_user'
)
session = self.runner.session_service.create_session_sync(
app_name='test_app', user_id='test_user'
)
self.session_id = session.id
return session
return asyncio.run(
self.runner.session_service.get_session(
app_name='test_app', user_id='test_user', session_id=self.session_id
)
return self.runner.session_service.get_session_sync(
app_name='test_app', user_id='test_user', session_id=self.session_id
)
def run(self, new_message: types.ContentUnion) -> list[Event]:
@ -199,6 +195,16 @@ class InMemoryRunner:
)
)
async def run_async(self, new_message: types.ContentUnion) -> list[Event]:
events = []
async for event in self.runner.run_async(
user_id=self.session.user_id,
session_id=self.session.id,
new_message=get_user_content(new_message),
):
events.append(event)
return events
def run_live(self, live_request_queue: LiveRequestQueue) -> list[Event]:
collected_responses = []