Samir Boulahtit
4b56eb7ab1
Some checks failed
Phase 1 of the loyalty production launch plan: config & security hardening, dropped-data fix, DB integrity guards, rate limiting, and constant-time auth compare. 362 tests pass. - 1.4 Persist customer birth_date (new column + migration). Enrollment form was collecting it but the value was silently dropped because create_customer_for_enrollment never received it. Backfills existing customers without overwriting. - 1.1 LOYALTY_GOOGLE_SERVICE_ACCOUNT_JSON validated at startup (file must exist and be readable; ~ expanded). Adds is_google_wallet_enabled and is_apple_wallet_enabled derived flags. Prod path documented as ~/apps/orion/google-wallet-sa.json. - 1.5 CHECK constraints on loyalty_cards (points_balance, stamp_count non-negative) and loyalty_programs (min_purchase, points_per_euro, welcome_bonus non-negative; stamps_target >= 1). Mirrored as CheckConstraint in models. Pre-flight scan showed zero violations. - 1.3 @rate_limit on store mutating endpoints: stamp 60/min, redeem/points-earn 30-60/min, void/adjust 20/min, pin unlock 10/min. - 1.2 Constant-time hmac.compare_digest for Apple Wallet auth token (pulled forward from Phase 9 — code is safe whenever Apple ships). See app/modules/loyalty/docs/production-launch-plan.md for the full launch plan and remaining phases. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
408 lines
14 KiB
Python
408 lines
14 KiB
Python
"""Unit tests for CardService."""
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
from app.modules.customers.models.customer import Customer
|
|
from app.modules.loyalty.exceptions import (
|
|
CustomerIdentifierRequiredException,
|
|
CustomerNotFoundByEmailException,
|
|
)
|
|
from app.modules.loyalty.models import LoyaltyCard, LoyaltyTransaction
|
|
from app.modules.loyalty.models.loyalty_transaction import TransactionType
|
|
from app.modules.loyalty.services.card_service import CardService
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.loyalty
|
|
class TestCardService:
|
|
"""Test suite for CardService."""
|
|
|
|
def setup_method(self):
|
|
self.service = CardService()
|
|
|
|
def test_service_instantiation(self):
|
|
"""Service can be instantiated."""
|
|
assert self.service is not None
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.loyalty
|
|
class TestGetCardByNumber:
|
|
"""Tests for card number lookup with SQL normalization (func.replace)."""
|
|
|
|
def setup_method(self):
|
|
self.service = CardService()
|
|
|
|
def test_find_card_by_exact_number(self, db, test_loyalty_card):
|
|
"""Find card by its exact stored card number."""
|
|
result = self.service.get_card_by_number(db, test_loyalty_card.card_number)
|
|
assert result is not None
|
|
assert result.id == test_loyalty_card.id
|
|
|
|
def test_find_card_with_dashes_in_query(self, db, test_loyalty_program, test_customer, test_store):
|
|
"""Query with dashes matches card stored without dashes."""
|
|
card = LoyaltyCard(
|
|
merchant_id=test_loyalty_program.merchant_id,
|
|
program_id=test_loyalty_program.id,
|
|
customer_id=test_customer.id,
|
|
enrolled_at_store_id=test_store.id,
|
|
card_number="ABCD1234",
|
|
is_active=True,
|
|
)
|
|
db.add(card)
|
|
db.commit()
|
|
|
|
result = self.service.get_card_by_number(db, "ABCD-1234")
|
|
assert result is not None
|
|
assert result.id == card.id
|
|
|
|
def test_find_card_with_spaces_in_query(self, db, test_loyalty_program, test_customer, test_store):
|
|
"""Query with spaces matches card stored without spaces."""
|
|
card = LoyaltyCard(
|
|
merchant_id=test_loyalty_program.merchant_id,
|
|
program_id=test_loyalty_program.id,
|
|
customer_id=test_customer.id,
|
|
enrolled_at_store_id=test_store.id,
|
|
card_number="WXYZ5678",
|
|
is_active=True,
|
|
)
|
|
db.add(card)
|
|
db.commit()
|
|
|
|
result = self.service.get_card_by_number(db, "WXYZ 5678")
|
|
assert result is not None
|
|
assert result.id == card.id
|
|
|
|
def test_card_not_found_returns_none(self, db):
|
|
"""Non-existent card number returns None."""
|
|
result = self.service.get_card_by_number(db, "NONEXISTENT-999")
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.loyalty
|
|
class TestSearchCardForStore:
|
|
"""Tests for free-text card search scoped to store's merchant."""
|
|
|
|
def setup_method(self):
|
|
self.service = CardService()
|
|
|
|
def test_search_by_card_number(self, db, test_loyalty_card, test_store):
|
|
"""Find card by card number."""
|
|
result = self.service.search_card_for_store(
|
|
db, test_store.id, test_loyalty_card.card_number
|
|
)
|
|
assert result is not None
|
|
assert result.id == test_loyalty_card.id
|
|
|
|
def test_search_by_customer_email(self, db, test_loyalty_card, test_store, test_customer):
|
|
"""Find card by customer email."""
|
|
result = self.service.search_card_for_store(
|
|
db, test_store.id, test_customer.email
|
|
)
|
|
assert result is not None
|
|
assert result.id == test_loyalty_card.id
|
|
|
|
def test_search_returns_none_for_wrong_merchant(self, db, test_loyalty_card, test_store):
|
|
"""Card from a different merchant is not returned."""
|
|
from app.modules.tenancy.models import Merchant, Store, User
|
|
from middleware.auth import AuthManager
|
|
|
|
auth = AuthManager()
|
|
uid = str(uuid.uuid4())[:8]
|
|
|
|
other_owner = User(
|
|
email=f"other_{uid}@test.com",
|
|
username=f"other_{uid}",
|
|
hashed_password=auth.hash_password("pass123"),
|
|
role="merchant_owner",
|
|
is_active=True,
|
|
)
|
|
db.add(other_owner)
|
|
db.flush()
|
|
|
|
other_merchant = Merchant(
|
|
name="Other Merchant",
|
|
owner_user_id=other_owner.id,
|
|
contact_email="other@test.com",
|
|
is_active=True,
|
|
)
|
|
db.add(other_merchant)
|
|
db.flush()
|
|
|
|
other_store = Store(
|
|
store_code=f"OTHER{uid}",
|
|
subdomain=f"other{uid}",
|
|
name="Other Store",
|
|
merchant_id=other_merchant.id,
|
|
is_active=True,
|
|
)
|
|
db.add(other_store)
|
|
db.commit()
|
|
|
|
# Search from the other store — should not find the card
|
|
result = self.service.search_card_for_store(
|
|
db, other_store.id, test_loyalty_card.card_number
|
|
)
|
|
assert result is None
|
|
|
|
def test_search_returns_none_for_nonexistent(self, db, test_store):
|
|
"""No match returns None."""
|
|
result = self.service.search_card_for_store(db, test_store.id, "nobody@nowhere.com")
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.loyalty
|
|
class TestResolveCustomerId:
|
|
"""Tests for customer ID resolution including self-enrollment creation."""
|
|
|
|
def setup_method(self):
|
|
self.service = CardService()
|
|
|
|
def test_resolve_by_customer_id(self, db, test_customer, test_store):
|
|
"""Returns immediately when customer_id is provided."""
|
|
result = self.service.resolve_customer_id(
|
|
db, customer_id=test_customer.id, email=None, store_id=test_store.id
|
|
)
|
|
assert result == test_customer.id
|
|
|
|
def test_resolve_by_email(self, db, test_customer, test_store):
|
|
"""Finds existing customer by email."""
|
|
result = self.service.resolve_customer_id(
|
|
db, customer_id=None, email=test_customer.email, store_id=test_store.id
|
|
)
|
|
assert result == test_customer.id
|
|
|
|
def test_resolve_creates_customer_when_missing(self, db, test_store):
|
|
"""create_if_missing=True creates a new customer."""
|
|
email = f"newcustomer-{uuid.uuid4().hex[:8]}@test.com"
|
|
result = self.service.resolve_customer_id(
|
|
db,
|
|
customer_id=None,
|
|
email=email,
|
|
store_id=test_store.id,
|
|
create_if_missing=True,
|
|
customer_name="Jane Doe",
|
|
customer_phone="+352123456",
|
|
)
|
|
assert result is not None
|
|
|
|
customer = db.query(Customer).filter(Customer.id == result).first()
|
|
assert customer is not None
|
|
assert customer.email == email
|
|
assert customer.first_name == "Jane"
|
|
assert customer.last_name == "Doe"
|
|
assert customer.phone == "+352123456"
|
|
assert customer.hashed_password.startswith("!enrollment!")
|
|
assert customer.customer_number is not None
|
|
assert customer.is_active is True
|
|
|
|
def test_resolve_splits_name_into_first_last(self, db, test_store):
|
|
"""Full name with space is split into first_name and last_name."""
|
|
email = f"split-{uuid.uuid4().hex[:8]}@test.com"
|
|
result = self.service.resolve_customer_id(
|
|
db,
|
|
customer_id=None,
|
|
email=email,
|
|
store_id=test_store.id,
|
|
create_if_missing=True,
|
|
customer_name="John Michael Smith",
|
|
)
|
|
customer = db.query(Customer).filter(Customer.id == result).first()
|
|
assert customer.first_name == "John"
|
|
assert customer.last_name == "Michael Smith"
|
|
|
|
def test_resolve_raises_when_email_not_found(self, db, test_store):
|
|
"""create_if_missing=False raises when email not found."""
|
|
with pytest.raises(CustomerNotFoundByEmailException):
|
|
self.service.resolve_customer_id(
|
|
db,
|
|
customer_id=None,
|
|
email="nonexistent@test.com",
|
|
store_id=test_store.id,
|
|
create_if_missing=False,
|
|
)
|
|
|
|
def test_resolve_raises_when_no_identifier(self, db, test_store):
|
|
"""Raises when neither customer_id nor email provided."""
|
|
with pytest.raises(CustomerIdentifierRequiredException):
|
|
self.service.resolve_customer_id(
|
|
db, customer_id=None, email=None, store_id=test_store.id
|
|
)
|
|
|
|
def test_resolve_persists_birthday_on_create(self, db, test_store):
|
|
"""Birthday provided at self-enrollment is saved on the new customer."""
|
|
from datetime import date
|
|
|
|
email = f"birthday-create-{uuid.uuid4().hex[:8]}@test.com"
|
|
result = self.service.resolve_customer_id(
|
|
db,
|
|
customer_id=None,
|
|
email=email,
|
|
store_id=test_store.id,
|
|
create_if_missing=True,
|
|
customer_name="Anna Test",
|
|
customer_birthday=date(1992, 6, 15),
|
|
)
|
|
customer = db.query(Customer).filter(Customer.id == result).first()
|
|
assert customer.birth_date == date(1992, 6, 15)
|
|
|
|
def test_resolve_backfills_birthday_on_existing_customer(
|
|
self, db, test_store
|
|
):
|
|
"""An existing customer without a birthday is backfilled from form."""
|
|
from datetime import date
|
|
|
|
email = f"backfill-{uuid.uuid4().hex[:8]}@test.com"
|
|
# Pre-create a customer with no birth_date
|
|
existing = self.service.resolve_customer_id(
|
|
db,
|
|
customer_id=None,
|
|
email=email,
|
|
store_id=test_store.id,
|
|
create_if_missing=True,
|
|
customer_name="No Birthday",
|
|
)
|
|
customer = db.query(Customer).filter(Customer.id == existing).first()
|
|
assert customer.birth_date is None
|
|
|
|
# Re-enroll with birthday — should backfill
|
|
self.service.resolve_customer_id(
|
|
db,
|
|
customer_id=None,
|
|
email=email,
|
|
store_id=test_store.id,
|
|
create_if_missing=True,
|
|
customer_birthday=date(1985, 3, 22),
|
|
)
|
|
db.refresh(customer)
|
|
assert customer.birth_date == date(1985, 3, 22)
|
|
|
|
def test_resolve_does_not_overwrite_existing_birthday(self, db, test_store):
|
|
"""Birthday already on the customer is not overwritten by form input."""
|
|
from datetime import date
|
|
|
|
email = f"no-overwrite-{uuid.uuid4().hex[:8]}@test.com"
|
|
# Create with original birthday
|
|
cid = self.service.resolve_customer_id(
|
|
db,
|
|
customer_id=None,
|
|
email=email,
|
|
store_id=test_store.id,
|
|
create_if_missing=True,
|
|
customer_birthday=date(1990, 1, 1),
|
|
)
|
|
# Re-enroll with a different birthday
|
|
self.service.resolve_customer_id(
|
|
db,
|
|
customer_id=None,
|
|
email=email,
|
|
store_id=test_store.id,
|
|
create_if_missing=True,
|
|
customer_birthday=date(2000, 1, 1),
|
|
)
|
|
customer = db.query(Customer).filter(Customer.id == cid).first()
|
|
assert customer.birth_date == date(1990, 1, 1)
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.loyalty
|
|
class TestGetStoreTransactions:
|
|
"""Tests for merchant-scoped transaction retrieval with eager loading."""
|
|
|
|
def setup_method(self):
|
|
self.service = CardService()
|
|
|
|
def test_returns_transactions_for_merchant(
|
|
self, db, test_loyalty_card, test_loyalty_transaction
|
|
):
|
|
"""Returns transactions belonging to the merchant."""
|
|
transactions, total = self.service.get_store_transactions(
|
|
db, test_loyalty_card.merchant_id
|
|
)
|
|
assert total >= 1
|
|
assert len(transactions) >= 1
|
|
assert any(t.id == test_loyalty_transaction.id for t in transactions)
|
|
|
|
def test_eager_loads_customer(
|
|
self, db, test_loyalty_card, test_loyalty_transaction
|
|
):
|
|
"""Verifies card.customer is accessible without lazy-load error."""
|
|
transactions, _ = self.service.get_store_transactions(
|
|
db, test_loyalty_card.merchant_id
|
|
)
|
|
tx = transactions[0]
|
|
# This should not raise a lazy-loading error
|
|
assert tx.card is not None
|
|
assert tx.card.customer is not None
|
|
assert tx.card.customer.email is not None
|
|
|
|
def test_pagination(self, db, test_loyalty_card, test_store):
|
|
"""Skip and limit work correctly."""
|
|
from datetime import UTC, datetime
|
|
|
|
# Create 3 transactions
|
|
txns = [
|
|
LoyaltyTransaction(
|
|
merchant_id=test_loyalty_card.merchant_id,
|
|
card_id=test_loyalty_card.id,
|
|
store_id=test_store.id,
|
|
transaction_type=TransactionType.POINTS_EARNED.value,
|
|
points_delta=10 * (i + 1),
|
|
transaction_at=datetime.now(UTC),
|
|
)
|
|
for i in range(3)
|
|
]
|
|
db.add_all(txns)
|
|
db.commit()
|
|
|
|
transactions, total = self.service.get_store_transactions(
|
|
db, test_loyalty_card.merchant_id, skip=0, limit=2
|
|
)
|
|
assert total >= 3
|
|
assert len(transactions) == 2
|
|
|
|
|
|
# ============================================================================
|
|
# Item 9: Reactivate card creates audit transaction
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.loyalty
|
|
class TestReactivateCardAudit:
|
|
"""Tests for reactivate_card creating CARD_REACTIVATED transaction."""
|
|
|
|
def setup_method(self):
|
|
self.service = CardService()
|
|
|
|
def test_reactivate_creates_transaction(self, db, test_loyalty_card):
|
|
"""Reactivating a card creates a CARD_REACTIVATED transaction."""
|
|
# Deactivate first
|
|
test_loyalty_card.is_active = False
|
|
db.commit()
|
|
|
|
self.service.reactivate_card(db, test_loyalty_card.id)
|
|
|
|
tx = (
|
|
db.query(LoyaltyTransaction)
|
|
.filter(
|
|
LoyaltyTransaction.card_id == test_loyalty_card.id,
|
|
LoyaltyTransaction.transaction_type == TransactionType.CARD_REACTIVATED.value,
|
|
)
|
|
.first()
|
|
)
|
|
assert tx is not None
|
|
assert tx.merchant_id == test_loyalty_card.merchant_id
|
|
|
|
def test_reactivate_sets_card_active(self, db, test_loyalty_card):
|
|
"""Reactivating a card sets is_active to True."""
|
|
test_loyalty_card.is_active = False
|
|
db.commit()
|
|
|
|
card = self.service.reactivate_card(db, test_loyalty_card.id)
|
|
assert card.is_active is True
|