Coverage for src / lexigram / admin / auth / guard_chain.py: 57%

1"""AdminGuardChain — implements GuardChainProtocol for lexigram-admin.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7from lexigram.contracts.exceptions.security import GuardDeniedError 

8from lexigram.di.decorators import inject 

9from lexigram.logging import get_logger 

10 

11if TYPE_CHECKING: 

12 from lexigram.contracts.web.guard import GuardProtocol 

13 

14logger = get_logger(__name__) 

15 

16 

17@inject 

18class AdminGuardChain: 

19 """Executes a sequence of guards, short-circuiting on first denial. 

20 

21 Implements GuardChainProtocol so it can be resolved from the container. 

22 

23 Args: 

24 guards: Initial list of guards to add to the chain. 

25 """ 

26 

27 def __init__(self, guards: list[GuardProtocol] | None = None) -> None: 

28 self._guards: list[GuardProtocol] = list(guards or []) 

29 

30 def add(self, guard: GuardProtocol) -> AdminGuardChain: 

31 """Add a guard to the chain (fluent API).""" 

32 self._guards.append(guard) 

33 return self 

34 

35 async def execute(self, context: dict[str, Any]) -> None: 

36 """Execute all guards in order. Raises GuardDeniedError on first denial. 

37 

38 Args: 

39 context: Arbitrary request context forwarded to each guard. 

40 

41 Raises: 

42 GuardDeniedError: If any guard's can_activate returns False. 

43 """ 

44 for guard in self._guards: 

45 allowed = await guard.can_activate(context) # type: ignore[arg-type] 

46 if not allowed: 

47 guard_name = type(guard).__name__ 

48 logger.warning("GuardProtocol %s denied access", guard_name) 

49 raise GuardDeniedError( 

50 f"Access denied by guard: {guard_name}", 

51 ) 

52 

53 

54__all__ = ["AdminGuardChain"]