Coverage for src/lexigram/auth/storage/in_memory_stores.py: 60%

1"""In-memory session store for development and testing.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7if TYPE_CHECKING: 

8 from lexigram.auth.models.session import UserSession 

9 

10 

11class InMemorySessionStore: 

12 """In-memory session storage for development and testing. 

13 

14 This store is NOT suitable for production as it does not persist data 

15 across restarts and does not support multi-instance deployments. 

16 """ 

17 

18 def __init__(self) -> None: 

19 self._sessions: dict[str, UserSession] = {} 

20 

21 async def create(self, session: UserSession) -> str: 

22 """Create a new session.""" 

23 self._sessions[session.session_id] = session 

24 return session.session_id 

25 

26 async def get(self, session_id: str) -> UserSession | None: 

27 """Get a session by ID.""" 

28 return self._sessions.get(session_id) 

29 

30 async def update(self, session_id: str, **updates: Any) -> None: 

31 """Update session fields.""" 

32 session = self._sessions.get(session_id) 

33 if session: 

34 for key, value in updates.items(): 

35 if hasattr(session, key): 

36 object.__setattr__(session, key, value) 

37 

38 async def delete(self, session_id: str) -> bool: 

39 """Delete a session.""" 

40 if session_id in self._sessions: 

41 del self._sessions[session_id] 

42 return True 

43 return False 

44 

45 async def delete_all_for_user(self, user_id: str) -> int: 

46 """Delete all sessions for a user.""" 

47 to_delete = [sid for sid, s in self._sessions.items() if s.user_id == user_id] 

48 for sid in to_delete: 

49 del self._sessions[sid] 

50 return len(to_delete) 

51 

52 async def list_for_user(self, user_id: str) -> list[UserSession]: 

53 """List all sessions for a user.""" 

54 return [ 

55 s 

56 for s in self._sessions.values() 

57 if s.user_id == user_id and s.is_active and not s.is_expired() 

58 ] 

59 

60 

61class InMemoryMFAStore: 

62 """In-memory MFA storage for development and testing. 

63 

64 This store is NOT suitable for production. 

65 """ 

66 

67 def __init__(self) -> None: 

68 self._mfa: dict[str, dict[str, Any]] = {} 

69 

70 async def get(self, user_id: str) -> dict[str, Any] | None: 

71 """Get MFA config for a user.""" 

72 return self._mfa.get(user_id) 

73 

74 async def save(self, mfa_data: dict[str, object]) -> None: 

75 """Save MFA config for a user.""" 

76 user_id = mfa_data.get("user_id") 

77 if isinstance(user_id, str): 

78 self._mfa[user_id] = mfa_data 

79 

80 async def delete(self, user_id: str) -> bool: 

81 """Delete MFA config for a user.""" 

82 if user_id in self._mfa: 

83 del self._mfa[user_id] 

84 return True 

85 return False 

86 

87 

88__all__ = [ 

89 "InMemoryMFAStore", 

90 "InMemorySessionStore", 

91]