107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
"""Unit tests for app.services.otp using an in-memory Redis fake."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from app.services import otp
|
|
|
|
|
|
class FakeRedis:
|
|
"""Minimal subset of redis-py used by services.otp."""
|
|
|
|
def __init__(self) -> None:
|
|
self._store: dict[str, tuple[Any, float | None]] = {}
|
|
|
|
def _expired(self, key: str) -> bool:
|
|
item = self._store.get(key)
|
|
if item is None:
|
|
return True
|
|
_, expires = item
|
|
if expires is not None and expires <= time.time():
|
|
self._store.pop(key, None)
|
|
return True
|
|
return False
|
|
|
|
def set(self, key: str, value: Any, ex: int | None = None) -> None:
|
|
expires = time.time() + ex if ex else None
|
|
self._store[key] = (value, expires)
|
|
|
|
def get(self, key: str) -> Any:
|
|
if self._expired(key):
|
|
return None
|
|
return self._store[key][0]
|
|
|
|
def ttl(self, key: str) -> int:
|
|
if self._expired(key):
|
|
return -2
|
|
_, expires = self._store[key]
|
|
if expires is None:
|
|
return -1
|
|
return max(int(expires - time.time()), 0)
|
|
|
|
def delete(self, key: str) -> None:
|
|
self._store.pop(key, None)
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_redis(monkeypatch):
|
|
fake = FakeRedis()
|
|
monkeypatch.setattr(otp, "redis_client", fake)
|
|
return fake
|
|
|
|
|
|
def test_issue_code_returns_six_digits(fake_redis):
|
|
code = otp.issue_code("alice@example.com", "register")
|
|
assert len(code) == 6
|
|
assert code.isdigit()
|
|
|
|
|
|
def test_verify_code_succeeds_and_consumes(fake_redis):
|
|
code = otp.issue_code("alice@example.com", "register")
|
|
otp.verify_code("alice@example.com", "register", code)
|
|
with pytest.raises(otp.OtpExpired):
|
|
otp.verify_code("alice@example.com", "register", code)
|
|
|
|
|
|
def test_verify_code_rejects_wrong_code(fake_redis):
|
|
otp.issue_code("alice@example.com", "register")
|
|
with pytest.raises(otp.OtpInvalid):
|
|
otp.verify_code("alice@example.com", "register", "000000")
|
|
|
|
|
|
def test_verify_code_locks_after_max_attempts(fake_redis):
|
|
code = otp.issue_code("alice@example.com", "register")
|
|
for _ in range(otp.MAX_ATTEMPTS):
|
|
with pytest.raises(otp.OtpInvalid):
|
|
otp.verify_code("alice@example.com", "register", "000000")
|
|
# After max attempts the next call should raise OtpAttemptsExceeded and clear the code.
|
|
with pytest.raises(otp.OtpAttemptsExceeded):
|
|
otp.verify_code("alice@example.com", "register", code)
|
|
with pytest.raises(otp.OtpExpired):
|
|
otp.verify_code("alice@example.com", "register", code)
|
|
|
|
|
|
def test_issue_code_enforces_resend_cooldown(fake_redis):
|
|
otp.issue_code("alice@example.com", "register")
|
|
with pytest.raises(otp.OtpResendRateLimited) as excinfo:
|
|
otp.issue_code("alice@example.com", "register")
|
|
assert excinfo.value.retry_after_seconds > 0
|
|
|
|
|
|
def test_issue_code_emails_are_case_insensitive(fake_redis):
|
|
code = otp.issue_code("Alice@Example.com", "register")
|
|
otp.verify_code("alice@example.com", "register", code)
|
|
|
|
|
|
def test_purposes_are_isolated(fake_redis):
|
|
register_code = otp.issue_code("alice@example.com", "register")
|
|
reset_code = otp.issue_code("alice@example.com", "reset_password")
|
|
assert register_code != reset_code
|
|
otp.verify_code("alice@example.com", "register", register_code)
|
|
# Reset code should still be valid after consuming the register code.
|
|
otp.verify_code("alice@example.com", "reset_password", reset_code)
|