feat: implement password reset for shop customers
Add complete password reset functionality: Database: - Add password_reset_tokens migration with token hash, expiry, used_at - Create PasswordResetToken model with secure token hashing (SHA256) - One active token per customer (old tokens invalidated on new request) - 1-hour token expiry for security API: - Implement forgot_password endpoint with email lookup - Implement reset_password endpoint with token validation - No email enumeration (same response for all requests) - Password minimum 8 characters validation Frontend: - Add reset-password.html template with Alpine.js - Support for invalid/expired token states - Success state with login redirect - Dark mode support Email: - Add password_reset email templates (en, fr, de, lb) - Uses existing EmailService with template rendering Testing: - Add comprehensive pytest tests (19 tests) - Test token creation, validation, expiry, reuse prevention - Test endpoint success and error cases Removes critical launch blocker for password reset functionality. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
451
tests/integration/api/v1/shop/test_password_reset.py
Normal file
451
tests/integration/api/v1/shop/test_password_reset.py
Normal file
@@ -0,0 +1,451 @@
|
||||
# tests/integration/api/v1/shop/test_password_reset.py
|
||||
"""Integration tests for shop password reset API endpoints.
|
||||
|
||||
Tests the /api/v1/shop/auth/forgot-password and /api/v1/shop/auth/reset-password endpoints.
|
||||
"""
|
||||
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from models.database.customer import Customer
|
||||
from models.database.password_reset_token import PasswordResetToken
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def shop_customer(db, test_vendor):
|
||||
"""Create a test customer for shop API tests."""
|
||||
from middleware.auth import AuthManager
|
||||
|
||||
auth_manager = AuthManager()
|
||||
customer = Customer(
|
||||
vendor_id=test_vendor.id,
|
||||
email="customer@example.com",
|
||||
hashed_password=auth_manager.hash_password("oldpassword123"),
|
||||
first_name="Test",
|
||||
last_name="Customer",
|
||||
customer_number="CUST001",
|
||||
is_active=True,
|
||||
)
|
||||
db.add(customer)
|
||||
db.commit()
|
||||
db.refresh(customer)
|
||||
return customer
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def inactive_customer(db, test_vendor):
|
||||
"""Create an inactive customer for testing."""
|
||||
from middleware.auth import AuthManager
|
||||
|
||||
auth_manager = AuthManager()
|
||||
customer = Customer(
|
||||
vendor_id=test_vendor.id,
|
||||
email="inactive@example.com",
|
||||
hashed_password=auth_manager.hash_password("password123"),
|
||||
first_name="Inactive",
|
||||
last_name="Customer",
|
||||
customer_number="CUST002",
|
||||
is_active=False,
|
||||
)
|
||||
db.add(customer)
|
||||
db.commit()
|
||||
db.refresh(customer)
|
||||
return customer
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def valid_reset_token(db, shop_customer):
|
||||
"""Create a valid password reset token."""
|
||||
token = PasswordResetToken.create_for_customer(db, shop_customer.id)
|
||||
db.commit()
|
||||
return token
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def expired_reset_token(db, shop_customer):
|
||||
"""Create an expired password reset token."""
|
||||
import secrets
|
||||
|
||||
token = secrets.token_urlsafe(32)
|
||||
token_hash = PasswordResetToken.hash_token(token)
|
||||
|
||||
reset_token = PasswordResetToken(
|
||||
customer_id=shop_customer.id,
|
||||
token_hash=token_hash,
|
||||
expires_at=datetime.utcnow() - timedelta(hours=2), # Already expired
|
||||
)
|
||||
db.add(reset_token)
|
||||
db.commit()
|
||||
return token
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def used_reset_token(db, shop_customer):
|
||||
"""Create a used password reset token."""
|
||||
import secrets
|
||||
|
||||
token = secrets.token_urlsafe(32)
|
||||
token_hash = PasswordResetToken.hash_token(token)
|
||||
|
||||
reset_token = PasswordResetToken(
|
||||
customer_id=shop_customer.id,
|
||||
token_hash=token_hash,
|
||||
expires_at=datetime.utcnow() + timedelta(hours=1),
|
||||
used_at=datetime.utcnow(), # Already used
|
||||
)
|
||||
db.add(reset_token)
|
||||
db.commit()
|
||||
return token
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.shop
|
||||
class TestForgotPasswordAPI:
|
||||
"""Test forgot password endpoint at /api/v1/shop/auth/forgot-password."""
|
||||
|
||||
def test_forgot_password_existing_customer(
|
||||
self, client, db, test_vendor, shop_customer
|
||||
):
|
||||
"""Test password reset request for existing customer."""
|
||||
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
|
||||
mock_getattr.return_value = test_vendor
|
||||
|
||||
# Mock email service to avoid actual email sending
|
||||
with patch("app.api.v1.shop.auth.EmailService") as mock_email_service:
|
||||
mock_instance = MagicMock()
|
||||
mock_email_service.return_value = mock_instance
|
||||
|
||||
response = client.post(
|
||||
"/api/v1/shop/auth/forgot-password",
|
||||
params={"email": shop_customer.email},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "password reset link has been sent" in data["message"].lower()
|
||||
|
||||
# Verify email was sent
|
||||
mock_instance.send_template.assert_called_once()
|
||||
call_kwargs = mock_instance.send_template.call_args.kwargs
|
||||
assert call_kwargs["template_code"] == "password_reset"
|
||||
assert call_kwargs["to_email"] == shop_customer.email
|
||||
|
||||
def test_forgot_password_nonexistent_email(self, client, db, test_vendor):
|
||||
"""Test password reset request for non-existent email (same response)."""
|
||||
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
|
||||
mock_getattr.return_value = test_vendor
|
||||
|
||||
response = client.post(
|
||||
"/api/v1/shop/auth/forgot-password",
|
||||
params={"email": "nonexistent@example.com"},
|
||||
)
|
||||
|
||||
# Should return same success message to prevent email enumeration
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "password reset link has been sent" in data["message"].lower()
|
||||
|
||||
def test_forgot_password_inactive_customer(
|
||||
self, client, db, test_vendor, inactive_customer
|
||||
):
|
||||
"""Test password reset request for inactive customer."""
|
||||
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
|
||||
mock_getattr.return_value = test_vendor
|
||||
|
||||
response = client.post(
|
||||
"/api/v1/shop/auth/forgot-password",
|
||||
params={"email": inactive_customer.email},
|
||||
)
|
||||
|
||||
# Should return same success message (inactive customers can't reset)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "password reset link has been sent" in data["message"].lower()
|
||||
|
||||
def test_forgot_password_creates_token(
|
||||
self, client, db, test_vendor, shop_customer
|
||||
):
|
||||
"""Test that forgot password creates a token in the database."""
|
||||
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
|
||||
mock_getattr.return_value = test_vendor
|
||||
|
||||
with patch("app.api.v1.shop.auth.EmailService"):
|
||||
response = client.post(
|
||||
"/api/v1/shop/auth/forgot-password",
|
||||
params={"email": shop_customer.email},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
# Verify token was created
|
||||
token = (
|
||||
db.query(PasswordResetToken)
|
||||
.filter(PasswordResetToken.customer_id == shop_customer.id)
|
||||
.first()
|
||||
)
|
||||
assert token is not None
|
||||
assert token.used_at is None
|
||||
assert token.expires_at > datetime.utcnow()
|
||||
|
||||
def test_forgot_password_invalidates_old_tokens(
|
||||
self, client, db, test_vendor, shop_customer, valid_reset_token
|
||||
):
|
||||
"""Test that requesting new token invalidates old ones."""
|
||||
# Get the old token record
|
||||
old_token_count = (
|
||||
db.query(PasswordResetToken)
|
||||
.filter(
|
||||
PasswordResetToken.customer_id == shop_customer.id,
|
||||
PasswordResetToken.used_at.is_(None),
|
||||
)
|
||||
.count()
|
||||
)
|
||||
assert old_token_count == 1
|
||||
|
||||
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
|
||||
mock_getattr.return_value = test_vendor
|
||||
|
||||
with patch("app.api.v1.shop.auth.EmailService"):
|
||||
response = client.post(
|
||||
"/api/v1/shop/auth/forgot-password",
|
||||
params={"email": shop_customer.email},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
# Old token should be deleted, new one created
|
||||
new_token_count = (
|
||||
db.query(PasswordResetToken)
|
||||
.filter(
|
||||
PasswordResetToken.customer_id == shop_customer.id,
|
||||
PasswordResetToken.used_at.is_(None),
|
||||
)
|
||||
.count()
|
||||
)
|
||||
assert new_token_count == 1
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.shop
|
||||
class TestResetPasswordAPI:
|
||||
"""Test reset password endpoint at /api/v1/shop/auth/reset-password."""
|
||||
|
||||
def test_reset_password_success(
|
||||
self, client, db, test_vendor, shop_customer, valid_reset_token
|
||||
):
|
||||
"""Test successful password reset."""
|
||||
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
|
||||
mock_getattr.return_value = test_vendor
|
||||
|
||||
new_password = "newpassword123"
|
||||
response = client.post(
|
||||
"/api/v1/shop/auth/reset-password",
|
||||
params={
|
||||
"reset_token": valid_reset_token,
|
||||
"new_password": new_password,
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "password reset successfully" in data["message"].lower()
|
||||
|
||||
# Verify password was changed
|
||||
db.refresh(shop_customer)
|
||||
from middleware.auth import AuthManager
|
||||
|
||||
auth_manager = AuthManager()
|
||||
assert auth_manager.verify_password(
|
||||
new_password, shop_customer.hashed_password
|
||||
)
|
||||
|
||||
def test_reset_password_token_marked_used(
|
||||
self, client, db, test_vendor, shop_customer, valid_reset_token
|
||||
):
|
||||
"""Test that token is marked as used after successful reset."""
|
||||
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
|
||||
mock_getattr.return_value = test_vendor
|
||||
|
||||
response = client.post(
|
||||
"/api/v1/shop/auth/reset-password",
|
||||
params={
|
||||
"reset_token": valid_reset_token,
|
||||
"new_password": "newpassword123",
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
# Verify token is marked as used
|
||||
token_record = (
|
||||
db.query(PasswordResetToken)
|
||||
.filter(PasswordResetToken.customer_id == shop_customer.id)
|
||||
.first()
|
||||
)
|
||||
assert token_record.used_at is not None
|
||||
|
||||
def test_reset_password_invalid_token(self, client, db, test_vendor):
|
||||
"""Test reset with invalid token."""
|
||||
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
|
||||
mock_getattr.return_value = test_vendor
|
||||
|
||||
response = client.post(
|
||||
"/api/v1/shop/auth/reset-password",
|
||||
params={
|
||||
"reset_token": "invalid_token_12345",
|
||||
"new_password": "newpassword123",
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_reset_password_expired_token(
|
||||
self, client, db, test_vendor, shop_customer, expired_reset_token
|
||||
):
|
||||
"""Test reset with expired token."""
|
||||
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
|
||||
mock_getattr.return_value = test_vendor
|
||||
|
||||
response = client.post(
|
||||
"/api/v1/shop/auth/reset-password",
|
||||
params={
|
||||
"reset_token": expired_reset_token,
|
||||
"new_password": "newpassword123",
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_reset_password_used_token(
|
||||
self, client, db, test_vendor, shop_customer, used_reset_token
|
||||
):
|
||||
"""Test reset with already used token."""
|
||||
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
|
||||
mock_getattr.return_value = test_vendor
|
||||
|
||||
response = client.post(
|
||||
"/api/v1/shop/auth/reset-password",
|
||||
params={
|
||||
"reset_token": used_reset_token,
|
||||
"new_password": "newpassword123",
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_reset_password_short_password(
|
||||
self, client, db, test_vendor, shop_customer, valid_reset_token
|
||||
):
|
||||
"""Test reset with password that's too short."""
|
||||
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
|
||||
mock_getattr.return_value = test_vendor
|
||||
|
||||
response = client.post(
|
||||
"/api/v1/shop/auth/reset-password",
|
||||
params={
|
||||
"reset_token": valid_reset_token,
|
||||
"new_password": "short", # Less than 8 chars
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_reset_password_cannot_reuse_token(
|
||||
self, client, db, test_vendor, shop_customer, valid_reset_token
|
||||
):
|
||||
"""Test that token cannot be reused after successful reset."""
|
||||
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
|
||||
mock_getattr.return_value = test_vendor
|
||||
|
||||
# First reset should succeed
|
||||
response1 = client.post(
|
||||
"/api/v1/shop/auth/reset-password",
|
||||
params={
|
||||
"reset_token": valid_reset_token,
|
||||
"new_password": "newpassword123",
|
||||
},
|
||||
)
|
||||
assert response1.status_code == 200
|
||||
|
||||
# Second reset with same token should fail
|
||||
response2 = client.post(
|
||||
"/api/v1/shop/auth/reset-password",
|
||||
params={
|
||||
"reset_token": valid_reset_token,
|
||||
"new_password": "anotherpassword123",
|
||||
},
|
||||
)
|
||||
assert response2.status_code == 400
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.shop
|
||||
class TestPasswordResetTokenModel:
|
||||
"""Test PasswordResetToken model functionality."""
|
||||
|
||||
def test_token_hash_is_deterministic(self):
|
||||
"""Test that hashing the same token produces the same hash."""
|
||||
token = "test_token_12345"
|
||||
hash1 = PasswordResetToken.hash_token(token)
|
||||
hash2 = PasswordResetToken.hash_token(token)
|
||||
assert hash1 == hash2
|
||||
|
||||
def test_different_tokens_produce_different_hashes(self):
|
||||
"""Test that different tokens produce different hashes."""
|
||||
hash1 = PasswordResetToken.hash_token("token1")
|
||||
hash2 = PasswordResetToken.hash_token("token2")
|
||||
assert hash1 != hash2
|
||||
|
||||
def test_create_for_customer_returns_plaintext(self, db, shop_customer):
|
||||
"""Test that create_for_customer returns plaintext token."""
|
||||
token = PasswordResetToken.create_for_customer(db, shop_customer.id)
|
||||
db.commit()
|
||||
|
||||
# Token should be URL-safe base64
|
||||
assert token is not None
|
||||
assert len(token) > 20 # secrets.token_urlsafe(32) produces ~43 chars
|
||||
|
||||
def test_find_valid_token_works_with_plaintext(self, db, shop_customer):
|
||||
"""Test that find_valid_token works with plaintext token."""
|
||||
plaintext_token = PasswordResetToken.create_for_customer(db, shop_customer.id)
|
||||
db.commit()
|
||||
|
||||
found = PasswordResetToken.find_valid_token(db, plaintext_token)
|
||||
assert found is not None
|
||||
assert found.customer_id == shop_customer.id
|
||||
|
||||
def test_find_valid_token_returns_none_for_invalid(self, db):
|
||||
"""Test that find_valid_token returns None for invalid token."""
|
||||
found = PasswordResetToken.find_valid_token(db, "invalid_token")
|
||||
assert found is None
|
||||
|
||||
def test_mark_used_sets_timestamp(self, db, shop_customer):
|
||||
"""Test that mark_used sets the used_at timestamp."""
|
||||
plaintext_token = PasswordResetToken.create_for_customer(db, shop_customer.id)
|
||||
db.commit()
|
||||
|
||||
token_record = PasswordResetToken.find_valid_token(db, plaintext_token)
|
||||
assert token_record.used_at is None
|
||||
|
||||
token_record.mark_used(db)
|
||||
db.commit()
|
||||
|
||||
assert token_record.used_at is not None
|
||||
|
||||
def test_used_token_not_found_by_find_valid(self, db, shop_customer):
|
||||
"""Test that used tokens are not returned by find_valid_token."""
|
||||
plaintext_token = PasswordResetToken.create_for_customer(db, shop_customer.id)
|
||||
db.commit()
|
||||
|
||||
token_record = PasswordResetToken.find_valid_token(db, plaintext_token)
|
||||
token_record.mark_used(db)
|
||||
db.commit()
|
||||
|
||||
# Should not find the used token
|
||||
found = PasswordResetToken.find_valid_token(db, plaintext_token)
|
||||
assert found is None
|
||||
Reference in New Issue
Block a user