import pytest from unittest.mock import AsyncMock, MagicMock from voice_assistant.agent.adapter import AgentAdapter def test_deep_agent_implements_port(): from voice_assistant.agent.deep_agent import DeepAgentsAdapter assert issubclass(DeepAgentsAdapter, AgentAdapter) assert callable(getattr(DeepAgentsAdapter, "stream", None)) assert callable(getattr(DeepAgentsAdapter, "aclose", None)) def test_deep_agent_session_config(): from voice_assistant.agent.deep_agent import DeepAgentsAdapter adapter = object.__new__(DeepAgentsAdapter) assert adapter._session_config("session-a") == { "configurable": { "thread_id": "session-a", } } assert adapter._session_config("session-b") == { "configurable": { "thread_id": "session-b", } } with pytest.raises(ValueError): adapter._session_config("") def test_deep_agent_cleanup_is_idempotent(): import asyncio from voice_assistant.agent.deep_agent import DeepAgentsAdapter adapter = object.__new__(DeepAgentsAdapter) adapter._model = MagicMock() adapter._model.aclose = AsyncMock() adapter._closed = False async def scenario(): await adapter.aclose() await adapter.aclose() asyncio.run(scenario()) adapter._model.aclose.assert_awaited_once() assert adapter._closed is True def test_deep_agent_stream_can_be_cancelled(): import asyncio from voice_assistant.agent.deep_agent import DeepAgentsAdapter async def scenario(): adapter = object.__new__(DeepAgentsAdapter) adapter._closed = False adapter._stream_event_timeout = 1.0 started = asyncio.Event() cancelled = asyncio.Event() class FakeAgent: async def astream(self, *args, **kwargs): started.set() try: yield ( FakeMessage("Primeiro fragmento"), {"langgraph_node": "model"}, ) await asyncio.Event().wait() except asyncio.CancelledError: cancelled.set() raise class FakeMessage: def __init__(self, content): self.content = content self.tool_call_chunks = [] adapter._graph = FakeAgent() async def consume(): async for _ in adapter.stream( session_id="session-cancel", turn_id=1, text="Teste de cancelamento", ): pass task = asyncio.create_task(consume()) await asyncio.wait_for(started.wait(), timeout=2.0) task.cancel() try: await task except asyncio.CancelledError: pass assert cancelled.is_set() assert task.done() asyncio.run(scenario()) def test_deep_agent_isolates_session_history(): import asyncio from deepagents import create_deep_agent from langchain_core.messages import HumanMessage from langchain_core.language_models import FakeListChatModel from langgraph.checkpoint.memory import MemorySaver async def scenario(): checkpointer = MemorySaver() class ToolCompatibleFakeChatModel(FakeListChatModel): def bind_tools(self, tools, *, tool_choice=None, **kwargs): return self model = ToolCompatibleFakeChatModel( responses=[ "Resposta A1", "Resposta B1", "Resposta A2", ] ) agent = create_deep_agent( model=model, tools=[], system_prompt="Responda de forma breve.", checkpointer=checkpointer, ) async def ask(session_id, text): return await agent.ainvoke( { "messages": [ HumanMessage(content=text) ] }, config={ "configurable": { "thread_id": session_id, } }, ) await ask("session-a", "Destino: Japão") await ask("session-b", "Destino: Canadá") await ask("session-a", "Qual era meu destino?") def history(session_id): snapshot = agent.get_state({ "configurable": { "thread_id": session_id, } }) return snapshot.values.get("messages", []) history_a = history("session-a") history_b = history("session-b") texts_a = [ message.content for message in history_a if isinstance(message, HumanMessage) ] texts_b = [ message.content for message in history_b if isinstance(message, HumanMessage) ] assert texts_a == [ "Destino: Japão", "Qual era meu destino?", ] assert texts_b == [ "Destino: Canadá", ] assert all( "Canadá" not in text for text in texts_a ) assert all( "Japão" not in text for text in texts_b ) asyncio.run(scenario()) def test_deep_agent_checkpoint_survives_cancellation(): import asyncio from deepagents import create_deep_agent from langchain_core.language_models import FakeListChatModel from langchain_core.messages import HumanMessage from langgraph.checkpoint.memory import MemorySaver class CancellableFakeModel(FakeListChatModel): def bind_tools(self, tools, *, tool_choice=None, **kwargs): return self async def _agenerate( self, messages, stop=None, run_manager=None, **kwargs, ): last_user_message = next( ( message for message in reversed(messages) if isinstance(message, HumanMessage) ), None, ) if ( last_user_message is not None and last_user_message.content == "Interromper esta resposta" ): started.set() await asyncio.Event().wait() return await super()._agenerate( messages, stop=stop, run_manager=run_manager, **kwargs, ) async def scenario(): nonlocal started started = asyncio.Event() model = CancellableFakeModel( responses=[ "Contexto inicial confirmado.", "Contexto preservado.", ] ) agent = create_deep_agent( model=model, tools=[], system_prompt="Responda brevemente.", checkpointer=MemorySaver(), ) config = { "configurable": { "thread_id": "session-cancellation-test", } } async def ask(text): return await agent.ainvoke( { "messages": [ HumanMessage(content=text) ] }, config=config, ) # First turn completes successfully. await ask("Meu destino é Japão") # Second turn begins but never completes. task = asyncio.create_task( ask("Interromper esta resposta") ) await asyncio.wait_for( started.wait(), timeout=2.0, ) task.cancel() try: await task except asyncio.CancelledError: pass assert task.done() assert task.cancelled() # The same thread must remain usable. await asyncio.wait_for( ask("Qual era meu destino?"), timeout=5.0, ) snapshot = await agent.aget_state(config) messages = snapshot.values.get("messages", []) human_messages = [ message.content for message in messages if isinstance(message, HumanMessage) ] assert "Meu destino é Japão" in human_messages assert "Qual era meu destino?" in human_messages # An interrupted execution must not be recorded # as a completed assistant response. assert not any( getattr(message, "content", None) == "Resposta interrompida concluída." for message in messages ) started = None asyncio.run(scenario()) @pytest.mark.parametrize( ("options", "expected_auth"), [ ({}, ("INSTANCE_PRINCIPAL", "DEFAULT", "~/.oci/config")), ( {"auth_type": "API_KEY", "auth_profile": "LOCAL", "auth_file_location": "/config/oci"}, ("API_KEY", "LOCAL", "/config/oci"), ), ( {"auth_type": "SECURITY_TOKEN", "auth_profile": "SESSION"}, ("SECURITY_TOKEN", "SESSION", "~/.oci/config"), ), ({"auth_type": "RESOURCE_PRINCIPAL"}, ("RESOURCE_PRINCIPAL", "DEFAULT", "~/.oci/config")), ], ) def test_deep_agent_factory_configures_authentication(monkeypatch, options, expected_auth): import voice_assistant.agent.deep_agent as module from voice_assistant.config.loader import load_provider_catalog from voice_assistant.config.models import AppConfig from voice_assistant.providers import create_agent_adapter model_factory = MagicMock() agent_factory = MagicMock() monkeypatch.setattr(module, "ChatOCIGenAI", model_factory) monkeypatch.setattr(module, "create_deep_agent", agent_factory) config = AppConfig.model_validate({ "stt": {"provider": "nvidia"}, "tts": {"provider": "voxcpm"}, "agent": { "provider": "deepagents", "options": { "region": "sa-vinhedo-1", "compartment_id": "test-compartment", "model_id": "test-model", "temperature": 0.3, "max_completion_tokens": 256, "timeout_seconds": 60.0, **options, }, }, }) adapter = create_agent_adapter(config=config, catalog=load_provider_catalog()) model_factory.assert_called_once() kwargs = model_factory.call_args.kwargs assert (kwargs["auth_type"], kwargs["auth_profile"], kwargs["auth_file_location"]) == expected_auth assert adapter._model is model_factory.return_value assert agent_factory.call_args.kwargs["model"] is adapter._model