mirror of
https://github.com/EvolutionAPI/adk-python.git
synced 2025-12-19 03:42:22 -06:00
fix(tests): use mock GCS client in artifact service tests to avoid real credentials
Copybara import of the project: -- ade1d98e030a966183f56cb5c9c1b04cf51f5337 by Thiago Neves <thiagohneves@gmail.com>: fix(tests): use mock GCS client in artifact service tests to avoid real credentials -- becd2925feebf60196129b029a0ab8d490f7b19e by Thiago Neves <thiagohneves@gmail.com>: test(agents): add unit tests for live_request_queue, readonly_context, and run_config COPYBARA_INTEGRATE_REVIEW=https://github.com/google/adk-python/pull/641 from thiagoneves:feature/increase-test-coverage 0f7a9fc55d97902e190a394f099324fbeb1541af PiperOrigin-RevId: 756798390
This commit is contained in:
committed by
Copybara-Service
parent
dc5c23c675
commit
ac97fc638f
52
tests/unittests/agents/test_live_request_queue.py
Normal file
52
tests/unittests/agents/test_live_request_queue.py
Normal file
@@ -0,0 +1,52 @@
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, AsyncMock, patch
|
||||
from google.adk.agents.live_request_queue import LiveRequest, LiveRequestQueue
|
||||
from google.genai import types
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_close_queue():
|
||||
queue = LiveRequestQueue()
|
||||
|
||||
with patch.object(queue._queue, "put_nowait") as mock_put_nowait:
|
||||
queue.close()
|
||||
mock_put_nowait.assert_called_once_with(LiveRequest(close=True))
|
||||
|
||||
|
||||
def test_send_content():
|
||||
queue = LiveRequestQueue()
|
||||
content = MagicMock(spec=types.Content)
|
||||
|
||||
with patch.object(queue._queue, "put_nowait") as mock_put_nowait:
|
||||
queue.send_content(content)
|
||||
mock_put_nowait.assert_called_once_with(LiveRequest(content=content))
|
||||
|
||||
|
||||
def test_send_realtime():
|
||||
queue = LiveRequestQueue()
|
||||
blob = MagicMock(spec=types.Blob)
|
||||
|
||||
with patch.object(queue._queue, "put_nowait") as mock_put_nowait:
|
||||
queue.send_realtime(blob)
|
||||
mock_put_nowait.assert_called_once_with(LiveRequest(blob=blob))
|
||||
|
||||
|
||||
def test_send():
|
||||
queue = LiveRequestQueue()
|
||||
req = LiveRequest(content=MagicMock(spec=types.Content))
|
||||
|
||||
with patch.object(queue._queue, "put_nowait") as mock_put_nowait:
|
||||
queue.send(req)
|
||||
mock_put_nowait.assert_called_once_with(req)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get():
|
||||
queue = LiveRequestQueue()
|
||||
res = MagicMock(spec=types.Content)
|
||||
|
||||
with patch.object(queue._queue, "get", return_value=res) as mock_get:
|
||||
result = await queue.get()
|
||||
|
||||
assert result == res
|
||||
mock_get.assert_called_once()
|
||||
Reference in New Issue
Block a user