Coverage for src/lexigram/admin/auth/store/email_otp_sql.py: 87%

1"""SQL-backed implementation of AdminEmailOtpStoreProtocol. 

2 

3Owns all DDL and DML for the ``admin_email_otps`` table. The service layer 

4depends only on ``AdminEmailOtpStoreProtocol`` 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 

12import uuid 

13 

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_email_otps" 

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 AdminEmailOtpSqlStore: 

38 """SQL-backed store for email one-time-password codes. 

39 

40 Implements ``AdminEmailOtpStoreProtocol`` via structural subtyping. 

41 Manages the ``admin_email_otps`` table including DDL bootstrap and 

42 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 OTP 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 id TEXT PRIMARY KEY, 

67 user_id TEXT NOT NULL, 

68 code_hash VARCHAR(64) NOT NULL, 

69 expires_at TIMESTAMPTZ NOT NULL, 

70 used_at TIMESTAMPTZ, 

71 created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() 

72 ) 

73 """ 

74 else: 

75 create_sql = f""" 

76 CREATE TABLE IF NOT EXISTS {_TABLE} ( 

77 id TEXT PRIMARY KEY, 

78 user_id TEXT NOT NULL, 

79 code_hash VARCHAR(64) NOT NULL, 

80 expires_at TIMESTAMP NOT NULL, 

81 used_at TIMESTAMP, 

82 created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP 

83 ) 

84 """ 

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

86 self._initialized = True 

87 

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

89 # AdminEmailOtpStoreProtocol 

90 # ------------------------------------------------------------------ 

91 

92 async def save(self, user_id: str, code_hash: str, expires_at: datetime) -> None: 

93 """Persist a new emailed code (see protocol docs).""" 

94 await self._db.execute( 

95 f"INSERT INTO {_TABLE} (id, user_id, code_hash, expires_at) " # noqa: S608 — table name is module constant "admin_email_otps", never user input 

96 "VALUES (?, ?, ?, ?)", 

97 [str(uuid.uuid4()), user_id, code_hash, expires_at], 

98 ) 

99 

100 async def consume(self, user_id: str, code_hash: str) -> bool: 

101 """Atomically consume a matching unexpired code (see protocol docs).""" 

102 result = await self._db.execute( 

103 f""" 

104 UPDATE {_TABLE} SET used_at = {now_expr(self._db)} 

105 WHERE user_id = ? 

106 AND code_hash = ? 

107 AND used_at IS NULL 

108 AND expires_at > {now_expr(self._db)} 

109 """, # noqa: S608 — table name is module constant, now_expr yields fixed NOW()/CURRENT_TIMESTAMP 

110 [user_id, code_hash], 

111 ) 

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

113 if row_count is not None: 

114 return int(row_count) > 0 

115 return True 

116 

117 async def last_sent_at(self, user_id: str) -> datetime | None: 

118 """Return the creation time of the most recent code (see protocol docs).""" 

119 result = await self._db.execute_query( 

120 f"SELECT created_at FROM {_TABLE} " # noqa: S608 — table name is module constant "admin_email_otps", never user input 

121 "WHERE user_id = ? ORDER BY created_at DESC LIMIT 1", 

122 [user_id], 

123 ) 

124 row = None 

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

126 row = result.rows[0] 

127 elif isinstance(result, list) and result: 

128 row = result[0] 

129 elif isinstance(result, dict): 

130 row = result 

131 if not row: 

132 return None 

133 return _parse_dt(row.get("created_at")) 

134 

135 

136__all__ = ["AdminEmailOtpSqlStore"]