Coverage for tests / unit / cli / test_serve.py: 97%
71 statements
« prev ^ index » next coverage.py v7.13.3, created at 2026-02-16 21:32 +0000
1from asyncio import TimeoutError, wait_for
2from pathlib import Path
3from unittest.mock import AsyncMock, MagicMock, patch
5from pytest import mark
6from starlette.responses import HTMLResponse
8from graphable.cli.commands.serve import Server, serve_command
11class TestServer:
12 def test_init(self):
13 path = Path("test.json")
14 server = Server(path, tag="v1")
15 assert server.path == path
16 assert server.tag == "v1"
17 assert len(server.connections) == 0
18 assert len(server.app.routes) == 2
20 @mark.anyio
21 @patch("graphable.cli.commands.serve.load_graph")
22 @patch("graphable.cli.commands.serve.create_topology_html")
23 async def test_index_success(self, mock_html, mock_load):
24 mock_load.return_value = MagicMock()
25 mock_html.return_value = "<html></html>"
27 server = Server(Path("test.json"))
28 request = MagicMock()
29 response = await server.index(request)
31 assert isinstance(response, HTMLResponse)
32 assert response.body == b"<html></html>"
33 assert response.status_code == 200
35 @mark.anyio
36 @patch("graphable.cli.commands.serve.load_graph")
37 async def test_index_error(self, mock_load):
38 mock_load.side_effect = Exception("Load error")
40 server = Server(Path("test.json"))
41 request = MagicMock()
42 response = await server.index(request)
44 assert isinstance(response, HTMLResponse)
45 assert b"Error loading graph" in response.body
46 assert b"Load error" in response.body
47 assert response.status_code == 500
49 @mark.anyio
50 async def test_websocket_endpoint(self):
51 server = Server(Path("test.json"))
52 websocket = AsyncMock()
54 # Simulate a single receive then disconnect
55 websocket.receive_text.side_effect = ["ping", Exception("Disconnect")]
57 await server.websocket_endpoint(websocket)
59 websocket.accept.assert_called_once()
60 assert websocket not in server.connections
62 @mark.anyio
63 @patch("graphable.cli.commands.serve.awatch")
64 async def test_watch_file(self, mock_awatch):
65 # Mock awatch to yield once then stop
66 mock_changes = AsyncMock()
67 mock_changes.__aiter__.return_value = [[(1, "test.json")]]
68 mock_awatch.return_value = mock_changes
70 server = Server(Path("test.json"))
71 ws = AsyncMock()
72 server.connections.add(ws)
74 # Run watch_file in a way we can stop it
75 try:
76 await wait_for(server.watch_file(), timeout=0.1)
77 except (
78 TimeoutError,
79 TypeError,
80 ): # TypeError might happen due to mock yielding
81 pass
83 ws.send_text.assert_called_with("reload")
86@patch("graphable.cli.commands.serve.UvicornServer")
87@patch("graphable.cli.commands.serve.Config")
88@patch("graphable.cli.commands.serve.get_event_loop")
89def test_serve_command(mock_loop, mock_config, mock_uv_server):
90 mock_loop_instance = MagicMock()
91 mock_loop.return_value = mock_loop_instance
93 serve_command(Path("test.json"), port=1234, tag="v2")
95 mock_config.assert_called_once()
96 args, kwargs = mock_config.call_args
97 assert kwargs["port"] == 1234
99 assert mock_loop_instance.create_task.called
100 assert mock_loop_instance.run_until_complete.called