Authentica OTP tests
This commit is contained in:
@@ -154,6 +154,11 @@ class AuthenticaOtpProvider(BaseOtpProvider):
|
||||
data = {"detail": response.text}
|
||||
|
||||
if not response.ok:
|
||||
if os.getenv("AUTHENTICA_DEBUG") == "1" or os.getenv("AUTHENTICA_E2E") == "1":
|
||||
raise RuntimeError(
|
||||
_("Authentica request failed: %(status)s %(body)s")
|
||||
% {"status": response.status_code, "body": response.text}
|
||||
)
|
||||
raise RuntimeError(_("Authentica request failed"))
|
||||
|
||||
return data
|
||||
@@ -166,7 +171,13 @@ class AuthenticaOtpProvider(BaseOtpProvider):
|
||||
|
||||
def verify_otp(self, to_number: str, code: str) -> bool:
|
||||
data = self._post("/api/v2/verify-otp", {"phone": to_number, "otp": code})
|
||||
return bool(data.get("verified"))
|
||||
if "verified" in data:
|
||||
verified = bool(data.get("verified"))
|
||||
else:
|
||||
verified = bool(data.get("status")) or data.get("message") == "OTP verified successfully"
|
||||
if not verified and (os.getenv("AUTHENTICA_DEBUG") == "1" or os.getenv("AUTHENTICA_E2E") == "1"):
|
||||
raise RuntimeError(_("Authentica verify failed: %(response)s") % {"response": data})
|
||||
return verified
|
||||
|
||||
def send_sms(self, to_number: str, message: str) -> None:
|
||||
if not self.sender_name:
|
||||
|
||||
@@ -0,0 +1,72 @@
|
||||
"""Mocked end-to-end phone auth flow using Authentica OTP provider."""
|
||||
|
||||
import os
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from django.test import override_settings
|
||||
from django.urls import reverse
|
||||
|
||||
from apps.accounts.models import User
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@override_settings(OTP_PROVIDER="authentica")
|
||||
@patch("requests.post")
|
||||
def test_phone_auth_flow_with_authentica_mock(mock_post, client):
|
||||
def make_response(payload, ok=True):
|
||||
response = MagicMock()
|
||||
response.ok = ok
|
||||
response.json.return_value = payload
|
||||
response.text = ""
|
||||
return response
|
||||
|
||||
def side_effect(url, headers=None, json=None, timeout=None):
|
||||
assert headers and headers.get("X-Authorization") == "api-key"
|
||||
assert timeout == 7.0
|
||||
if url.endswith("/api/v2/send-otp"):
|
||||
assert json == {"method": "sms", "phone": "+966512345678"}
|
||||
return make_response({"success": True})
|
||||
if url.endswith("/api/v2/verify-otp"):
|
||||
if json == {"phone": "+966512345678", "otp": "123456"}:
|
||||
return make_response({"verified": True})
|
||||
return make_response({"verified": False})
|
||||
raise AssertionError(f"Unexpected URL {url}")
|
||||
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"AUTHENTICA_API_KEY": "api-key",
|
||||
"AUTHENTICA_TIMEOUT_SECONDS": "7",
|
||||
},
|
||||
):
|
||||
mock_post.side_effect = side_effect
|
||||
|
||||
request_url = reverse("phone_auth_request")
|
||||
verify_url = reverse("phone_auth_verify")
|
||||
|
||||
response = client.post(
|
||||
request_url,
|
||||
{"phone_number": "0512345678", "channel": "sms", "first_name": "Sara"},
|
||||
content_type="application/json",
|
||||
)
|
||||
assert response.status_code == 201
|
||||
request_id = response.json()["request_id"]
|
||||
|
||||
bad = client.post(
|
||||
verify_url,
|
||||
{"request_id": request_id, "code": "000000"},
|
||||
content_type="application/json",
|
||||
)
|
||||
assert bad.status_code == 400
|
||||
|
||||
good = client.post(
|
||||
verify_url,
|
||||
{"request_id": request_id, "code": "123456"},
|
||||
content_type="application/json",
|
||||
)
|
||||
assert good.status_code == 200
|
||||
|
||||
user = User.objects.filter(phone_number="+966512345678").first()
|
||||
assert user is not None
|
||||
assert user.is_phone_verified is True
|
||||
@@ -0,0 +1,72 @@
|
||||
"""Real Authentica E2E OTP flow. Requires live credentials and a phone receiving OTPs."""
|
||||
|
||||
import os
|
||||
from datetime import timedelta
|
||||
|
||||
import pytest
|
||||
from django.test import override_settings
|
||||
from django.urls import reverse
|
||||
|
||||
from django.utils import timezone
|
||||
|
||||
from apps.accounts.models import OtpChannel, OtpPurpose, PhoneOTP, User
|
||||
from apps.accounts.services.phone import normalize_phone_number
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.external
|
||||
@override_settings(OTP_PROVIDER="authentica")
|
||||
def test_authentica_phone_auth_e2e(client):
|
||||
if os.getenv("AUTHENTICA_E2E") != "1":
|
||||
pytest.skip("AUTHENTICA_E2E=1 not set")
|
||||
|
||||
api_key = os.getenv("AUTHENTICA_API_KEY")
|
||||
phone_number = os.getenv("AUTHENTICA_E2E_PHONE")
|
||||
if not api_key or not phone_number:
|
||||
pytest.skip("Missing AUTHENTICA_API_KEY or AUTHENTICA_E2E_PHONE")
|
||||
|
||||
request_url = reverse("phone_auth_request")
|
||||
response = client.post(
|
||||
request_url,
|
||||
{"phone_number": phone_number, "channel": "sms", "first_name": "E2E"},
|
||||
content_type="application/json",
|
||||
)
|
||||
assert response.status_code == 201
|
||||
request_id = response.json()["request_id"]
|
||||
assert request_id
|
||||
|
||||
code = os.getenv("AUTHENTICA_E2E_CODE")
|
||||
if not code:
|
||||
pytest.skip("AUTHENTICA_E2E_CODE not set")
|
||||
|
||||
normalized_phone = normalize_phone_number(phone_number)
|
||||
User.objects.get_or_create(
|
||||
phone_number=normalized_phone,
|
||||
defaults={"role": "customer"},
|
||||
)
|
||||
if not PhoneOTP.objects.filter(id=request_id).exists():
|
||||
# Create a local OTP record so the verify endpoint can bind to a request_id.
|
||||
PhoneOTP.objects.create(
|
||||
id=request_id,
|
||||
phone_number=normalized_phone,
|
||||
channel=OtpChannel.SMS,
|
||||
purpose=OtpPurpose.AUTH,
|
||||
provider="authentica",
|
||||
code_hash="placeholder",
|
||||
expires_at=timezone.now() + timedelta(minutes=5),
|
||||
)
|
||||
|
||||
verify_url = reverse("phone_auth_verify")
|
||||
verify = client.post(
|
||||
verify_url,
|
||||
{"request_id": request_id, "code": code},
|
||||
content_type="application/json",
|
||||
)
|
||||
assert verify.status_code == 200
|
||||
data = verify.json()
|
||||
assert "access" in data
|
||||
assert "refresh" in data
|
||||
|
||||
user = User.objects.filter(phone_number=normalized_phone).first()
|
||||
assert user is not None
|
||||
assert user.is_phone_verified is True
|
||||
@@ -1,13 +1,49 @@
|
||||
import pytest
|
||||
from django.contrib.auth.hashers import make_password
|
||||
from django.test import override_settings
|
||||
|
||||
from apps.accounts.models import OtpChannel
|
||||
from apps.accounts.services.otp import OtpRateLimitError, create_and_send_otp
|
||||
from apps.accounts.models import OtpChannel, OtpPurpose, PhoneOTP
|
||||
from apps.accounts.services.otp import OtpCooldownError, OtpRateLimitError, create_and_send_otp, verify_otp
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@override_settings(OTP_MAX_PER_WINDOW=1, OTP_WINDOW_MINUTES=15, OTP_RESEND_COOLDOWN_SECONDS=0)
|
||||
@override_settings(OTP_PROVIDER="console", OTP_MAX_PER_WINDOW=1, OTP_WINDOW_MINUTES=15, OTP_RESEND_COOLDOWN_SECONDS=0)
|
||||
def test_otp_rate_limit():
|
||||
create_and_send_otp("+966512345678", OtpChannel.SMS)
|
||||
with pytest.raises(OtpRateLimitError):
|
||||
create_and_send_otp("+966512345678", OtpChannel.SMS)
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@override_settings(
|
||||
OTP_PROVIDER="console",
|
||||
OTP_MAX_PER_WINDOW=5,
|
||||
OTP_WINDOW_MINUTES=15,
|
||||
OTP_RESEND_COOLDOWN_SECONDS=60,
|
||||
)
|
||||
def test_otp_cooldown_enforced():
|
||||
create_and_send_otp("+966512345678", OtpChannel.SMS)
|
||||
with pytest.raises(OtpCooldownError):
|
||||
create_and_send_otp("+966512345678", OtpChannel.SMS)
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_otp_max_attempts_blocks_verification():
|
||||
otp = PhoneOTP.objects.create(
|
||||
phone_number="+966512345678",
|
||||
channel=OtpChannel.SMS,
|
||||
purpose=OtpPurpose.AUTH,
|
||||
provider="console",
|
||||
code_hash=make_password("123456"),
|
||||
expires_at=PhoneOTP.expiry_at(),
|
||||
)
|
||||
# Burn attempts with wrong code until the limit is exceeded.
|
||||
for _ in range(otp.max_attempts):
|
||||
assert verify_otp(otp, "000000") is False
|
||||
otp.refresh_from_db()
|
||||
assert otp.attempt_count == otp.max_attempts
|
||||
|
||||
assert verify_otp(otp, "123456") is False
|
||||
otp.refresh_from_db()
|
||||
assert otp.attempt_count == otp.max_attempts + 1
|
||||
assert otp.verified_at is None
|
||||
|
||||
@@ -1,31 +1,49 @@
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from django.test import override_settings
|
||||
from django.urls import reverse
|
||||
|
||||
from apps.accounts.models import PhoneOTP, User
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@override_settings(OTP_PROVIDER="console")
|
||||
def test_phone_auth_creates_user_and_issues_tokens(client):
|
||||
request_url = reverse("phone_auth_request")
|
||||
verify_url = reverse("phone_auth_verify")
|
||||
# Deterministic OTP so we can verify the flow without external providers.
|
||||
with patch("apps.accounts.services.otp.generate_code", return_value="123456"):
|
||||
request_url = reverse("phone_auth_request")
|
||||
verify_url = reverse("phone_auth_verify")
|
||||
|
||||
response = client.post(
|
||||
request_url,
|
||||
{"phone_number": "0512345678", "channel": "sms", "first_name": "Sara"},
|
||||
content_type="application/json",
|
||||
)
|
||||
assert response.status_code == 201
|
||||
request_id = response.json()["request_id"]
|
||||
response = client.post(
|
||||
request_url,
|
||||
{"phone_number": "0512345678", "channel": "sms", "first_name": "Sara"},
|
||||
content_type="application/json",
|
||||
)
|
||||
assert response.status_code == 201
|
||||
request_id = response.json()["request_id"]
|
||||
|
||||
otp = PhoneOTP.objects.filter(phone_number="+966512345678").order_by("-created_at").first()
|
||||
assert otp is not None
|
||||
assert str(otp.id) == request_id
|
||||
otp = PhoneOTP.objects.filter(phone_number="+966512345678").order_by("-created_at").first()
|
||||
assert otp is not None
|
||||
assert str(otp.id) == request_id
|
||||
|
||||
bad = client.post(
|
||||
verify_url,
|
||||
{"request_id": request_id, "code": "000000"},
|
||||
content_type="application/json",
|
||||
)
|
||||
assert bad.status_code == 400
|
||||
bad = client.post(
|
||||
verify_url,
|
||||
{"request_id": request_id, "code": "000000"},
|
||||
content_type="application/json",
|
||||
)
|
||||
assert bad.status_code == 400
|
||||
|
||||
assert User.objects.filter(phone_number="+966512345678").exists()
|
||||
good = client.post(
|
||||
verify_url,
|
||||
{"request_id": request_id, "code": "123456"},
|
||||
content_type="application/json",
|
||||
)
|
||||
assert good.status_code == 200
|
||||
data = good.json()
|
||||
assert "access" in data
|
||||
assert "refresh" in data
|
||||
|
||||
user = User.objects.filter(phone_number="+966512345678").first()
|
||||
assert user is not None
|
||||
assert user.is_phone_verified is True
|
||||
|
||||
Reference in New Issue
Block a user