Coverage for src/lexigram/admin/dashboard/widget_stream.py: 93%

1"""Widget-stream SSE route: live push for dashboard widgets. 

2 

3Wraps SubjectAdminEventHub in the sanctioned sse_from_stream bridge 

4(resolved from the container as ``ReactiveSseBridgeProtocol``, provided by 

5lexigram-web), narrowing the caller-requested `resources` filter to 

6only resources the caller is authorized to list via PermissionService. 

7""" 

8 

9from __future__ import annotations 

10 

11from typing import TYPE_CHECKING, Any, cast 

12 

13from lexigram.admin.realtime import SubjectAdminEventHub 

14from lexigram.contracts.web.sse import ReactiveSseBridgeProtocol 

15from lexigram.reactive import Stream 

16from lexigram.serialization import dumps_str 

17 

18if TYPE_CHECKING: 

19 from collections.abc import Awaitable, Callable 

20 

21 from starlette.requests import Request 

22 from starlette.responses import Response 

23 

24 from lexigram.admin.rbac.service import PermissionService 

25 

26 

27async def authorized_resources( 

28 user: Any, 

29 requested: str | None, 

30 permission_service: PermissionService, 

31) -> list[str] | None: 

32 """Narrow a caller-supplied resources= filter to ones the caller may list. 

33 

34 Fail-closed on resources with no registered permission schema: 

35 ``PermissionService.can_list`` itself returns ``True`` for unknown 

36 resources (management-UI semantics: no permission model = public), 

37 which is the wrong default on a channel boundary — a name without a 

38 schema is treated as unauthorized here. 

39 

40 Args: 

41 user: The authenticated request user (or None). 

42 requested: Raw comma-separated `resources` query param value. 

43 permission_service: Resolves per-resource `can_list` authorization. 

44 

45 Returns: 

46 The allowed subset, or None when nothing was requested or every 

47 requested resource was denied — both mean "apply no resource 

48 filter," and None is unambiguous where an empty list could be 

49 misread as "match nothing." 

50 """ 

51 if not requested: 

52 return None 

53 candidates = [r.strip() for r in requested.split(",") if r.strip()] 

54 allowed = [ 

55 r 

56 for r in candidates 

57 if permission_service.get_schema(r) is not None 

58 and await permission_service.can_list(user, r) 

59 ] 

60 return allowed or None 

61 

62 

63def build_widget_event_stream_handler( 

64 widget_hub: SubjectAdminEventHub, 

65 permission_service: PermissionService, 

66 sse_bridge: ReactiveSseBridgeProtocol | None = None, 

67) -> Callable[[Request], Awaitable[Response]]: 

68 """Build the ASGI route handler for GET /admin/_sse/widgets. 

69 

70 Args: 

71 widget_hub: Hub to subscribe to for live admin events. 

72 permission_service: Used to authorize the caller's resources= filter. 

73 sse_bridge: Optional :class:`ReactiveSseBridgeProtocol` bridging the 

74 reactive stream into an SSE response (resolved from the 

75 container by the mount pipeline; provided by lexigram-web). 

76 Required to serve requests — when omitted the returned handler 

77 raises ``RuntimeError``. 

78 

79 Returns: 

80 An async Starlette-style route handler. 

81 

82 Raises: 

83 RuntimeError: When a request arrives and no *sse_bridge* was given. 

84 """ 

85 

86 def _require_bridge() -> ReactiveSseBridgeProtocol: 

87 if sse_bridge is None: 

88 raise RuntimeError( 

89 "no sse_bridge provided; resolve ReactiveSseBridgeProtocol " 

90 "from the container (lexigram-web) before building the " 

91 "widget stream handler" 

92 ) 

93 return sse_bridge 

94 

95 async def widget_event_stream(request: Request) -> Response: 

96 user = getattr(request.state, "user", None) 

97 user_id = getattr(user, "user_id", None) if user else None 

98 tenant_id = getattr(request.state, "tenant_id", None) or None 

99 resources = await authorized_resources( 

100 user, request.query_params.get("resources"), permission_service 

101 ) 

102 

103 def serialize(event: Any) -> str: 

104 return dumps_str(event.to_dict()) 

105 

106 return cast( 

107 "Response", 

108 _require_bridge()( 

109 Stream( 

110 widget_hub.subscribe( 

111 user_id=user_id, 

112 resources=resources, 

113 tenant_id=tenant_id, 

114 ) 

115 ), 

116 serializer=serialize, 

117 ), 

118 ) 

119 

120 return widget_event_stream 

121 

122 

123__all__ = ["authorized_resources", "build_widget_event_stream_handler"]