generated from coulomb/repo-seed
tegwick
019f6e7dc7
Wire LLMConnectAdapter behind the existing LLMAdapter seam with config-driven selection, graceful degradation, --offline mode, and bounded session context. Add unit tests, integration docs, and update README/SCOPE/AGENTS.
111 lines
3.8 KiB
Python
111 lines
3.8 KiB
Python
"""LLMConnectAdapter unit tests with mocked llm-connect (CYA-WP-0008)."""
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from cya.config import LLMSettings
|
|
from cya.llm.adapter import AssistanceRequest
|
|
from cya.llm.connect_adapter import LLMConnectAdapter
|
|
|
|
|
|
def _mock_llm_response(content: str = "Try: git status"):
|
|
resp = MagicMock()
|
|
resp.content = content
|
|
resp.model = "mock/model"
|
|
resp.usage = {"total_tokens": 42}
|
|
resp.finish_reason = "stop"
|
|
return resp
|
|
|
|
|
|
@patch("llm_connect.create_adapter")
|
|
@patch("llm_connect.config.resolve_api_key", return_value="test-key")
|
|
def test_complete_delegates_to_llm_connect(mock_resolve, mock_create):
|
|
client = MagicMock()
|
|
client.execute_prompt.return_value = _mock_llm_response()
|
|
mock_create.return_value = client
|
|
|
|
settings = LLMSettings(adapter="connect", backend="mock", model="mock/model", configured=True)
|
|
adapter = LLMConnectAdapter(settings)
|
|
response = adapter.complete(
|
|
AssistanceRequest(user_request="show git status", context={"cwd": "/tmp"})
|
|
)
|
|
|
|
assert "git status" in response.suggestion.lower()
|
|
assert response.metadata.get("adapter") == "LLMConnectAdapter"
|
|
assert response.metadata.get("degraded") is not True
|
|
client.execute_prompt.assert_called_once()
|
|
|
|
|
|
def test_graceful_degrade_when_llm_connect_missing(monkeypatch):
|
|
import builtins
|
|
|
|
real_import = builtins.__import__
|
|
|
|
def _import(name, *args, **kwargs):
|
|
if name == "llm_connect" or name.startswith("llm_connect."):
|
|
raise ImportError("no llm_connect")
|
|
return real_import(name, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(builtins, "__import__", _import)
|
|
settings = LLMSettings(adapter="connect", backend="openrouter", configured=True)
|
|
adapter = LLMConnectAdapter(settings)
|
|
response = adapter.complete(AssistanceRequest(user_request="hello"))
|
|
|
|
assert response.metadata.get("degraded") is True
|
|
assert "llm-connect" in response.suggestion
|
|
|
|
|
|
@patch("llm_connect.create_adapter")
|
|
@patch("llm_connect.config.resolve_api_key", return_value=None)
|
|
def test_graceful_degrade_when_api_key_missing(mock_resolve, mock_create):
|
|
settings = LLMSettings(adapter="connect", backend="openrouter", configured=True)
|
|
adapter = LLMConnectAdapter(settings)
|
|
response = adapter.complete(AssistanceRequest(user_request="hello"))
|
|
|
|
assert response.metadata.get("degraded") is True
|
|
assert "API key" in response.suggestion
|
|
mock_create.assert_not_called()
|
|
|
|
|
|
@patch("llm_connect.create_adapter")
|
|
@patch("llm_connect.config.resolve_api_key", return_value="test-key")
|
|
def test_session_turns_included_in_prompt(mock_resolve, mock_create):
|
|
client = MagicMock()
|
|
client.execute_prompt.return_value = _mock_llm_response("ok")
|
|
mock_create.return_value = client
|
|
|
|
settings = LLMSettings(adapter="connect", backend="mock", configured=True)
|
|
adapter = LLMConnectAdapter(settings)
|
|
adapter.complete(
|
|
AssistanceRequest(
|
|
user_request="follow up",
|
|
context={
|
|
"session_turns": [{"user": "first", "assistant": "reply"}],
|
|
},
|
|
)
|
|
)
|
|
|
|
_prompt_arg, _ = client.execute_prompt.call_args[0]
|
|
assert "first" in _prompt_arg
|
|
assert "follow up" in _prompt_arg
|
|
|
|
|
|
@pytest.mark.llm_live
|
|
def test_live_openrouter_smoke():
|
|
"""Manual verification only — skipped unless OPENROUTER_API_KEY is set."""
|
|
import os
|
|
|
|
if not os.environ.get("OPENROUTER_API_KEY"):
|
|
pytest.skip("OPENROUTER_API_KEY not set")
|
|
|
|
settings = LLMSettings(
|
|
adapter="connect",
|
|
backend="openrouter",
|
|
model="anthropic/claude-sonnet-4",
|
|
configured=True,
|
|
)
|
|
adapter = LLMConnectAdapter(settings)
|
|
response = adapter.complete(AssistanceRequest(user_request="Reply with exactly: pong"))
|
|
assert response.metadata.get("degraded") is not True
|
|
assert response.suggestion |