Increase local_server test coverage

2026-01-22 13:40:59 +00:00
parent 557eafc163
commit 8f252adc27
4 changed files with 357 additions and 72 deletions
@@ -6,7 +6,12 @@ import pytest
 from aiohttp import web

 from textual_webterm.config import App, Config
-from textual_webterm.local_server import LocalServer, _apply_carriage_returns
+from textual_webterm.local_server import (
+    LocalClientConnector,
+    LocalServer,
+    _apply_carriage_returns,
+    _rewrite_svg_fonts,
+)


 class TestGetStaticPath:
@@ -369,3 +374,251 @@ class TestWebSocketProtocol:
        assert ping[0] == "ping"
        assert pong[0] == "pong"
        assert ping[1] == pong[1]
+
+
+class TestLocalServerMoreCoverage:
+    @pytest.fixture
+    def server_with_no_apps(self, tmp_path):
+        config = Config(apps=[])
+        config_file = tmp_path / "config.toml"
+        config_file.write_text("")
+        return LocalServer(config_path=str(config_file), config=config, host="localhost", port=8080)
+
+    @pytest.mark.asyncio
+    async def test_handle_health_check(self, server_with_no_apps):
+        resp = await server_with_no_apps._handle_health_check(MagicMock())
+        assert resp.text == "Local server is running"
+
+    def test_select_app_for_route_picks_default(self, server_with_no_apps, monkeypatch):
+        default_app = App(name="D", slug="d", path=".", command="echo d", terminal=True)
+        monkeypatch.setattr(server_with_no_apps.session_manager, "get_default_app", lambda: default_app)
+        assert server_with_no_apps._select_app_for_route("missing") == default_app
+
+    @pytest.mark.asyncio
+    async def test_handle_session_data_no_ws_noop(self, server_with_no_apps):
+        await server_with_no_apps.handle_session_data("rk", b"data")
+
+    @pytest.mark.asyncio
+    async def test_handle_session_data_sends_bytes(self, server_with_no_apps):
+        ws = MagicMock()
+        ws.send_bytes = AsyncMock()
+        server_with_no_apps._websocket_connections["rk"] = ws
+        await server_with_no_apps.handle_session_data("rk", b"data")
+        ws.send_bytes.assert_awaited_once_with(b"data")
+
+    @pytest.mark.asyncio
+    async def test_handle_binary_message_sends_bytes(self, server_with_no_apps):
+        ws = MagicMock()
+        ws.send_bytes = AsyncMock()
+        server_with_no_apps._websocket_connections["rk"] = ws
+        await server_with_no_apps.handle_binary_message("rk", b"bin")
+        ws.send_bytes.assert_awaited_once_with(b"bin")
+
+    @pytest.mark.asyncio
+    async def test_handle_session_close_ends_session_and_closes_ws(self, server_with_no_apps, monkeypatch):
+        ws = MagicMock()
+        ws.close = AsyncMock()
+        server_with_no_apps._websocket_connections["rk"] = ws
+        monkeypatch.setattr(server_with_no_apps.session_manager, "on_session_end", MagicMock())
+        await server_with_no_apps.handle_session_close("sid", "rk")
+        server_with_no_apps.session_manager.on_session_end.assert_called_once_with("sid")
+        ws.close.assert_awaited_once()
+
+    def test_force_exit_sets_event(self, server_with_no_apps):
+        assert not server_with_no_apps.exit_event.is_set()
+        server_with_no_apps.force_exit()
+        assert server_with_no_apps.exit_event.is_set()
+
+    def test_get_static_path_import_error_returns_none(self, monkeypatch):
+        import builtins
+
+        from textual_webterm.local_server import _get_static_path
+
+        real_import = builtins.__import__
+
+        def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
+            if name == "textual_serve":
+                raise ImportError("nope")
+            return real_import(name, globals, locals, fromlist, level)
+
+        monkeypatch.setattr(builtins, "__import__", fake_import)
+        assert _get_static_path() is None
+
+    def test_add_terminal_windows_noop(self, server_with_no_apps, monkeypatch):
+        from textual_webterm import constants as constants_mod
+
+        monkeypatch.setattr(constants_mod, "WINDOWS", True)
+        server_with_no_apps.add_terminal("T", "cmd", "slug")
+        assert "slug" not in server_with_no_apps.session_manager.apps_by_slug
+
+    @pytest.mark.asyncio
+    async def test_handle_screenshot_404_when_no_running_session(self, server_with_no_apps, monkeypatch):
+        request = MagicMock()
+        request.query = {}
+        monkeypatch.setattr(server_with_no_apps.session_manager, "get_first_running_session", lambda: None)
+        with pytest.raises(web.HTTPNotFound):
+            await server_with_no_apps._handle_screenshot(request)
+
+    @pytest.mark.asyncio
+    async def test_handle_screenshot_404_when_session_missing_buffer(self, server_with_no_apps, monkeypatch):
+        request = MagicMock()
+        request.query = {"route_key": "rk"}
+        monkeypatch.setattr(server_with_no_apps.session_manager, "get_session_by_route_key", lambda _rk: object())
+        with pytest.raises(web.HTTPNotFound):
+            await server_with_no_apps._handle_screenshot(request)
+
+    @pytest.mark.asyncio
+    async def test_get_ws_url_falls_back_when_no_host_header(self, server_with_no_apps):
+        request = MagicMock()
+        request.headers = {}
+        request.secure = False
+        url = server_with_no_apps._get_ws_url_from_request(request, "rk")
+        assert url.startswith("ws://")
+
+    @pytest.mark.asyncio
+    async def test_root_terminal_page_includes_assets_and_dataset(self, server_with_no_apps, monkeypatch):
+        server_with_no_apps.session_manager.apps_by_slug["rk"] = App(
+            name="Known",
+            slug="rk",
+            path=".",
+            command="echo",
+            terminal=True,
+        )
+        request = MagicMock()
+        request.query = {"route_key": "rk"}
+        request.headers = {"Host": "localhost:8080"}
+        request.secure = False
+
+        resp = await server_with_no_apps._handle_root(request)
+        assert "/static/css/xterm.css" in resp.text
+        assert "/static-webterm/monospace.css" in resp.text
+        assert "data-session-websocket-url" in resp.text
+        assert "data-font-size" in resp.text
+
+    def test_rewrite_svg_fonts_removes_font_face_and_forces_stack(self):
+        svg = (
+            '<svg xmlns="http://www.w3.org/2000/svg">'
+            '<style>@font-face{src:url(https://cdnjs.cloudflare.com/x);} text{font-family:Fira Code;}</style>'
+            '<text>hi</text>'
+            '</svg>'
+        )
+        out = _rewrite_svg_fonts(svg)
+        assert "@font-face" not in out
+        assert "cdnjs.cloudflare.com" not in out
+        assert "ui-monospace" in out
+
+    def test_rewrite_svg_fonts_injects_style_if_missing(self):
+        svg = '<svg xmlns="http://www.w3.org/2000/svg"><text>hi</text></svg>'
+        out = _rewrite_svg_fonts(svg)
+        assert "ui-monospace" in out
+
+    @pytest.mark.asyncio
+    async def test_cached_screenshot_etag_returns_304(self, server_with_no_apps):
+        request = MagicMock()
+        request.headers = {"If-None-Match": "abc"}
+        server_with_no_apps._screenshot_cache["rk"] = (0.0, "<svg></svg>")
+        server_with_no_apps._screenshot_cache_etag["rk"] = "abc"
+
+        with pytest.raises(web.HTTPNotModified):
+            server_with_no_apps._get_cached_screenshot_response(request, "rk")
+
+    @pytest.mark.asyncio
+    async def test_cached_screenshot_etag_sets_headers(self, server_with_no_apps):
+        request = MagicMock()
+        request.headers = {}
+        server_with_no_apps._screenshot_cache["rk"] = (0.0, "<svg></svg>")
+        server_with_no_apps._screenshot_cache_etag["rk"] = "abc"
+
+        resp = server_with_no_apps._get_cached_screenshot_response(request, "rk")
+        assert resp is not None
+        assert resp.headers.get("ETag") == "abc"
+
+    def test_screenshot_cache_ttl_backs_off(self, server_with_no_apps, monkeypatch):
+        # Drive each tier by controlling now and last_activity.
+        server_with_no_apps._route_last_activity["rk"] = 99.0
+        assert server_with_no_apps._get_screenshot_cache_ttl("rk", now=100.0) == 1.0
+        assert server_with_no_apps._get_screenshot_cache_ttl("rk", now=110.0) == 5.0
+        assert server_with_no_apps._get_screenshot_cache_ttl("rk", now=200.0) == 15.0
+        assert server_with_no_apps._get_screenshot_cache_ttl("rk", now=1000.0) == 60.0
+
+    @pytest.mark.asyncio
+    async def test_handle_screenshot_uses_cache_when_no_new_activity(self, server_with_no_apps, monkeypatch):
+        request = MagicMock()
+        request.query = {"route_key": "rk"}
+        request.headers = {}
+
+        session = MagicMock()
+        session.get_replay_buffer = AsyncMock(return_value=b"SHOULD_NOT_BE_READ")
+        monkeypatch.setattr(server_with_no_apps.session_manager, "get_session_by_route_key", lambda _rk: session)
+
+        server_with_no_apps._screenshot_cache["rk"] = (0.0, "<svg>cached</svg>")
+        server_with_no_apps._screenshot_cache_etag["rk"] = "etag"
+        server_with_no_apps._route_last_activity["rk"] = 5.0
+        server_with_no_apps._screenshot_last_rendered_activity["rk"] = 5.0
+
+        resp = await server_with_no_apps._handle_screenshot(request)
+        assert "cached" in resp.text
+        session.get_replay_buffer.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_handle_screenshot_invalid_width_height_defaults(self, server_with_no_apps, monkeypatch):
+        request = MagicMock()
+        request.query = {"route_key": "rk", "width": "nope", "height": "nope"}
+        request.headers = {}
+
+        session = MagicMock()
+        session.get_replay_buffer = AsyncMock(return_value=b"hello\n")
+        monkeypatch.setattr(server_with_no_apps.session_manager, "get_session_by_route_key", lambda _rk: session)
+
+        resp = await server_with_no_apps._handle_screenshot(request)
+        assert resp.content_type == "image/svg+xml"
+        assert "<svg" in resp.text
+
+    @pytest.mark.asyncio
+    async def test_handle_root_no_apps_available(self, server_with_no_apps):
+        request = MagicMock()
+        request.query = {}
+        resp = await server_with_no_apps._handle_root(request)
+        assert "No Apps Available" in resp.text
+
+    @pytest.mark.asyncio
+    async def test_dispatch_ws_message_ping_sends_pong(self, server_with_no_apps):
+        ws = MagicMock()
+        ws.send_json = AsyncMock()
+        created = await server_with_no_apps._dispatch_ws_message(["ping", "x"], "rk", ws, False)
+        assert created is False
+        ws.send_json.assert_awaited_once_with(["pong", "x"])
+
+    @pytest.mark.asyncio
+    async def test_dispatch_ws_message_stdin_sends_bytes_to_session(self, server_with_no_apps, monkeypatch):
+        session = MagicMock()
+        session.send_bytes = AsyncMock()
+        monkeypatch.setattr(server_with_no_apps.session_manager, "get_session_by_route_key", lambda _rk: session)
+
+        ws = MagicMock()
+        created = await server_with_no_apps._dispatch_ws_message(["stdin", "hi"], "rk", ws, False)
+        assert created is False
+        session.send_bytes.assert_awaited_once_with(b"hi")
+
+    @pytest.mark.asyncio
+    async def test_connector_methods_forward_to_server(self):
+        server = MagicMock()
+        server.mark_route_activity = MagicMock()
+        server.handle_session_data = AsyncMock()
+        server.handle_binary_message = AsyncMock()
+        server.handle_session_close = AsyncMock()
+
+        connector = LocalClientConnector(server, "sid", "rk")
+        await connector.on_data(b"data")
+        server.mark_route_activity.assert_called_once_with("rk")
+        server.handle_session_data.assert_awaited_once_with("rk", b"data")
+
+        await connector.on_meta({"type": "open_url", "url": "https://example.com"})
+        await connector.on_meta({"type": "deliver_file_start", "path": "/tmp/x"})
+        await connector.on_meta({"type": "unknown"})
+
+        await connector.on_binary_encoded_message(b"bin")
+        server.handle_binary_message.assert_awaited_once_with("rk", b"bin")
+
+        await connector.on_close()
+        server.handle_session_close.assert_awaited_once_with("sid", "rk")
@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+import json
+
+import pytest
+from aiohttp import WSMsgType, web
+from aiohttp.test_utils import TestClient, TestServer
+
+from textual_webterm.config import App, Config
+from textual_webterm.local_server import LocalServer
+
+
+async def _make_client(server: LocalServer) -> TestClient:
+    app = web.Application()
+    app.add_routes(server._build_routes())
+    test_server = TestServer(app)
+    client = TestClient(test_server)
+    await client.start_server()
+    return client
+
+
+@pytest.mark.asyncio
+async def test_websocket_creates_session_on_resize(tmp_path):
+    config = Config(apps=[App(name="Test", slug="test", path=".", command="echo test", terminal=True)])
+    config_file = tmp_path / "config.toml"
+    config_file.write_text("")
+    server = LocalServer(config_path=str(config_file), config=config)
+
+    # Avoid spawning any real processes.
+    created = {"args": None}
+
+    async def fake_create(route_key: str, width: int, height: int) -> None:
+        created["args"] = (route_key, width, height)
+
+    server._create_terminal_session = fake_create  # type: ignore[method-assign]
+
+    client = await _make_client(server)
+    try:
+        ws = await client.ws_connect("/ws/test")
+        await ws.send_str(json.dumps(["resize", {"width": 90, "height": 25}]))
+        await ws.close()
+    finally:
+        await client.close()
+
+    assert created["args"] == ("test", 90, 25)
+
+
+@pytest.mark.asyncio
+async def test_websocket_ping_pong(tmp_path):
+    config = Config(apps=[App(name="Test", slug="test", path=".", command="echo test", terminal=True)])
+    config_file = tmp_path / "config.toml"
+    config_file.write_text("")
+    server = LocalServer(config_path=str(config_file), config=config)
+
+    client = await _make_client(server)
+    try:
+        ws = await client.ws_connect("/ws/test")
+        await ws.send_str(json.dumps(["ping", "123"]))
+
+        msg = await ws.receive(timeout=1)
+        assert msg.type == WSMsgType.TEXT
+        assert json.loads(msg.data) == ["pong", "123"]
+
+        await ws.close()
+    finally:
+        await client.close()
+
+
+@pytest.mark.asyncio
+async def test_websocket_ignores_invalid_envelopes(tmp_path):
+    config = Config(apps=[App(name="Test", slug="test", path=".", command="echo test", terminal=True)])
+    config_file = tmp_path / "config.toml"
+    config_file.write_text("")
+    server = LocalServer(config_path=str(config_file), config=config)
+
+    client = await _make_client(server)
+    try:
+        ws = await client.ws_connect("/ws/test")
+        await ws.send_str("not json")
+        await ws.send_str(json.dumps({"not": "a list"}))
+        await ws.send_str(json.dumps([]))
+        await ws.close()
+    finally:
+        await client.close()