Coverage for src/lexigram/auth/services/result_pattern_service.py: 59%

1"""Authentication service using Result pattern for error handling. 

2 

3This service demonstrates the Result[T, AuthError] pattern for auth operations. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import TYPE_CHECKING 

9 

10from lexigram.auth.exceptions import InvalidCredentialsError, PasswordPolicyError 

11from lexigram.contracts.auth.exceptions import AuthError 

12from lexigram.logging import get_logger 

13from lexigram.result import Err, Ok, Result 

14 

15if TYPE_CHECKING: 

16 from lexigram.auth.models.token import AuthToken 

17 from lexigram.contracts.events.protocols import EventBusProtocol 

18 from lexigram.contracts.infra.cache.protocols import CacheBackendProtocol 

19 

20logger = get_logger(__name__) 

21 

22 

23class AuthServiceWithResultPattern: 

24 """Authentication service using Result pattern. 

25 

26 This service wraps auth operations and returns Result[T, AuthError] 

27 instead of raising exceptions or returning bare types. 

28 """ 

29 

30 def __init__( 

31 self, 

32 cache: CacheBackendProtocol | None = None, 

33 event_bus: EventBusProtocol | None = None, 

34 ) -> None: 

35 """Initialize the auth service with optional dependencies. 

36 

37 Args: 

38 cache: Cache backend for token validation 

39 event_bus: Event bus for auth events 

40 """ 

41 self.cache = cache 

42 self.event_bus = event_bus 

43 

44 async def validate_password( 

45 self, 

46 password: str, 

47 min_length: int = 8, 

48 require_uppercase: bool = True, 

49 require_digits: bool = True, 

50 ) -> Result[None, AuthError]: 

51 """Validate password against policy requirements. 

52 

53 Args: 

54 password: Password to validate 

55 min_length: Minimum password length 

56 require_uppercase: Whether to require uppercase letters 

57 require_digits: Whether to require digits 

58 

59 Returns: 

60 Ok(None) if valid, Err(PasswordPolicyError) if invalid 

61 """ 

62 try: 

63 errors = [] 

64 

65 if len(password) < min_length: 

66 errors.append(f"Password must be at least {min_length} characters") 

67 

68 if require_uppercase and not any(c.isupper() for c in password): 

69 errors.append("Password must contain uppercase letters") 

70 

71 if require_digits and not any(c.isdigit() for c in password): 

72 errors.append("Password must contain digits") 

73 

74 if errors: 

75 message = "; ".join(errors) 

76 return Err(PasswordPolicyError(message)) 

77 

78 return Ok(None) 

79 except Exception as e: # noqa: BLE001 # demo service - broad catch for Result pattern demonstration 

80 logger.error("password_validation_failed: %s", e) 

81 return Err(AuthError(f"Password validation failed: {e}")) 

82 

83 async def verify_credentials( 

84 self, 

85 username: str, 

86 password: str, 

87 stored_hash: str | None = None, 

88 ) -> Result[bool, AuthError]: 

89 """Verify user credentials against stored hash. 

90 

91 Args: 

92 username: Username to verify 

93 password: Password to check 

94 stored_hash: Stored password hash (for demo purposes) 

95 

96 Returns: 

97 Ok(True) if credentials valid, Err(InvalidCredentialsError) if not 

98 """ 

99 try: 

100 if not username or not password: 

101 return Err( 

102 InvalidCredentialsError("Username and password are required") 

103 ) 

104 

105 # Simple demo verification (replace with actual hash verification) 

106 if stored_hash is None: 

107 return Err(InvalidCredentialsError("No credentials on file for user")) 

108 

109 # In production: use proper password hasher 

110 # password_valid = await self.password_hasher.verify(password, stored_hash) 

111 password_valid = password == stored_hash 

112 

113 if not password_valid: 

114 return Err(InvalidCredentialsError("Invalid password")) 

115 

116 return Ok(True) 

117 except Exception as e: # noqa: BLE001 # demo service - broad catch for Result pattern demonstration 

118 logger.error("credential_verification_failed: %s", e) 

119 return Err(AuthError(f"Credential verification failed: {e}")) 

120 

121 async def get_cached_token( 

122 self, token_id: str 

123 ) -> Result[AuthToken | None, AuthError]: 

124 """Get cached auth token. 

125 

126 Args: 

127 token_id: Token identifier 

128 

129 Returns: 

130 Ok(token) if found, Ok(None) if not found, Err(AuthError) on failure 

131 """ 

132 try: 

133 if not self.cache: 

134 return Ok(None) 

135 

136 cache_result = await self.cache.get(f"token:{token_id}") 

137 if cache_result.is_err(): 

138 logger.warning("cache_get_failed: %s", cache_result.unwrap_err()) 

139 return Ok(None) 

140 

141 cached_data = cache_result.unwrap() 

142 if cached_data is None: 

143 return Ok(None) 

144 

145 # In production: deserialize properly 

146 return Ok(cached_data) 

147 except Exception as e: # noqa: BLE001 # demo service - broad catch for Result pattern demonstration 

148 logger.error("token_cache_retrieval_failed: %s", e) 

149 return Err(AuthError(f"Token retrieval failed: {e}")) 

150 

151 async def cache_token( 

152 self, 

153 token_id: str, 

154 token: AuthToken, 

155 ttl: int = 3600, 

156 ) -> Result[None, AuthError]: 

157 """Cache auth token with TTL. 

158 

159 Args: 

160 token_id: Token identifier 

161 token: Token to cache 

162 ttl: Time to live in seconds 

163 

164 Returns: 

165 Ok(None) if successful, Err(AuthError) on failure 

166 """ 

167 try: 

168 if not self.cache: 

169 return Ok(None) 

170 

171 cache_result = await self.cache.set( 

172 f"token:{token_id}", 

173 token, 

174 ttl=ttl, 

175 ) 

176 

177 if cache_result.is_err(): 

178 logger.warning("cache_set_failed: %s", cache_result.unwrap_err()) 

179 return Err(AuthError("Failed to cache token")) 

180 

181 return Ok(None) 

182 except Exception as e: # noqa: BLE001 # demo service - broad catch for Result pattern demonstration 

183 logger.error("token_cache_failed: %s", e) 

184 return Err(AuthError(f"Token caching failed: {e}")) 

185 

186 async def invalidate_token(self, token_id: str) -> Result[None, AuthError]: 

187 """Invalidate a cached token. 

188 

189 Args: 

190 token_id: Token identifier to invalidate 

191 

192 Returns: 

193 Ok(None) if successful, Err(AuthError) on failure 

194 """ 

195 try: 

196 if not self.cache: 

197 return Ok(None) 

198 

199 delete_result = await self.cache.delete(f"token:{token_id}") 

200 if delete_result.is_err(): 

201 logger.warning("cache_delete_failed: %s", delete_result.unwrap_err()) 

202 return Err(AuthError("Failed to invalidate token")) 

203 

204 return Ok(None) 

205 except Exception as e: # noqa: BLE001 # demo service - broad catch for Result pattern demonstration 

206 logger.error("token_invalidation_failed: %s", e) 

207 return Err(AuthError(f"Token invalidation failed: {e}")) 

208 

209 

210__all__ = ["AuthServiceWithResultPattern"]