131 lines
5.2 KiB
Python
131 lines
5.2 KiB
Python
import sys
|
|
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 == "hello there"
|
|
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 == "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_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")
|