Some checks failed
- Fix rate limiter to extract real client IP and handle sync/async endpoints - Rate-limit public enrollment (10/min) and program info (30/min) endpoints - Add 409 Conflict to non-retryable status codes in retry decorator - Cache private key in get_save_url() to avoid re-reading JSON per call - Make update_class() return bool success status with error-level logging - Move Google Wallet config from core to loyalty module config - Document time.sleep() safety in retry decorator (threadpool execution) - Add per-card retry (1 retry, 2s delay) to wallet sync task - Add logo URL reachability check (HEAD request) to validate_config() - Add 26 comprehensive unit tests for GoogleWalletService Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
392 lines
14 KiB
Python
392 lines
14 KiB
Python
# app/modules/loyalty/tests/unit/test_wallet_service.py
|
|
"""
|
|
Unit tests for WalletService — wallet object creation during enrollment.
|
|
|
|
Tests that:
|
|
- Google Wallet objects are created and DB fields populated on enrollment
|
|
- Apple Wallet serial numbers are set when apple_pass_type_id is configured
|
|
- Wallet creation failures don't crash enrollment
|
|
- Wallet creation is skipped when not configured
|
|
"""
|
|
|
|
import uuid
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from app.modules.loyalty.models import LoyaltyCard, LoyaltyProgram
|
|
from app.modules.loyalty.models.loyalty_program import LoyaltyType
|
|
from app.modules.loyalty.services.wallet_service import WalletService
|
|
|
|
# ============================================================================
|
|
# Fixtures
|
|
# Uses test_merchant and test_customer from shared fixtures (store_fixtures,
|
|
# customer_fixtures) which handle owner_user_id and other required fields.
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def wt_program(db, test_merchant):
|
|
"""Create a loyalty program for wallet tests."""
|
|
program = LoyaltyProgram(
|
|
merchant_id=test_merchant.id,
|
|
loyalty_type=LoyaltyType.POINTS.value,
|
|
points_per_euro=10,
|
|
welcome_bonus_points=0,
|
|
minimum_redemption_points=100,
|
|
cooldown_minutes=0,
|
|
max_daily_stamps=5,
|
|
require_staff_pin=False,
|
|
card_name="Wallet Test Rewards",
|
|
card_color="#4F46E5",
|
|
is_active=True,
|
|
)
|
|
db.add(program)
|
|
db.commit()
|
|
db.refresh(program)
|
|
return program
|
|
|
|
|
|
@pytest.fixture
|
|
def wt_program_with_apple(db, test_merchant):
|
|
"""Create a loyalty program with Apple Wallet configured."""
|
|
program = LoyaltyProgram(
|
|
merchant_id=test_merchant.id,
|
|
loyalty_type=LoyaltyType.POINTS.value,
|
|
points_per_euro=10,
|
|
welcome_bonus_points=0,
|
|
minimum_redemption_points=100,
|
|
cooldown_minutes=0,
|
|
max_daily_stamps=5,
|
|
require_staff_pin=False,
|
|
card_name="Apple Wallet Test",
|
|
card_color="#000000",
|
|
apple_pass_type_id="pass.com.test.loyalty",
|
|
is_active=True,
|
|
)
|
|
db.add(program)
|
|
db.commit()
|
|
db.refresh(program)
|
|
return program
|
|
|
|
|
|
@pytest.fixture
|
|
def wt_card(db, wt_program, test_customer):
|
|
"""Create a loyalty card for wallet tests."""
|
|
card = LoyaltyCard(
|
|
merchant_id=wt_program.merchant_id,
|
|
program_id=wt_program.id,
|
|
customer_id=test_customer.id,
|
|
card_number=f"WT-{uuid.uuid4().hex[:8].upper()}",
|
|
is_active=True,
|
|
)
|
|
db.add(card)
|
|
db.commit()
|
|
db.refresh(card)
|
|
return card
|
|
|
|
|
|
@pytest.fixture
|
|
def wt_card_apple(db, wt_program_with_apple, test_customer):
|
|
"""Create a loyalty card for Apple Wallet tests."""
|
|
card = LoyaltyCard(
|
|
merchant_id=wt_program_with_apple.merchant_id,
|
|
program_id=wt_program_with_apple.id,
|
|
customer_id=test_customer.id,
|
|
card_number=f"WTA-{uuid.uuid4().hex[:8].upper()}",
|
|
is_active=True,
|
|
)
|
|
db.add(card)
|
|
db.commit()
|
|
db.refresh(card)
|
|
return card
|
|
|
|
|
|
# ============================================================================
|
|
# WalletService instantiation
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.loyalty
|
|
class TestWalletService:
|
|
"""Test suite for WalletService."""
|
|
|
|
def setup_method(self):
|
|
self.service = WalletService()
|
|
|
|
def test_service_instantiation(self):
|
|
"""Service can be instantiated."""
|
|
assert self.service is not None
|
|
|
|
|
|
# ============================================================================
|
|
# Google Wallet object creation
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.loyalty
|
|
class TestCreateWalletObjectsGoogle:
|
|
"""Tests for Google Wallet object creation during enrollment."""
|
|
|
|
@patch("app.modules.loyalty.services.google_wallet_service.google_wallet_service")
|
|
def test_google_wallet_object_created_when_configured(
|
|
self, mock_gw, db, wt_card
|
|
):
|
|
"""Google Wallet object is created when service is configured."""
|
|
mock_gw.is_configured = True
|
|
mock_gw.create_object.return_value = "test_object_id"
|
|
|
|
service = WalletService()
|
|
results = service.create_wallet_objects(db, wt_card)
|
|
|
|
assert results["google_wallet"] is True
|
|
mock_gw.create_object.assert_called_once_with(db, wt_card)
|
|
|
|
@patch("app.modules.loyalty.services.google_wallet_service.google_wallet_service")
|
|
def test_google_wallet_skipped_when_not_configured(
|
|
self, mock_gw, db, wt_card
|
|
):
|
|
"""Google Wallet is skipped when not configured."""
|
|
mock_gw.is_configured = False
|
|
|
|
service = WalletService()
|
|
results = service.create_wallet_objects(db, wt_card)
|
|
|
|
assert results["google_wallet"] is False
|
|
mock_gw.create_object.assert_not_called()
|
|
|
|
@patch("app.modules.loyalty.services.google_wallet_service.google_wallet_service")
|
|
def test_google_wallet_failure_does_not_crash(
|
|
self, mock_gw, db, wt_card
|
|
):
|
|
"""Enrollment continues even if Google Wallet creation fails."""
|
|
mock_gw.is_configured = True
|
|
mock_gw.create_object.side_effect = Exception("Google API error")
|
|
|
|
service = WalletService()
|
|
results = service.create_wallet_objects(db, wt_card)
|
|
|
|
assert results["google_wallet"] is False
|
|
# Card should still be valid
|
|
db.refresh(wt_card)
|
|
assert wt_card.is_active is True
|
|
|
|
|
|
# ============================================================================
|
|
# Apple Wallet serial number setup
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.loyalty
|
|
class TestCreateWalletObjectsApple:
|
|
"""Tests for Apple Wallet serial number assignment during enrollment."""
|
|
|
|
def test_apple_serial_number_set_when_configured(self, db, wt_card_apple):
|
|
"""Apple serial number is set when program has apple_pass_type_id."""
|
|
assert wt_card_apple.apple_serial_number is None
|
|
|
|
service = WalletService()
|
|
with patch(
|
|
"app.modules.loyalty.services.google_wallet_service.google_wallet_service"
|
|
) as mock_gw:
|
|
mock_gw.is_configured = False
|
|
results = service.create_wallet_objects(db, wt_card_apple)
|
|
|
|
assert results["apple_wallet"] is True
|
|
db.refresh(wt_card_apple)
|
|
assert wt_card_apple.apple_serial_number is not None
|
|
assert wt_card_apple.apple_serial_number.startswith(f"card_{wt_card_apple.id}_")
|
|
|
|
def test_apple_serial_number_not_set_when_unconfigured(self, db, wt_card):
|
|
"""Apple serial number not set when program lacks apple_pass_type_id."""
|
|
service = WalletService()
|
|
with patch(
|
|
"app.modules.loyalty.services.google_wallet_service.google_wallet_service"
|
|
) as mock_gw:
|
|
mock_gw.is_configured = False
|
|
results = service.create_wallet_objects(db, wt_card)
|
|
|
|
assert results["apple_wallet"] is False
|
|
db.refresh(wt_card)
|
|
assert wt_card.apple_serial_number is None
|
|
|
|
|
|
# ============================================================================
|
|
# GoogleWalletService.create_object — DB field population
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.loyalty
|
|
class TestGoogleWalletCreateObject:
|
|
"""Tests that GoogleWalletService.create_object populates DB fields correctly."""
|
|
|
|
@patch("app.modules.loyalty.services.google_wallet_service.config")
|
|
def test_create_object_sets_google_object_id(self, mock_config, db, wt_card, wt_program):
|
|
"""create_object sets card.google_object_id in the database."""
|
|
from app.modules.loyalty.services.google_wallet_service import (
|
|
GoogleWalletService,
|
|
)
|
|
|
|
mock_config.google_issuer_id = "1234567890"
|
|
mock_config.google_service_account_json = "/fake/path.json"
|
|
|
|
# Pre-set class_id on program so create_class isn't called
|
|
wt_program.google_class_id = "1234567890.loyalty_program_1"
|
|
db.commit()
|
|
|
|
service = GoogleWalletService()
|
|
service._credentials = MagicMock() # Skip credential loading
|
|
|
|
mock_http = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 201
|
|
mock_http.post.return_value = mock_response
|
|
service._http_client = mock_http
|
|
|
|
result = service.create_object(db, wt_card)
|
|
|
|
expected_object_id = f"1234567890.loyalty_card_{wt_card.id}"
|
|
assert result == expected_object_id
|
|
|
|
db.refresh(wt_card)
|
|
assert wt_card.google_object_id == expected_object_id
|
|
|
|
@patch("app.modules.loyalty.services.google_wallet_service.config")
|
|
def test_create_object_handles_409_conflict(self, mock_config, db, wt_card, wt_program):
|
|
"""create_object handles 409 (already exists) by still setting the object_id."""
|
|
from app.modules.loyalty.services.google_wallet_service import (
|
|
GoogleWalletService,
|
|
)
|
|
|
|
mock_config.google_issuer_id = "1234567890"
|
|
mock_config.google_service_account_json = "/fake/path.json"
|
|
|
|
wt_program.google_class_id = "1234567890.loyalty_program_1"
|
|
db.commit()
|
|
|
|
service = GoogleWalletService()
|
|
service._credentials = MagicMock()
|
|
|
|
mock_http = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 409 # Already exists
|
|
mock_http.post.return_value = mock_response
|
|
service._http_client = mock_http
|
|
|
|
service.create_object(db, wt_card)
|
|
|
|
db.refresh(wt_card)
|
|
assert wt_card.google_object_id == f"1234567890.loyalty_card_{wt_card.id}"
|
|
|
|
@patch("app.modules.loyalty.services.google_wallet_service.config")
|
|
def test_create_class_sets_google_class_id(self, mock_config, db, wt_program):
|
|
"""create_class sets program.google_class_id in the database."""
|
|
from app.modules.loyalty.services.google_wallet_service import (
|
|
GoogleWalletService,
|
|
)
|
|
|
|
mock_config.google_issuer_id = "1234567890"
|
|
mock_config.google_service_account_json = "/fake/path.json"
|
|
|
|
assert wt_program.google_class_id is None
|
|
|
|
service = GoogleWalletService()
|
|
service._credentials = MagicMock()
|
|
|
|
mock_http = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 201
|
|
mock_http.post.return_value = mock_response
|
|
service._http_client = mock_http
|
|
|
|
result = service.create_class(db, wt_program)
|
|
|
|
expected_class_id = f"1234567890.loyalty_program_{wt_program.id}"
|
|
assert result == expected_class_id
|
|
|
|
db.refresh(wt_program)
|
|
assert wt_program.google_class_id == expected_class_id
|
|
|
|
|
|
# ============================================================================
|
|
# End-to-end: enrollment → wallet creation
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.loyalty
|
|
class TestEnrollmentWalletCreation:
|
|
"""Tests that enrollment triggers wallet object creation and populates DB fields."""
|
|
|
|
@patch("app.modules.loyalty.services.google_wallet_service.config")
|
|
def test_enrollment_creates_google_wallet_object(
|
|
self, mock_config, db, wt_program, test_merchant, test_customer
|
|
):
|
|
"""Full enrollment flow creates Google Wallet class + object in DB."""
|
|
from app.modules.loyalty.services.google_wallet_service import (
|
|
google_wallet_service,
|
|
)
|
|
|
|
mock_config.google_issuer_id = "1234567890"
|
|
mock_config.google_service_account_json = "/fake/path.json"
|
|
|
|
# Mock the HTTP client to simulate Google API success
|
|
mock_http = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 201
|
|
mock_http.post.return_value = mock_response
|
|
google_wallet_service._http_client = mock_http
|
|
google_wallet_service._credentials = MagicMock()
|
|
|
|
# Enroll via card_service
|
|
from app.modules.loyalty.services.card_service import card_service
|
|
card = card_service.enroll_customer(db, test_customer.id, test_merchant.id)
|
|
|
|
# Verify wallet DB fields are populated
|
|
db.refresh(card)
|
|
assert card.google_object_id is not None
|
|
assert card.google_object_id == f"1234567890.loyalty_card_{card.id}"
|
|
|
|
# Verify program class_id is set
|
|
db.refresh(wt_program)
|
|
assert wt_program.google_class_id is not None
|
|
assert wt_program.google_class_id == f"1234567890.loyalty_program_{wt_program.id}"
|
|
|
|
# Clean up singleton state
|
|
google_wallet_service._http_client = None
|
|
google_wallet_service._credentials = None
|
|
|
|
def test_enrollment_succeeds_without_wallet_config(
|
|
self, db, wt_program, test_merchant, test_customer
|
|
):
|
|
"""Enrollment works even when Google Wallet is not configured."""
|
|
from app.modules.loyalty.services.card_service import card_service
|
|
card = card_service.enroll_customer(db, test_customer.id, test_merchant.id)
|
|
|
|
db.refresh(card)
|
|
assert card.id is not None
|
|
assert card.is_active is True
|
|
# Wallet fields should be None since not configured
|
|
assert card.google_object_id is None
|
|
|
|
@patch("app.modules.loyalty.services.google_wallet_service.config")
|
|
def test_enrollment_with_apple_wallet_sets_serial(
|
|
self, mock_config, db, wt_program_with_apple, test_customer
|
|
):
|
|
"""Enrollment sets apple_serial_number when program has apple_pass_type_id."""
|
|
mock_config.google_issuer_id = None
|
|
mock_config.google_service_account_json = None
|
|
|
|
from app.modules.loyalty.services.card_service import card_service
|
|
card = card_service.enroll_customer(
|
|
db, test_customer.id, wt_program_with_apple.merchant_id
|
|
)
|
|
|
|
db.refresh(card)
|
|
assert card.apple_serial_number is not None
|
|
assert card.apple_serial_number.startswith(f"card_{card.id}_")
|