274 lines
8.9 KiB
Python
274 lines
8.9 KiB
Python
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())
|