import asyncio import threading from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock import pytest from voice_assistant.agent.oci_deep_agent import OCIDeepAgentsAdapter from voice_assistant.config.loader import load_provider_catalog from voice_assistant.config.models import AppConfig from voice_assistant.providers import create_agent_adapter def options(**overrides): return { "region": "sa-vinhedo-1", "compartment_id": "test", "model_id": "test-model", "temperature": 0.3, "max_completion_tokens": 256, "timeout_seconds": 1.0, **overrides, } def fake_model(): model = MagicMock() model.aclose = AsyncMock() return model def patch_runtime(monkeypatch, model_factory=None, graph_factory=None): import voice_assistant.agent.oci_deep_agent as module model_factory = model_factory or MagicMock(return_value=fake_model()) graph_factory = graph_factory or MagicMock() monkeypatch.setattr(module, "ChatOCIGenAI", model_factory) monkeypatch.setattr(module, "create_deep_agent", graph_factory) return model_factory, graph_factory @pytest.mark.parametrize( ("auth", "expected"), [ ({}, ("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_catalog_selects_oci_composition_and_initializes_once( monkeypatch, auth, expected ): model_factory, graph_factory = patch_runtime(monkeypatch) config = AppConfig.model_validate( { "stt": {"provider": "nvidia"}, "tts": {"provider": "voxcpm"}, "agent": {"provider": "deepagents", "options": options(**auth)}, } ) async def scenario(): adapter = create_agent_adapter(config=config, catalog=load_provider_catalog()) assert isinstance(adapter, OCIDeepAgentsAdapter) model_factory.assert_not_called() await asyncio.gather(adapter.initialize(), adapter.initialize()) model_factory.assert_called_once() kwargs = model_factory.call_args.kwargs assert ( kwargs["auth_type"], kwargs["auth_profile"], kwargs["auth_file_location"], ) == expected assert graph_factory.call_args.kwargs["model"] is model_factory.return_value await adapter.aclose() await adapter.aclose() model_factory.return_value.aclose.assert_awaited_once() model_factory.return_value.client.base_client.session.close.assert_called_once() asyncio.run(scenario()) def test_model_authentication_does_not_block_event_loop(monkeypatch): entered, release = threading.Event(), threading.Event() model = fake_model() def factory(**kwargs): entered.set() assert release.wait(2) return model patch_runtime(monkeypatch, factory) async def scenario(): adapter = OCIDeepAgentsAdapter(**options()) task = asyncio.create_task(adapter.initialize()) try: async with asyncio.timeout(1): while not entered.is_set(): await asyncio.sleep(0.001) await asyncio.sleep(0.01) assert not task.done() finally: release.set() await task await adapter.aclose() asyncio.run(scenario()) @pytest.mark.parametrize("abandon", ["timeout", "cancel", "close"]) def test_abandoned_initialization_retains_ownership_and_closes_late_model( monkeypatch, abandon ): entered, release = threading.Event(), threading.Event() model = fake_model() def factory(**kwargs): entered.set() assert release.wait(2) return model _, graph_factory = patch_runtime(monkeypatch, factory) async def scenario(): adapter = OCIDeepAgentsAdapter( **options(initialization_timeout_seconds=0.05, cleanup_timeout_seconds=0.01) ) task = asyncio.create_task(adapter.initialize()) try: async with asyncio.timeout(1): while not entered.is_set(): await asyncio.sleep(0.001) if abandon == "cancel": task.cancel() with pytest.raises(asyncio.CancelledError): await task elif abandon == "timeout": with pytest.raises(TimeoutError): await task else: with pytest.raises(TimeoutError, match="ownership retained"): await adapter.aclose() with pytest.raises(TimeoutError, match="ownership retained"): await adapter.aclose() assert not adapter._initialization.done() finally: release.set() await asyncio.wait_for(adapter._initialization, 1) await asyncio.gather(task, return_exceptions=True) await adapter.aclose() model.aclose.assert_awaited_once() model.client.base_client.session.close.assert_called_once() graph_factory.assert_not_called() asyncio.run(scenario()) def test_graph_failure_closes_partially_initialized_model(monkeypatch): model = fake_model() patch_runtime( monkeypatch, MagicMock(return_value=model), MagicMock(side_effect=ValueError("graph failed")), ) async def scenario(): adapter = OCIDeepAgentsAdapter(**options()) with pytest.raises(ValueError, match="graph failed"): await adapter.initialize() await adapter.aclose() model.aclose.assert_awaited_once() model.client.base_client.session.close.assert_called_once() asyncio.run(scenario()) def test_composition_closes_graph_stream_on_early_exit(monkeypatch): closed = asyncio.Event() async def source(*args, **kwargs): try: yield ( SimpleNamespace(content="Olá", tool_call_chunks=[]), {"langgraph_node": "model"}, ) await asyncio.Event().wait() finally: closed.set() graph = SimpleNamespace(astream=source) patch_runtime(monkeypatch, graph_factory=MagicMock(return_value=graph)) async def scenario(): adapter = OCIDeepAgentsAdapter(**options()) stream = adapter.stream("session", 1, "Olá") assert await anext(stream) == "Olá" await stream.aclose() assert closed.is_set() await adapter.aclose() asyncio.run(scenario()) @pytest.mark.parametrize( "field", ["timeout_seconds", "initialization_timeout_seconds", "cleanup_timeout_seconds"], ) @pytest.mark.parametrize("value", [0, -1, float("inf"), float("nan"), True]) def test_invalid_deadline_fails_before_creating_model(monkeypatch, field, value): model_factory, _ = patch_runtime(monkeypatch) with pytest.raises(ValueError, match="finite and positive"): OCIDeepAgentsAdapter(**options(**{field: value})) model_factory.assert_not_called() def test_gateway_awaits_initialization_before_creating_other_providers(monkeypatch): from voice_assistant.agent.mock import MockAgentAdapter from voice_assistant.audio_gateway.app import main from voice_assistant.tts.mock import MockTTSAdapter order = [] class Agent(MockAgentAdapter): async def initialize(self): order.append("initialized") def tts_factory(**kwargs): assert order == ["initialized"] return MockTTSAdapter() monkeypatch.setattr(main, "create_agent_adapter", lambda **kwargs: Agent()) monkeypatch.setattr(main, "create_tts_adapter", tts_factory) async def scenario(): session = await main.create_voice_session(AsyncMock(), None) await session.aclose() asyncio.run(scenario()) def test_gateway_closes_agent_when_initialization_fails(monkeypatch): from voice_assistant.agent.mock import MockAgentAdapter from voice_assistant.audio_gateway.app import main agent = MockAgentAdapter() agent.initialize = AsyncMock(side_effect=TimeoutError("initialization deadline")) agent.aclose = AsyncMock() tts_factory = MagicMock() monkeypatch.setattr(main, "create_agent_adapter", lambda **kwargs: agent) monkeypatch.setattr(main, "create_tts_adapter", tts_factory) async def scenario(): with pytest.raises(TimeoutError, match="initialization deadline"): await main.create_voice_session(AsyncMock(), None) agent.aclose.assert_awaited_once() tts_factory.assert_not_called() asyncio.run(scenario())