Coverage for src/lexigram/admin/auth/store/password_reset_token_sql.py: 0%

1"""SQL-backed implementation of AdminPasswordResetTokenStoreProtocol. 

2 

3Owns all DDL and DML for the ``admin_password_reset_tokens`` table. The 

4service layer depends only on ``AdminPasswordResetTokenStoreProtocol`` from 

5``lexigram.admin.auth.protocols`` — never on this class directly. 

6""" 

7 

8from __future__ import annotations 

9 

10from datetime import UTC, datetime 

11from typing import Any 

12 

13from lexigram.admin.auth.types import AdminPasswordResetToken 

14from lexigram.admin.sql_dialect import is_postgres, now_expr 

15from lexigram.contracts.data import DatabaseProviderProtocol 

16from lexigram.di.decorators import inject 

17from lexigram.logging import get_logger 

18 

19logger = get_logger(__name__) 

20 

21_TABLE = "admin_password_reset_tokens" 

22 

23 

24def _parse_dt(value: Any) -> datetime | None: 

25 """Parse a provider-returned timestamp into a UTC-aware datetime.""" 

26 if value is None: 

27 return None 

28 if isinstance(value, datetime): 

29 return value 

30 parsed = datetime.fromisoformat(str(value)) 

31 if parsed.tzinfo is None: 

32 return parsed.replace(tzinfo=UTC) 

33 return parsed 

34 

35 

36@inject 

37class AdminPasswordResetTokenSqlStore: 

38 """SQL-backed store for password reset tokens. 

39 

40 Implements ``AdminPasswordResetTokenStoreProtocol`` via structural 

41 subtyping. Manages the ``admin_password_reset_tokens`` table including 

42 DDL bootstrap and single-use consumption semantics. 

43 """ 

44 

45 def __init__(self, db: DatabaseProviderProtocol) -> None: 

46 """Initialise with a resolved database provider. 

47 

48 Args: 

49 db: Framework database provider exposing ``execute`` and 

50 ``execute_query``. 

51 """ 

52 self._db = db 

53 self._initialized = False 

54 

55 # ------------------------------------------------------------------ 

56 # Schema bootstrap (DDL) 

57 # ------------------------------------------------------------------ 

58 

59 async def ensure_schema(self) -> None: 

60 """Create the token table if it does not exist (idempotent).""" 

61 if self._initialized: 

62 return 

63 if is_postgres(self._db): 

64 create_sql = f""" 

65 CREATE TABLE IF NOT EXISTS {_TABLE} ( 

66 token_hash VARCHAR(64) PRIMARY KEY, 

67 email VARCHAR(255) NOT NULL, 

68 expires_at TIMESTAMPTZ NOT NULL, 

69 consumed_at TIMESTAMPTZ, 

70 created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() 

71 ) 

72 """ 

73 else: 

74 create_sql = f""" 

75 CREATE TABLE IF NOT EXISTS {_TABLE} ( 

76 token_hash VARCHAR(64) PRIMARY KEY, 

77 email VARCHAR(255) NOT NULL, 

78 expires_at TIMESTAMP NOT NULL, 

79 consumed_at TIMESTAMP, 

80 created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP 

81 ) 

82 """ 

83 await self._db.execute(create_sql, []) 

84 self._initialized = True 

85 

86 # ------------------------------------------------------------------ 

87 # AdminPasswordResetTokenStoreProtocol 

88 # ------------------------------------------------------------------ 

89 

90 async def create(self, email: str, token_hash: str, expires_at: datetime) -> None: 

91 """Persist a new token record (see protocol docs).""" 

92 await self._db.execute( 

93 f"INSERT INTO {_TABLE} (token_hash, email, expires_at) VALUES (?, ?, ?)", # noqa: S608 — table name is module constant "admin_password_reset_tokens", never user input 

94 [token_hash, email, expires_at], 

95 ) 

96 

97 async def find_by_hash(self, token_hash: str) -> AdminPasswordResetToken | None: 

98 """Look up a token by sha256 hash (see protocol docs).""" 

99 result = await self._db.execute_query( 

100 f"SELECT token_hash, email, expires_at, consumed_at FROM {_TABLE} WHERE token_hash = ?", # noqa: S608 — table name is module constant "admin_password_reset_tokens", never user input 

101 [token_hash], 

102 ) 

103 row = None 

104 if hasattr(result, "rows") and result.rows: 

105 row = result.rows[0] 

106 elif isinstance(result, list) and result: 

107 row = result[0] 

108 elif isinstance(result, dict): 

109 row = result 

110 if not row: 

111 return None 

112 return AdminPasswordResetToken( 

113 email=str(row.get("email", "")), 

114 token_hash=str(row.get("token_hash", token_hash)), 

115 expires_at=_parse_dt(row.get("expires_at")) or datetime.now(UTC), 

116 consumed_at=_parse_dt(row.get("consumed_at")), 

117 ) 

118 

119 async def mark_consumed(self, token_hash: str) -> bool: 

120 """Atomically verify-and-consume a token in one statement. 

121 

122 Returns False when the token is missing, already consumed, or 

123 expired — the caller cannot distinguish which without a separate 

124 lookup. 

125 """ 

126 result = await self._db.execute( 

127 f"UPDATE {_TABLE} SET consumed_at = {now_expr(self._db)} " # noqa: S608 — table name is module constant, now_expr yields fixed NOW()/CURRENT_TIMESTAMP 

128 "WHERE token_hash = ? AND consumed_at IS NULL " 

129 f"AND expires_at > {now_expr(self._db)}", 

130 [token_hash], 

131 ) 

132 row_count = getattr(result, "row_count", None) 

133 if row_count is not None: 

134 return int(row_count) > 0 

135 return True 

136 

137 

138__all__ = ["AdminPasswordResetTokenSqlStore"]