Coverage for custom_components/dev_shell_server/websocket_api.py: 54%

1"""Websocket commands: dev_shell_server/exec, dev_shell_server/reset, 

2dev_shell_server/sessions (all admin only).""" 

3 

4from __future__ import annotations 

5 

6from typing import Any 

7 

8import voluptuous as vol 

9from homeassistant.components import websocket_api 

10from homeassistant.core import HomeAssistant, callback 

11 

12from .const import DEFAULT_SESSION, DOMAIN 

13from .session import SessionManager 

14 

15 

16@callback 

17def async_register(hass: HomeAssistant) -> None: 

18 websocket_api.async_register_command(hass, ws_exec) 

19 websocket_api.async_register_command(hass, ws_reset) 

20 websocket_api.async_register_command(hass, ws_sessions) 

21 

22 

23def _manager( 

24 hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any] 

25) -> SessionManager | None: 

26 if (manager := hass.data.get(DOMAIN)) is None: 

27 connection.send_error( 

28 msg["id"], "not_loaded", "Dev Shell integration is not loaded" 

29 ) 

30 return manager 

31 

32 

33@websocket_api.require_admin 

34# ty resolves vol.Required/vol.Optional's own Marker subclassing too narrowly 

35# to see this inline schema dict as a VolDictType - a known voluptuous/ 

36# static-typing friction point across the HA ecosystem, not specific to this 

37# schema (mypy, via homeassistant-stubs, is fine with it as-is). 

38@websocket_api.websocket_command({ 

39 vol.Required("type"): "dev_shell_server/exec", 

40 vol.Required("code"): str, 

41 vol.Optional("session", default=DEFAULT_SESSION): str, 

42 vol.Optional("timeout"): vol.Coerce(float), 

43 vol.Optional("color", default=False): bool, 

44 vol.Optional("width", default=88): vol.Coerce(int), 

45}) 

46@websocket_api.async_response 

47async def ws_exec( 

48 hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any] 

49) -> None: 

50 if (manager := _manager(hass, connection, msg)) is None: 

51 return 

52 result = await manager.get(msg["session"]).run( 

53 msg["code"], msg.get("timeout"), color=msg["color"], width=msg["width"] 

54 ) 

55 connection.send_result(msg["id"], result.as_dict()) 

56 

57 

58@websocket_api.require_admin 

59@websocket_api.websocket_command({ 

60 vol.Required("type"): "dev_shell_server/reset", 

61 vol.Optional("session", default=DEFAULT_SESSION): str, 

62}) 

63@callback 

64def ws_reset( 

65 hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any] 

66) -> None: 

67 if (manager := _manager(hass, connection, msg)) is None: 

68 return 

69 connection.send_result(msg["id"], {"reset": manager.reset(msg["session"])}) 

70 

71 

72@websocket_api.require_admin 

73@websocket_api.websocket_command({vol.Required("type"): "dev_shell_server/sessions"}) 

74@callback 

75def ws_sessions( 

76 hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any] 

77) -> None: 

78 if (manager := _manager(hass, connection, msg)) is None: 

79 return 

80 connection.send_result(msg["id"], {"sessions": manager.describe()})