test: add loyalty module tests for today's bug fixes
Some checks failed
CI / ruff (push) Successful in 11s
CI / dependency-scanning (push) Successful in 28s
CI / docs (push) Has been skipped
CI / pytest (push) Failing after 46m26s
CI / validate (push) Successful in 23s
CI / deploy (push) Has been skipped

Covers card lookup route ordering, func.replace normalization, customer
name in transactions, self-enrollment creation, and earn points endpoint.
54 tests total (was 1).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-24 23:28:22 +01:00
parent 92a434530f
commit 3df75e2e78
4 changed files with 745 additions and 2 deletions

View File

@@ -1,3 +1,184 @@
# app/modules/loyalty/tests/conftest.py # app/modules/loyalty/tests/conftest.py
# Module-specific fixtures for loyalty tests. """
# Core fixtures (db, client, etc.) are inherited from the root conftest.py. Module-specific fixtures for loyalty tests.
Core fixtures (db, client, etc.) are inherited from the root conftest.py.
"""
import uuid
from datetime import UTC, datetime
import pytest
from app.modules.loyalty.models import LoyaltyCard, LoyaltyProgram
from app.modules.loyalty.models.loyalty_program import LoyaltyType
from app.modules.tenancy.models import Merchant, Platform, Store, User
from app.modules.tenancy.models.store import StoreUser
from app.modules.tenancy.models.store_platform import StorePlatform
@pytest.fixture
def loyalty_platform(db):
"""Create a platform for loyalty store tests."""
platform = Platform(
code=f"loyalty_{uuid.uuid4().hex[:8]}",
name="Loyalty Test Platform",
is_active=True,
)
db.add(platform)
db.commit()
db.refresh(platform)
return platform
@pytest.fixture
def loyalty_store_setup(db, loyalty_platform):
"""
Full store setup for loyalty integration tests.
Creates: User -> Merchant -> Store -> StoreUser -> StorePlatform
+ LoyaltyProgram + LoyaltyCard (with customer)
Returns dict with all objects for easy access.
"""
from app.modules.customers.models.customer import Customer
from middleware.auth import AuthManager
auth = AuthManager()
uid = uuid.uuid4().hex[:8]
# Owner user
owner = User(
email=f"loyaltyowner_{uid}@test.com",
username=f"loyaltyowner_{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)
# Merchant
merchant = Merchant(
name=f"Loyalty 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 = Store(
merchant_id=merchant.id,
store_code=f"LOYTEST_{uid.upper()}",
subdomain=f"loytest{uid}",
name=f"Loyalty Test Store {uid}",
is_active=True,
is_verified=True,
)
db.add(store)
db.commit()
db.refresh(store)
# StoreUser
store_user = StoreUser(
store_id=store.id,
user_id=owner.id,
is_active=True,
)
db.add(store_user)
db.commit()
# StorePlatform
sp = StorePlatform(
store_id=store.id,
platform_id=loyalty_platform.id,
)
db.add(sp)
db.commit()
# Customer
customer = Customer(
email=f"customer_{uid}@test.com",
first_name="Test",
last_name="Customer",
hashed_password="!test!unused", # noqa: SEC-001
customer_number=f"CUST-{uid.upper()}",
store_id=store.id,
is_active=True,
)
db.add(customer)
db.commit()
db.refresh(customer)
# Loyalty Program
program = LoyaltyProgram(
merchant_id=merchant.id,
loyalty_type=LoyaltyType.POINTS.value,
points_per_euro=10,
welcome_bonus_points=50,
minimum_redemption_points=100,
minimum_purchase_cents=0,
points_expiration_days=365,
cooldown_minutes=0,
max_daily_stamps=10,
require_staff_pin=False,
card_name="Test Rewards",
card_color="#4F46E5",
is_active=True,
points_rewards=[
{"id": "reward_1", "name": "€5 off", "points_required": 100, "is_active": True},
],
)
db.add(program)
db.commit()
db.refresh(program)
# Loyalty Card
card = LoyaltyCard(
merchant_id=merchant.id,
program_id=program.id,
customer_id=customer.id,
enrolled_at_store_id=store.id,
card_number=f"TESTCARD-{uid.upper()}",
points_balance=100,
total_points_earned=150,
points_redeemed=50,
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 loyalty_store_headers(client, loyalty_store_setup):
"""
Get real JWT auth headers by logging in via store auth endpoint.
"""
owner = loyalty_store_setup["owner"]
response = client.post(
"/api/v1/store/auth/login",
json={
"email_or_username": owner.username,
"password": "storepass123",
},
)
assert response.status_code == 200, f"Store login failed: {response.text}"
token = response.json()["access_token"]
return {"Authorization": f"Bearer {token}"}

View File

@@ -0,0 +1,215 @@
# app/modules/loyalty/tests/integration/test_store_api.py
"""
Integration tests for store loyalty API endpoints.
Tests the endpoints fixed today:
- GET /cards/lookup (route ordering fix)
- GET /cards/{card_id} (card detail)
- GET /transactions (customer_name in response)
- POST /points/earn (endpoint path rename)
Authentication: Uses real JWT tokens via store login endpoint.
"""
from datetime import UTC, datetime
import pytest
from app.modules.loyalty.models import LoyaltyTransaction
from app.modules.loyalty.models.loyalty_transaction import TransactionType
BASE = "/api/v1/store/loyalty"
# ============================================================================
# Card Lookup Tests
# ============================================================================
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.loyalty
class TestCardLookup:
"""Tests for GET /cards/lookup endpoint."""
def test_lookup_by_card_number(
self, client, loyalty_store_headers, loyalty_store_setup
):
"""Find card by card number."""
card = loyalty_store_setup["card"]
response = client.get(
f"{BASE}/cards/lookup",
params={"q": card.card_number},
headers=loyalty_store_headers,
)
assert response.status_code == 200
data = response.json()
assert data["card_number"] == card.card_number
def test_lookup_by_email(
self, client, loyalty_store_headers, loyalty_store_setup
):
"""Find card by customer email."""
customer = loyalty_store_setup["customer"]
response = client.get(
f"{BASE}/cards/lookup",
params={"q": customer.email},
headers=loyalty_store_headers,
)
assert response.status_code == 200
data = response.json()
assert data["customer_email"] == customer.email
def test_lookup_not_found(self, client, loyalty_store_headers):
"""Non-existent search returns 404."""
response = client.get(
f"{BASE}/cards/lookup",
params={"q": "nonexistent@nowhere.com"},
headers=loyalty_store_headers,
)
assert response.status_code == 404
def test_lookup_route_not_captured_by_card_id(
self, client, loyalty_store_headers
):
"""GET /cards/lookup is NOT matched by /cards/{card_id} (route ordering fix)."""
response = client.get(
f"{BASE}/cards/lookup",
params={"q": "test"},
headers=loyalty_store_headers,
)
# Should be 404 (not found) not 422 (validation error from parsing "lookup" as int)
assert response.status_code != 422
# ============================================================================
# Card Detail Tests
# ============================================================================
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.loyalty
class TestCardDetail:
"""Tests for GET /cards/{card_id} endpoint."""
def test_get_card_detail(
self, client, loyalty_store_headers, loyalty_store_setup
):
"""Card detail includes customer_name."""
card = loyalty_store_setup["card"]
customer = loyalty_store_setup["customer"]
response = client.get(
f"{BASE}/cards/{card.id}",
headers=loyalty_store_headers,
)
assert response.status_code == 200
data = response.json()
assert data["card_number"] == card.card_number
assert data["customer_name"] == f"{customer.first_name} {customer.last_name}"
def test_get_card_not_found(self, client, loyalty_store_headers):
"""Non-existent card returns 404."""
response = client.get(
f"{BASE}/cards/999999",
headers=loyalty_store_headers,
)
assert response.status_code == 404
# ============================================================================
# Store Transactions Tests
# ============================================================================
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.loyalty
class TestStoreTransactions:
"""Tests for GET /transactions endpoint."""
def test_list_transactions(
self, client, loyalty_store_headers, loyalty_store_setup, db
):
"""Transactions endpoint returns data."""
card = loyalty_store_setup["card"]
store = loyalty_store_setup["store"]
# Create a transaction
tx = LoyaltyTransaction(
merchant_id=card.merchant_id,
card_id=card.id,
store_id=store.id,
transaction_type=TransactionType.POINTS_EARNED.value,
points_delta=50,
points_balance_after=150,
transaction_at=datetime.now(UTC),
)
db.add(tx)
db.commit()
response = client.get(
f"{BASE}/transactions",
headers=loyalty_store_headers,
)
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
assert len(data["transactions"]) >= 1
def test_transactions_include_customer_name(
self, client, loyalty_store_headers, loyalty_store_setup, db
):
"""Transaction responses include customer_name (not 'Unknown')."""
card = loyalty_store_setup["card"]
customer = loyalty_store_setup["customer"]
store = loyalty_store_setup["store"]
tx = LoyaltyTransaction(
merchant_id=card.merchant_id,
card_id=card.id,
store_id=store.id,
transaction_type=TransactionType.POINTS_EARNED.value,
points_delta=25,
transaction_at=datetime.now(UTC),
)
db.add(tx)
db.commit()
response = client.get(
f"{BASE}/transactions",
headers=loyalty_store_headers,
)
assert response.status_code == 200
transactions = response.json()["transactions"]
assert len(transactions) >= 1
# Customer name should be populated, not None
assert transactions[0]["customer_name"] == f"{customer.first_name} {customer.last_name}"
# ============================================================================
# Earn Points Tests
# ============================================================================
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.loyalty
class TestEarnPoints:
"""Tests for POST /points/earn endpoint."""
def test_earn_points_endpoint_exists(
self, client, loyalty_store_headers, loyalty_store_setup
):
"""POST /points/earn returns success (not 404)."""
card = loyalty_store_setup["card"]
response = client.post(
f"{BASE}/points/earn",
json={
"card_id": card.id,
"purchase_amount_cents": 1000,
},
headers=loyalty_store_headers,
)
# Should not be 404 (endpoint exists after rename)
assert response.status_code != 404

View File

@@ -1,7 +1,16 @@
"""Unit tests for CardService.""" """Unit tests for CardService."""
import uuid
import pytest 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 from app.modules.loyalty.services.card_service import CardService
@@ -16,3 +25,266 @@ class TestCardService:
def test_service_instantiation(self): def test_service_instantiation(self):
"""Service can be instantiated.""" """Service can be instantiated."""
assert self.service is not None 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("!loyalty-enroll!")
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
)
@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
for i in range(3):
t = 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),
)
db.add(t)
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

View File

@@ -0,0 +1,75 @@
"""Unit tests for loyalty schemas."""
from datetime import UTC, datetime
import pytest
from app.modules.loyalty.schemas.card import (
CardEnrollRequest,
TransactionResponse,
)
@pytest.mark.unit
@pytest.mark.loyalty
class TestCardEnrollRequest:
"""Tests for enrollment request schema validation."""
def test_valid_with_email(self):
"""Email-only enrollment is valid."""
req = CardEnrollRequest(email="test@example.com")
assert req.email == "test@example.com"
assert req.customer_id is None
def test_valid_with_customer_id(self):
"""Customer ID enrollment is valid."""
req = CardEnrollRequest(customer_id=42)
assert req.customer_id == 42
assert req.email is None
def test_self_enrollment_fields(self):
"""Self-enrollment accepts name, phone, birthday."""
req = CardEnrollRequest(
email="self@example.com",
customer_name="Jane Doe",
customer_phone="+352123456",
customer_birthday="1990-01-15",
)
assert req.customer_name == "Jane Doe"
assert req.customer_phone == "+352123456"
assert req.customer_birthday == "1990-01-15"
@pytest.mark.unit
@pytest.mark.loyalty
class TestTransactionResponse:
"""Tests for transaction response schema."""
def test_customer_name_field_defaults_none(self):
"""customer_name field exists and defaults to None."""
now = datetime.now(UTC)
tx = TransactionResponse(
id=1,
card_id=1,
transaction_type="points_earned",
stamps_delta=0,
points_delta=50,
transaction_at=now,
created_at=now,
)
assert tx.customer_name is None
def test_customer_name_can_be_set(self):
"""customer_name can be explicitly set."""
now = datetime.now(UTC)
tx = TransactionResponse(
id=1,
card_id=1,
transaction_type="card_created",
stamps_delta=0,
points_delta=0,
customer_name="John Doe",
transaction_at=now,
created_at=now,
)
assert tx.customer_name == "John Doe"