Coverage for src/lexigram/admin/auth/store/auth_provider.py: 62%

1""" 

2AuthProvider adapter for admin user store. 

3""" 

4 

5from __future__ import annotations 

6 

7from typing import TYPE_CHECKING 

8 

9from lexigram.admin.auth.store.base import AbstractAdminUserStore 

10from lexigram.di.decorators import inject 

11from lexigram.logging import get_logger 

12 

13if TYPE_CHECKING: 

14 from lexigram.contracts import AuthenticatedUserProtocol 

15 from lexigram.contracts.auth.protocols import PasswordHasherProtocol 

16 

17logger = get_logger(__name__) 

18 

19 

20@inject 

21class AuthProviderAdminUserStore(AbstractAdminUserStore): 

22 """Adapter that delegates admin user operations to an existing AuthProvider. 

23 

24 This allows the admin subsystem to avoid implementing user lifecycle and 

25 authentication itself when a canonical AuthProvider (lexigram-auth) is 

26 available in the application's DI container. 

27 

28 The adapter expects an AuthProvider instance (which exposes `authenticate_user`, 

29 `user_store` with `get_user_by_username`, `get_user_by_email`, and `delete_user`). 

30 """ 

31 

32 def __init__( 

33 self, 

34 auth_provider: object, 

35 password_hasher: PasswordHasherProtocol | None = None, 

36 ): 

37 self.auth_provider = auth_provider 

38 self._password_hasher = password_hasher 

39 # No local cache; operate directly against auth_provider for canonical behaviour 

40 

41 async def get_by_id(self, user_id: str) -> AuthenticatedUserProtocol | None: 

42 # AuthProvider user_store implementations commonly support get_user_by_id 

43 user_store = getattr(self.auth_provider, "user_store", None) 

44 if user_store and hasattr(user_store, "get_user_by_id"): 

45 return await user_store.get_user_by_id(user_id) 

46 # Fallback to username lookup 

47 if user_store and hasattr(user_store, "get_user_by_username"): 

48 return await user_store.get_user_by_username(user_id) 

49 return None 

50 

51 async def get_by_email(self, email: str) -> AuthenticatedUserProtocol | None: 

52 user_store = getattr(self.auth_provider, "user_store", None) 

53 if user_store and hasattr(user_store, "get_user_by_email"): 

54 return await user_store.get_user_by_email(email) 

55 return None 

56 

57 async def get_by_username(self, username: str) -> AuthenticatedUserProtocol | None: 

58 user_store = getattr(self.auth_provider, "user_store", None) 

59 if user_store and hasattr(user_store, "get_user_by_username"): 

60 return await user_store.get_user_by_username(username) 

61 return None 

62 

63 async def authenticate( 

64 self, email: str, password: str 

65 ) -> AuthenticatedUserProtocol | None: 

66 # Prefer provider.authenticate_user (legacy providers) … 

67 if hasattr(self.auth_provider, "authenticate_user"): 

68 result = await self.auth_provider.authenticate_user(email, password) 

69 return result.unwrap_or(None) 

70 # … or provider.service.authenticate_user (G4-A1.3 split: method lives on AuthenticationService) 

71 svc = getattr(self.auth_provider, "service", None) 

72 if svc is not None and hasattr(svc, "authenticate_user"): 

73 result = await svc.authenticate_user(email, password) 

74 return result.unwrap_or(None) 

75 # Fallback to user_store verification if available 

76 user_store = getattr(self.auth_provider, "user_store", None) 

77 if user_store and hasattr(user_store, "get_user_by_email"): 

78 user = await user_store.get_user_by_email(email) 

79 if not user: 

80 return None 

81 hashed = getattr(user, "hashed_password", None) 

82 if not hashed or self._password_hasher is None: 

83 return None 

84 if await self._password_hasher.verify(password, hashed): 

85 return user 

86 return None 

87 

88 async def count(self) -> int: 

89 """Return the total number of users via the underlying auth provider store.""" 

90 user_store = getattr(self.auth_provider, "user_store", None) 

91 if user_store: 

92 # Prefer common count methods used by user stores 

93 if hasattr(user_store, "count_users"): 

94 return await user_store.count_users() 

95 if hasattr(user_store, "count"): 

96 return await user_store.count() 

97 return 0