From 828cbcc822615774e445ffab84dbcc42f4e85c6c Mon Sep 17 00:00:00 2001 From: mohammad Date: Sat, 28 Feb 2026 17:31:03 +0300 Subject: [PATCH] Authentica OTP tests --- .gitignore | 1 + README.md | 1 + backend/README.md | 18 +++++ backend/apps/accounts/services/otp.py | 13 +++- .../tests/test_authentica_e2e_mock.py | 72 +++++++++++++++++++ .../tests/test_authentica_e2e_real.py | 72 +++++++++++++++++++ .../apps/accounts/tests/test_otp_limits.py | 42 ++++++++++- .../apps/accounts/tests/test_phone_auth.py | 56 ++++++++++----- backend/pytest.ini | 4 +- backend/salon_api/settings.py | 14 ++-- 10 files changed, 265 insertions(+), 28 deletions(-) create mode 100644 backend/apps/accounts/tests/test_authentica_e2e_mock.py create mode 100644 backend/apps/accounts/tests/test_authentica_e2e_real.py diff --git a/.gitignore b/.gitignore index 3229932..959762a 100644 --- a/.gitignore +++ b/.gitignore @@ -17,3 +17,4 @@ dist/ # OS .DS_Store +backend/tmp_authentica_request_id.txt diff --git a/README.md b/README.md index 2f10055..e2cc029 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,7 @@ After migrations, you can seed demo data: ### Tests - From project root with venv active: `venv/bin/python3 -m pytest` (run from `backend/` so `pytest.ini` is picked up) +- External provider tests are skipped by default; run explicitly when needed: `PYTEST_ADDOPTS='' venv/bin/python3 -m pytest -m external` ### Core API endpoints (current scaffold) diff --git a/backend/README.md b/backend/README.md index b6bcc6d..a8b7455 100644 --- a/backend/README.md +++ b/backend/README.md @@ -8,6 +8,24 @@ ## Near-Term Focus - Hardening Authentica integration (timeouts, retries, async delivery) and aligning notification provider choices. + +**Authentica E2E** +Run the real Authentica OTP flow only when explicitly enabled. + +Env vars (in `backend/.env` or shell): +- `AUTHENTICA_E2E=1` +- `AUTHENTICA_API_KEY=...` +- `AUTHENTICA_E2E_PHONE=...` (must receive OTP) +- `AUTHENTICA_E2E_CODE=...` (required; no interactive prompt) + +Command: +```bash +cd backend +PYTEST_ADDOPTS='' python3 -m pytest apps/accounts/tests -m external +``` + +Suggested flow: +1. Trigger the E2E test to send the OTP, then set `AUTHENTICA_E2E_CODE` and re-run if needed. - Decide and document payment lifecycle scope (capture/refund supported vs explicitly out of scope). - Add timeouts/logging for external calls or introduce minimal async jobs for OTP/notifications. - Keep booking, payment, and notification orchestration in service layers, not views. diff --git a/backend/apps/accounts/services/otp.py b/backend/apps/accounts/services/otp.py index 7fa77ee..75fe1db 100644 --- a/backend/apps/accounts/services/otp.py +++ b/backend/apps/accounts/services/otp.py @@ -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: diff --git a/backend/apps/accounts/tests/test_authentica_e2e_mock.py b/backend/apps/accounts/tests/test_authentica_e2e_mock.py new file mode 100644 index 0000000..8fe1ba3 --- /dev/null +++ b/backend/apps/accounts/tests/test_authentica_e2e_mock.py @@ -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 diff --git a/backend/apps/accounts/tests/test_authentica_e2e_real.py b/backend/apps/accounts/tests/test_authentica_e2e_real.py new file mode 100644 index 0000000..e6b908b --- /dev/null +++ b/backend/apps/accounts/tests/test_authentica_e2e_real.py @@ -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 diff --git a/backend/apps/accounts/tests/test_otp_limits.py b/backend/apps/accounts/tests/test_otp_limits.py index 4660dc8..1158dca 100644 --- a/backend/apps/accounts/tests/test_otp_limits.py +++ b/backend/apps/accounts/tests/test_otp_limits.py @@ -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 diff --git a/backend/apps/accounts/tests/test_phone_auth.py b/backend/apps/accounts/tests/test_phone_auth.py index d2a4ec7..ee7d187 100644 --- a/backend/apps/accounts/tests/test_phone_auth.py +++ b/backend/apps/accounts/tests/test_phone_auth.py @@ -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 diff --git a/backend/pytest.ini b/backend/pytest.ini index 6db2981..2be90e3 100644 --- a/backend/pytest.ini +++ b/backend/pytest.ini @@ -1,4 +1,6 @@ [pytest] DJANGO_SETTINGS_MODULE = salon_api.settings python_files = tests.py test_*.py *_tests.py -addopts = -q +addopts = -q -m "not external" +markers = + external: hits real third-party services (requires explicit env to run) diff --git a/backend/salon_api/settings.py b/backend/salon_api/settings.py index 6ca6513..0a484a1 100644 --- a/backend/salon_api/settings.py +++ b/backend/salon_api/settings.py @@ -1,4 +1,5 @@ import os +import sys from pathlib import Path from datetime import timedelta from urllib.parse import urlparse @@ -78,11 +79,14 @@ def parse_database_url(database_url: str): } -DATABASE_URL = os.getenv("DATABASE_URL") -if DATABASE_URL: - parsed_db = parse_database_url(DATABASE_URL) +running_tests = "PYTEST_CURRENT_TEST" in os.environ or any("pytest" in arg for arg in sys.argv) +test_database_url = os.getenv("TEST_DATABASE_URL") +database_url = os.getenv("DATABASE_URL") + +if running_tests: + parsed_db = parse_database_url(test_database_url) if test_database_url else None else: - parsed_db = None + parsed_db = parse_database_url(database_url) if database_url else None DATABASES = { "default": parsed_db @@ -136,6 +140,8 @@ CORS_ALLOWED_ORIGINS = [ ] OTP_PROVIDER = os.getenv("OTP_PROVIDER", "console") +if running_tests: + OTP_PROVIDER = os.getenv("TEST_OTP_PROVIDER", "console") OTP_EXPIRY_MINUTES = int(os.getenv("OTP_EXPIRY_MINUTES", "5")) OTP_MAX_PER_WINDOW = int(os.getenv("OTP_MAX_PER_WINDOW", "5")) OTP_WINDOW_MINUTES = int(os.getenv("OTP_WINDOW_MINUTES", "15"))