feat: add SQL query tool, platform debug, loyalty settings, and multi-module improvements
Some checks failed
Some checks failed
- Add admin SQL query tool with saved queries, schema explorer presets, and collapsible category sections (dev_tools module) - Add platform debug tool for admin diagnostics - Add loyalty settings page with owner-only access control - Fix loyalty settings owner check (use currentUser instead of window.__userData) - Replace HTTPException with AuthorizationException in loyalty routes - Expand loyalty module with PIN service, Apple Wallet, program management - Improve store login with platform detection and multi-platform support - Update billing feature gates and subscription services - Add store platform sync improvements and remove is_primary column - Add unit tests for loyalty (PIN, points, stamps, program services) - Update i18n translations across dev_tools locales Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -12,12 +12,15 @@ Tests:
|
||||
Authentication: Uses real JWT tokens via store login endpoint.
|
||||
"""
|
||||
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
|
||||
import pytest
|
||||
|
||||
from app.modules.loyalty.models import LoyaltyTransaction
|
||||
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.tenancy.models import User
|
||||
|
||||
BASE = "/api/v1/store/loyalty"
|
||||
|
||||
@@ -217,42 +220,20 @@ class TestEarnPoints:
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Store Program CRUD Removed
|
||||
# Store Program CRUD
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.loyalty
|
||||
class TestStoreProgramCrudRemoved:
|
||||
"""Verify POST/PATCH /program endpoints are removed from store API."""
|
||||
class TestStoreProgramCrud:
|
||||
"""Tests for store program CRUD endpoints (merchant_owner only)."""
|
||||
|
||||
def test_create_program_removed(
|
||||
self, client, loyalty_store_headers
|
||||
):
|
||||
"""POST /program no longer exists on store API."""
|
||||
response = client.post(
|
||||
f"{BASE}/program",
|
||||
json={"loyalty_type": "points"},
|
||||
headers=loyalty_store_headers,
|
||||
)
|
||||
assert response.status_code == 405 # Method Not Allowed
|
||||
|
||||
def test_update_program_removed(
|
||||
self, client, loyalty_store_headers
|
||||
):
|
||||
"""PATCH /program no longer exists on store API."""
|
||||
response = client.patch(
|
||||
f"{BASE}/program",
|
||||
json={"card_name": "Should Not Work"},
|
||||
headers=loyalty_store_headers,
|
||||
)
|
||||
assert response.status_code == 405 # Method Not Allowed
|
||||
|
||||
def test_get_program_still_works(
|
||||
def test_get_program_works(
|
||||
self, client, loyalty_store_headers, loyalty_store_setup
|
||||
):
|
||||
"""GET /program still works (read-only)."""
|
||||
"""GET /program returns program."""
|
||||
response = client.get(
|
||||
f"{BASE}/program",
|
||||
headers=loyalty_store_headers,
|
||||
@@ -261,6 +242,19 @@ class TestStoreProgramCrudRemoved:
|
||||
data = response.json()
|
||||
assert data["id"] == loyalty_store_setup["program"].id
|
||||
|
||||
def test_update_program_via_put(
|
||||
self, client, loyalty_store_headers, loyalty_store_setup
|
||||
):
|
||||
"""PUT /program updates the program (merchant_owner)."""
|
||||
response = client.put(
|
||||
f"{BASE}/program",
|
||||
json={"card_name": "Updated Card"},
|
||||
headers=loyalty_store_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["card_name"] == "Updated Card"
|
||||
|
||||
def test_stats_still_works(
|
||||
self, client, loyalty_store_headers, loyalty_store_setup
|
||||
):
|
||||
@@ -270,3 +264,201 @@ class TestStoreProgramCrudRemoved:
|
||||
headers=loyalty_store_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_stats_has_new_fields(
|
||||
self, client, loyalty_store_headers, loyalty_store_setup
|
||||
):
|
||||
"""GET /stats returns all new fields."""
|
||||
response = client.get(
|
||||
f"{BASE}/stats",
|
||||
headers=loyalty_store_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "new_this_month" in data
|
||||
assert "total_points_balance" in data
|
||||
assert "avg_points_per_member" in data
|
||||
assert "transactions_30d" in data
|
||||
assert "points_issued_30d" in data
|
||||
assert "points_redeemed_30d" in data
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Stamp Earn/Redeem Integration Tests
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def stamp_store_setup(db, loyalty_platform):
|
||||
"""Setup with a stamps-type program for integration tests."""
|
||||
from app.modules.customers.models.customer import Customer
|
||||
from app.modules.tenancy.models.store import StoreUser
|
||||
from app.modules.tenancy.models.store_platform import StorePlatform
|
||||
from middleware.auth import AuthManager
|
||||
|
||||
auth = AuthManager()
|
||||
uid = uuid.uuid4().hex[:8]
|
||||
|
||||
owner = User(
|
||||
email=f"stampint_{uid}@test.com",
|
||||
username=f"stampint_{uid}",
|
||||
hashed_password=auth.hash_password("storepass123"),
|
||||
role="merchant_owner",
|
||||
is_active=True,
|
||||
is_email_verified=True,
|
||||
)
|
||||
db.add(owner)
|
||||
db.commit()
|
||||
db.refresh(owner)
|
||||
|
||||
from app.modules.tenancy.models import Merchant, Store
|
||||
|
||||
merchant = Merchant(
|
||||
name=f"Stamp Int Merchant {uid}",
|
||||
owner_user_id=owner.id,
|
||||
contact_email=owner.email,
|
||||
is_active=True,
|
||||
is_verified=True,
|
||||
)
|
||||
db.add(merchant)
|
||||
db.commit()
|
||||
db.refresh(merchant)
|
||||
|
||||
store = Store(
|
||||
merchant_id=merchant.id,
|
||||
store_code=f"STINT_{uid.upper()}",
|
||||
subdomain=f"stint{uid}",
|
||||
name=f"Stamp Int Store {uid}",
|
||||
is_active=True,
|
||||
is_verified=True,
|
||||
)
|
||||
db.add(store)
|
||||
db.commit()
|
||||
db.refresh(store)
|
||||
|
||||
store_user = StoreUser(store_id=store.id, user_id=owner.id, is_active=True)
|
||||
db.add(store_user)
|
||||
db.commit()
|
||||
|
||||
sp = StorePlatform(store_id=store.id, platform_id=loyalty_platform.id)
|
||||
db.add(sp)
|
||||
db.commit()
|
||||
|
||||
customer = Customer(
|
||||
email=f"stampcust_{uid}@test.com",
|
||||
first_name="Stamp",
|
||||
last_name="IntCustomer",
|
||||
hashed_password="!unused!", # noqa: SEC001
|
||||
customer_number=f"SIC-{uid.upper()}",
|
||||
store_id=store.id,
|
||||
is_active=True,
|
||||
)
|
||||
db.add(customer)
|
||||
db.commit()
|
||||
db.refresh(customer)
|
||||
|
||||
program = LoyaltyProgram(
|
||||
merchant_id=merchant.id,
|
||||
loyalty_type=LoyaltyType.HYBRID.value,
|
||||
stamps_target=3,
|
||||
stamps_reward_description="Free item",
|
||||
stamps_reward_value_cents=500,
|
||||
points_per_euro=10,
|
||||
welcome_bonus_points=0,
|
||||
minimum_redemption_points=100,
|
||||
minimum_purchase_cents=0,
|
||||
cooldown_minutes=0,
|
||||
max_daily_stamps=50,
|
||||
require_staff_pin=False,
|
||||
card_name="Int Stamp Card",
|
||||
card_color="#FF0000",
|
||||
is_active=True,
|
||||
points_rewards=[
|
||||
{"id": "r1", "name": "5 EUR off", "points_required": 100, "is_active": True},
|
||||
],
|
||||
)
|
||||
db.add(program)
|
||||
db.commit()
|
||||
db.refresh(program)
|
||||
|
||||
card = LoyaltyCard(
|
||||
merchant_id=merchant.id,
|
||||
program_id=program.id,
|
||||
customer_id=customer.id,
|
||||
enrolled_at_store_id=store.id,
|
||||
card_number=f"SINTCARD-{uid.upper()}",
|
||||
stamp_count=0,
|
||||
total_stamps_earned=0,
|
||||
stamps_redeemed=0,
|
||||
points_balance=0,
|
||||
total_points_earned=0,
|
||||
points_redeemed=0,
|
||||
is_active=True,
|
||||
last_activity_at=datetime.now(UTC),
|
||||
)
|
||||
db.add(card)
|
||||
db.commit()
|
||||
db.refresh(card)
|
||||
|
||||
return {
|
||||
"owner": owner,
|
||||
"merchant": merchant,
|
||||
"store": store,
|
||||
"platform": loyalty_platform,
|
||||
"customer": customer,
|
||||
"program": program,
|
||||
"card": card,
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def stamp_store_headers(client, stamp_store_setup):
|
||||
"""JWT auth headers for stamp store setup."""
|
||||
owner = stamp_store_setup["owner"]
|
||||
response = client.post(
|
||||
"/api/v1/store/auth/login",
|
||||
json={"email_or_username": owner.username, "password": "storepass123"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
token = response.json()["access_token"]
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.loyalty
|
||||
class TestStampEarnRedeem:
|
||||
"""Integration tests for stamp earn/redeem via store API."""
|
||||
|
||||
def test_stamp_earn(self, client, stamp_store_headers, stamp_store_setup):
|
||||
"""POST /stamp adds a stamp."""
|
||||
card = stamp_store_setup["card"]
|
||||
response = client.post(
|
||||
f"{BASE}/stamp",
|
||||
json={"card_id": card.id},
|
||||
headers=stamp_store_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["success"] is True
|
||||
assert data["stamp_count"] == 1
|
||||
|
||||
def test_stamp_redeem(self, client, stamp_store_headers, stamp_store_setup, db):
|
||||
"""POST /stamp/redeem redeems stamps."""
|
||||
card = stamp_store_setup["card"]
|
||||
program = stamp_store_setup["program"]
|
||||
|
||||
# Give enough stamps
|
||||
card.stamp_count = program.stamps_target
|
||||
card.total_stamps_earned = program.stamps_target
|
||||
db.commit()
|
||||
|
||||
response = client.post(
|
||||
f"{BASE}/stamp/redeem",
|
||||
json={"card_id": card.id},
|
||||
headers=stamp_store_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["success"] is True
|
||||
assert data["stamp_count"] == 0
|
||||
|
||||
0
app/modules/loyalty/tests/unit/__init__.py
Normal file
0
app/modules/loyalty/tests/unit/__init__.py
Normal file
@@ -1,8 +1,20 @@
|
||||
"""Unit tests for PinService."""
|
||||
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
|
||||
import pytest
|
||||
|
||||
from app.modules.loyalty.exceptions import (
|
||||
InvalidStaffPinException,
|
||||
StaffPinLockedException,
|
||||
)
|
||||
from app.modules.loyalty.models import LoyaltyProgram, StaffPin
|
||||
from app.modules.loyalty.models.loyalty_program import LoyaltyType
|
||||
from app.modules.loyalty.schemas.pin import PinCreate
|
||||
from app.modules.loyalty.services.pin_service import PinService
|
||||
from app.modules.tenancy.models import Merchant, Store, User
|
||||
from app.modules.tenancy.models.store import StoreUser
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@@ -16,3 +28,208 @@ class TestPinService:
|
||||
def test_service_instantiation(self):
|
||||
"""Service can be instantiated."""
|
||||
assert self.service is not None
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Fixtures
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def pin_setup(db):
|
||||
"""Create a full setup for PIN tests."""
|
||||
from middleware.auth import AuthManager
|
||||
|
||||
auth = AuthManager()
|
||||
uid = uuid.uuid4().hex[:8]
|
||||
|
||||
owner = User(
|
||||
email=f"pinowner_{uid}@test.com",
|
||||
username=f"pinowner_{uid}",
|
||||
hashed_password=auth.hash_password("testpass"),
|
||||
role="merchant_owner",
|
||||
is_active=True,
|
||||
is_email_verified=True,
|
||||
)
|
||||
db.add(owner)
|
||||
db.commit()
|
||||
db.refresh(owner)
|
||||
|
||||
merchant = Merchant(
|
||||
name=f"PIN Merchant {uid}",
|
||||
owner_user_id=owner.id,
|
||||
contact_email=owner.email,
|
||||
is_active=True,
|
||||
is_verified=True,
|
||||
)
|
||||
db.add(merchant)
|
||||
db.commit()
|
||||
db.refresh(merchant)
|
||||
|
||||
store = Store(
|
||||
merchant_id=merchant.id,
|
||||
store_code=f"PIN_{uid.upper()}",
|
||||
subdomain=f"pin{uid}",
|
||||
name=f"PIN Store {uid}",
|
||||
is_active=True,
|
||||
is_verified=True,
|
||||
)
|
||||
db.add(store)
|
||||
db.commit()
|
||||
db.refresh(store)
|
||||
|
||||
store_user = StoreUser(store_id=store.id, user_id=owner.id, is_active=True)
|
||||
db.add(store_user)
|
||||
db.commit()
|
||||
|
||||
program = LoyaltyProgram(
|
||||
merchant_id=merchant.id,
|
||||
loyalty_type=LoyaltyType.POINTS.value,
|
||||
points_per_euro=10,
|
||||
cooldown_minutes=0,
|
||||
max_daily_stamps=10,
|
||||
require_staff_pin=True,
|
||||
card_name="PIN Card",
|
||||
card_color="#00FF00",
|
||||
is_active=True,
|
||||
)
|
||||
db.add(program)
|
||||
db.commit()
|
||||
db.refresh(program)
|
||||
|
||||
return {
|
||||
"merchant": merchant,
|
||||
"store": store,
|
||||
"program": program,
|
||||
}
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Create / Unlock Tests
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.loyalty
|
||||
class TestCreatePin:
|
||||
"""Tests for create_pin."""
|
||||
|
||||
def setup_method(self):
|
||||
self.service = PinService()
|
||||
|
||||
def test_create_pin(self, db, pin_setup):
|
||||
"""Create a staff PIN."""
|
||||
program = pin_setup["program"]
|
||||
store = pin_setup["store"]
|
||||
|
||||
data = PinCreate(name="Alice", staff_id="EMP001", pin="1234")
|
||||
pin = self.service.create_pin(db, program.id, store.id, data)
|
||||
|
||||
assert pin.id is not None
|
||||
assert pin.name == "Alice"
|
||||
assert pin.staff_id == "EMP001"
|
||||
assert pin.verify_pin("1234")
|
||||
|
||||
def test_unlock_pin(self, db, pin_setup):
|
||||
"""Unlock a locked PIN."""
|
||||
program = pin_setup["program"]
|
||||
store = pin_setup["store"]
|
||||
|
||||
data = PinCreate(name="Bob", staff_id="EMP002", pin="5678")
|
||||
pin = self.service.create_pin(db, program.id, store.id, data)
|
||||
|
||||
# Lock it
|
||||
pin.failed_attempts = 5
|
||||
from datetime import timedelta
|
||||
pin.locked_until = datetime.now(UTC) + timedelta(minutes=30)
|
||||
db.commit()
|
||||
|
||||
assert pin.is_locked
|
||||
|
||||
# Unlock
|
||||
unlocked = self.service.unlock_pin(db, pin.id)
|
||||
assert unlocked.failed_attempts == 0
|
||||
assert unlocked.locked_until is None
|
||||
assert not unlocked.is_locked
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Verify PIN Tests
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.loyalty
|
||||
class TestVerifyPin:
|
||||
"""Tests for verify_pin."""
|
||||
|
||||
def setup_method(self):
|
||||
self.service = PinService()
|
||||
|
||||
def test_verify_pin_success(self, db, pin_setup):
|
||||
"""Correct PIN verifies successfully."""
|
||||
program = pin_setup["program"]
|
||||
store = pin_setup["store"]
|
||||
|
||||
data = PinCreate(name="Charlie", staff_id="EMP003", pin="1111")
|
||||
self.service.create_pin(db, program.id, store.id, data)
|
||||
|
||||
result = self.service.verify_pin(db, program.id, "1111", store_id=store.id)
|
||||
assert result.name == "Charlie"
|
||||
|
||||
def test_verify_pin_wrong_single_failure(self, db, pin_setup):
|
||||
"""Wrong PIN records failure on ONE pin only, not all."""
|
||||
program = pin_setup["program"]
|
||||
store = pin_setup["store"]
|
||||
|
||||
# Create two PINs
|
||||
self.service.create_pin(db, program.id, store.id, PinCreate(name="A", pin="1111"))
|
||||
self.service.create_pin(db, program.id, store.id, PinCreate(name="B", pin="2222"))
|
||||
|
||||
# Wrong PIN
|
||||
with pytest.raises(InvalidStaffPinException):
|
||||
self.service.verify_pin(db, program.id, "9999", store_id=store.id)
|
||||
|
||||
# Only one PIN should have failed_attempts incremented
|
||||
pins = self.service.list_pins(db, program.id, store_id=store.id, is_active=True)
|
||||
failed_counts = [p.failed_attempts for p in pins]
|
||||
assert sum(failed_counts) == 1 # Only 1 PIN got the failure, not both
|
||||
|
||||
def test_verify_pin_lockout(self, db, pin_setup):
|
||||
"""After max failures, PIN gets locked."""
|
||||
program = pin_setup["program"]
|
||||
store = pin_setup["store"]
|
||||
|
||||
self.service.create_pin(db, program.id, store.id, PinCreate(name="Lock", pin="3333"))
|
||||
|
||||
# Fail 5 times (default max)
|
||||
for _ in range(5):
|
||||
try:
|
||||
self.service.verify_pin(db, program.id, "9999", store_id=store.id)
|
||||
except (InvalidStaffPinException, StaffPinLockedException):
|
||||
pass
|
||||
|
||||
# Next attempt should be locked
|
||||
with pytest.raises((InvalidStaffPinException, StaffPinLockedException)):
|
||||
self.service.verify_pin(db, program.id, "9999", store_id=store.id)
|
||||
|
||||
def test_verify_skips_locked_pins(self, db, pin_setup):
|
||||
"""Locked PINs are skipped during verification."""
|
||||
program = pin_setup["program"]
|
||||
store = pin_setup["store"]
|
||||
|
||||
from datetime import timedelta
|
||||
|
||||
# Create a locked PIN and an unlocked one
|
||||
data1 = PinCreate(name="Locked", pin="1111")
|
||||
pin1 = self.service.create_pin(db, program.id, store.id, data1)
|
||||
pin1.locked_until = datetime.now(UTC) + timedelta(minutes=30)
|
||||
pin1.failed_attempts = 5
|
||||
db.commit()
|
||||
|
||||
data2 = PinCreate(name="Active", pin="2222")
|
||||
self.service.create_pin(db, program.id, store.id, data2)
|
||||
|
||||
# Should find the active PIN
|
||||
result = self.service.verify_pin(db, program.id, "2222", store_id=store.id)
|
||||
assert result.name == "Active"
|
||||
|
||||
@@ -1,8 +1,20 @@
|
||||
"""Unit tests for PointsService."""
|
||||
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
|
||||
import pytest
|
||||
|
||||
from app.modules.loyalty.exceptions import (
|
||||
InsufficientPointsException,
|
||||
InvalidRewardException,
|
||||
)
|
||||
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.points_service import PointsService
|
||||
from app.modules.tenancy.models import Merchant, Store, User
|
||||
from app.modules.tenancy.models.store import StoreUser
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@@ -16,3 +28,347 @@ class TestPointsService:
|
||||
def test_service_instantiation(self):
|
||||
"""Service can be instantiated."""
|
||||
assert self.service is not None
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Fixtures
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def points_setup(db):
|
||||
"""Create a full setup for points tests."""
|
||||
from app.modules.customers.models.customer import Customer
|
||||
from middleware.auth import AuthManager
|
||||
|
||||
auth = AuthManager()
|
||||
uid = uuid.uuid4().hex[:8]
|
||||
|
||||
owner = User(
|
||||
email=f"ptsowner_{uid}@test.com",
|
||||
username=f"ptsowner_{uid}",
|
||||
hashed_password=auth.hash_password("testpass"),
|
||||
role="merchant_owner",
|
||||
is_active=True,
|
||||
is_email_verified=True,
|
||||
)
|
||||
db.add(owner)
|
||||
db.commit()
|
||||
db.refresh(owner)
|
||||
|
||||
merchant = Merchant(
|
||||
name=f"Points Merchant {uid}",
|
||||
owner_user_id=owner.id,
|
||||
contact_email=owner.email,
|
||||
is_active=True,
|
||||
is_verified=True,
|
||||
)
|
||||
db.add(merchant)
|
||||
db.commit()
|
||||
db.refresh(merchant)
|
||||
|
||||
store = Store(
|
||||
merchant_id=merchant.id,
|
||||
store_code=f"PTS_{uid.upper()}",
|
||||
subdomain=f"pts{uid}",
|
||||
name=f"Points Store {uid}",
|
||||
is_active=True,
|
||||
is_verified=True,
|
||||
)
|
||||
db.add(store)
|
||||
db.commit()
|
||||
db.refresh(store)
|
||||
|
||||
store_user = StoreUser(store_id=store.id, user_id=owner.id, is_active=True)
|
||||
db.add(store_user)
|
||||
db.commit()
|
||||
|
||||
customer = Customer(
|
||||
email=f"ptscust_{uid}@test.com",
|
||||
first_name="Points",
|
||||
last_name="Customer",
|
||||
hashed_password="!unused!", # noqa: SEC001
|
||||
customer_number=f"PC-{uid.upper()}",
|
||||
store_id=store.id,
|
||||
is_active=True,
|
||||
)
|
||||
db.add(customer)
|
||||
db.commit()
|
||||
db.refresh(customer)
|
||||
|
||||
program = LoyaltyProgram(
|
||||
merchant_id=merchant.id,
|
||||
loyalty_type=LoyaltyType.POINTS.value,
|
||||
points_per_euro=10,
|
||||
welcome_bonus_points=0,
|
||||
minimum_redemption_points=50,
|
||||
minimum_purchase_cents=100,
|
||||
cooldown_minutes=0,
|
||||
max_daily_stamps=10,
|
||||
require_staff_pin=False,
|
||||
card_name="Points Card",
|
||||
card_color="#0000FF",
|
||||
is_active=True,
|
||||
points_rewards=[
|
||||
{"id": "r1", "name": "5 EUR off", "points_required": 100, "is_active": True},
|
||||
{"id": "r2", "name": "10 EUR off", "points_required": 200, "is_active": True},
|
||||
{"id": "r3", "name": "Inactive Reward", "points_required": 50, "is_active": False},
|
||||
],
|
||||
)
|
||||
db.add(program)
|
||||
db.commit()
|
||||
db.refresh(program)
|
||||
|
||||
card = LoyaltyCard(
|
||||
merchant_id=merchant.id,
|
||||
program_id=program.id,
|
||||
customer_id=customer.id,
|
||||
enrolled_at_store_id=store.id,
|
||||
card_number=f"PTSCARD-{uid.upper()}",
|
||||
stamp_count=0,
|
||||
total_stamps_earned=0,
|
||||
stamps_redeemed=0,
|
||||
points_balance=500,
|
||||
total_points_earned=500,
|
||||
points_redeemed=0,
|
||||
is_active=True,
|
||||
last_activity_at=datetime.now(UTC),
|
||||
)
|
||||
db.add(card)
|
||||
db.commit()
|
||||
db.refresh(card)
|
||||
|
||||
return {
|
||||
"merchant": merchant,
|
||||
"store": store,
|
||||
"customer": customer,
|
||||
"program": program,
|
||||
"card": card,
|
||||
}
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Earn Points Tests
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.loyalty
|
||||
class TestEarnPoints:
|
||||
"""Tests for earn_points."""
|
||||
|
||||
def setup_method(self):
|
||||
self.service = PointsService()
|
||||
|
||||
def test_earn_points_calculation(self, db, points_setup):
|
||||
"""Points calculated correctly from purchase amount."""
|
||||
card = points_setup["card"]
|
||||
store = points_setup["store"]
|
||||
|
||||
result = self.service.earn_points(
|
||||
db,
|
||||
store_id=store.id,
|
||||
card_id=card.id,
|
||||
purchase_amount_cents=2000, # 20 EUR
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["points_earned"] == 200 # 20 EUR * 10 pts/EUR
|
||||
assert result["points_balance"] == 700 # 500 + 200
|
||||
|
||||
def test_earn_points_minimum_purchase(self, db, points_setup):
|
||||
"""Below minimum purchase returns 0 points."""
|
||||
card = points_setup["card"]
|
||||
store = points_setup["store"]
|
||||
|
||||
result = self.service.earn_points(
|
||||
db,
|
||||
store_id=store.id,
|
||||
card_id=card.id,
|
||||
purchase_amount_cents=50, # Below min of 100
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["points_earned"] == 0
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Redeem Points Tests
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.loyalty
|
||||
class TestRedeemPoints:
|
||||
"""Tests for redeem_points."""
|
||||
|
||||
def setup_method(self):
|
||||
self.service = PointsService()
|
||||
|
||||
def test_redeem_points_success(self, db, points_setup):
|
||||
"""Successfully redeem points for a reward."""
|
||||
card = points_setup["card"]
|
||||
store = points_setup["store"]
|
||||
|
||||
result = self.service.redeem_points(
|
||||
db,
|
||||
store_id=store.id,
|
||||
card_id=card.id,
|
||||
reward_id="r1",
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["points_spent"] == 100
|
||||
assert result["points_balance"] == 400 # 500 - 100
|
||||
|
||||
def test_redeem_points_insufficient(self, db, points_setup):
|
||||
"""Redeeming without enough points raises exception."""
|
||||
card = points_setup["card"]
|
||||
store = points_setup["store"]
|
||||
|
||||
# Set balance low
|
||||
card.points_balance = 50
|
||||
db.commit()
|
||||
|
||||
with pytest.raises(InsufficientPointsException):
|
||||
self.service.redeem_points(
|
||||
db,
|
||||
store_id=store.id,
|
||||
card_id=card.id,
|
||||
reward_id="r1", # needs 100
|
||||
)
|
||||
|
||||
def test_redeem_inactive_reward(self, db, points_setup):
|
||||
"""Redeeming an inactive reward raises exception."""
|
||||
card = points_setup["card"]
|
||||
store = points_setup["store"]
|
||||
|
||||
with pytest.raises(InvalidRewardException):
|
||||
self.service.redeem_points(
|
||||
db,
|
||||
store_id=store.id,
|
||||
card_id=card.id,
|
||||
reward_id="r3", # inactive
|
||||
)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Void Points Tests
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.loyalty
|
||||
class TestVoidPoints:
|
||||
"""Tests for void_points."""
|
||||
|
||||
def setup_method(self):
|
||||
self.service = PointsService()
|
||||
|
||||
def test_void_by_transaction(self, db, points_setup):
|
||||
"""Void points by original transaction ID."""
|
||||
card = points_setup["card"]
|
||||
store = points_setup["store"]
|
||||
|
||||
# Earn some points
|
||||
earn_result = self.service.earn_points(
|
||||
db, store_id=store.id, card_id=card.id, purchase_amount_cents=1000,
|
||||
)
|
||||
assert earn_result["points_earned"] == 100
|
||||
|
||||
# Find the earn transaction
|
||||
tx = (
|
||||
db.query(LoyaltyTransaction)
|
||||
.filter(
|
||||
LoyaltyTransaction.card_id == card.id,
|
||||
LoyaltyTransaction.transaction_type == TransactionType.POINTS_EARNED.value,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
void_result = self.service.void_points(
|
||||
db, store_id=store.id, card_id=card.id, original_transaction_id=tx.id,
|
||||
)
|
||||
|
||||
assert void_result["success"] is True
|
||||
assert void_result["points_voided"] == 100
|
||||
|
||||
def test_void_by_order_reference(self, db, points_setup):
|
||||
"""Void points by order reference."""
|
||||
card = points_setup["card"]
|
||||
store = points_setup["store"]
|
||||
|
||||
# Earn with order reference
|
||||
self.service.earn_points(
|
||||
db,
|
||||
store_id=store.id,
|
||||
card_id=card.id,
|
||||
purchase_amount_cents=2000,
|
||||
order_reference="ORDER-VOID-TEST",
|
||||
)
|
||||
|
||||
void_result = self.service.void_points(
|
||||
db,
|
||||
store_id=store.id,
|
||||
card_id=card.id,
|
||||
order_reference="ORDER-VOID-TEST",
|
||||
)
|
||||
|
||||
assert void_result["success"] is True
|
||||
assert void_result["points_voided"] == 200
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Adjust Points Tests
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.loyalty
|
||||
class TestAdjustPoints:
|
||||
"""Tests for adjust_points."""
|
||||
|
||||
def setup_method(self):
|
||||
self.service = PointsService()
|
||||
|
||||
def test_adjust_positive(self, db, points_setup):
|
||||
"""Add points via adjustment."""
|
||||
card = points_setup["card"]
|
||||
|
||||
result = self.service.adjust_points(
|
||||
db,
|
||||
card_id=card.id,
|
||||
points_delta=50,
|
||||
reason="Goodwill bonus",
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["points_balance"] == 550 # 500 + 50
|
||||
|
||||
def test_adjust_negative(self, db, points_setup):
|
||||
"""Remove points via adjustment."""
|
||||
card = points_setup["card"]
|
||||
|
||||
result = self.service.adjust_points(
|
||||
db,
|
||||
card_id=card.id,
|
||||
points_delta=-100,
|
||||
reason="Correction",
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["points_balance"] == 400 # 500 - 100
|
||||
|
||||
def test_adjust_floor_at_zero(self, db, points_setup):
|
||||
"""Negative adjustment doesn't go below zero."""
|
||||
card = points_setup["card"]
|
||||
|
||||
result = self.service.adjust_points(
|
||||
db,
|
||||
card_id=card.id,
|
||||
points_delta=-9999,
|
||||
reason="Full correction",
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["points_balance"] == 0
|
||||
|
||||
@@ -352,3 +352,98 @@ class TestDeleteProgram:
|
||||
"""Deleting non-existent program raises exception."""
|
||||
with pytest.raises(LoyaltyProgramNotFoundException):
|
||||
self.service.delete_program(db, 999999)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Stats
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.loyalty
|
||||
class TestGetProgramStats:
|
||||
"""Tests for get_program_stats."""
|
||||
|
||||
def setup_method(self):
|
||||
self.service = ProgramService()
|
||||
|
||||
def test_stats_returns_all_fields(self, db, ps_program):
|
||||
"""Stats response includes all required fields."""
|
||||
stats = self.service.get_program_stats(db, ps_program.id)
|
||||
|
||||
assert "total_cards" in stats
|
||||
assert "active_cards" in stats
|
||||
assert "new_this_month" in stats
|
||||
assert "total_points_balance" in stats
|
||||
assert "avg_points_per_member" in stats
|
||||
assert "transactions_30d" in stats
|
||||
assert "points_issued_30d" in stats
|
||||
assert "points_redeemed_30d" in stats
|
||||
assert "points_this_month" in stats
|
||||
assert "points_redeemed_this_month" in stats
|
||||
assert "estimated_liability_cents" in stats
|
||||
|
||||
def test_stats_empty_program(self, db, ps_program):
|
||||
"""Stats for program with no cards."""
|
||||
stats = self.service.get_program_stats(db, ps_program.id)
|
||||
|
||||
assert stats["total_cards"] == 0
|
||||
assert stats["active_cards"] == 0
|
||||
assert stats["new_this_month"] == 0
|
||||
assert stats["total_points_balance"] == 0
|
||||
assert stats["avg_points_per_member"] == 0
|
||||
|
||||
def test_stats_with_cards(self, db, ps_program, ps_merchant):
|
||||
"""Stats reflect actual card data."""
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from app.modules.customers.models.customer import Customer
|
||||
from app.modules.loyalty.models import LoyaltyCard
|
||||
from app.modules.tenancy.models import Store
|
||||
|
||||
uid_store = uuid.uuid4().hex[:8]
|
||||
store = Store(
|
||||
merchant_id=ps_merchant.id,
|
||||
store_code=f"STAT_{uid_store.upper()}",
|
||||
subdomain=f"stat{uid_store}",
|
||||
name=f"Stats Store {uid_store}",
|
||||
is_active=True,
|
||||
is_verified=True,
|
||||
)
|
||||
db.add(store)
|
||||
db.flush()
|
||||
|
||||
# Create cards with customers
|
||||
for i in range(3):
|
||||
uid = uuid.uuid4().hex[:8]
|
||||
customer = Customer(
|
||||
email=f"stat_{uid}@test.com",
|
||||
first_name="Stat",
|
||||
last_name=f"Customer{i}",
|
||||
hashed_password="!unused!", # noqa: SEC001
|
||||
customer_number=f"SC-{uid.upper()}",
|
||||
store_id=store.id,
|
||||
is_active=True,
|
||||
)
|
||||
db.add(customer)
|
||||
db.flush()
|
||||
|
||||
card = LoyaltyCard(
|
||||
merchant_id=ps_merchant.id,
|
||||
program_id=ps_program.id,
|
||||
customer_id=customer.id,
|
||||
card_number=f"STAT-{i}-{uuid.uuid4().hex[:6]}",
|
||||
points_balance=100 * (i + 1),
|
||||
total_points_earned=100 * (i + 1),
|
||||
is_active=True,
|
||||
last_activity_at=datetime.now(UTC),
|
||||
)
|
||||
db.add(card)
|
||||
db.commit()
|
||||
|
||||
stats = self.service.get_program_stats(db, ps_program.id)
|
||||
|
||||
assert stats["total_cards"] == 3
|
||||
assert stats["active_cards"] == 3
|
||||
assert stats["total_points_balance"] == 600 # 100+200+300
|
||||
assert stats["avg_points_per_member"] == 200.0 # 600/3
|
||||
|
||||
@@ -1,8 +1,21 @@
|
||||
"""Unit tests for StampService."""
|
||||
|
||||
import uuid
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
import pytest
|
||||
|
||||
from app.modules.loyalty.exceptions import (
|
||||
DailyStampLimitException,
|
||||
InsufficientStampsException,
|
||||
StampCooldownException,
|
||||
)
|
||||
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.stamp_service import StampService
|
||||
from app.modules.tenancy.models import Merchant, Store, User
|
||||
from app.modules.tenancy.models.store import StoreUser
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@@ -16,3 +29,259 @@ class TestStampService:
|
||||
def test_service_instantiation(self):
|
||||
"""Service can be instantiated."""
|
||||
assert self.service is not None
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Fixtures
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def stamp_setup(db):
|
||||
"""Create a full setup for stamp tests with stamps-type program."""
|
||||
from app.modules.customers.models.customer import Customer
|
||||
from middleware.auth import AuthManager
|
||||
|
||||
auth = AuthManager()
|
||||
uid = uuid.uuid4().hex[:8]
|
||||
|
||||
owner = User(
|
||||
email=f"stampowner_{uid}@test.com",
|
||||
username=f"stampowner_{uid}",
|
||||
hashed_password=auth.hash_password("testpass"),
|
||||
role="merchant_owner",
|
||||
is_active=True,
|
||||
is_email_verified=True,
|
||||
)
|
||||
db.add(owner)
|
||||
db.commit()
|
||||
db.refresh(owner)
|
||||
|
||||
merchant = Merchant(
|
||||
name=f"Stamp Merchant {uid}",
|
||||
owner_user_id=owner.id,
|
||||
contact_email=owner.email,
|
||||
is_active=True,
|
||||
is_verified=True,
|
||||
)
|
||||
db.add(merchant)
|
||||
db.commit()
|
||||
db.refresh(merchant)
|
||||
|
||||
store = Store(
|
||||
merchant_id=merchant.id,
|
||||
store_code=f"STAMP_{uid.upper()}",
|
||||
subdomain=f"stamp{uid}",
|
||||
name=f"Stamp Store {uid}",
|
||||
is_active=True,
|
||||
is_verified=True,
|
||||
)
|
||||
db.add(store)
|
||||
db.commit()
|
||||
db.refresh(store)
|
||||
|
||||
store_user = StoreUser(store_id=store.id, user_id=owner.id, is_active=True)
|
||||
db.add(store_user)
|
||||
db.commit()
|
||||
|
||||
customer = Customer(
|
||||
email=f"stampcust_{uid}@test.com",
|
||||
first_name="Stamp",
|
||||
last_name="Customer",
|
||||
hashed_password="!unused!", # noqa: SEC001
|
||||
customer_number=f"SC-{uid.upper()}",
|
||||
store_id=store.id,
|
||||
is_active=True,
|
||||
)
|
||||
db.add(customer)
|
||||
db.commit()
|
||||
db.refresh(customer)
|
||||
|
||||
program = LoyaltyProgram(
|
||||
merchant_id=merchant.id,
|
||||
loyalty_type=LoyaltyType.STAMPS.value,
|
||||
stamps_target=5,
|
||||
stamps_reward_description="Free coffee",
|
||||
stamps_reward_value_cents=500,
|
||||
cooldown_minutes=0,
|
||||
max_daily_stamps=10,
|
||||
require_staff_pin=False,
|
||||
card_name="Stamp Card",
|
||||
card_color="#FF0000",
|
||||
is_active=True,
|
||||
points_per_euro=1,
|
||||
)
|
||||
db.add(program)
|
||||
db.commit()
|
||||
db.refresh(program)
|
||||
|
||||
card = LoyaltyCard(
|
||||
merchant_id=merchant.id,
|
||||
program_id=program.id,
|
||||
customer_id=customer.id,
|
||||
enrolled_at_store_id=store.id,
|
||||
card_number=f"STAMPCARD-{uid.upper()}",
|
||||
stamp_count=0,
|
||||
total_stamps_earned=0,
|
||||
stamps_redeemed=0,
|
||||
points_balance=0,
|
||||
total_points_earned=0,
|
||||
points_redeemed=0,
|
||||
is_active=True,
|
||||
last_activity_at=datetime.now(UTC),
|
||||
)
|
||||
db.add(card)
|
||||
db.commit()
|
||||
db.refresh(card)
|
||||
|
||||
return {
|
||||
"merchant": merchant,
|
||||
"store": store,
|
||||
"customer": customer,
|
||||
"program": program,
|
||||
"card": card,
|
||||
}
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Add Stamp Tests
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.loyalty
|
||||
class TestAddStamp:
|
||||
"""Tests for add_stamp."""
|
||||
|
||||
def setup_method(self):
|
||||
self.service = StampService()
|
||||
|
||||
def test_add_stamp_success(self, db, stamp_setup):
|
||||
"""Successfully add a stamp to a card."""
|
||||
card = stamp_setup["card"]
|
||||
store = stamp_setup["store"]
|
||||
|
||||
result = self.service.add_stamp(db, store_id=store.id, card_id=card.id)
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["stamp_count"] == 1
|
||||
assert result["stamps_target"] == 5
|
||||
assert result["stamps_until_reward"] == 4
|
||||
|
||||
def test_add_stamp_cooldown_violation(self, db, stamp_setup):
|
||||
"""Stamp within cooldown period raises exception."""
|
||||
card = stamp_setup["card"]
|
||||
store = stamp_setup["store"]
|
||||
program = stamp_setup["program"]
|
||||
|
||||
# Set cooldown
|
||||
program.cooldown_minutes = 15
|
||||
db.commit()
|
||||
|
||||
# Add first stamp
|
||||
self.service.add_stamp(db, store_id=store.id, card_id=card.id)
|
||||
|
||||
# Second stamp should fail (cooldown)
|
||||
with pytest.raises(StampCooldownException):
|
||||
self.service.add_stamp(db, store_id=store.id, card_id=card.id)
|
||||
|
||||
def test_add_stamp_daily_limit(self, db, stamp_setup):
|
||||
"""Exceeding daily stamp limit raises exception."""
|
||||
card = stamp_setup["card"]
|
||||
store = stamp_setup["store"]
|
||||
program = stamp_setup["program"]
|
||||
|
||||
# Set max 2 daily stamps
|
||||
program.max_daily_stamps = 2
|
||||
db.commit()
|
||||
|
||||
# Add 2 stamps
|
||||
self.service.add_stamp(db, store_id=store.id, card_id=card.id)
|
||||
self.service.add_stamp(db, store_id=store.id, card_id=card.id)
|
||||
|
||||
# Third should fail
|
||||
with pytest.raises(DailyStampLimitException):
|
||||
self.service.add_stamp(db, store_id=store.id, card_id=card.id)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Redeem Stamps Tests
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.loyalty
|
||||
class TestRedeemStamps:
|
||||
"""Tests for redeem_stamps."""
|
||||
|
||||
def setup_method(self):
|
||||
self.service = StampService()
|
||||
|
||||
def test_redeem_stamps_success(self, db, stamp_setup):
|
||||
"""Successfully redeem stamps for a reward."""
|
||||
card = stamp_setup["card"]
|
||||
store = stamp_setup["store"]
|
||||
program = stamp_setup["program"]
|
||||
|
||||
# Give enough stamps
|
||||
card.stamp_count = program.stamps_target
|
||||
card.total_stamps_earned = program.stamps_target
|
||||
db.commit()
|
||||
|
||||
result = self.service.redeem_stamps(db, store_id=store.id, card_id=card.id)
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["stamp_count"] == 0
|
||||
assert result["reward_description"] == "Free coffee"
|
||||
|
||||
def test_redeem_stamps_insufficient(self, db, stamp_setup):
|
||||
"""Redeeming without enough stamps raises exception."""
|
||||
card = stamp_setup["card"]
|
||||
store = stamp_setup["store"]
|
||||
|
||||
# Card has 0 stamps, needs 5
|
||||
with pytest.raises(InsufficientStampsException):
|
||||
self.service.redeem_stamps(db, store_id=store.id, card_id=card.id)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Void Stamps Tests
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.loyalty
|
||||
class TestVoidStamps:
|
||||
"""Tests for void_stamps."""
|
||||
|
||||
def setup_method(self):
|
||||
self.service = StampService()
|
||||
|
||||
def test_void_stamps_by_transaction(self, db, stamp_setup):
|
||||
"""Void stamps by original transaction ID."""
|
||||
card = stamp_setup["card"]
|
||||
store = stamp_setup["store"]
|
||||
|
||||
# Add a stamp first
|
||||
result = self.service.add_stamp(db, store_id=store.id, card_id=card.id)
|
||||
assert result["stamp_count"] == 1
|
||||
|
||||
# Find the transaction
|
||||
tx = (
|
||||
db.query(LoyaltyTransaction)
|
||||
.filter(
|
||||
LoyaltyTransaction.card_id == card.id,
|
||||
LoyaltyTransaction.transaction_type == TransactionType.STAMP_EARNED.value,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
void_result = self.service.void_stamps(
|
||||
db,
|
||||
store_id=store.id,
|
||||
card_id=card.id,
|
||||
original_transaction_id=tx.id,
|
||||
)
|
||||
|
||||
assert void_result["success"] is True
|
||||
assert void_result["stamp_count"] == 0
|
||||
|
||||
Reference in New Issue
Block a user