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_console 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-console-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" post.assert_called_once() args, kwargs = post.call_args assert args[0] == "http://test-server:5002/desk/converse" assert kwargs["json"] == {"session_id": "pet-console-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-console-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_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-console-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-console-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") class _StreamResponse: def __init__(self, events): self._lines = [__import__("json").dumps(e) for e in events] self.closed = False def raise_for_status(self): pass def iter_lines(self, decode_unicode=True): return iter(self._lines) def close(self): self.closed = True def test_converse_stream_hands_each_sentence_over_and_marks_the_reply_streamed(): said = [] events = [ {"type": "say", "text": "First."}, {"type": "say", "text": "Second."}, {"type": "reply", "text": "First. Second.", "already_spoken": True}, ] with patch.object(server_client.requests, "post", return_value=_StreamResponse(events)): reply = server_client.converse_stream("hi", on_say=said.append) assert said == ["First.", "Second."] assert reply == server_client.Reply("First. Second.", streamed=True) def test_converse_stream_finishes_a_tool_call_through_the_plain_relay(): events = [{"type": "command", "command": "filectl {}", "token": "t1"}] posts = [_StreamResponse(events), _mock_response({"type": "reply", "text": "done"})] on_command = MagicMock(return_value="out") with patch.object(server_client.requests, "post", side_effect=posts) as post: reply = server_client.converse_stream("hi", on_say=lambda _s: None, on_command=on_command) on_command.assert_called_once_with("filectl {}") assert post.call_args_list[1].args[0].endswith("/desk/tool_result") assert reply == server_client.Reply("done", streamed=False) def test_converse_stream_failing_before_any_text_raises_for_the_fallback(): with patch.object(server_client.requests, "post", side_effect=ConnectionError("down")): with pytest.raises(server_client.ServerError): server_client.converse_stream("hi", on_say=lambda _s: None) def test_run_local_command_runs_in_the_current_directory(tmp_path, monkeypatch): monkeypatch.chdir(tmp_path) (tmp_path / "marker.txt").write_text("x") output = server_client.run_local_command("dir /b" if sys.platform == "win32" else "ls") assert output.startswith("[exit 0]") assert "marker.txt" in output def test_a_final_reply_repeating_the_streamed_sentences_is_not_shown_twice(): """Seen live: sentences streamed, then the same text again as the reply, without already_spoken — with line breaks the stream had dropped.""" events = [ {"type": "say", "text": "Let me check that."}, {"type": "say", "text": "The folder contains:"}, {"type": "reply", "text": "Let me check that.\n\nThe folder contains:\n- a\n- b"}, ] with patch.object(server_client.requests, "post", return_value=_StreamResponse(events)): reply = server_client.converse_stream("hi", on_say=lambda _s: None) assert reply == server_client.Reply("- a\n- b", streamed=False) def test_an_exact_repeat_counts_as_streamed(): events = [{"type": "say", "text": "Done."}, {"type": "reply", "text": "Done."}] with patch.object(server_client.requests, "post", return_value=_StreamResponse(events)): reply = server_client.converse_stream("hi", on_say=lambda _s: None) assert reply.streamed and reply.text == "" def test_a_different_final_reply_is_shown_in_full(): events = [{"type": "say", "text": "Checking."}, {"type": "reply", "text": "All good."}] with patch.object(server_client.requests, "post", return_value=_StreamResponse(events)): reply = server_client.converse_stream("hi", on_say=lambda _s: None) assert reply == server_client.Reply("All good.", streamed=False) def test_list_chats_returns_the_list_with_the_api_key(): chats = [{"session_id": "chat-3f9a2b1c9e04", "label": "Fix the build", "last_message": "try again", "last_active": "2026-09-25T23:46:31+00:00"}] with patch.object(server_client.requests, "get") as get: get.return_value = _mock_response({"chats": chats}) assert server_client.list_chats() == chats args, kwargs = get.call_args assert args[0] == "http://test-server:5002/desk/chats" assert kwargs["headers"] == {"X-Desk-Api-Key": "test-key"} def test_list_chats_defaults_to_empty_and_wraps_failures(): with patch.object(server_client.requests, "get") as get: get.return_value = _mock_response({}) assert server_client.list_chats() == [] with patch.object(server_client.requests, "get") as get: get.return_value = _mock_response({}, ok=False) # e.g. a 401 with pytest.raises(server_client.ServerError, match="couldn't list chats"): server_client.list_chats() def test_create_chat_posts_and_returns_the_new_session_id(): with patch.object(server_client.requests, "post") as post: post.return_value = _mock_response({"session_id": "chat-3f9a2b1c9e04"}) assert server_client.create_chat() == "chat-3f9a2b1c9e04" args, kwargs = post.call_args assert args[0] == "http://test-server:5002/desk/chats" assert kwargs["headers"] == {"X-Desk-Api-Key": "test-key"} def test_create_chat_without_a_session_id_is_an_error(): with patch.object(server_client.requests, "post") as post: post.return_value = _mock_response({}) with pytest.raises(server_client.ServerError, match="session_id"): server_client.create_chat() with patch.object(server_client.requests, "post", side_effect=ConnectionError("down")): with pytest.raises(server_client.ServerError): server_client.create_chat() def test_every_call_follows_a_reassigned_session_id(monkeypatch): """Switching chats is just `config.SESSION_ID = new_id`; the next request must carry it, with no call site holding on to the old one.""" monkeypatch.setattr(server_client.config, "SESSION_ID", "chat-new") with patch.object(server_client.requests, "post") as post: post.return_value = _mock_response({"type": "reply", "text": "hi"}) server_client.converse("hi") assert post.call_args.kwargs["json"]["session_id"] == "chat-new"