Coverage for src/lexigram/auth/admin/handlers/active_sessions.py: 95%

1"""Active sessions widget handler.""" 

2 

3from __future__ import annotations 

4 

5from datetime import UTC, datetime 

6 

7from lexigram.contracts.admin import ( 

8 SessionCountProtocol, 

9 Stat, 

10 StatContent, 

11 WidgetParams, 

12) 

13from lexigram.contracts.admin.errors import AdminError 

14from lexigram.contracts.auth import SessionRepositoryProtocol 

15from lexigram.result import Ok, Result 

16 

17 

18class ActiveSessionsWidgetHandler: 

19 """Handler for the active sessions widget. 

20 

21 Args: 

22 session_repository: optional session repository. Degrades to a zero 

23 count when absent or when it lacks the session-count capability. 

24 """ 

25 

26 def __init__( 

27 self, session_repository: SessionRepositoryProtocol | None = None 

28 ) -> None: 

29 """Initialize handler with the session repository. 

30 

31 Args: 

32 session_repository: repository implementing SessionRepositoryProtocol. 

33 When None, the widget reports a zero count. 

34 """ 

35 self._session_repository = session_repository 

36 

37 async def get_data(self, params: WidgetParams) -> Result[StatContent, AdminError]: 

38 """Fetch active sessions data. 

39 

40 Mirrors the widget template: the count is rendered statically with 

41 neutral styling and the peak line is shown only when non-zero — no 

42 tone/threshold ladder exists in the template. 

43 

44 Infrastructure failures propagate as exceptions. 

45 

46 Args: 

47 params: Widget parameters (unused for this widget). 

48 

49 Returns: 

50 Result containing StatContent or AdminError. 

51 """ 

52 count = 0 

53 if isinstance(self._session_repository, SessionCountProtocol): 

54 cutoff = datetime.now(UTC) 

55 count = await self._session_repository.count_active(cutoff) 

56 return Ok(self._build_content(count=count, peak_today=0)) 

57 

58 def _build_content(self, count: int, peak_today: int) -> StatContent: 

59 """Build the StatContent mirroring the widget template. 

60 

61 Args: 

62 count: Number of currently active sessions. 

63 peak_today: Peak session count today. 

64 

65 Returns: 

66 StatContent with neutral styling and a conditional peak stat. 

67 """ 

68 stats: list[Stat] = [ 

69 Stat(label="Currently Active", value=str(count)), 

70 ] 

71 if peak_today > 0: 

72 stats.append(Stat(label="Peak today", value=str(peak_today))) 

73 return StatContent(stats=tuple(stats)) 

74 

75 

76__all__ = ["ActiveSessionsWidgetHandler"]