Samir Boulahtit
994c6419f0
- Fix 404 on /admin/loyalty/* and /vendor/loyalty/* by applying ROUTE_CONFIG custom_prefix when registering page routers in main.py (admin, vendor, and storefront registrations all updated) - Move loyalty alembic migrations from central alembic/versions/ into app/modules/loyalty/migrations/versions/ with proper naming convention - Add migrations_path="migrations" to loyalty module definition so the auto-discovery system finds them - Add unit tests for Apple/Google Wallet Code 128 barcode configuration (6 Apple tests, 4 Google tests) - Add integration tests for module migration auto-discovery (4 tests) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
466 lines
18 KiB
Python
466 lines
18 KiB
Python
# 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)
|
|
|
|
|
|
# =============================================================================
|
|
# Apple Wallet Barcode Tests
|
|
# =============================================================================
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.service
|
|
class TestAppleWalletBarcode:
|
|
"""Tests for Apple Wallet pass barcode configuration (Code 128)."""
|
|
|
|
def _build_pass(self, card):
|
|
from app.modules.loyalty.services.apple_wallet_service import AppleWalletService
|
|
|
|
service = AppleWalletService()
|
|
return service._build_pass_json(card, card.program)
|
|
|
|
def test_primary_barcode_is_code128(self, db, test_loyalty_card):
|
|
"""Primary barcode format must be Code 128 for retail scanners."""
|
|
pass_data = self._build_pass(test_loyalty_card)
|
|
assert pass_data["barcode"]["format"] == "PKBarcodeFormatCode128"
|
|
|
|
def test_primary_barcode_uses_card_number_without_dashes(self, db, test_loyalty_card):
|
|
"""Barcode message is card number with dashes stripped (digits only)."""
|
|
pass_data = self._build_pass(test_loyalty_card)
|
|
expected = test_loyalty_card.card_number.replace("-", "")
|
|
assert pass_data["barcode"]["message"] == expected
|
|
assert "-" not in pass_data["barcode"]["message"]
|
|
|
|
def test_primary_barcode_alttext_shows_formatted_number(self, db, test_loyalty_card):
|
|
"""altText displays the human-readable card number with dashes."""
|
|
pass_data = self._build_pass(test_loyalty_card)
|
|
assert pass_data["barcode"]["altText"] == test_loyalty_card.card_number
|
|
|
|
def test_barcodes_array_code128_first_qr_second(self, db, test_loyalty_card):
|
|
"""Barcodes array has Code 128 first (primary) and QR second (fallback)."""
|
|
pass_data = self._build_pass(test_loyalty_card)
|
|
barcodes = pass_data["barcodes"]
|
|
assert len(barcodes) == 2
|
|
assert barcodes[0]["format"] == "PKBarcodeFormatCode128"
|
|
assert barcodes[1]["format"] == "PKBarcodeFormatQR"
|
|
|
|
def test_barcodes_array_code128_matches_primary(self, db, test_loyalty_card):
|
|
"""First entry in barcodes array matches the primary barcode."""
|
|
pass_data = self._build_pass(test_loyalty_card)
|
|
expected = test_loyalty_card.card_number.replace("-", "")
|
|
assert pass_data["barcodes"][0]["message"] == expected
|
|
assert pass_data["barcodes"][0]["altText"] == test_loyalty_card.card_number
|
|
|
|
def test_barcodes_array_qr_uses_qr_code_data(self, db, test_loyalty_card):
|
|
"""QR fallback uses the qr_code_data token, not the card number."""
|
|
pass_data = self._build_pass(test_loyalty_card)
|
|
assert pass_data["barcodes"][1]["message"] == test_loyalty_card.qr_code_data
|
|
|
|
|
|
# =============================================================================
|
|
# Google Wallet Barcode Tests
|
|
# =============================================================================
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.service
|
|
class TestGoogleWalletBarcode:
|
|
"""Tests for Google Wallet object barcode configuration (CODE_128)."""
|
|
|
|
def _build_object(self, card):
|
|
from app.modules.loyalty.services.google_wallet_service import GoogleWalletService
|
|
|
|
service = GoogleWalletService()
|
|
return service._build_object_data(card, f"test_issuer.loyalty_card_{card.id}")
|
|
|
|
def test_barcode_type_is_code128(self, db, test_loyalty_card):
|
|
"""Barcode type must be CODE_128 for retail scanners."""
|
|
obj = self._build_object(test_loyalty_card)
|
|
assert obj["barcode"]["type"] == "CODE_128"
|
|
|
|
def test_barcode_value_uses_card_number_without_dashes(self, db, test_loyalty_card):
|
|
"""Barcode value is card number with dashes stripped."""
|
|
obj = self._build_object(test_loyalty_card)
|
|
expected = test_loyalty_card.card_number.replace("-", "")
|
|
assert obj["barcode"]["value"] == expected
|
|
assert "-" not in obj["barcode"]["value"]
|
|
|
|
def test_barcode_alternate_text_shows_formatted_number(self, db, test_loyalty_card):
|
|
"""alternateText displays the human-readable card number."""
|
|
obj = self._build_object(test_loyalty_card)
|
|
assert obj["barcode"]["alternateText"] == test_loyalty_card.card_number
|
|
|
|
def test_barcode_has_all_required_fields(self, db, test_loyalty_card):
|
|
"""Barcode object contains type, value, and alternateText."""
|
|
obj = self._build_object(test_loyalty_card)
|
|
barcode = obj["barcode"]
|
|
assert "type" in barcode
|
|
assert "value" in barcode
|
|
assert "alternateText" in barcode
|
|
|
|
|
|
# =============================================================================
|
|
# Module Migration Discovery Tests
|
|
# =============================================================================
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestLoyaltyMigrationDiscovery:
|
|
"""Tests for loyalty module migration auto-discovery."""
|
|
|
|
def test_loyalty_migrations_discovered(self):
|
|
"""Loyalty module migrations are found by the discovery system."""
|
|
from app.modules.migrations import discover_module_migrations
|
|
|
|
paths = discover_module_migrations()
|
|
loyalty_paths = [p for p in paths if "loyalty" in str(p)]
|
|
assert len(loyalty_paths) == 1
|
|
assert loyalty_paths[0].exists()
|
|
|
|
def test_loyalty_migrations_in_all_paths(self):
|
|
"""Loyalty migrations are included in get_all_migration_paths."""
|
|
from app.modules.migrations import get_all_migration_paths
|
|
|
|
paths = get_all_migration_paths()
|
|
path_strs = [str(p) for p in paths]
|
|
assert any("loyalty" in p for p in path_strs)
|
|
# Core migrations should still be first
|
|
assert "alembic" in str(paths[0])
|
|
|
|
def test_loyalty_migration_files_exist(self):
|
|
"""Loyalty migration version files exist in the module directory."""
|
|
from app.modules.migrations import discover_module_migrations
|
|
|
|
paths = discover_module_migrations()
|
|
loyalty_path = [p for p in paths if "loyalty" in str(p)][0]
|
|
migration_files = list(loyalty_path.glob("loyalty_*.py"))
|
|
assert len(migration_files) >= 2
|
|
|
|
def test_loyalty_migrations_follow_naming_convention(self):
|
|
"""Loyalty migration files follow the loyalty_ prefix convention."""
|
|
from app.modules.migrations import discover_module_migrations
|
|
|
|
paths = discover_module_migrations()
|
|
loyalty_path = [p for p in paths if "loyalty" in str(p)][0]
|
|
for f in loyalty_path.glob("*.py"):
|
|
if f.name == "__init__.py":
|
|
continue
|
|
assert f.name.startswith("loyalty_"), f"{f.name} should start with 'loyalty_'"
|