test(loyalty): add comprehensive test suite for loyalty module
Add tests for the loyalty module Phase 2 implementation: Fixtures (tests/fixtures/loyalty_fixtures.py): - test_loyalty_program: Points-based program with rewards - test_loyalty_program_no_expiration: Program without point expiry - test_loyalty_card: Active customer card - test_loyalty_card_inactive: Card for expiration testing - test_loyalty_transaction: Sample transaction - test_staff_pin: Staff PIN for verification tests Unit Tests (tests/unit/services/test_loyalty_services.py): - ProgramService: Company queries, listing, filtering - CardService: Lookup, enrollment, balance operations - PointsService: Earn, redeem, void, adjust operations - PinService: Creation, verification, lockout Integration Tests (tests/integration/api/v1/loyalty/): - Vendor API: Program settings, card management, PIN operations - Storefront API: Endpoint existence verification Task Tests (tests/integration/tasks/test_loyalty_tasks.py): - Point expiration for inactive cards - Transaction record creation on expiration - No expiration for active cards or zero balances - Voided total tracking Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
320
tests/unit/services/test_loyalty_services.py
Normal file
320
tests/unit/services/test_loyalty_services.py
Normal file
@@ -0,0 +1,320 @@
|
||||
# tests/unit/services/test_loyalty_services.py
|
||||
"""
|
||||
Unit tests for Loyalty module services.
|
||||
|
||||
Tests cover:
|
||||
- Program service: CRUD operations, company-based queries
|
||||
- Card service: Enrollment, lookup, balance operations
|
||||
- Points service: Earn, redeem, void operations
|
||||
- PIN service: Verification, lockout
|
||||
"""
|
||||
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from app.modules.loyalty.exceptions import (
|
||||
LoyaltyCardNotFoundException,
|
||||
LoyaltyException,
|
||||
LoyaltyProgramNotFoundException,
|
||||
)
|
||||
from app.modules.loyalty.models import LoyaltyCard, LoyaltyProgram, LoyaltyTransaction
|
||||
from app.modules.loyalty.models.loyalty_program import LoyaltyType
|
||||
from app.modules.loyalty.models.loyalty_transaction import TransactionType
|
||||
from app.modules.loyalty.services import (
|
||||
card_service,
|
||||
pin_service,
|
||||
points_service,
|
||||
program_service,
|
||||
)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Program Service Tests
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.service
|
||||
class TestProgramService:
|
||||
"""Tests for program_service."""
|
||||
|
||||
def test_get_program_by_company(self, db, test_loyalty_program):
|
||||
"""Test getting a program by company ID."""
|
||||
program = program_service.get_program_by_company(
|
||||
db, test_loyalty_program.company_id
|
||||
)
|
||||
assert program is not None
|
||||
assert program.id == test_loyalty_program.id
|
||||
assert program.company_id == test_loyalty_program.company_id
|
||||
|
||||
def test_get_program_by_company_not_found(self, db):
|
||||
"""Test getting a program for non-existent company."""
|
||||
program = program_service.get_program_by_company(db, 99999)
|
||||
assert program is None
|
||||
|
||||
def test_get_program_by_vendor(self, db, test_loyalty_program, test_vendor):
|
||||
"""Test getting a program by vendor ID."""
|
||||
program = program_service.get_program_by_vendor(db, test_vendor.id)
|
||||
assert program is not None
|
||||
assert program.company_id == test_vendor.company_id
|
||||
|
||||
def test_list_programs(self, db, test_loyalty_program):
|
||||
"""Test listing all programs with pagination."""
|
||||
programs, total = program_service.list_programs(db, skip=0, limit=10)
|
||||
assert total >= 1
|
||||
assert any(p.id == test_loyalty_program.id for p in programs)
|
||||
|
||||
def test_list_programs_active_only(self, db, test_loyalty_program):
|
||||
"""Test listing only active programs."""
|
||||
programs, total = program_service.list_programs(
|
||||
db, skip=0, limit=10, active_only=True
|
||||
)
|
||||
assert all(p.is_active for p in programs)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Card Service Tests
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.service
|
||||
class TestCardService:
|
||||
"""Tests for card_service."""
|
||||
|
||||
def test_get_card_by_id(self, db, test_loyalty_card):
|
||||
"""Test getting a card by ID."""
|
||||
card = card_service.get_card(db, test_loyalty_card.id)
|
||||
assert card is not None
|
||||
assert card.id == test_loyalty_card.id
|
||||
|
||||
def test_get_card_not_found(self, db):
|
||||
"""Test getting a non-existent card."""
|
||||
with pytest.raises(LoyaltyCardNotFoundException):
|
||||
card_service.get_card(db, 99999)
|
||||
|
||||
def test_get_card_by_number(self, db, test_loyalty_card):
|
||||
"""Test getting a card by card number."""
|
||||
card = card_service.get_card_by_number(
|
||||
db, test_loyalty_card.company_id, test_loyalty_card.card_number
|
||||
)
|
||||
assert card is not None
|
||||
assert card.card_number == test_loyalty_card.card_number
|
||||
|
||||
def test_get_card_by_customer_email(self, db, test_loyalty_card):
|
||||
"""Test getting a card by customer email."""
|
||||
card = card_service.get_card_by_customer_email(
|
||||
db, test_loyalty_card.company_id, test_loyalty_card.customer_email
|
||||
)
|
||||
assert card is not None
|
||||
assert card.customer_email == test_loyalty_card.customer_email
|
||||
|
||||
def test_lookup_card(self, db, test_loyalty_card):
|
||||
"""Test looking up a card by various identifiers."""
|
||||
# By card number
|
||||
card = card_service.lookup_card(
|
||||
db, test_loyalty_card.company_id, test_loyalty_card.card_number
|
||||
)
|
||||
assert card is not None
|
||||
assert card.id == test_loyalty_card.id
|
||||
|
||||
# By email
|
||||
card = card_service.lookup_card(
|
||||
db, test_loyalty_card.company_id, test_loyalty_card.customer_email
|
||||
)
|
||||
assert card is not None
|
||||
assert card.id == test_loyalty_card.id
|
||||
|
||||
def test_enroll_customer(self, db, test_loyalty_program, test_vendor):
|
||||
"""Test enrolling a new customer."""
|
||||
card = card_service.enroll_customer(
|
||||
db,
|
||||
vendor_id=test_vendor.id,
|
||||
customer_email="newmember@test.com",
|
||||
customer_name="New Member",
|
||||
customer_phone="+352123456789",
|
||||
)
|
||||
db.commit()
|
||||
|
||||
assert card is not None
|
||||
assert card.customer_email == "newmember@test.com"
|
||||
assert card.company_id == test_vendor.company_id
|
||||
# Check welcome bonus was applied
|
||||
assert card.points_balance == test_loyalty_program.welcome_bonus_points
|
||||
|
||||
def test_enroll_customer_duplicate(self, db, test_loyalty_card, test_vendor):
|
||||
"""Test enrolling an existing customer raises error."""
|
||||
with pytest.raises(LoyaltyException) as exc_info:
|
||||
card_service.enroll_customer(
|
||||
db,
|
||||
vendor_id=test_vendor.id,
|
||||
customer_email=test_loyalty_card.customer_email,
|
||||
customer_name="Duplicate",
|
||||
)
|
||||
assert "already enrolled" in str(exc_info.value).lower()
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Points Service Tests
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.service
|
||||
class TestPointsService:
|
||||
"""Tests for points_service."""
|
||||
|
||||
def test_earn_points(self, db, test_loyalty_card, test_vendor, test_staff_pin):
|
||||
"""Test earning points."""
|
||||
initial_balance = test_loyalty_card.points_balance
|
||||
purchase_amount_cents = 5000 # €50
|
||||
|
||||
result = points_service.earn_points(
|
||||
db,
|
||||
card_id=test_loyalty_card.id,
|
||||
vendor_id=test_vendor.id,
|
||||
purchase_amount_cents=purchase_amount_cents,
|
||||
staff_pin_id=test_staff_pin.id,
|
||||
)
|
||||
db.commit()
|
||||
|
||||
assert result is not None
|
||||
assert result["points_earned"] > 0
|
||||
assert result["new_balance"] > initial_balance
|
||||
|
||||
def test_redeem_points(self, db, test_loyalty_card, test_vendor, test_staff_pin):
|
||||
"""Test redeeming points."""
|
||||
# Ensure card has enough points
|
||||
test_loyalty_card.points_balance = 200
|
||||
db.commit()
|
||||
|
||||
initial_balance = test_loyalty_card.points_balance
|
||||
|
||||
result = points_service.redeem_points(
|
||||
db,
|
||||
card_id=test_loyalty_card.id,
|
||||
vendor_id=test_vendor.id,
|
||||
reward_id="reward_1", # 100 points
|
||||
staff_pin_id=test_staff_pin.id,
|
||||
)
|
||||
db.commit()
|
||||
|
||||
assert result is not None
|
||||
assert result["points_redeemed"] == 100
|
||||
assert result["new_balance"] == initial_balance - 100
|
||||
|
||||
def test_redeem_points_insufficient_balance(
|
||||
self, db, test_loyalty_card, test_vendor, test_staff_pin
|
||||
):
|
||||
"""Test redeeming points with insufficient balance."""
|
||||
test_loyalty_card.points_balance = 50 # Less than minimum
|
||||
db.commit()
|
||||
|
||||
with pytest.raises(LoyaltyException) as exc_info:
|
||||
points_service.redeem_points(
|
||||
db,
|
||||
card_id=test_loyalty_card.id,
|
||||
vendor_id=test_vendor.id,
|
||||
reward_id="reward_1", # 100 points needed
|
||||
staff_pin_id=test_staff_pin.id,
|
||||
)
|
||||
assert "insufficient" in str(exc_info.value).lower()
|
||||
|
||||
def test_void_points(self, db, test_loyalty_card, test_vendor, test_staff_pin):
|
||||
"""Test voiding points (for returns)."""
|
||||
initial_balance = test_loyalty_card.points_balance
|
||||
points_to_void = 50
|
||||
|
||||
result = points_service.void_points(
|
||||
db,
|
||||
card_id=test_loyalty_card.id,
|
||||
vendor_id=test_vendor.id,
|
||||
points_to_void=points_to_void,
|
||||
reason="Customer return",
|
||||
staff_pin_id=test_staff_pin.id,
|
||||
)
|
||||
db.commit()
|
||||
|
||||
assert result is not None
|
||||
assert result["points_voided"] == points_to_void
|
||||
assert result["new_balance"] == initial_balance - points_to_void
|
||||
|
||||
def test_adjust_points(self, db, test_loyalty_card, test_vendor):
|
||||
"""Test manual points adjustment."""
|
||||
initial_balance = test_loyalty_card.points_balance
|
||||
adjustment = 25
|
||||
|
||||
result = points_service.adjust_points(
|
||||
db,
|
||||
card_id=test_loyalty_card.id,
|
||||
vendor_id=test_vendor.id,
|
||||
points_delta=adjustment,
|
||||
reason="Manual correction",
|
||||
)
|
||||
db.commit()
|
||||
|
||||
assert result is not None
|
||||
assert result["new_balance"] == initial_balance + adjustment
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# PIN Service Tests
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.service
|
||||
class TestPinService:
|
||||
"""Tests for pin_service."""
|
||||
|
||||
def test_verify_pin_success(self, db, test_staff_pin):
|
||||
"""Test successful PIN verification."""
|
||||
result = pin_service.verify_pin(
|
||||
db,
|
||||
pin_id=test_staff_pin.id,
|
||||
pin="1234",
|
||||
)
|
||||
assert result is True
|
||||
|
||||
def test_verify_pin_wrong_pin(self, db, test_staff_pin):
|
||||
"""Test PIN verification with wrong PIN."""
|
||||
result = pin_service.verify_pin(
|
||||
db,
|
||||
pin_id=test_staff_pin.id,
|
||||
pin="9999",
|
||||
)
|
||||
assert result is False
|
||||
|
||||
def test_verify_pin_increments_failed_attempts(self, db, test_staff_pin):
|
||||
"""Test that failed verification increments attempt counter."""
|
||||
initial_attempts = test_staff_pin.failed_attempts or 0
|
||||
|
||||
pin_service.verify_pin(db, pin_id=test_staff_pin.id, pin="wrong")
|
||||
db.refresh(test_staff_pin)
|
||||
|
||||
assert test_staff_pin.failed_attempts == initial_attempts + 1
|
||||
|
||||
def test_create_pin(self, db, test_loyalty_program, test_vendor):
|
||||
"""Test creating a new staff PIN."""
|
||||
pin = pin_service.create_pin(
|
||||
db,
|
||||
program_id=test_loyalty_program.id,
|
||||
vendor_id=test_vendor.id,
|
||||
staff_name="New Staff Member",
|
||||
pin="5678",
|
||||
)
|
||||
db.commit()
|
||||
|
||||
assert pin is not None
|
||||
assert pin.staff_name == "New Staff Member"
|
||||
assert pin.is_active is True
|
||||
# Verify PIN works
|
||||
assert pin_service.verify_pin(db, pin.id, "5678") is True
|
||||
|
||||
def test_list_pins_for_vendor(self, db, test_staff_pin, test_vendor):
|
||||
"""Test listing PINs for a vendor."""
|
||||
pins = pin_service.list_pins_for_vendor(db, test_vendor.id)
|
||||
assert len(pins) >= 1
|
||||
assert any(p.id == test_staff_pin.id for p in pins)
|
||||
Reference in New Issue
Block a user