Coverage for src/lilbee/server/routes/setup.py: 100%

1"""Setup routes: status and bootstrap for optional runtime components. 

2 

3Currently exposes Playwright Chromium bootstrap (needed for /crawl). The 

4SSE event sequence mirrors what the TUI does in 

5``TaskBarController.ensure_chromium`` so a stream consumer can render a 

6matching ``setup`` progress indicator. 

7 

8Endpoints: 

9 GET /setup/crawler/status → { installed, package_installed, 

10 chromium_installed, component, browsers_path } 

11 POST /setup/crawler → text/event-stream of setup_start → 

12 setup_progress → setup_done → done 

13""" 

14 

15from __future__ import annotations 

16 

17import asyncio 

18from collections.abc import AsyncGenerator 

19from typing import Any 

20 

21from litestar import get, post 

22from litestar.response import Stream 

23 

24from lilbee.crawler import ( 

25 bootstrap_chromium, 

26 chromium_installed, 

27 crawler_available, 

28 crawler_browsers_path, 

29) 

30from lilbee.server.handlers import SseStream 

31from lilbee.server.handlers.sse import SSE_MEDIA_TYPE 

32 

33 

34@get("/setup/crawler/status") 

35async def setup_crawler_status_route() -> dict[str, Any]: 

36 """Return whether the crawler is fully ready (Python package + Chromium).""" 

37 package_installed = crawler_available() 

38 chromium_ok = chromium_installed() 

39 return { 

40 "installed": package_installed and chromium_ok, 

41 "package_installed": package_installed, 

42 "chromium_installed": chromium_ok, 

43 "component": "chromium", 

44 "browsers_path": str(crawler_browsers_path()), 

45 } 

46 

47 

48async def _bootstrap_crawler_stream() -> AsyncGenerator[str, None]: 

49 sse = SseStream() 

50 

51 async def _run() -> None: 

52 try: 

53 await bootstrap_chromium(on_progress=sse.callback) 

54 finally: 

55 sse.queue.put_nowait(None) 

56 

57 task = asyncio.create_task(_run()) 

58 async for event in sse.drain(task, "Crawler setup stream"): 

59 yield event 

60 # This copy used to skip the cancel check and emit a done frame even to a 

61 # client that had already disconnected; the shared helper does not. 

62 frame = sse.terminal_frame(task, lambda _: {}) 

63 if frame is not None: 

64 yield frame 

65 

66 

67@post("/setup/crawler", media_type=SSE_MEDIA_TYPE) 

68async def setup_crawler_route() -> Stream: 

69 """Stream the Chromium bootstrap subprocess as SSE events.""" 

70 return Stream(_bootstrap_crawler_stream(), media_type=SSE_MEDIA_TYPE) 

71 

72 

73__all__ = [ 

74 "setup_crawler_route", 

75 "setup_crawler_status_route", 

76]