1773 lines
56 KiB
Python
1773 lines
56 KiB
Python
import json
|
|
import subprocess
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import httpx
|
|
import pytest
|
|
from agents.models import AgentSession, AgentSessionLogData
|
|
from agents.providers import (
|
|
_parse_opencode_events,
|
|
agent_prompt,
|
|
friendly_error_message,
|
|
gemini_agent_prompt,
|
|
groq_agent_prompt,
|
|
is_transient_error,
|
|
list_gemini_agent_models,
|
|
list_groq_agent_models,
|
|
list_mistral_agent_models,
|
|
list_openrouter_free_models,
|
|
mistral_agent_prompt,
|
|
opencode_agent_prompt,
|
|
openrouter_agent_prompt,
|
|
retry_after_from,
|
|
)
|
|
from django.conf import settings
|
|
from django.contrib.auth import get_user_model
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
from scrobbles.models import Scrobble
|
|
from scrobbles.notifications import (
|
|
AgentSessionFailedNtfyNotification,
|
|
AgentSessionResponseNtfyNotification,
|
|
)
|
|
from scrobbles.scrobblers import (
|
|
manual_scrobble_agent_follow_up,
|
|
manual_scrobble_agent_session,
|
|
)
|
|
from scrobbles.tasks import (
|
|
AGENT_SESSION_AUTO_COMPLETE_SECONDS,
|
|
scrobble_agent_session_complete,
|
|
scrobble_agent_session_prompt,
|
|
)
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
@pytest.fixture
|
|
def user(db):
|
|
return User.objects.create_user(username="agentuser", password="pw")
|
|
|
|
|
|
def _mk_scrobble(user, *, in_progress=True, turns=None, title="First prompt"):
|
|
agent_session, _ = AgentSession.find_or_create(
|
|
provider=settings.AGENT_PROVIDER, model=settings.LLM_MODEL
|
|
)
|
|
scrobble = Scrobble.create_or_update(
|
|
agent_session,
|
|
user.id,
|
|
{
|
|
"user_id": user.id,
|
|
"timestamp": timezone.now(),
|
|
"playback_position_seconds": 0,
|
|
"source": "Vrobbler",
|
|
},
|
|
skip_in_progress_check=True,
|
|
)
|
|
log = scrobble.log if isinstance(scrobble.log, dict) else {}
|
|
log["provider"] = settings.AGENT_PROVIDER
|
|
log["model"] = settings.LLM_MODEL
|
|
log["title"] = title
|
|
log["turns"] = turns or []
|
|
scrobble.log = log
|
|
scrobble.in_progress = in_progress
|
|
scrobble.save(update_fields=["log", "in_progress"])
|
|
return scrobble
|
|
|
|
|
|
# --- providers ---
|
|
|
|
|
|
def test_parse_opencode_events_joins_text_parts():
|
|
raw = "\n".join(
|
|
[
|
|
json.dumps({"type": "step_start"}),
|
|
json.dumps({"type": "text", "part": {"type": "text", "text": "hello"}}),
|
|
json.dumps({"type": "text", "part": {"type": "text", "text": " world"}}),
|
|
json.dumps({"type": "step_finish"}),
|
|
"not-json",
|
|
]
|
|
)
|
|
assert _parse_opencode_events(raw) == "hello\n world"
|
|
|
|
|
|
def test_opencode_agent_prompt(settings):
|
|
settings.OPENCODE_CMD = "opencode"
|
|
events = json.dumps(
|
|
{"type": "text", "part": {"type": "text", "text": "the answer"}}
|
|
)
|
|
with patch(
|
|
"agents.providers.subprocess.run",
|
|
return_value=MagicMock(returncode=0, stdout=events, stderr=""),
|
|
) as mock_run:
|
|
result = opencode_agent_prompt("what is 2+2?")
|
|
|
|
mock_run.assert_called_once()
|
|
cmd = mock_run.call_args.args[0]
|
|
assert cmd[:2] == ["opencode", "run"]
|
|
assert cmd[2:5] == ["--format", "json", "--dir"]
|
|
assert cmd[5].startswith("/tmp/")
|
|
assert cmd[6] == "what is 2+2?"
|
|
assert result == {"text": "the answer", "provider": "opencode", "model": "opencode"}
|
|
|
|
|
|
def test_opencode_agent_prompt_nonzero_exit():
|
|
with patch(
|
|
"agents.providers.subprocess.run",
|
|
return_value=MagicMock(returncode=1, stdout="", stderr="boom"),
|
|
):
|
|
with pytest.raises(ValueError, match="opencode exited with 1"):
|
|
opencode_agent_prompt("hi")
|
|
|
|
|
|
def test_opencode_agent_prompt_timeout():
|
|
with patch(
|
|
"agents.providers.subprocess.run",
|
|
side_effect=subprocess.TimeoutExpired("cmd", 900),
|
|
):
|
|
with pytest.raises(ValueError, match="timed out"):
|
|
opencode_agent_prompt("hi")
|
|
|
|
|
|
def test_opencode_agent_prompt_no_text_output():
|
|
with patch(
|
|
"agents.providers.subprocess.run",
|
|
return_value=MagicMock(
|
|
returncode=0, stdout=json.dumps({"type": "step_start"}), stderr=""
|
|
),
|
|
):
|
|
with pytest.raises(ValueError, match="no text output"):
|
|
opencode_agent_prompt("hi")
|
|
|
|
|
|
def test_gemini_agent_prompt(settings):
|
|
settings.GOOGLE_AI_API_KEY = "test-key"
|
|
settings.LLM_MODEL = "gemini-2.5-flash"
|
|
fake = MagicMock()
|
|
fake.raise_for_status = MagicMock()
|
|
fake.json.return_value = {
|
|
"candidates": [{"content": {"parts": [{"text": " the answer "}]}}]
|
|
}
|
|
with patch("agents.providers.httpx.post", return_value=fake) as mock_post:
|
|
result = gemini_agent_prompt("what is 2+2?")
|
|
|
|
mock_post.assert_called_once()
|
|
kwargs = mock_post.call_args.kwargs
|
|
assert kwargs["params"] == {"key": "test-key"}
|
|
assert kwargs["json"] == {"contents": [{"parts": [{"text": "what is 2+2?"}]}]}
|
|
assert result == {
|
|
"text": "the answer",
|
|
"provider": "gemini",
|
|
"model": "gemini-2.5-flash",
|
|
}
|
|
|
|
|
|
def test_gemini_agent_prompt_missing_key(settings):
|
|
settings.GOOGLE_AI_API_KEY = ""
|
|
with pytest.raises(ValueError, match="VROBBLER_GOOGLE_AI_API_KEY is not set"):
|
|
gemini_agent_prompt("hi")
|
|
|
|
|
|
def test_gemini_agent_prompt_bad_response(settings):
|
|
settings.GOOGLE_AI_API_KEY = "test-key"
|
|
fake = MagicMock()
|
|
fake.raise_for_status = MagicMock()
|
|
fake.json.return_value = {"unexpected": True}
|
|
with patch("agents.providers.httpx.post", return_value=fake):
|
|
with pytest.raises(ValueError, match="Unexpected Gemini response"):
|
|
gemini_agent_prompt("hi")
|
|
|
|
|
|
def test_gemini_agent_prompt_with_history(settings):
|
|
settings.GOOGLE_AI_API_KEY = "test-key"
|
|
fake = MagicMock()
|
|
fake.raise_for_status = MagicMock()
|
|
fake.json.return_value = {"candidates": [{"content": {"parts": [{"text": "12"}]}}]}
|
|
history = [
|
|
{"prompt": "what is 2+2?", "response": "4"},
|
|
{"prompt": "no response yet", "response": None},
|
|
]
|
|
with patch("agents.providers.httpx.post", return_value=fake) as mock_post:
|
|
gemini_agent_prompt("times 3?", history=history)
|
|
|
|
contents = mock_post.call_args.kwargs["json"]["contents"]
|
|
assert contents == [
|
|
{"parts": [{"text": "what is 2+2?"}]},
|
|
{"parts": [{"text": "4"}]},
|
|
{"parts": [{"text": "times 3?"}]},
|
|
]
|
|
|
|
|
|
def test_opencode_agent_prompt_with_history(settings):
|
|
settings.OPENCODE_CMD = "opencode"
|
|
with patch(
|
|
"agents.providers.subprocess.run",
|
|
return_value=MagicMock(
|
|
returncode=0,
|
|
stdout=json.dumps({"type": "text", "part": {"type": "text", "text": "x"}}),
|
|
stderr="",
|
|
),
|
|
) as mock_run:
|
|
opencode_agent_prompt(
|
|
"times 3?", history=[{"prompt": "what is 2+2?", "response": "4"}]
|
|
)
|
|
|
|
prompt = mock_run.call_args.args[0][6]
|
|
assert "Previous conversation:" in prompt
|
|
assert "User: what is 2+2?" in prompt
|
|
assert "Assistant: 4" in prompt
|
|
assert "times 3?" in prompt
|
|
|
|
|
|
def test_agent_prompt_unknown_provider(settings):
|
|
settings.AGENT_PROVIDER = "nope"
|
|
with pytest.raises(ValueError, match="Unknown agent provider"):
|
|
agent_prompt("hi")
|
|
|
|
|
|
def test_agent_prompt_defaults_to_configured_provider(settings):
|
|
settings.AGENT_PROVIDER = "gemini"
|
|
settings.GOOGLE_AI_API_KEY = "test-key"
|
|
with patch("agents.providers.gemini_agent_prompt", return_value={"text": "x"}) as m:
|
|
agent_prompt("hi")
|
|
m.assert_called_once_with("hi", history=None, model=None)
|
|
|
|
|
|
def test_agent_prompt_openrouter_dispatches(settings):
|
|
settings.AGENT_PROVIDER = "openrouter"
|
|
settings.OPENROUTER_API_KEY = "sk-test"
|
|
with patch(
|
|
"agents.providers.openrouter_agent_prompt", return_value={"text": "x"}
|
|
) as m:
|
|
agent_prompt("hi", model="free/model-a")
|
|
m.assert_called_once_with("hi", history=None, model="free/model-a")
|
|
|
|
|
|
def test_agent_prompt_groq_dispatches(settings):
|
|
settings.AGENT_PROVIDER = "groq"
|
|
settings.GROQ_API_KEY = "gsk-test"
|
|
with patch("agents.providers.groq_agent_prompt", return_value={"text": "x"}) as m:
|
|
agent_prompt("hi", model="llama-3.3-70b-versatile")
|
|
m.assert_called_once_with("hi", history=None, model="llama-3.3-70b-versatile")
|
|
|
|
|
|
def test_agent_prompt_mistral_dispatches(settings):
|
|
settings.AGENT_PROVIDER = "mistral"
|
|
settings.MISTRAL_API_KEY = "msk-test"
|
|
with patch(
|
|
"agents.providers.mistral_agent_prompt", return_value={"text": "x"}
|
|
) as m:
|
|
agent_prompt("hi", model="mistral-small-latest")
|
|
m.assert_called_once_with("hi", history=None, model="mistral-small-latest")
|
|
|
|
|
|
def test_openrouter_agent_prompt(settings):
|
|
settings.OPENROUTER_API_KEY = "sk-test"
|
|
settings.LLM_MODEL = "some/model"
|
|
fake = MagicMock()
|
|
fake.raise_for_status = MagicMock()
|
|
fake.json.return_value = {"choices": [{"message": {"content": " the answer "}}]}
|
|
with patch("agents.providers.httpx.post", return_value=fake) as mock_post:
|
|
result = openrouter_agent_prompt("what is 2+2?")
|
|
|
|
mock_post.assert_called_once()
|
|
kwargs = mock_post.call_args.kwargs
|
|
assert kwargs["headers"] == {"Authorization": "Bearer sk-test"}
|
|
assert kwargs["json"] == {
|
|
"model": "some/model",
|
|
"messages": [{"role": "user", "content": "what is 2+2?"}],
|
|
}
|
|
assert result == {
|
|
"text": "the answer",
|
|
"provider": "openrouter",
|
|
"model": "some/model",
|
|
}
|
|
|
|
|
|
def test_openrouter_agent_prompt_missing_key(settings):
|
|
settings.OPENROUTER_API_KEY = ""
|
|
with pytest.raises(ValueError, match="VROBBLER_OPENROUTER_API_KEY is not set"):
|
|
openrouter_agent_prompt("hi")
|
|
|
|
|
|
def test_openrouter_agent_prompt_bad_response(settings):
|
|
settings.OPENROUTER_API_KEY = "sk-test"
|
|
fake = MagicMock()
|
|
fake.raise_for_status = MagicMock()
|
|
fake.json.return_value = {"unexpected": True}
|
|
with patch("agents.providers.httpx.post", return_value=fake):
|
|
with pytest.raises(ValueError, match="Unexpected OpenRouter response"):
|
|
openrouter_agent_prompt("hi")
|
|
|
|
|
|
def test_openrouter_agent_prompt_with_history(settings):
|
|
settings.OPENROUTER_API_KEY = "sk-test"
|
|
fake = MagicMock()
|
|
fake.raise_for_status = MagicMock()
|
|
fake.json.return_value = {"choices": [{"message": {"content": "12"}}]}
|
|
history = [
|
|
{"prompt": "what is 2+2?", "response": "4"},
|
|
{"prompt": "no response yet", "response": None},
|
|
]
|
|
with patch("agents.providers.httpx.post", return_value=fake) as mock_post:
|
|
openrouter_agent_prompt("times 3?", history=history, model="some/model")
|
|
|
|
messages = mock_post.call_args.kwargs["json"]["messages"]
|
|
assert messages == [
|
|
{"role": "user", "content": "what is 2+2?"},
|
|
{"role": "assistant", "content": "4"},
|
|
{"role": "user", "content": "times 3?"},
|
|
]
|
|
|
|
|
|
def test_groq_agent_prompt(settings):
|
|
settings.GROQ_API_KEY = "gsk-test"
|
|
settings.LLM_MODEL = "llama-3.3-70b-versatile"
|
|
fake = MagicMock()
|
|
fake.raise_for_status = MagicMock()
|
|
fake.json.return_value = {"choices": [{"message": {"content": " the answer "}}]}
|
|
with patch("agents.providers.httpx.post", return_value=fake) as mock_post:
|
|
result = groq_agent_prompt("what is 2+2?")
|
|
|
|
mock_post.assert_called_once()
|
|
assert (
|
|
mock_post.call_args.args[0] == "https://api.groq.com/openai/v1/chat/completions"
|
|
)
|
|
kwargs = mock_post.call_args.kwargs
|
|
assert kwargs["headers"] == {"Authorization": "Bearer gsk-test"}
|
|
assert kwargs["json"] == {
|
|
"model": "llama-3.3-70b-versatile",
|
|
"messages": [{"role": "user", "content": "what is 2+2?"}],
|
|
}
|
|
assert result == {
|
|
"text": "the answer",
|
|
"provider": "groq",
|
|
"model": "llama-3.3-70b-versatile",
|
|
}
|
|
|
|
|
|
def test_groq_agent_prompt_missing_key(settings):
|
|
settings.GROQ_API_KEY = ""
|
|
with pytest.raises(ValueError, match="VROBBLER_GROQ_API_KEY is not set"):
|
|
groq_agent_prompt("hi")
|
|
|
|
|
|
def test_groq_agent_prompt_bad_response(settings):
|
|
settings.GROQ_API_KEY = "gsk-test"
|
|
fake = MagicMock()
|
|
fake.raise_for_status = MagicMock()
|
|
fake.json.return_value = {"unexpected": True}
|
|
with patch("agents.providers.httpx.post", return_value=fake):
|
|
with pytest.raises(ValueError, match="Unexpected Groq response"):
|
|
groq_agent_prompt("hi")
|
|
|
|
|
|
def test_groq_agent_prompt_with_history(settings):
|
|
settings.GROQ_API_KEY = "gsk-test"
|
|
fake = MagicMock()
|
|
fake.raise_for_status = MagicMock()
|
|
fake.json.return_value = {"choices": [{"message": {"content": "12"}}]}
|
|
history = [
|
|
{"prompt": "what is 2+2?", "response": "4"},
|
|
{"prompt": "no response yet", "response": None},
|
|
]
|
|
with patch("agents.providers.httpx.post", return_value=fake) as mock_post:
|
|
groq_agent_prompt("times 3?", history=history, model="llama-3.3-70b-versatile")
|
|
|
|
messages = mock_post.call_args.kwargs["json"]["messages"]
|
|
assert messages == [
|
|
{"role": "user", "content": "what is 2+2?"},
|
|
{"role": "assistant", "content": "4"},
|
|
{"role": "user", "content": "times 3?"},
|
|
]
|
|
|
|
|
|
def test_mistral_agent_prompt(settings):
|
|
settings.MISTRAL_API_KEY = "msk-test"
|
|
settings.LLM_MODEL = "mistral-small-latest"
|
|
fake = MagicMock()
|
|
fake.raise_for_status = MagicMock()
|
|
fake.json.return_value = {"choices": [{"message": {"content": " the answer "}}]}
|
|
with patch("agents.providers.httpx.post", return_value=fake) as mock_post:
|
|
result = mistral_agent_prompt("what is 2+2?")
|
|
|
|
mock_post.assert_called_once()
|
|
assert mock_post.call_args.args[0] == "https://api.mistral.ai/v1/chat/completions"
|
|
kwargs = mock_post.call_args.kwargs
|
|
assert kwargs["headers"] == {"Authorization": "Bearer msk-test"}
|
|
assert kwargs["json"] == {
|
|
"model": "mistral-small-latest",
|
|
"messages": [{"role": "user", "content": "what is 2+2?"}],
|
|
}
|
|
assert result == {
|
|
"text": "the answer",
|
|
"provider": "mistral",
|
|
"model": "mistral-small-latest",
|
|
}
|
|
|
|
|
|
def test_mistral_agent_prompt_missing_key(settings):
|
|
settings.MISTRAL_API_KEY = ""
|
|
with pytest.raises(ValueError, match="VROBBLER_MISTRAL_API_KEY is not set"):
|
|
mistral_agent_prompt("hi")
|
|
|
|
|
|
def test_mistral_agent_prompt_bad_response(settings):
|
|
settings.MISTRAL_API_KEY = "msk-test"
|
|
fake = MagicMock()
|
|
fake.raise_for_status = MagicMock()
|
|
fake.json.return_value = {"unexpected": True}
|
|
with patch("agents.providers.httpx.post", return_value=fake):
|
|
with pytest.raises(ValueError, match="Unexpected Mistral response"):
|
|
mistral_agent_prompt("hi")
|
|
|
|
|
|
def test_mistral_agent_prompt_with_history(settings):
|
|
settings.MISTRAL_API_KEY = "msk-test"
|
|
fake = MagicMock()
|
|
fake.raise_for_status = MagicMock()
|
|
fake.json.return_value = {"choices": [{"message": {"content": "12"}}]}
|
|
history = [
|
|
{"prompt": "what is 2+2?", "response": "4"},
|
|
{"prompt": "no response yet", "response": None},
|
|
]
|
|
with patch("agents.providers.httpx.post", return_value=fake) as mock_post:
|
|
mistral_agent_prompt("times 3?", history=history, model="mistral-small-latest")
|
|
|
|
messages = mock_post.call_args.kwargs["json"]["messages"]
|
|
assert messages == [
|
|
{"role": "user", "content": "what is 2+2?"},
|
|
{"role": "assistant", "content": "4"},
|
|
{"role": "user", "content": "times 3?"},
|
|
]
|
|
|
|
|
|
def test_list_openrouter_free_models_filters_free():
|
|
fake = MagicMock()
|
|
fake.raise_for_status = MagicMock()
|
|
fake.json.return_value = {
|
|
"data": [
|
|
{
|
|
"id": "paid/model",
|
|
"name": "Paid",
|
|
"pricing": {"prompt": "0.001", "completion": "0.002"},
|
|
},
|
|
{
|
|
"id": "free/model-a",
|
|
"name": "Bee",
|
|
"pricing": {"prompt": "0", "completion": "0"},
|
|
},
|
|
{
|
|
"id": "free/model-b",
|
|
"name": "Alpha",
|
|
"pricing": {"prompt": "0", "completion": "0"},
|
|
},
|
|
]
|
|
}
|
|
with patch("agents.providers.httpx.get", return_value=fake) as mock_get:
|
|
models = list_openrouter_free_models()
|
|
|
|
mock_get.assert_called_once()
|
|
assert models == [
|
|
{"provider": "openrouter", "id": "free/model-b", "name": "Alpha"},
|
|
{"provider": "openrouter", "id": "free/model-a", "name": "Bee"},
|
|
]
|
|
|
|
|
|
def test_list_openrouter_free_models_skips_missing_pricing():
|
|
fake = MagicMock()
|
|
fake.raise_for_status = MagicMock()
|
|
fake.json.return_value = {
|
|
"data": [
|
|
{"id": "no-pricing/model", "name": "No Pricing"},
|
|
]
|
|
}
|
|
with patch("agents.providers.httpx.get", return_value=fake):
|
|
assert list_openrouter_free_models() == []
|
|
|
|
|
|
def test_list_gemini_agent_models(settings):
|
|
settings.GOOGLE_AI_API_KEY = "test-key"
|
|
settings.GEMINI_AGENT_MODELS = "gemini-2.5-flash, gemini-3.6-flash"
|
|
assert list_gemini_agent_models() == [
|
|
{
|
|
"provider": "gemini",
|
|
"id": "gemini-2.5-flash",
|
|
"name": "gemini-2.5-flash",
|
|
},
|
|
{
|
|
"provider": "gemini",
|
|
"id": "gemini-3.6-flash",
|
|
"name": "gemini-3.6-flash",
|
|
},
|
|
]
|
|
|
|
|
|
def test_list_gemini_agent_models_no_key(settings):
|
|
settings.GOOGLE_AI_API_KEY = ""
|
|
assert list_gemini_agent_models() == []
|
|
|
|
|
|
def test_list_groq_agent_models(settings):
|
|
settings.GROQ_API_KEY = "gsk-test"
|
|
settings.GROQ_AGENT_MODELS = "llama-3.3-70b-versatile, openai/gpt-oss-20b"
|
|
assert list_groq_agent_models() == [
|
|
{
|
|
"provider": "groq",
|
|
"id": "llama-3.3-70b-versatile",
|
|
"name": "llama-3.3-70b-versatile",
|
|
},
|
|
{
|
|
"provider": "groq",
|
|
"id": "openai/gpt-oss-20b",
|
|
"name": "openai/gpt-oss-20b",
|
|
},
|
|
]
|
|
|
|
|
|
def test_list_groq_agent_models_no_key(settings):
|
|
settings.GROQ_API_KEY = ""
|
|
assert list_groq_agent_models() == []
|
|
|
|
|
|
def test_list_mistral_agent_models(settings):
|
|
settings.MISTRAL_API_KEY = "msk-test"
|
|
settings.MISTRAL_AGENT_MODELS = "mistral-small-latest, codestral-latest"
|
|
assert list_mistral_agent_models() == [
|
|
{
|
|
"provider": "mistral",
|
|
"id": "mistral-small-latest",
|
|
"name": "mistral-small-latest",
|
|
},
|
|
{
|
|
"provider": "mistral",
|
|
"id": "codestral-latest",
|
|
"name": "codestral-latest",
|
|
},
|
|
]
|
|
|
|
|
|
def test_list_mistral_agent_models_no_key(settings):
|
|
settings.MISTRAL_API_KEY = ""
|
|
assert list_mistral_agent_models() == []
|
|
|
|
|
|
# --- retry helpers ---
|
|
|
|
|
|
def _http_error(status_code, headers=None):
|
|
request = httpx.Request("POST", "https://openrouter.ai/api/v1/chat/completions")
|
|
response = httpx.Response(status_code, request=request, headers=headers or {})
|
|
return httpx.HTTPStatusError("provider error", request=request, response=response)
|
|
|
|
|
|
def test_is_transient_error_429():
|
|
assert is_transient_error(_http_error(429)) is True
|
|
|
|
|
|
def test_is_transient_error_5xx():
|
|
assert is_transient_error(_http_error(500)) is True
|
|
assert is_transient_error(_http_error(502)) is True
|
|
assert is_transient_error(_http_error(503)) is True
|
|
assert is_transient_error(_http_error(504)) is True
|
|
|
|
|
|
def test_is_transient_error_4xx_not_retryable():
|
|
assert is_transient_error(_http_error(400)) is False
|
|
assert is_transient_error(_http_error(401)) is False
|
|
assert is_transient_error(_http_error(403)) is False
|
|
|
|
|
|
def test_is_transient_error_transport():
|
|
assert is_transient_error(httpx.ConnectError("boom")) is True
|
|
assert is_transient_error(httpx.ReadTimeout("boom")) is True
|
|
|
|
|
|
def test_is_transient_error_plain_exception():
|
|
assert is_transient_error(ValueError("boom")) is False
|
|
|
|
|
|
def test_retry_after_from_header():
|
|
assert retry_after_from(_http_error(429, {"Retry-After": "45"})) == 45
|
|
|
|
|
|
def test_retry_after_from_missing():
|
|
assert retry_after_from(_http_error(429)) is None
|
|
assert retry_after_from(ValueError("boom")) is None
|
|
|
|
|
|
def test_retry_after_from_invalid_header():
|
|
assert retry_after_from(_http_error(429, {"Retry-After": "soon"})) is None
|
|
|
|
|
|
def test_friendly_error_message_http():
|
|
msg = friendly_error_message(_http_error(429))
|
|
assert msg == "Provider returned HTTP 429 (Too Many Requests)"
|
|
|
|
|
|
def test_friendly_error_message_plain():
|
|
assert friendly_error_message(ValueError("boom")) == "boom"
|
|
|
|
|
|
# --- scrobbler ---
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_manual_scrobble_agent_session_creates_new_session(mock_delay, user):
|
|
scrobble = manual_scrobble_agent_session("What is the capital of France?", user.id)
|
|
|
|
assert Scrobble.objects.filter(id=scrobble.id).exists()
|
|
assert scrobble.media_type == Scrobble.MediaType.AGENT_SESSION
|
|
assert scrobble.in_progress is True
|
|
assert scrobble.agent_session.provider == settings.AGENT_PROVIDER
|
|
assert scrobble.agent_session.model == settings.LLM_MODEL
|
|
assert (
|
|
scrobble.agent_session.title
|
|
== f"{settings.AGENT_PROVIDER} {settings.LLM_MODEL}"
|
|
)
|
|
assert scrobble.log["title"] == "What is the capital of France?"
|
|
turns = scrobble.log["turns"]
|
|
assert len(turns) == 1
|
|
assert turns[0]["prompt"] == "What is the capital of France?"
|
|
assert turns[0]["response"] is None
|
|
mock_delay.assert_called_once_with(scrobble.id, turns[0]["prompt_id"])
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_manual_scrobble_agent_session_reuses_agent_session(mock_delay, user):
|
|
agent_session, created = AgentSession.find_or_create(
|
|
provider=settings.AGENT_PROVIDER, model=settings.LLM_MODEL
|
|
)
|
|
assert created is True
|
|
|
|
manual_scrobble_agent_session("First question?", user.id)
|
|
|
|
agent_session_again, created = AgentSession.find_or_create(
|
|
provider=settings.AGENT_PROVIDER, model=settings.LLM_MODEL
|
|
)
|
|
assert created is False
|
|
assert agent_session_again.id == agent_session.id
|
|
assert AgentSession.objects.count() == 1
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_manual_scrobble_agent_session_appends_to_in_progress(mock_delay, user):
|
|
_mk_scrobble(
|
|
user,
|
|
in_progress=True,
|
|
turns=[
|
|
{"prompt_id": "first-1", "prompt": "First prompt", "response": "answer"}
|
|
],
|
|
)
|
|
|
|
scrobble = manual_scrobble_agent_session("And the second question?", user.id)
|
|
|
|
turns = scrobble.log["turns"]
|
|
assert len(turns) == 2
|
|
assert turns[0]["prompt"] == "First prompt"
|
|
assert turns[1]["prompt"] == "And the second question?"
|
|
assert turns[1]["response"] is None
|
|
assert scrobble.log["title"] == "First prompt"
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_manual_scrobble_agent_session_new_session_after_completed(mock_delay, user):
|
|
_mk_scrobble(user, in_progress=False)
|
|
|
|
scrobble = manual_scrobble_agent_session("A new session?", user.id)
|
|
|
|
assert scrobble.media_type == Scrobble.MediaType.AGENT_SESSION
|
|
assert len(scrobble.log["turns"]) == 1
|
|
assert Scrobble.objects.filter(user_id=user.id).count() == 2
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_manual_scrobble_agent_session_opencode_model_empty(mock_delay, user, settings):
|
|
settings.AGENT_PROVIDER = "opencode"
|
|
|
|
scrobble = manual_scrobble_agent_session("hi", user.id)
|
|
|
|
assert scrobble.agent_session.provider == "opencode"
|
|
assert scrobble.agent_session.model == "opencode"
|
|
assert scrobble.agent_session.title == "opencode opencode"
|
|
|
|
|
|
# --- celery task ---
|
|
|
|
|
|
@patch("agents.providers.agent_prompt")
|
|
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
|
|
def test_scrobble_agent_session_prompt_fills_turn(
|
|
mock_complete, mock_agent_prompt, user
|
|
):
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": None,
|
|
}
|
|
],
|
|
)
|
|
mock_agent_prompt.return_value = {
|
|
"text": "hi back",
|
|
"provider": settings.AGENT_PROVIDER,
|
|
"model": settings.LLM_MODEL,
|
|
}
|
|
|
|
scrobble_agent_session_prompt(scrobble.id, "abc-123")
|
|
|
|
scrobble.refresh_from_db()
|
|
mock_agent_prompt.assert_called_once_with(
|
|
"hello", provider=settings.AGENT_PROVIDER, history=[], model=settings.LLM_MODEL
|
|
)
|
|
assert scrobble.log["turns"][0]["response"] == "hi back"
|
|
assert scrobble.log["provider"] == settings.AGENT_PROVIDER
|
|
assert scrobble.in_progress is True
|
|
assert scrobble.played_to_completion is False
|
|
mock_complete.assert_called_once_with(
|
|
args=[scrobble.id], countdown=AGENT_SESSION_AUTO_COMPLETE_SECONDS
|
|
)
|
|
|
|
|
|
@patch("agents.providers.agent_prompt")
|
|
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
|
|
def test_scrobble_agent_session_prompt_passes_history(
|
|
mock_complete, mock_agent_prompt, user
|
|
):
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
turns=[
|
|
{
|
|
"prompt_id": "first-1",
|
|
"prompt": "what is 2+2?",
|
|
"response": "4",
|
|
},
|
|
{
|
|
"prompt_id": "err-1",
|
|
"prompt": "boom?",
|
|
"response": "Error: boom",
|
|
"error": True,
|
|
},
|
|
{
|
|
"prompt_id": "cur-1",
|
|
"prompt": "and times 3?",
|
|
"response": None,
|
|
},
|
|
],
|
|
)
|
|
|
|
mock_agent_prompt.return_value = {
|
|
"text": "12",
|
|
"provider": settings.AGENT_PROVIDER,
|
|
"model": settings.LLM_MODEL,
|
|
}
|
|
scrobble_agent_session_prompt(scrobble.id, "cur-1")
|
|
|
|
mock_agent_prompt.assert_called_once_with(
|
|
"and times 3?",
|
|
provider=settings.AGENT_PROVIDER,
|
|
history=[{"prompt": "what is 2+2?", "response": "4"}],
|
|
model=settings.LLM_MODEL,
|
|
)
|
|
|
|
|
|
@patch("agents.providers.agent_prompt")
|
|
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
|
|
def test_scrobble_agent_session_prompt_stores_error(
|
|
mock_complete, mock_agent_prompt, user
|
|
):
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": None,
|
|
}
|
|
],
|
|
)
|
|
mock_agent_prompt.side_effect = ValueError("boom")
|
|
|
|
scrobble_agent_session_prompt(scrobble.id, "abc-123")
|
|
|
|
scrobble.refresh_from_db()
|
|
assert scrobble.log["turns"][0]["response"] == "Error: boom"
|
|
assert scrobble.log["turns"][0]["error"] is True
|
|
assert scrobble.in_progress is True
|
|
mock_complete.assert_called_once_with(
|
|
args=[scrobble.id], countdown=AGENT_SESSION_AUTO_COMPLETE_SECONDS
|
|
)
|
|
|
|
|
|
@patch("agents.providers.agent_prompt")
|
|
@patch("scrobbles.notifications.AgentSessionResponseNtfyNotification")
|
|
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
|
|
def test_scrobble_agent_session_prompt_sends_ntfy_on_response(
|
|
mock_complete, mock_notification, mock_agent_prompt, user
|
|
):
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": None,
|
|
}
|
|
],
|
|
)
|
|
mock_agent_prompt.return_value = {
|
|
"text": "hi back",
|
|
"provider": settings.AGENT_PROVIDER,
|
|
"model": settings.LLM_MODEL,
|
|
}
|
|
|
|
scrobble_agent_session_prompt(scrobble.id, "abc-123")
|
|
|
|
mock_notification.assert_called_once()
|
|
notification = mock_notification.return_value
|
|
notification.send.assert_called_once_with()
|
|
|
|
|
|
@patch("agents.providers.agent_prompt")
|
|
@patch("scrobbles.notifications.AgentSessionResponseNtfyNotification")
|
|
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
|
|
def test_scrobble_agent_session_prompt_no_ntfy_on_error(
|
|
mock_complete, mock_notification, mock_agent_prompt, user
|
|
):
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": None,
|
|
}
|
|
],
|
|
)
|
|
mock_agent_prompt.side_effect = ValueError("boom")
|
|
|
|
scrobble_agent_session_prompt(scrobble.id, "abc-123")
|
|
|
|
mock_notification.assert_not_called()
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.apply_async")
|
|
@patch("agents.providers.agent_prompt")
|
|
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
|
|
def test_scrobble_agent_session_prompt_retries_transient_error(
|
|
mock_complete, mock_agent_prompt, mock_retry, user, settings
|
|
):
|
|
settings.CELERY_TASK_ALWAYS_EAGER = False
|
|
settings.AGENT_PROVIDER_MAX_ATTEMPTS = 3
|
|
settings.AGENT_PROVIDER_RETRY_BACKOFF_BASE = 60
|
|
settings.AGENT_PROVIDER_RETRY_MAX_BACKOFF = 300
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": None,
|
|
}
|
|
],
|
|
)
|
|
mock_agent_prompt.side_effect = _http_error(429, {"Retry-After": "30"})
|
|
|
|
scrobble_agent_session_prompt(scrobble.id, "abc-123")
|
|
|
|
scrobble.refresh_from_db()
|
|
turn = scrobble.log["turns"][0]
|
|
assert turn["response"] is None
|
|
assert turn["error"] is True
|
|
assert turn["retryable"] is True
|
|
assert turn["attempts"] == 1
|
|
mock_retry.assert_called_once_with(args=[scrobble.id, "abc-123"], countdown=30)
|
|
mock_complete.assert_not_called()
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.apply_async")
|
|
@patch("agents.providers.agent_prompt")
|
|
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
|
|
def test_scrobble_agent_session_prompt_retry_backoff_without_retry_after(
|
|
mock_complete, mock_agent_prompt, mock_retry, user, settings
|
|
):
|
|
settings.CELERY_TASK_ALWAYS_EAGER = False
|
|
settings.AGENT_PROVIDER_MAX_ATTEMPTS = 3
|
|
settings.AGENT_PROVIDER_RETRY_BACKOFF_BASE = 60
|
|
settings.AGENT_PROVIDER_RETRY_MAX_BACKOFF = 300
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": None,
|
|
}
|
|
],
|
|
)
|
|
mock_agent_prompt.side_effect = _http_error(503)
|
|
|
|
scrobble_agent_session_prompt(scrobble.id, "abc-123")
|
|
|
|
mock_retry.assert_called_once_with(args=[scrobble.id, "abc-123"], countdown=60)
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.apply_async")
|
|
@patch("agents.providers.agent_prompt")
|
|
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
|
|
def test_scrobble_agent_session_prompt_gives_up_after_max_attempts(
|
|
mock_complete, mock_agent_prompt, mock_retry, user, settings
|
|
):
|
|
settings.CELERY_TASK_ALWAYS_EAGER = False
|
|
settings.AGENT_PROVIDER_MAX_ATTEMPTS = 2
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": None,
|
|
"error": True,
|
|
"retryable": True,
|
|
"attempts": 1,
|
|
}
|
|
],
|
|
)
|
|
mock_agent_prompt.side_effect = _http_error(429, {"Retry-After": "30"})
|
|
|
|
scrobble_agent_session_prompt(scrobble.id, "abc-123")
|
|
|
|
scrobble.refresh_from_db()
|
|
turn = scrobble.log["turns"][0]
|
|
assert turn["response"] == "Error: Provider returned HTTP 429 (Too Many Requests)"
|
|
assert turn["error"] is True
|
|
assert turn["retryable"] is False
|
|
assert turn["attempts"] == 2
|
|
mock_retry.assert_not_called()
|
|
mock_complete.assert_called_once_with(
|
|
args=[scrobble.id], countdown=AGENT_SESSION_AUTO_COMPLETE_SECONDS
|
|
)
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.apply_async")
|
|
@patch("agents.providers.agent_prompt")
|
|
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
|
|
def test_scrobble_agent_session_prompt_eager_mode_is_terminal(
|
|
mock_complete, mock_agent_prompt, mock_retry, user, settings
|
|
):
|
|
settings.CELERY_TASK_ALWAYS_EAGER = True
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": None,
|
|
}
|
|
],
|
|
)
|
|
mock_agent_prompt.side_effect = _http_error(429, {"Retry-After": "30"})
|
|
|
|
scrobble_agent_session_prompt(scrobble.id, "abc-123")
|
|
|
|
scrobble.refresh_from_db()
|
|
turn = scrobble.log["turns"][0]
|
|
assert turn["response"] == "Error: Provider returned HTTP 429 (Too Many Requests)"
|
|
assert turn["retryable"] is False
|
|
mock_retry.assert_not_called()
|
|
|
|
|
|
@patch("agents.providers.agent_prompt")
|
|
def test_scrobble_agent_session_prompt_skips_answered_turn(mock_agent_prompt, user):
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": "hi back",
|
|
}
|
|
],
|
|
)
|
|
|
|
scrobble_agent_session_prompt(scrobble.id, "abc-123")
|
|
|
|
mock_agent_prompt.assert_not_called()
|
|
scrobble.refresh_from_db()
|
|
assert scrobble.log["turns"][0]["response"] == "hi back"
|
|
|
|
|
|
@patch("scrobbles.notifications.AgentSessionFailedNtfyNotification")
|
|
@patch("agents.providers.agent_prompt")
|
|
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
|
|
def test_scrobble_agent_session_prompt_sends_ntfy_on_terminal_error(
|
|
mock_complete, mock_agent_prompt, mock_notification, user
|
|
):
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": None,
|
|
}
|
|
],
|
|
)
|
|
mock_agent_prompt.side_effect = ValueError("boom")
|
|
|
|
scrobble_agent_session_prompt(scrobble.id, "abc-123")
|
|
|
|
mock_notification.assert_called_once()
|
|
mock_notification.return_value.send.assert_called_once_with()
|
|
mock_complete.assert_called_once_with(
|
|
args=[scrobble.id], countdown=AGENT_SESSION_AUTO_COMPLETE_SECONDS
|
|
)
|
|
|
|
|
|
@patch("agents.providers.agent_prompt")
|
|
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
|
|
def test_scrobble_agent_session_prompt_success_clears_retry_state(
|
|
mock_complete, mock_agent_prompt, user
|
|
):
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": None,
|
|
"error": True,
|
|
"retryable": True,
|
|
"attempts": 1,
|
|
}
|
|
],
|
|
)
|
|
mock_agent_prompt.return_value = {
|
|
"text": "hi back",
|
|
"provider": settings.AGENT_PROVIDER,
|
|
"model": settings.LLM_MODEL,
|
|
}
|
|
|
|
scrobble_agent_session_prompt(scrobble.id, "abc-123")
|
|
|
|
scrobble.refresh_from_db()
|
|
turn = scrobble.log["turns"][0]
|
|
assert turn["response"] == "hi back"
|
|
assert "error" not in turn
|
|
assert "retryable" not in turn
|
|
assert "attempts" not in turn
|
|
|
|
|
|
# --- auto-complete ---
|
|
|
|
|
|
def test_scrobble_agent_session_complete_marks_done(user):
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
in_progress=True,
|
|
turns=[{"prompt_id": "abc-123", "prompt": "hello", "response": "hi back"}],
|
|
)
|
|
|
|
scrobble_agent_session_complete(scrobble.id)
|
|
|
|
scrobble.refresh_from_db()
|
|
assert scrobble.in_progress is False
|
|
assert scrobble.played_to_completion is True
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
|
|
def test_scrobble_agent_session_complete_reschedules_when_pending(mock_complete, user):
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
in_progress=True,
|
|
turns=[{"prompt_id": "abc-123", "prompt": "hello", "response": None}],
|
|
)
|
|
|
|
scrobble_agent_session_complete(scrobble.id)
|
|
|
|
scrobble.refresh_from_db()
|
|
assert scrobble.in_progress is True
|
|
mock_complete.assert_called_once_with(
|
|
args=[scrobble.id], countdown=AGENT_SESSION_AUTO_COMPLETE_SECONDS
|
|
)
|
|
|
|
|
|
def test_scrobble_agent_session_complete_unknown_scrobble(user):
|
|
scrobble_agent_session_complete(999999)
|
|
|
|
|
|
# --- views ---
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_manual_scrobble_view_routes_plain_text_to_agent(mock_delay, client, user):
|
|
client.force_login(user)
|
|
response = client.post(
|
|
reverse("scrobbles:lookup-manual-scrobble"),
|
|
{"item_id": "what is the meaning of life?"},
|
|
)
|
|
|
|
assert response.status_code == 302
|
|
assert response.url == reverse("agents:agent_model_select")
|
|
assert client.session["agent_prompt"] == "what is the meaning of life?"
|
|
assert not Scrobble.objects.filter(user_id=user.id).exists()
|
|
|
|
|
|
@patch("agents.views.list_openrouter_free_models")
|
|
@patch("agents.views.list_gemini_agent_models")
|
|
@patch("agents.views.list_groq_agent_models")
|
|
@patch("agents.views.list_mistral_agent_models")
|
|
def test_agent_model_select_view_lists_models(
|
|
mock_mistral, mock_groq, mock_gemini, mock_or, client, user
|
|
):
|
|
mock_gemini.return_value = [
|
|
{"provider": "gemini", "id": "gemini-3.6-flash", "name": "gemini-3.6-flash"}
|
|
]
|
|
mock_groq.return_value = [
|
|
{
|
|
"provider": "groq",
|
|
"id": "llama-3.3-70b-versatile",
|
|
"name": "llama-3.3-70b-versatile",
|
|
}
|
|
]
|
|
mock_mistral.return_value = [
|
|
{
|
|
"provider": "mistral",
|
|
"id": "mistral-small-latest",
|
|
"name": "mistral-small-latest",
|
|
}
|
|
]
|
|
mock_or.return_value = [
|
|
{"provider": "openrouter", "id": "free/model-a", "name": "Model A"},
|
|
{"provider": "openrouter", "id": "free/model-b", "name": "Model B"},
|
|
]
|
|
client.force_login(user)
|
|
session = client.session
|
|
session["agent_prompt"] = "what is 2+2?"
|
|
session.save()
|
|
response = client.get(reverse("agents:agent_model_select"))
|
|
|
|
assert response.status_code == 200
|
|
assert b"what is 2+2?" in response.content
|
|
assert b"free/model-a" in response.content
|
|
assert b"Model A" in response.content
|
|
assert b"gemini-3.6-flash" in response.content
|
|
assert b"Google Gemini" in response.content
|
|
assert b"llama-3.3-70b-versatile" in response.content
|
|
assert b"Groq" in response.content
|
|
assert b"mistral-small-latest" in response.content
|
|
assert b"Mistral" in response.content
|
|
expected = (
|
|
mock_gemini.return_value
|
|
+ mock_or.return_value
|
|
+ mock_groq.return_value
|
|
+ mock_mistral.return_value
|
|
)
|
|
assert client.session["agent_models"] == expected
|
|
|
|
|
|
@patch("agents.views.list_openrouter_free_models")
|
|
@patch("agents.views.list_gemini_agent_models")
|
|
@patch("agents.views.list_groq_agent_models")
|
|
@patch("agents.views.list_mistral_agent_models")
|
|
def test_agent_model_select_view_handles_errors(
|
|
mock_mistral, mock_groq, mock_gemini, mock_or, client, user
|
|
):
|
|
mock_gemini.return_value = []
|
|
mock_groq.return_value = []
|
|
mock_mistral.return_value = []
|
|
mock_or.side_effect = ValueError("boom")
|
|
client.force_login(user)
|
|
session = client.session
|
|
session["agent_prompt"] = "hi"
|
|
session.save()
|
|
response = client.get(reverse("agents:agent_model_select"))
|
|
|
|
assert response.status_code == 200
|
|
assert b"Could not load models from OpenRouter" in response.content
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_agent_scrobble_from_model_creates_session(mock_delay, client, user):
|
|
client.force_login(user)
|
|
session = client.session
|
|
session["agent_prompt"] = "what is 2+2?"
|
|
session.save()
|
|
response = client.post(
|
|
reverse("agents:agent_scrobble_from_model"),
|
|
{"model": "free/model-a", "prompt": "what is 2+2?"},
|
|
)
|
|
|
|
assert response.status_code == 302
|
|
scrobble = Scrobble.objects.filter(user_id=user.id).first()
|
|
assert scrobble is not None
|
|
assert scrobble.agent_session.provider == "openrouter"
|
|
assert scrobble.agent_session.model == "free/model-a"
|
|
assert scrobble.log["title"] == "what is 2+2?"
|
|
assert response.url == scrobble.get_absolute_url()
|
|
assert "agent_prompt" not in client.session
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_agent_scrobble_from_model_gemini_provider(mock_delay, client, user):
|
|
client.force_login(user)
|
|
session = client.session
|
|
session["agent_prompt"] = "what is 2+2?"
|
|
session.save()
|
|
response = client.post(
|
|
reverse("agents:agent_scrobble_from_model"),
|
|
{"provider": "gemini", "model": "gemini-3.6-flash", "prompt": "what is 2+2?"},
|
|
)
|
|
|
|
assert response.status_code == 302
|
|
scrobble = Scrobble.objects.filter(user_id=user.id).first()
|
|
assert scrobble is not None
|
|
assert scrobble.agent_session.provider == "gemini"
|
|
assert scrobble.agent_session.model == "gemini-3.6-flash"
|
|
assert response.url == scrobble.get_absolute_url()
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_agent_scrobble_from_model_missing_model(mock_delay, client, user):
|
|
client.force_login(user)
|
|
response = client.post(reverse("agents:agent_scrobble_from_model"), {"model": ""})
|
|
|
|
assert response.status_code == 302
|
|
assert response.url == reverse("agents:agent_model_select")
|
|
mock_delay.assert_not_called()
|
|
assert not Scrobble.objects.filter(user_id=user.id).exists()
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_manual_scrobble_agent_session_explicit_model(mock_delay, user):
|
|
scrobble = manual_scrobble_agent_session(
|
|
"hi", user.id, provider="openrouter", model="free/model-a"
|
|
)
|
|
|
|
assert scrobble.agent_session.provider == "openrouter"
|
|
assert scrobble.agent_session.model == "free/model-a"
|
|
assert scrobble.agent_session.title == "openrouter free/model-a"
|
|
assert scrobble.log["turns"][0]["prompt"] == "hi"
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_agent_session_partial_pending_polls(mock_delay, client, user):
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
turns=[{"prompt_id": "abc-123", "prompt": "hello", "response": None}],
|
|
)
|
|
client.force_login(user)
|
|
response = client.get(
|
|
reverse("scrobbles:agent-session-partial", args=[scrobble.id])
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert b"every 5s" in response.content
|
|
assert b"Thinking" in response.content
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_agent_session_partial_completed_stops_polling(mock_delay, client, user):
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
in_progress=False,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": "**hi back**",
|
|
}
|
|
],
|
|
)
|
|
client.force_login(user)
|
|
response = client.get(
|
|
reverse("scrobbles:agent-session-partial", args=[scrobble.id])
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert b"every 5s" not in response.content
|
|
assert b"hi back" in response.content
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_agent_session_list_view(mock_delay, client, user):
|
|
_mk_scrobble(user, in_progress=True)
|
|
client.force_login(user)
|
|
response = client.get(reverse("agents:agent_session_list"))
|
|
|
|
assert response.status_code == 200
|
|
assert settings.LLM_MODEL.encode() in response.content
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_homepage_shows_agent_sessions_with_first_prompt_title(
|
|
mock_delay, client, user
|
|
):
|
|
first_prompt = "tell me about the history of cheese making in europe"
|
|
_mk_scrobble(user, in_progress=False, title=first_prompt)
|
|
client.force_login(user)
|
|
response = client.get(reverse("vrobbler-home"))
|
|
|
|
assert response.status_code == 200
|
|
assert b"AgentSession" in response.content
|
|
assert first_prompt.encode() in response.content
|
|
assert settings.LLM_MODEL.encode() not in response.content
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_agent_session_detail_view(mock_delay, client, user):
|
|
scrobble = _mk_scrobble(user, in_progress=True)
|
|
client.force_login(user)
|
|
response = client.get(scrobble.media_obj.get_absolute_url())
|
|
|
|
assert response.status_code == 200
|
|
assert settings.LLM_MODEL.encode() in response.content
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_manual_scrobble_agent_follow_up_forks_completed_session(mock_delay, user):
|
|
original = _mk_scrobble(
|
|
user,
|
|
in_progress=False,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": "hi back",
|
|
}
|
|
],
|
|
)
|
|
|
|
result = manual_scrobble_agent_follow_up("tell me more", original.id, user.id)
|
|
|
|
assert result.id != original.id
|
|
assert result.agent_session_id == original.agent_session_id
|
|
assert result.in_progress is True
|
|
assert result.log["title"] == original.log["title"]
|
|
assert result.log["forked_from_id"] == original.id
|
|
|
|
original.refresh_from_db()
|
|
assert len(original.log["turns"]) == 1
|
|
assert original.in_progress is False
|
|
|
|
turns = result.log["turns"]
|
|
assert len(turns) == 2
|
|
assert turns[0] == original.log["turns"][0]
|
|
assert turns[1]["prompt"] == "tell me more"
|
|
assert turns[1]["response"] is None
|
|
mock_delay.assert_called_once_with(result.id, turns[1]["prompt_id"])
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_manual_scrobble_agent_follow_up_appends_turn_when_in_progress(
|
|
mock_delay, user
|
|
):
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
in_progress=True,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": "hi back",
|
|
}
|
|
],
|
|
)
|
|
|
|
result = manual_scrobble_agent_follow_up("tell me more", scrobble.id, user.id)
|
|
|
|
assert result.id == scrobble.id
|
|
scrobble.refresh_from_db()
|
|
turns = scrobble.log["turns"]
|
|
assert len(turns) == 2
|
|
assert turns[1]["prompt"] == "tell me more"
|
|
assert turns[1]["response"] is None
|
|
assert scrobble.in_progress is True
|
|
assert "forked_from_id" not in scrobble.log
|
|
mock_delay.assert_called_once_with(scrobble.id, turns[1]["prompt_id"])
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_manual_scrobble_agent_follow_up_unknown_scrobble(mock_delay, user):
|
|
assert manual_scrobble_agent_follow_up("tell me more", 999999, user.id) is None
|
|
mock_delay.assert_not_called()
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_agent_session_followup_post_forks_completed_session(mock_delay, client, user):
|
|
original = _mk_scrobble(
|
|
user,
|
|
in_progress=False,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": "hi back",
|
|
}
|
|
],
|
|
)
|
|
client.force_login(user)
|
|
response = client.post(
|
|
reverse("scrobbles:agent-session-followup", args=[original.id]),
|
|
{"prompt": "tell me more"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response["HX-Redirect"] == reverse(
|
|
"scrobbles:detail", args=[response.context["object"].id]
|
|
)
|
|
forked = response.context["object"]
|
|
assert forked.id != original.id
|
|
assert b"tell me more" in response.content
|
|
assert b"Thinking" in response.content
|
|
assert b"every 5s" in response.content
|
|
original.refresh_from_db()
|
|
assert len(original.log["turns"]) == 1
|
|
assert original.in_progress is False
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_agent_session_followup_post_appends_turn_when_in_progress(
|
|
mock_delay, client, user
|
|
):
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
in_progress=True,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": "hi back",
|
|
}
|
|
],
|
|
)
|
|
client.force_login(user)
|
|
response = client.post(
|
|
reverse("scrobbles:agent-session-followup", args=[scrobble.id]),
|
|
{"prompt": "tell me more"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "HX-Redirect" not in response
|
|
assert b"tell me more" in response.content
|
|
assert b"Thinking" in response.content
|
|
assert b"every 5s" in response.content
|
|
scrobble.refresh_from_db()
|
|
assert len(scrobble.log["turns"]) == 2
|
|
assert scrobble.in_progress is True
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_agent_session_followup_post_empty_prompt(mock_delay, client, user):
|
|
scrobble = _mk_scrobble(user, in_progress=False)
|
|
client.force_login(user)
|
|
response = client.post(
|
|
reverse("scrobbles:agent-session-followup", args=[scrobble.id]),
|
|
{"prompt": " "},
|
|
)
|
|
|
|
assert response.status_code == 302
|
|
mock_delay.assert_not_called()
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_agent_session_followup_form_shown_when_complete(mock_delay, client, user):
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
in_progress=False,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": "hi back",
|
|
}
|
|
],
|
|
)
|
|
client.force_login(user)
|
|
response = client.get(
|
|
reverse("scrobbles:agent-session-partial", args=[scrobble.id])
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert b"Ask a follow-up" in response.content
|
|
assert b"/follow-up/" in response.content
|
|
assert b"csrfmiddlewaretoken" in response.content
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_agent_session_retry_post_resets_and_dispatches(mock_delay, client, user):
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
in_progress=False,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": "Error: boom",
|
|
"error": True,
|
|
"retryable": False,
|
|
"attempts": 3,
|
|
}
|
|
],
|
|
)
|
|
client.force_login(user)
|
|
response = client.post(
|
|
reverse("scrobbles:agent-session-retry", args=[scrobble.id]),
|
|
{"prompt_id": "abc-123"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert b"Thinking" in response.content
|
|
assert b"every 5s" in response.content
|
|
scrobble.refresh_from_db()
|
|
turn = scrobble.log["turns"][0]
|
|
assert turn["response"] is None
|
|
assert turn["error"] is False
|
|
assert "retryable" not in turn
|
|
assert "attempts" not in turn
|
|
assert scrobble.in_progress is True
|
|
mock_delay.assert_called_once_with(scrobble.id, "abc-123")
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_agent_session_retry_post_skips_answered_turn(mock_delay, client, user):
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
in_progress=False,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": "hi back",
|
|
}
|
|
],
|
|
)
|
|
client.force_login(user)
|
|
response = client.post(
|
|
reverse("scrobbles:agent-session-retry", args=[scrobble.id]),
|
|
{"prompt_id": "abc-123"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
mock_delay.assert_not_called()
|
|
scrobble.refresh_from_db()
|
|
assert scrobble.log["turns"][0]["response"] == "hi back"
|
|
assert scrobble.in_progress is False
|
|
|
|
|
|
@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
|
|
def test_agent_session_retry_post_skips_already_retrying(mock_delay, client, user):
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
in_progress=True,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": None,
|
|
"error": True,
|
|
"retryable": True,
|
|
"attempts": 1,
|
|
}
|
|
],
|
|
)
|
|
client.force_login(user)
|
|
response = client.post(
|
|
reverse("scrobbles:agent-session-retry", args=[scrobble.id]),
|
|
{"prompt_id": "abc-123"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
mock_delay.assert_not_called()
|
|
scrobble.refresh_from_db()
|
|
assert scrobble.log["turns"][0]["response"] is None
|
|
assert scrobble.in_progress is True
|
|
|
|
|
|
# --- log data ---
|
|
|
|
|
|
def test_agent_session_logdata_pending():
|
|
log = AgentSessionLogData(
|
|
provider="gemini",
|
|
model="gemini-2.5-flash",
|
|
turns=[{"prompt_id": "1", "prompt": "hi", "response": None}],
|
|
)
|
|
assert log.pending is True
|
|
log.turns[0]["response"] = "hi back"
|
|
assert log.pending is False
|
|
assert AgentSessionLogData(provider="gemini").pending is False
|
|
|
|
|
|
def test_agent_session_logdata_as_html():
|
|
log = AgentSessionLogData(
|
|
provider="gemini",
|
|
model="gemini-2.5-flash",
|
|
turns=[
|
|
{"prompt_id": "1", "prompt": "**bold?**", "response": "yes"},
|
|
{"prompt_id": "2", "prompt": "again", "response": None},
|
|
],
|
|
)
|
|
html = log.as_html()
|
|
|
|
assert '<div class="agent-turn">' in html
|
|
assert "Prompt 1" in html
|
|
assert "<strong>bold?</strong>" in html
|
|
assert "yes" in html
|
|
assert 'id="response-1"' in html
|
|
assert "Prompt 2" in html
|
|
assert "Thinking" in html
|
|
assert 'id="response-2"' in html
|
|
|
|
|
|
@patch("scrobbles.notifications.requests.post")
|
|
def test_agent_session_response_ntfy_links_to_response(mock_post, user):
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": "hi back",
|
|
}
|
|
],
|
|
title="First prompt",
|
|
)
|
|
user.profile.ntfy_url = "https://ntfy.example.com/topic"
|
|
user.profile.ntfy_enabled = True
|
|
user.profile.save()
|
|
|
|
scrobble = Scrobble.objects.get(pk=scrobble.pk)
|
|
|
|
AgentSessionResponseNtfyNotification(scrobble, scrobble.log["turns"][0]).send()
|
|
|
|
mock_post.assert_called_once()
|
|
call = mock_post.call_args
|
|
assert call.args[0] == "https://ntfy.example.com/topic"
|
|
headers = call.kwargs["headers"]
|
|
assert (
|
|
headers["Click"]
|
|
== f"https://example.com/scrobbles/{scrobble.pk}/#response-abc-123"
|
|
)
|
|
assert headers["Title"] == "Agent Response Ready"
|
|
assert headers["Tags"] == "robot"
|
|
body = call.kwargs["data"].decode("utf-8")
|
|
assert "First prompt" in body
|
|
assert "hi back" in body
|
|
|
|
|
|
def test_agent_session_logdata_as_html_retrying():
|
|
log = AgentSessionLogData(
|
|
provider="gemini",
|
|
model="gemini-2.5-flash",
|
|
turns=[
|
|
{
|
|
"prompt_id": "1",
|
|
"prompt": "hi",
|
|
"response": None,
|
|
"error": True,
|
|
"retryable": True,
|
|
"attempts": 1,
|
|
}
|
|
],
|
|
)
|
|
html = log.as_html()
|
|
|
|
assert "Retrying" in html
|
|
assert "attempt 1 of" in html
|
|
assert "hx-post" not in html
|
|
|
|
|
|
def test_agent_session_logdata_as_html_retrying_attempts(settings):
|
|
settings.AGENT_PROVIDER_MAX_ATTEMPTS = 5
|
|
log = AgentSessionLogData(
|
|
turns=[
|
|
{
|
|
"prompt_id": "1",
|
|
"prompt": "hi",
|
|
"response": None,
|
|
"error": True,
|
|
"retryable": True,
|
|
"attempts": 3,
|
|
}
|
|
],
|
|
)
|
|
assert "attempt 3 of 5" in log.as_html()
|
|
|
|
|
|
def test_agent_session_logdata_as_html_retry_button():
|
|
log = AgentSessionLogData(
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": "Error: boom",
|
|
"error": True,
|
|
"retryable": False,
|
|
}
|
|
],
|
|
)
|
|
html = log.as_html(scrobble_id=42)
|
|
|
|
assert 'hx-post="/agent-session/42/retry/"' in html
|
|
assert 'value="abc-123"' in html
|
|
assert "Retry" in html
|
|
assert "hx-post" not in log.as_html()
|
|
|
|
|
|
def test_agent_session_logdata_as_html_escapes_prompt_id():
|
|
log = AgentSessionLogData(
|
|
turns=[
|
|
{
|
|
"prompt_id": '"><script>',
|
|
"prompt": "hello",
|
|
"response": "Error: boom",
|
|
"error": True,
|
|
}
|
|
],
|
|
)
|
|
html = log.as_html(scrobble_id=42)
|
|
|
|
assert ""><script>" in html
|
|
assert "<script>" not in html
|
|
|
|
|
|
@patch("scrobbles.notifications.requests.post")
|
|
def test_agent_session_failed_ntfy_links_to_response(mock_post, user):
|
|
scrobble = _mk_scrobble(
|
|
user,
|
|
turns=[
|
|
{
|
|
"prompt_id": "abc-123",
|
|
"prompt": "hello",
|
|
"response": "Error: boom",
|
|
"error": True,
|
|
"retryable": False,
|
|
}
|
|
],
|
|
title="First prompt",
|
|
)
|
|
user.profile.ntfy_url = "https://ntfy.example.com/topic"
|
|
user.profile.ntfy_enabled = True
|
|
user.profile.save()
|
|
|
|
scrobble = Scrobble.objects.get(pk=scrobble.pk)
|
|
|
|
AgentSessionFailedNtfyNotification(scrobble, scrobble.log["turns"][0]).send()
|
|
|
|
mock_post.assert_called_once()
|
|
call = mock_post.call_args
|
|
assert call.args[0] == "https://ntfy.example.com/topic"
|
|
headers = call.kwargs["headers"]
|
|
assert (
|
|
headers["Click"]
|
|
== f"https://example.com/scrobbles/{scrobble.pk}/#response-abc-123"
|
|
)
|
|
assert headers["Title"] == "Agent Response Failed"
|
|
assert headers["Tags"] == "warning"
|
|
assert headers["Priority"] == "high"
|
|
body = call.kwargs["data"].decode("utf-8")
|
|
assert "First prompt" in body
|
|
assert "Error: boom" in body
|