themajesticmagician
3ee67cb4d6
- Refactor `run_local_command` to manage subprocesses more effectively, ensuring child processes are terminated on timeout. - Introduce `_terminate` function to handle process group termination and capture output. - Implement `_command_output` to format command results with a character limit. - Add local intent recognition in `intents.py` to handle commands like "stop", "go to sleep", and "come here" without server interaction. - Normalize user input to match local intents while stripping filler words. - Update tests to cover new local intent functionality and ensure proper command handling. - Enhance speech processing to handle abbreviations and improve spoken output clarity.
196 lines
8.3 KiB
Python
196 lines
8.3 KiB
Python
import os
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
import pytest
|
|
|
|
from bolt_pet import server_client
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _configure(monkeypatch):
|
|
monkeypatch.setattr(server_client.config, "SERVER_URL", "http://test-server:5002")
|
|
monkeypatch.setattr(server_client.config, "API_KEY", "test-key")
|
|
monkeypatch.setattr(server_client.config, "SESSION_ID", "pet-test")
|
|
|
|
|
|
def _mock_response(json_data, ok=True):
|
|
resp = MagicMock()
|
|
resp.json.return_value = json_data
|
|
resp.raise_for_status = MagicMock() if ok else MagicMock(side_effect=Exception("boom"))
|
|
return resp
|
|
|
|
|
|
def test_converse_returns_reply_directly():
|
|
with patch.object(server_client.requests, "post") as post:
|
|
post.return_value = _mock_response({"type": "reply", "text": "hello there"})
|
|
result = server_client.converse("hi")
|
|
assert result.text == "hello there"
|
|
assert result.voice_id == "" # no speak_as on this reply
|
|
post.assert_called_once()
|
|
args, kwargs = post.call_args
|
|
assert args[0] == "http://test-server:5002/desk/converse"
|
|
assert kwargs["json"] == {"session_id": "pet-test", "text": "hi"}
|
|
assert kwargs["headers"] == {"X-Desk-Api-Key": "test-key"}
|
|
|
|
|
|
def test_converse_relays_a_command_then_returns_reply():
|
|
responses = [
|
|
_mock_response({"type": "command", "command": "echo hi", "token": "tok1"}),
|
|
_mock_response({"type": "reply", "text": "done"}),
|
|
]
|
|
with patch.object(server_client.requests, "post", side_effect=responses) as post:
|
|
on_command = MagicMock(return_value="[exit 0]\nhi")
|
|
result = server_client.converse("run echo hi", on_command=on_command)
|
|
assert result.text == "done"
|
|
on_command.assert_called_once_with("echo hi")
|
|
# second call was to /desk/tool_result with the command's output
|
|
second_call = post.call_args_list[1]
|
|
assert second_call.args[0] == "http://test-server:5002/desk/tool_result"
|
|
assert second_call.kwargs["json"] == {
|
|
"session_id": "pet-test", "token": "tok1", "output": "[exit 0]\nhi",
|
|
}
|
|
|
|
|
|
def test_converse_carries_a_speak_as_voice_back_with_the_reply():
|
|
"""The server tags a reply with the voice it picked (speak_as); this
|
|
client is what actually speaks in it, so the id has to survive the
|
|
return trip rather than being dropped with the rest of the payload."""
|
|
with patch.object(server_client.requests, "post") as post:
|
|
post.return_value = _mock_response({
|
|
"type": "reply", "text": "Ahoy there.",
|
|
"voice_id": "hnhGxwvHP8fc469w51rM", "voice_name": "Terence",
|
|
})
|
|
result = server_client.converse("talk like a pirate")
|
|
assert result == server_client.Reply("Ahoy there.", "hnhGxwvHP8fc469w51rM", "Terence")
|
|
|
|
|
|
def test_converse_raises_server_error_on_error_payload():
|
|
with patch.object(server_client.requests, "post") as post:
|
|
post.return_value = _mock_response({"type": "error", "error": "unauthorized"})
|
|
with pytest.raises(server_client.ServerError, match="unauthorized"):
|
|
server_client.converse("hi")
|
|
|
|
|
|
def test_converse_raises_server_error_when_unreachable():
|
|
with patch.object(server_client.requests, "post", side_effect=ConnectionError("no route")):
|
|
with pytest.raises(server_client.ServerError):
|
|
server_client.converse("hi")
|
|
|
|
|
|
def test_report_status_returns_reply_text_when_present():
|
|
with patch.object(server_client.requests, "post") as post:
|
|
post.return_value = _mock_response({"reply": "don't forget your 3pm"})
|
|
result = server_client.report_status()
|
|
assert result == "don't forget your 3pm"
|
|
|
|
|
|
def test_report_status_returns_none_when_nothing_pending():
|
|
with patch.object(server_client.requests, "post") as post:
|
|
post.return_value = _mock_response({"ok": True})
|
|
assert server_client.report_status() is None
|
|
|
|
|
|
def test_check_health_returns_parsed_json():
|
|
with patch.object(server_client.requests, "get") as get:
|
|
get.return_value = _mock_response({"ok": True, "service": "bolt-desk-api"})
|
|
result = server_client.check_health()
|
|
assert result == {"ok": True, "service": "bolt-desk-api"}
|
|
|
|
|
|
def test_list_outbox_files_returns_the_queue():
|
|
with patch.object(server_client.requests, "get") as get:
|
|
get.return_value = _mock_response(
|
|
{"files": [{"id": "abc", "name": "report.pdf", "size": 9}]}
|
|
)
|
|
result = server_client.list_outbox_files()
|
|
assert result == [{"id": "abc", "name": "report.pdf", "size": 9}]
|
|
args, kwargs = get.call_args
|
|
assert args[0] == "http://test-server:5002/desk/files"
|
|
assert kwargs["params"] == {"session_id": "pet-test"}
|
|
assert kwargs["headers"] == {"X-Desk-Api-Key": "test-key"}
|
|
|
|
|
|
def test_list_outbox_files_defaults_to_empty_list():
|
|
with patch.object(server_client.requests, "get") as get:
|
|
get.return_value = _mock_response({})
|
|
assert server_client.list_outbox_files() == []
|
|
|
|
|
|
def test_list_outbox_files_raises_server_error_when_unreachable():
|
|
with patch.object(server_client.requests, "get", side_effect=ConnectionError("no route")):
|
|
with pytest.raises(server_client.ServerError):
|
|
server_client.list_outbox_files()
|
|
|
|
|
|
def test_download_outbox_file_returns_raw_bytes():
|
|
with patch.object(server_client.requests, "get") as get:
|
|
resp = _mock_response({})
|
|
resp.content = b"%PDF fake bytes"
|
|
get.return_value = resp
|
|
result = server_client.download_outbox_file("abc")
|
|
assert result == b"%PDF fake bytes"
|
|
args, kwargs = get.call_args
|
|
assert args[0] == "http://test-server:5002/desk/files/abc"
|
|
assert kwargs["params"] == {"session_id": "pet-test"}
|
|
|
|
|
|
def test_download_outbox_file_raises_server_error_on_http_failure():
|
|
with patch.object(server_client.requests, "get") as get:
|
|
get.return_value = _mock_response({}, ok=False)
|
|
with pytest.raises(server_client.ServerError, match="abc"):
|
|
server_client.download_outbox_file("abc")
|
|
|
|
|
|
def test_converse_reports_a_relay_that_never_produced_a_reply():
|
|
"""Hitting the hop cap used to surface as "unknown server response", which
|
|
sent everyone looking at the payload shape instead of at a model that kept
|
|
calling tools and never answered."""
|
|
command = {"type": "command", "command": "echo hi", "token": "t"}
|
|
with patch.object(server_client.requests, "post") as post:
|
|
post.return_value = _mock_response(command)
|
|
with pytest.raises(server_client.ServerError, match="hop cap"):
|
|
server_client.converse("hi", on_command=lambda cmd: "ok")
|
|
|
|
|
|
# ── relayed shell commands ──────────────────────────────────────────────────
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _no_sudo_prompt(monkeypatch):
|
|
monkeypatch.setattr(server_client.config, "SUDO_ASKPASS_PROMPT", False)
|
|
|
|
|
|
@pytest.mark.skipif(os.name != "posix", reason="POSIX shell and process groups")
|
|
def test_a_successful_command_returns_its_output_and_exit_code():
|
|
output = server_client.run_local_command("echo hello; exit 3")
|
|
assert output.startswith("[exit 3]")
|
|
assert "hello" in output
|
|
|
|
|
|
@pytest.mark.skipif(os.name != "posix", reason="POSIX shell and process groups")
|
|
def test_a_timed_out_command_still_reports_what_it_printed():
|
|
"""A bare "timed out" tells the model nothing; the last line of output
|
|
usually says exactly what it was stuck waiting for."""
|
|
output = server_client.run_local_command("echo working on it; sleep 30", timeout=1)
|
|
assert "timed out after 1s" in output
|
|
assert "working on it" in output
|
|
|
|
|
|
@pytest.mark.skipif(os.name != "posix", reason="POSIX shell and process groups")
|
|
def test_a_timed_out_command_takes_its_children_with_it(tmp_path):
|
|
"""subprocess.run() would only kill the `sh`, leaving whatever it spawned
|
|
running for the rest of the session with no parent watching."""
|
|
marker = tmp_path / "ticks"
|
|
server_client.run_local_command(
|
|
f"(while true; do echo tick >> {marker}; sleep 0.05; done) & sleep 30",
|
|
timeout=1,
|
|
)
|
|
settled = marker.stat().st_size if marker.exists() else 0
|
|
time.sleep(0.4)
|
|
grew = (marker.stat().st_size if marker.exists() else 0) - settled
|
|
assert grew == 0, "a grandchild survived the timeout and is still writing"
|