db3d2a2e31
Closes bd-3ny. Added mousedown listener that dismisses the dropdown when clicking outside both the dropdown and textarea. Uses early return to avoid registering listeners when dropdown is already closed.
336 lines
12 KiB
Python
336 lines
12 KiB
Python
"""Tests for mixins/http.py edge cases.
|
|
|
|
Unit tests for HTTP routing and response handling.
|
|
"""
|
|
|
|
import io
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from amc_server.mixins.http import HttpMixin
|
|
|
|
|
|
class DummyHttpHandler(HttpMixin):
|
|
"""Minimal handler for testing HTTP mixin."""
|
|
|
|
def __init__(self):
|
|
self.response_code = None
|
|
self.headers_sent = {}
|
|
self.body_sent = b""
|
|
self.path = "/"
|
|
self.wfile = io.BytesIO()
|
|
|
|
def send_response(self, code):
|
|
self.response_code = code
|
|
|
|
def send_header(self, key, value):
|
|
self.headers_sent[key] = value
|
|
|
|
def end_headers(self):
|
|
pass
|
|
|
|
|
|
class TestSendBytesResponse(unittest.TestCase):
|
|
"""Tests for _send_bytes_response edge cases."""
|
|
|
|
def test_sends_correct_headers(self):
|
|
handler = DummyHttpHandler()
|
|
handler._send_bytes_response(200, b"test", content_type="text/plain")
|
|
|
|
self.assertEqual(handler.response_code, 200)
|
|
self.assertEqual(handler.headers_sent["Content-Type"], "text/plain")
|
|
self.assertEqual(handler.headers_sent["Content-Length"], "4")
|
|
|
|
def test_includes_extra_headers(self):
|
|
handler = DummyHttpHandler()
|
|
handler._send_bytes_response(
|
|
200, b"test",
|
|
extra_headers={"X-Custom": "value", "Cache-Control": "no-cache"}
|
|
)
|
|
|
|
self.assertEqual(handler.headers_sent["X-Custom"], "value")
|
|
self.assertEqual(handler.headers_sent["Cache-Control"], "no-cache")
|
|
|
|
def test_broken_pipe_returns_false(self):
|
|
handler = DummyHttpHandler()
|
|
handler.wfile.write = MagicMock(side_effect=BrokenPipeError())
|
|
|
|
result = handler._send_bytes_response(200, b"test")
|
|
self.assertFalse(result)
|
|
|
|
def test_connection_reset_returns_false(self):
|
|
handler = DummyHttpHandler()
|
|
handler.wfile.write = MagicMock(side_effect=ConnectionResetError())
|
|
|
|
result = handler._send_bytes_response(200, b"test")
|
|
self.assertFalse(result)
|
|
|
|
def test_os_error_returns_false(self):
|
|
handler = DummyHttpHandler()
|
|
handler.wfile.write = MagicMock(side_effect=OSError("write error"))
|
|
|
|
result = handler._send_bytes_response(200, b"test")
|
|
self.assertFalse(result)
|
|
|
|
|
|
class TestSendJson(unittest.TestCase):
|
|
"""Tests for _send_json edge cases."""
|
|
|
|
def test_includes_cors_header(self):
|
|
handler = DummyHttpHandler()
|
|
handler._send_json(200, {"key": "value"})
|
|
|
|
self.assertEqual(handler.headers_sent["Access-Control-Allow-Origin"], "*")
|
|
|
|
def test_sets_json_content_type(self):
|
|
handler = DummyHttpHandler()
|
|
handler._send_json(200, {"key": "value"})
|
|
|
|
self.assertEqual(handler.headers_sent["Content-Type"], "application/json")
|
|
|
|
def test_encodes_payload_as_json(self):
|
|
handler = DummyHttpHandler()
|
|
handler._send_json(200, {"key": "value"})
|
|
|
|
written = handler.wfile.getvalue()
|
|
self.assertEqual(json.loads(written), {"key": "value"})
|
|
|
|
|
|
class TestServeDashboardFile(unittest.TestCase):
|
|
"""Tests for _serve_dashboard_file edge cases."""
|
|
|
|
def test_nonexistent_file_returns_404(self):
|
|
handler = DummyHttpHandler()
|
|
handler.errors = []
|
|
|
|
def capture_error(code, message):
|
|
handler.errors.append((code, message))
|
|
|
|
handler._json_error = capture_error
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
with patch("amc_server.mixins.http.DASHBOARD_DIR", Path(tmpdir)):
|
|
handler._serve_dashboard_file("nonexistent.html")
|
|
|
|
self.assertEqual(len(handler.errors), 1)
|
|
self.assertEqual(handler.errors[0][0], 404)
|
|
|
|
def test_path_traversal_blocked(self):
|
|
handler = DummyHttpHandler()
|
|
handler.errors = []
|
|
|
|
def capture_error(code, message):
|
|
handler.errors.append((code, message))
|
|
|
|
handler._json_error = capture_error
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
# Create a file outside the dashboard dir that shouldn't be accessible
|
|
secret = Path(tmpdir).parent / "secret.txt"
|
|
|
|
with patch("amc_server.mixins.http.DASHBOARD_DIR", Path(tmpdir)):
|
|
handler._serve_dashboard_file("../secret.txt")
|
|
|
|
self.assertEqual(len(handler.errors), 1)
|
|
self.assertEqual(handler.errors[0][0], 403)
|
|
|
|
def test_correct_content_type_for_html(self):
|
|
handler = DummyHttpHandler()
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
html_file = Path(tmpdir) / "test.html"
|
|
html_file.write_text("<html></html>")
|
|
|
|
with patch("amc_server.mixins.http.DASHBOARD_DIR", Path(tmpdir)):
|
|
handler._serve_dashboard_file("test.html")
|
|
|
|
self.assertEqual(handler.headers_sent["Content-Type"], "text/html; charset=utf-8")
|
|
|
|
def test_correct_content_type_for_css(self):
|
|
handler = DummyHttpHandler()
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
css_file = Path(tmpdir) / "styles.css"
|
|
css_file.write_text("body {}")
|
|
|
|
with patch("amc_server.mixins.http.DASHBOARD_DIR", Path(tmpdir)):
|
|
handler._serve_dashboard_file("styles.css")
|
|
|
|
self.assertEqual(handler.headers_sent["Content-Type"], "text/css; charset=utf-8")
|
|
|
|
def test_correct_content_type_for_js(self):
|
|
handler = DummyHttpHandler()
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
js_file = Path(tmpdir) / "app.js"
|
|
js_file.write_text("console.log('hello')")
|
|
|
|
with patch("amc_server.mixins.http.DASHBOARD_DIR", Path(tmpdir)):
|
|
handler._serve_dashboard_file("app.js")
|
|
|
|
self.assertEqual(handler.headers_sent["Content-Type"], "application/javascript; charset=utf-8")
|
|
|
|
def test_unknown_extension_gets_octet_stream(self):
|
|
handler = DummyHttpHandler()
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
unknown_file = Path(tmpdir) / "data.xyz"
|
|
unknown_file.write_bytes(b"\x00\x01\x02")
|
|
|
|
with patch("amc_server.mixins.http.DASHBOARD_DIR", Path(tmpdir)):
|
|
handler._serve_dashboard_file("data.xyz")
|
|
|
|
self.assertEqual(handler.headers_sent["Content-Type"], "application/octet-stream")
|
|
|
|
def test_no_cache_headers_set(self):
|
|
handler = DummyHttpHandler()
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
html_file = Path(tmpdir) / "test.html"
|
|
html_file.write_text("<html></html>")
|
|
|
|
with patch("amc_server.mixins.http.DASHBOARD_DIR", Path(tmpdir)):
|
|
handler._serve_dashboard_file("test.html")
|
|
|
|
self.assertIn("no-cache", handler.headers_sent.get("Cache-Control", ""))
|
|
|
|
|
|
class TestDoGet(unittest.TestCase):
|
|
"""Tests for do_GET routing edge cases."""
|
|
|
|
def _make_handler(self, path):
|
|
handler = DummyHttpHandler()
|
|
handler.path = path
|
|
handler._serve_dashboard_file = MagicMock()
|
|
handler._serve_state = MagicMock()
|
|
handler._serve_stream = MagicMock()
|
|
handler._serve_events = MagicMock()
|
|
handler._serve_conversation = MagicMock()
|
|
handler._json_error = MagicMock()
|
|
return handler
|
|
|
|
def test_root_serves_index(self):
|
|
handler = self._make_handler("/")
|
|
handler.do_GET()
|
|
handler._serve_dashboard_file.assert_called_with("index.html")
|
|
|
|
def test_index_html_serves_index(self):
|
|
handler = self._make_handler("/index.html")
|
|
handler.do_GET()
|
|
handler._serve_dashboard_file.assert_called_with("index.html")
|
|
|
|
def test_static_file_served(self):
|
|
handler = self._make_handler("/components/App.js")
|
|
handler.do_GET()
|
|
handler._serve_dashboard_file.assert_called_with("components/App.js")
|
|
|
|
def test_path_traversal_in_static_blocked(self):
|
|
handler = self._make_handler("/../../etc/passwd")
|
|
handler.do_GET()
|
|
handler._json_error.assert_called_with(404, "Not Found")
|
|
|
|
def test_api_state_routed(self):
|
|
handler = self._make_handler("/api/state")
|
|
handler.do_GET()
|
|
handler._serve_state.assert_called_once()
|
|
|
|
def test_api_stream_routed(self):
|
|
handler = self._make_handler("/api/stream")
|
|
handler.do_GET()
|
|
handler._serve_stream.assert_called_once()
|
|
|
|
def test_api_events_routed_with_id(self):
|
|
handler = self._make_handler("/api/events/session-123")
|
|
handler.do_GET()
|
|
handler._serve_events.assert_called_with("session-123")
|
|
|
|
def test_api_events_url_decoded(self):
|
|
handler = self._make_handler("/api/events/session%20with%20spaces")
|
|
handler.do_GET()
|
|
handler._serve_events.assert_called_with("session with spaces")
|
|
|
|
def test_api_conversation_with_query_params(self):
|
|
handler = self._make_handler("/api/conversation/sess123?project_dir=/test&agent=codex")
|
|
handler.do_GET()
|
|
handler._serve_conversation.assert_called_with("sess123", "/test", "codex")
|
|
|
|
def test_api_conversation_defaults_to_claude(self):
|
|
handler = self._make_handler("/api/conversation/sess123")
|
|
handler.do_GET()
|
|
handler._serve_conversation.assert_called_with("sess123", "", "claude")
|
|
|
|
def test_unknown_api_path_returns_404(self):
|
|
handler = self._make_handler("/api/unknown")
|
|
handler.do_GET()
|
|
handler._json_error.assert_called_with(404, "Not Found")
|
|
|
|
|
|
class TestDoPost(unittest.TestCase):
|
|
"""Tests for do_POST routing edge cases."""
|
|
|
|
def _make_handler(self, path):
|
|
handler = DummyHttpHandler()
|
|
handler.path = path
|
|
handler._dismiss_dead_sessions = MagicMock()
|
|
handler._dismiss_session = MagicMock()
|
|
handler._respond_to_session = MagicMock()
|
|
handler._json_error = MagicMock()
|
|
return handler
|
|
|
|
def test_dismiss_dead_routed(self):
|
|
handler = self._make_handler("/api/dismiss-dead")
|
|
handler.do_POST()
|
|
handler._dismiss_dead_sessions.assert_called_once()
|
|
|
|
def test_dismiss_session_routed(self):
|
|
handler = self._make_handler("/api/dismiss/session-abc")
|
|
handler.do_POST()
|
|
handler._dismiss_session.assert_called_with("session-abc")
|
|
|
|
def test_dismiss_url_decoded(self):
|
|
handler = self._make_handler("/api/dismiss/session%2Fwith%2Fslash")
|
|
handler.do_POST()
|
|
handler._dismiss_session.assert_called_with("session/with/slash")
|
|
|
|
def test_respond_routed(self):
|
|
handler = self._make_handler("/api/respond/session-xyz")
|
|
handler.do_POST()
|
|
handler._respond_to_session.assert_called_with("session-xyz")
|
|
|
|
def test_unknown_post_path_returns_404(self):
|
|
handler = self._make_handler("/api/unknown")
|
|
handler.do_POST()
|
|
handler._json_error.assert_called_with(404, "Not Found")
|
|
|
|
|
|
class TestDoOptions(unittest.TestCase):
|
|
"""Tests for do_OPTIONS CORS preflight."""
|
|
|
|
def test_returns_204_with_cors_headers(self):
|
|
handler = DummyHttpHandler()
|
|
handler.do_OPTIONS()
|
|
|
|
self.assertEqual(handler.response_code, 204)
|
|
self.assertEqual(handler.headers_sent["Access-Control-Allow-Origin"], "*")
|
|
self.assertIn("POST", handler.headers_sent["Access-Control-Allow-Methods"])
|
|
self.assertIn("Content-Type", handler.headers_sent["Access-Control-Allow-Headers"])
|
|
|
|
|
|
class TestJsonError(unittest.TestCase):
|
|
"""Tests for _json_error helper."""
|
|
|
|
def test_sends_json_with_error(self):
|
|
handler = DummyHttpHandler()
|
|
handler._json_error(404, "Not Found")
|
|
|
|
written = handler.wfile.getvalue()
|
|
payload = json.loads(written)
|
|
self.assertEqual(payload, {"ok": False, "error": "Not Found"})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|