Coverage for src/lexigram/admin/controllers/impersonation.py: 96%

1"""Impersonation controller for the admin panel.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from starlette.requests import Request 

8from starlette.responses import RedirectResponse, Response 

9from starlette.routing import Route 

10 

11from lexigram.admin.auth.store import AdminUserStoreProtocol 

12from lexigram.admin.services.impersonation import ImpersonationService 

13from lexigram.di.decorators import inject 

14from lexigram.logging import get_logger 

15from lexigram.serialization import dumps_str 

16 

17logger = get_logger(__name__) 

18 

19 

20@inject 

21class ImpersonationController: 

22 """Handles starting and stopping superadmin impersonation sessions. 

23 

24 CSRF is validated by the global ``AdminCsrfMiddleware`` already applied 

25 to every admin POST route — this controller does not duplicate that 

26 check. 

27 """ 

28 

29 def __init__( 

30 self, 

31 service: ImpersonationService, 

32 user_store: AdminUserStoreProtocol, 

33 ) -> None: 

34 self._service = service 

35 self._user_store = user_store 

36 

37 def get_routes(self) -> list[Any]: 

38 """Build routes explicitly, with the literal ``/impersonate/stop`` 

39 path ordered before the parameterised ``/impersonate/{user_id}`` 

40 path — Starlette matches routes in list order, and a 

41 dispatched-first ``{user_id}`` route would otherwise swallow 

42 ``/impersonate/stop`` (``user_id="stop"``). 

43 """ 

44 return [ 

45 Route( 

46 "/impersonate/stop", 

47 endpoint=self.stop_impersonation, 

48 methods=["POST"], 

49 name="admin_impersonation_stop", 

50 ), 

51 Route( 

52 "/impersonate/{user_id}", 

53 endpoint=self.start_impersonation, 

54 methods=["POST"], 

55 name="admin_impersonation_start", 

56 ), 

57 ] 

58 

59 async def start_impersonation(self, request: Request) -> Response: 

60 """Start impersonating the target user (D4/D5).""" 

61 actor = getattr(request.state, "user", None) 

62 if actor is None: 

63 return self._toast_error("You must be signed in.", status_code=403) 

64 

65 target_user_id = str(request.path_params.get("user_id", "")) 

66 

67 target_roles: list[str] | None = None 

68 try: 

69 target = await self._user_store.get_user_by_id(target_user_id) 

70 except Exception: # noqa: BLE001 — best-effort role lookup 

71 target = None 

72 if target is not None: 

73 target_roles = list(getattr(target, "roles", []) or []) 

74 

75 result = await self._service.start( 

76 actor, 

77 target_user_id, 

78 request=request, 

79 target_roles=target_roles, 

80 ) 

81 

82 if result.is_err(): 

83 error = result.unwrap_err() 

84 # DomainError.message is the bare human-readable text; str() 

85 # appends the LEX_ERR_* code enrichment (docs URL + arrow), 

86 # which is not latin-1 encodable and so not header-safe. 

87 return self._toast_error(error.message, status_code=403) 

88 

89 response = Response(status_code=200) 

90 response.headers["HX-Redirect"] = "/admin/users" 

91 return response 

92 

93 async def stop_impersonation(self, request: Request) -> Response: 

94 """Stop the active impersonation session for the current actor (D4/D7).""" 

95 actor = getattr(request.state, "user", None) 

96 if actor is not None: 

97 result = await self._service.stop(actor, request) 

98 if result.is_err(): 

99 from lexigram.admin.state.context import AdminContextManager 

100 

101 async with AdminContextManager(request) as ctx: 

102 ctx.add_flash("No active impersonation session to stop.", "warning") 

103 return RedirectResponse(url="/admin/", status_code=302) 

104 

105 @staticmethod 

106 def _toast_error(message: str, *, status_code: int) -> Response: 

107 """Build an error response carrying an HX-Trigger toast event.""" 

108 response = Response(content=message, status_code=status_code) 

109 response.headers["HX-Trigger"] = dumps_str( 

110 {"show-toast": {"message": message, "type": "error"}} 

111 ) 

112 return response