feat(loyalty): restructure program CRUD by interface
Some checks failed
Some checks failed
Move program CRUD from store to merchant/admin interfaces. Store becomes view-only for program config while merchant gets full CRUD and admin gets override capabilities. Merchant portal: - New API endpoints (GET/POST/PATCH/DELETE /program) - New settings page with create/edit/delete form - Overview page now has Create/Edit Program buttons - Settings menu item added to sidebar Admin portal: - New CRUD endpoints (create for merchant, update, delete) - New activate/deactivate program endpoints - Programs list has edit and toggle buttons per row - Merchant detail has create/delete/toggle program actions Store portal: - Removed POST/PATCH /program endpoints (now read-only) - Removed settings page route and template - Terminal, cards, stats, enroll unchanged Tests: 112 passed (58 new) covering merchant API, admin CRUD, store endpoint removal, and program service unit tests. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -9,11 +9,14 @@ from datetime import UTC, datetime
|
||||
|
||||
import pytest
|
||||
|
||||
from app.api.deps import get_current_merchant_api, get_merchant_for_current_user
|
||||
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
|
||||
from main import app
|
||||
from models.schema.auth import UserContext
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -166,6 +169,95 @@ def loyalty_store_setup(db, loyalty_platform):
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def loyalty_merchant_setup(db, loyalty_platform):
|
||||
"""
|
||||
Merchant-only setup for loyalty integration tests (no program).
|
||||
|
||||
Creates: User -> Merchant (no program yet).
|
||||
Use this for testing program creation via merchant API.
|
||||
"""
|
||||
from middleware.auth import AuthManager
|
||||
|
||||
auth = AuthManager()
|
||||
uid = uuid.uuid4().hex[:8]
|
||||
|
||||
owner = User(
|
||||
email=f"merchowner_{uid}@test.com",
|
||||
username=f"merchowner_{uid}",
|
||||
hashed_password=auth.hash_password("merchpass123"),
|
||||
role="merchant_owner",
|
||||
is_active=True,
|
||||
is_email_verified=True,
|
||||
)
|
||||
db.add(owner)
|
||||
db.commit()
|
||||
db.refresh(owner)
|
||||
|
||||
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)
|
||||
|
||||
return {
|
||||
"owner": owner,
|
||||
"merchant": merchant,
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def loyalty_merchant_headers(loyalty_store_setup):
|
||||
"""
|
||||
Override auth dependencies to return merchant/user for the merchant owner.
|
||||
Uses the full loyalty_store_setup which includes a program.
|
||||
"""
|
||||
owner = loyalty_store_setup["owner"]
|
||||
merchant = loyalty_store_setup["merchant"]
|
||||
|
||||
user_context = UserContext(
|
||||
id=owner.id,
|
||||
email=owner.email,
|
||||
username=owner.username,
|
||||
role="merchant_owner",
|
||||
is_active=True,
|
||||
)
|
||||
|
||||
app.dependency_overrides[get_merchant_for_current_user] = lambda: merchant
|
||||
app.dependency_overrides[get_current_merchant_api] = lambda: user_context
|
||||
yield {"Authorization": "Bearer fake-token"}
|
||||
app.dependency_overrides.pop(get_merchant_for_current_user, None)
|
||||
app.dependency_overrides.pop(get_current_merchant_api, None)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def loyalty_merchant_headers_no_program(loyalty_merchant_setup):
|
||||
"""
|
||||
Override auth dependencies for a merchant that has no program yet.
|
||||
"""
|
||||
owner = loyalty_merchant_setup["owner"]
|
||||
merchant = loyalty_merchant_setup["merchant"]
|
||||
|
||||
user_context = UserContext(
|
||||
id=owner.id,
|
||||
email=owner.email,
|
||||
username=owner.username,
|
||||
role="merchant_owner",
|
||||
is_active=True,
|
||||
)
|
||||
|
||||
app.dependency_overrides[get_merchant_for_current_user] = lambda: merchant
|
||||
app.dependency_overrides[get_current_merchant_api] = lambda: user_context
|
||||
yield {"Authorization": "Bearer fake-token"}
|
||||
app.dependency_overrides.pop(get_merchant_for_current_user, None)
|
||||
app.dependency_overrides.pop(get_current_merchant_api, None)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def loyalty_store_headers(client, loyalty_store_setup):
|
||||
"""
|
||||
|
||||
385
app/modules/loyalty/tests/integration/test_admin_api.py
Normal file
385
app/modules/loyalty/tests/integration/test_admin_api.py
Normal file
@@ -0,0 +1,385 @@
|
||||
# app/modules/loyalty/tests/integration/test_admin_api.py
|
||||
"""
|
||||
Integration tests for admin loyalty CRUD API endpoints.
|
||||
|
||||
Tests the admin program management endpoints at:
|
||||
/api/v1/admin/loyalty/*
|
||||
|
||||
Authentication: Uses super_admin_headers fixture (real JWT login).
|
||||
"""
|
||||
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
|
||||
import pytest
|
||||
|
||||
from app.modules.loyalty.models import LoyaltyProgram
|
||||
from app.modules.loyalty.models.loyalty_program import LoyaltyType
|
||||
from app.modules.tenancy.models import Merchant, User
|
||||
|
||||
BASE = "/api/v1/admin/loyalty"
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Fixtures
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def admin_merchant(db):
|
||||
"""Create a merchant for admin CRUD tests."""
|
||||
from middleware.auth import AuthManager
|
||||
|
||||
auth = AuthManager()
|
||||
uid = uuid.uuid4().hex[:8]
|
||||
|
||||
owner = User(
|
||||
email=f"adminmerchowner_{uid}@test.com",
|
||||
username=f"adminmerchowner_{uid}",
|
||||
hashed_password=auth.hash_password("testpass123"),
|
||||
role="merchant_owner",
|
||||
is_active=True,
|
||||
is_email_verified=True,
|
||||
)
|
||||
db.add(owner)
|
||||
db.commit()
|
||||
db.refresh(owner)
|
||||
|
||||
merchant = Merchant(
|
||||
name=f"Admin Test 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)
|
||||
|
||||
return merchant
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def admin_program(db, admin_merchant):
|
||||
"""Create a loyalty program for admin CRUD tests."""
|
||||
program = LoyaltyProgram(
|
||||
merchant_id=admin_merchant.id,
|
||||
loyalty_type=LoyaltyType.POINTS.value,
|
||||
points_per_euro=5,
|
||||
welcome_bonus_points=25,
|
||||
minimum_redemption_points=50,
|
||||
minimum_purchase_cents=0,
|
||||
cooldown_minutes=0,
|
||||
max_daily_stamps=10,
|
||||
require_staff_pin=False,
|
||||
card_name="Admin Test Rewards",
|
||||
card_color="#1E90FF",
|
||||
is_active=True,
|
||||
points_rewards=[
|
||||
{"id": "reward_1", "name": "5 EUR off", "points_required": 50, "is_active": True},
|
||||
],
|
||||
)
|
||||
db.add(program)
|
||||
db.commit()
|
||||
db.refresh(program)
|
||||
return program
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# POST /merchants/{merchant_id}/program
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.loyalty
|
||||
class TestAdminCreateProgram:
|
||||
"""Tests for POST /api/v1/admin/loyalty/merchants/{merchant_id}/program."""
|
||||
|
||||
def test_create_program_for_merchant(
|
||||
self, client, super_admin_headers, admin_merchant
|
||||
):
|
||||
"""Admin can create a program for any merchant."""
|
||||
response = client.post(
|
||||
f"{BASE}/merchants/{admin_merchant.id}/program",
|
||||
json={
|
||||
"loyalty_type": "points",
|
||||
"points_per_euro": 8,
|
||||
"card_name": "Admin Created",
|
||||
"card_color": "#00FF00",
|
||||
},
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 201
|
||||
data = response.json()
|
||||
assert data["merchant_id"] == admin_merchant.id
|
||||
assert data["points_per_euro"] == 8
|
||||
assert data["card_name"] == "Admin Created"
|
||||
|
||||
def test_create_program_with_stamps(
|
||||
self, client, super_admin_headers, admin_merchant
|
||||
):
|
||||
"""Admin can create a stamps-type program."""
|
||||
response = client.post(
|
||||
f"{BASE}/merchants/{admin_merchant.id}/program",
|
||||
json={
|
||||
"loyalty_type": "stamps",
|
||||
"stamps_target": 10,
|
||||
"stamps_reward_description": "Free coffee",
|
||||
},
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 201
|
||||
data = response.json()
|
||||
assert data["loyalty_type"] == "stamps"
|
||||
assert data["stamps_target"] == 10
|
||||
|
||||
def test_create_program_requires_auth(
|
||||
self, client, admin_merchant
|
||||
):
|
||||
"""Unauthenticated request is rejected."""
|
||||
response = client.post(
|
||||
f"{BASE}/merchants/{admin_merchant.id}/program",
|
||||
json={"loyalty_type": "points"},
|
||||
)
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# PATCH /programs/{program_id}
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.loyalty
|
||||
class TestAdminUpdateProgram:
|
||||
"""Tests for PATCH /api/v1/admin/loyalty/programs/{program_id}."""
|
||||
|
||||
def test_update_program(
|
||||
self, client, super_admin_headers, admin_program
|
||||
):
|
||||
"""Admin can update any program."""
|
||||
response = client.patch(
|
||||
f"{BASE}/programs/{admin_program.id}",
|
||||
json={
|
||||
"points_per_euro": 15,
|
||||
"card_name": "Updated by Admin",
|
||||
},
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["points_per_euro"] == 15
|
||||
assert data["card_name"] == "Updated by Admin"
|
||||
|
||||
def test_update_program_partial(
|
||||
self, client, super_admin_headers, admin_program
|
||||
):
|
||||
"""Partial update only changes specified fields."""
|
||||
response = client.patch(
|
||||
f"{BASE}/programs/{admin_program.id}",
|
||||
json={"card_name": "Only Name Changed"},
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["card_name"] == "Only Name Changed"
|
||||
assert data["points_per_euro"] == 5 # unchanged
|
||||
|
||||
def test_update_nonexistent_program(
|
||||
self, client, super_admin_headers
|
||||
):
|
||||
"""Updating non-existent program returns 404."""
|
||||
response = client.patch(
|
||||
f"{BASE}/programs/999999",
|
||||
json={"card_name": "Ghost"},
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# DELETE /programs/{program_id}
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.loyalty
|
||||
class TestAdminDeleteProgram:
|
||||
"""Tests for DELETE /api/v1/admin/loyalty/programs/{program_id}."""
|
||||
|
||||
def test_delete_program(
|
||||
self, client, super_admin_headers, admin_program, db
|
||||
):
|
||||
"""Admin can delete any program."""
|
||||
program_id = admin_program.id
|
||||
|
||||
response = client.delete(
|
||||
f"{BASE}/programs/{program_id}",
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 204
|
||||
|
||||
# Verify deleted
|
||||
deleted = db.get(LoyaltyProgram, program_id)
|
||||
assert deleted is None
|
||||
|
||||
def test_delete_nonexistent_program(
|
||||
self, client, super_admin_headers
|
||||
):
|
||||
"""Deleting non-existent program returns 404."""
|
||||
response = client.delete(
|
||||
f"{BASE}/programs/999999",
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# POST /programs/{program_id}/activate
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.loyalty
|
||||
class TestAdminActivateProgram:
|
||||
"""Tests for POST /api/v1/admin/loyalty/programs/{program_id}/activate."""
|
||||
|
||||
def test_activate_inactive_program(
|
||||
self, client, super_admin_headers, admin_program, db
|
||||
):
|
||||
"""Admin can activate an inactive program."""
|
||||
# First deactivate
|
||||
admin_program.is_active = False
|
||||
db.commit()
|
||||
db.refresh(admin_program)
|
||||
|
||||
response = client.post(
|
||||
f"{BASE}/programs/{admin_program.id}/activate",
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["is_active"] is True
|
||||
|
||||
def test_activate_already_active_program(
|
||||
self, client, super_admin_headers, admin_program
|
||||
):
|
||||
"""Activating an already active program succeeds (idempotent)."""
|
||||
assert admin_program.is_active is True
|
||||
|
||||
response = client.post(
|
||||
f"{BASE}/programs/{admin_program.id}/activate",
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json()["is_active"] is True
|
||||
|
||||
def test_activate_nonexistent_program(
|
||||
self, client, super_admin_headers
|
||||
):
|
||||
"""Activating non-existent program returns 404."""
|
||||
response = client.post(
|
||||
f"{BASE}/programs/999999/activate",
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# POST /programs/{program_id}/deactivate
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.loyalty
|
||||
class TestAdminDeactivateProgram:
|
||||
"""Tests for POST /api/v1/admin/loyalty/programs/{program_id}/deactivate."""
|
||||
|
||||
def test_deactivate_active_program(
|
||||
self, client, super_admin_headers, admin_program
|
||||
):
|
||||
"""Admin can deactivate an active program."""
|
||||
response = client.post(
|
||||
f"{BASE}/programs/{admin_program.id}/deactivate",
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["is_active"] is False
|
||||
|
||||
def test_deactivate_already_inactive_program(
|
||||
self, client, super_admin_headers, admin_program, db
|
||||
):
|
||||
"""Deactivating an already inactive program succeeds (idempotent)."""
|
||||
admin_program.is_active = False
|
||||
db.commit()
|
||||
|
||||
response = client.post(
|
||||
f"{BASE}/programs/{admin_program.id}/deactivate",
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json()["is_active"] is False
|
||||
|
||||
def test_deactivate_nonexistent_program(
|
||||
self, client, super_admin_headers
|
||||
):
|
||||
"""Deactivating non-existent program returns 404."""
|
||||
response = client.post(
|
||||
f"{BASE}/programs/999999/deactivate",
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Existing Admin Endpoints Still Work
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.loyalty
|
||||
class TestAdminExistingEndpoints:
|
||||
"""Verify existing admin endpoints still work after CRUD additions."""
|
||||
|
||||
def test_list_programs(
|
||||
self, client, super_admin_headers, admin_program
|
||||
):
|
||||
"""GET /programs returns list including created program."""
|
||||
response = client.get(
|
||||
f"{BASE}/programs",
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "programs" in data
|
||||
assert "total" in data
|
||||
assert data["total"] >= 1
|
||||
|
||||
def test_get_program_by_id(
|
||||
self, client, super_admin_headers, admin_program
|
||||
):
|
||||
"""GET /programs/{id} returns specific program."""
|
||||
response = client.get(
|
||||
f"{BASE}/programs/{admin_program.id}",
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["id"] == admin_program.id
|
||||
|
||||
def test_get_program_stats(
|
||||
self, client, super_admin_headers, admin_program
|
||||
):
|
||||
"""GET /programs/{id}/stats returns statistics."""
|
||||
response = client.get(
|
||||
f"{BASE}/programs/{admin_program.id}/stats",
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
267
app/modules/loyalty/tests/integration/test_merchant_api.py
Normal file
267
app/modules/loyalty/tests/integration/test_merchant_api.py
Normal file
@@ -0,0 +1,267 @@
|
||||
# app/modules/loyalty/tests/integration/test_merchant_api.py
|
||||
"""
|
||||
Integration tests for merchant loyalty API endpoints.
|
||||
|
||||
Tests the merchant program CRUD endpoints at:
|
||||
/api/v1/merchants/loyalty/*
|
||||
|
||||
Authentication: Uses dependency overrides (merch_auth_headers pattern).
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
BASE = "/api/v1/merchants/loyalty"
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# GET /program
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.loyalty
|
||||
class TestMerchantGetProgram:
|
||||
"""Tests for GET /api/v1/merchants/loyalty/program."""
|
||||
|
||||
def test_get_program_success(
|
||||
self, client, loyalty_merchant_headers, loyalty_store_setup
|
||||
):
|
||||
"""Returns the merchant's loyalty program."""
|
||||
response = client.get(
|
||||
f"{BASE}/program", headers=loyalty_merchant_headers
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["id"] == loyalty_store_setup["program"].id
|
||||
assert data["merchant_id"] == loyalty_store_setup["merchant"].id
|
||||
assert data["points_per_euro"] == 10
|
||||
assert data["is_active"] is True
|
||||
|
||||
def test_get_program_includes_display_fields(
|
||||
self, client, loyalty_merchant_headers, loyalty_store_setup
|
||||
):
|
||||
"""Response includes computed display fields."""
|
||||
response = client.get(
|
||||
f"{BASE}/program", headers=loyalty_merchant_headers
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "display_name" in data
|
||||
assert "is_stamps_enabled" in data
|
||||
assert "is_points_enabled" in data
|
||||
|
||||
def test_get_program_not_found(
|
||||
self, client, loyalty_merchant_headers_no_program
|
||||
):
|
||||
"""Returns 404 when merchant has no program."""
|
||||
response = client.get(
|
||||
f"{BASE}/program", headers=loyalty_merchant_headers_no_program
|
||||
)
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# POST /program
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.loyalty
|
||||
class TestMerchantCreateProgram:
|
||||
"""Tests for POST /api/v1/merchants/loyalty/program."""
|
||||
|
||||
def test_create_program_success(
|
||||
self, client, loyalty_merchant_headers_no_program, loyalty_merchant_setup
|
||||
):
|
||||
"""Create a new loyalty program for the merchant."""
|
||||
response = client.post(
|
||||
f"{BASE}/program",
|
||||
json={
|
||||
"loyalty_type": "points",
|
||||
"points_per_euro": 5,
|
||||
"card_name": "My Rewards",
|
||||
"card_color": "#FF5733",
|
||||
},
|
||||
headers=loyalty_merchant_headers_no_program,
|
||||
)
|
||||
assert response.status_code == 201
|
||||
data = response.json()
|
||||
assert data["loyalty_type"] == "points"
|
||||
assert data["points_per_euro"] == 5
|
||||
assert data["card_name"] == "My Rewards"
|
||||
assert data["card_color"] == "#FF5733"
|
||||
assert data["merchant_id"] == loyalty_merchant_setup["merchant"].id
|
||||
|
||||
def test_create_program_with_rewards(
|
||||
self, client, loyalty_merchant_headers_no_program
|
||||
):
|
||||
"""Create program with point rewards configured."""
|
||||
response = client.post(
|
||||
f"{BASE}/program",
|
||||
json={
|
||||
"loyalty_type": "points",
|
||||
"points_per_euro": 10,
|
||||
"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,
|
||||
},
|
||||
],
|
||||
},
|
||||
headers=loyalty_merchant_headers_no_program,
|
||||
)
|
||||
assert response.status_code == 201
|
||||
data = response.json()
|
||||
assert len(data["points_rewards"]) == 2
|
||||
assert data["points_rewards"][0]["name"] == "5 EUR off"
|
||||
|
||||
def test_create_program_defaults(
|
||||
self, client, loyalty_merchant_headers_no_program
|
||||
):
|
||||
"""Creating with minimal data uses sensible defaults."""
|
||||
response = client.post(
|
||||
f"{BASE}/program",
|
||||
json={"loyalty_type": "points"},
|
||||
headers=loyalty_merchant_headers_no_program,
|
||||
)
|
||||
assert response.status_code == 201
|
||||
data = response.json()
|
||||
assert data["points_per_euro"] == 1
|
||||
assert data["is_active"] is True
|
||||
|
||||
def test_create_program_duplicate_fails(
|
||||
self, client, loyalty_merchant_headers, loyalty_store_setup
|
||||
):
|
||||
"""Cannot create a second program for the same merchant."""
|
||||
response = client.post(
|
||||
f"{BASE}/program",
|
||||
json={"loyalty_type": "points"},
|
||||
headers=loyalty_merchant_headers,
|
||||
)
|
||||
# Should fail - merchant already has a program
|
||||
assert response.status_code in (400, 409, 422)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# PATCH /program
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.loyalty
|
||||
class TestMerchantUpdateProgram:
|
||||
"""Tests for PATCH /api/v1/merchants/loyalty/program."""
|
||||
|
||||
def test_update_program_success(
|
||||
self, client, loyalty_merchant_headers, loyalty_store_setup
|
||||
):
|
||||
"""Update program fields."""
|
||||
response = client.patch(
|
||||
f"{BASE}/program",
|
||||
json={
|
||||
"points_per_euro": 20,
|
||||
"card_name": "Updated Rewards",
|
||||
},
|
||||
headers=loyalty_merchant_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["points_per_euro"] == 20
|
||||
assert data["card_name"] == "Updated Rewards"
|
||||
|
||||
def test_update_program_partial(
|
||||
self, client, loyalty_merchant_headers, loyalty_store_setup
|
||||
):
|
||||
"""Partial update only modifies specified fields."""
|
||||
response = client.patch(
|
||||
f"{BASE}/program",
|
||||
json={"card_name": "New Name Only"},
|
||||
headers=loyalty_merchant_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["card_name"] == "New Name Only"
|
||||
# Other fields should be unchanged
|
||||
assert data["points_per_euro"] == 10
|
||||
|
||||
def test_update_program_deactivate(
|
||||
self, client, loyalty_merchant_headers, loyalty_store_setup
|
||||
):
|
||||
"""Can deactivate program via update."""
|
||||
response = client.patch(
|
||||
f"{BASE}/program",
|
||||
json={"is_active": False},
|
||||
headers=loyalty_merchant_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json()["is_active"] is False
|
||||
|
||||
def test_update_program_not_found(
|
||||
self, client, loyalty_merchant_headers_no_program
|
||||
):
|
||||
"""Update returns 404 when no program exists."""
|
||||
response = client.patch(
|
||||
f"{BASE}/program",
|
||||
json={"card_name": "No Program"},
|
||||
headers=loyalty_merchant_headers_no_program,
|
||||
)
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# DELETE /program
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.loyalty
|
||||
class TestMerchantDeleteProgram:
|
||||
"""Tests for DELETE /api/v1/merchants/loyalty/program."""
|
||||
|
||||
def test_delete_program_success(
|
||||
self, client, loyalty_merchant_headers, loyalty_store_setup, db
|
||||
):
|
||||
"""Delete the merchant's loyalty program."""
|
||||
program_id = loyalty_store_setup["program"].id
|
||||
|
||||
response = client.delete(
|
||||
f"{BASE}/program", headers=loyalty_merchant_headers
|
||||
)
|
||||
assert response.status_code == 204
|
||||
|
||||
# Verify program is deleted
|
||||
from app.modules.loyalty.models import LoyaltyProgram
|
||||
|
||||
deleted = db.get(LoyaltyProgram, program_id)
|
||||
assert deleted is None
|
||||
|
||||
def test_delete_program_not_found(
|
||||
self, client, loyalty_merchant_headers_no_program
|
||||
):
|
||||
"""Delete returns 404 when no program exists."""
|
||||
response = client.delete(
|
||||
f"{BASE}/program", headers=loyalty_merchant_headers_no_program
|
||||
)
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_delete_then_get_returns_404(
|
||||
self, client, loyalty_merchant_headers, loyalty_store_setup
|
||||
):
|
||||
"""After deletion, GET returns 404."""
|
||||
client.delete(f"{BASE}/program", headers=loyalty_merchant_headers)
|
||||
response = client.get(
|
||||
f"{BASE}/program", headers=loyalty_merchant_headers
|
||||
)
|
||||
assert response.status_code == 404
|
||||
@@ -2,11 +2,12 @@
|
||||
"""
|
||||
Integration tests for store loyalty API endpoints.
|
||||
|
||||
Tests the endpoints fixed today:
|
||||
Tests:
|
||||
- GET /cards/lookup (route ordering fix)
|
||||
- GET /cards/{card_id} (card detail)
|
||||
- GET /transactions (customer_name in response)
|
||||
- POST /points/earn (endpoint path rename)
|
||||
- Store program CRUD endpoints removed (POST/PATCH /program)
|
||||
|
||||
Authentication: Uses real JWT tokens via store login endpoint.
|
||||
"""
|
||||
@@ -213,3 +214,59 @@ class TestEarnPoints:
|
||||
)
|
||||
# Should not be 404 (endpoint exists after rename)
|
||||
assert response.status_code != 404
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Store Program CRUD Removed
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.loyalty
|
||||
class TestStoreProgramCrudRemoved:
|
||||
"""Verify POST/PATCH /program endpoints are removed from store API."""
|
||||
|
||||
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(
|
||||
self, client, loyalty_store_headers, loyalty_store_setup
|
||||
):
|
||||
"""GET /program still works (read-only)."""
|
||||
response = client.get(
|
||||
f"{BASE}/program",
|
||||
headers=loyalty_store_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["id"] == loyalty_store_setup["program"].id
|
||||
|
||||
def test_stats_still_works(
|
||||
self, client, loyalty_store_headers, loyalty_store_setup
|
||||
):
|
||||
"""GET /stats still works."""
|
||||
response = client.get(
|
||||
f"{BASE}/stats",
|
||||
headers=loyalty_store_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
|
||||
@@ -1,8 +1,18 @@
|
||||
"""Unit tests for ProgramService."""
|
||||
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
|
||||
from app.modules.loyalty.exceptions import (
|
||||
LoyaltyProgramAlreadyExistsException,
|
||||
LoyaltyProgramNotFoundException,
|
||||
)
|
||||
from app.modules.loyalty.models import LoyaltyProgram
|
||||
from app.modules.loyalty.models.loyalty_program import LoyaltyType
|
||||
from app.modules.loyalty.schemas.program import ProgramCreate, ProgramUpdate
|
||||
from app.modules.loyalty.services.program_service import ProgramService
|
||||
from app.modules.tenancy.models import Merchant, User
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@@ -16,3 +26,329 @@ class TestProgramService:
|
||||
def test_service_instantiation(self):
|
||||
"""Service can be instantiated."""
|
||||
assert self.service is not None
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Fixtures
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ps_merchant(db):
|
||||
"""Create a merchant for program service tests."""
|
||||
from middleware.auth import AuthManager
|
||||
|
||||
auth = AuthManager()
|
||||
uid = uuid.uuid4().hex[:8]
|
||||
|
||||
owner = User(
|
||||
email=f"psowner_{uid}@test.com",
|
||||
username=f"psowner_{uid}",
|
||||
hashed_password=auth.hash_password("testpass123"),
|
||||
role="merchant_owner",
|
||||
is_active=True,
|
||||
is_email_verified=True,
|
||||
)
|
||||
db.add(owner)
|
||||
db.commit()
|
||||
db.refresh(owner)
|
||||
|
||||
merchant = Merchant(
|
||||
name=f"PS Test 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)
|
||||
|
||||
return merchant
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ps_program(db, ps_merchant):
|
||||
"""Create a program for program service tests."""
|
||||
program = LoyaltyProgram(
|
||||
merchant_id=ps_merchant.id,
|
||||
loyalty_type=LoyaltyType.POINTS.value,
|
||||
points_per_euro=10,
|
||||
welcome_bonus_points=50,
|
||||
minimum_redemption_points=100,
|
||||
minimum_purchase_cents=0,
|
||||
cooldown_minutes=0,
|
||||
max_daily_stamps=10,
|
||||
require_staff_pin=False,
|
||||
card_name="PS Test Rewards",
|
||||
card_color="#4F46E5",
|
||||
is_active=True,
|
||||
points_rewards=[
|
||||
{"id": "reward_1", "name": "5 EUR off", "points_required": 100, "is_active": True},
|
||||
],
|
||||
)
|
||||
db.add(program)
|
||||
db.commit()
|
||||
db.refresh(program)
|
||||
return program
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Read Operations
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.loyalty
|
||||
class TestGetProgram:
|
||||
"""Tests for get_program and get_program_by_merchant."""
|
||||
|
||||
def setup_method(self):
|
||||
self.service = ProgramService()
|
||||
|
||||
def test_get_program_by_id(self, db, ps_program):
|
||||
"""Get a program by its ID."""
|
||||
result = self.service.get_program(db, ps_program.id)
|
||||
assert result is not None
|
||||
assert result.id == ps_program.id
|
||||
|
||||
def test_get_program_not_found(self, db):
|
||||
"""Returns None for non-existent program."""
|
||||
result = self.service.get_program(db, 999999)
|
||||
assert result is None
|
||||
|
||||
def test_get_program_by_merchant(self, db, ps_program, ps_merchant):
|
||||
"""Get a program by merchant ID."""
|
||||
result = self.service.get_program_by_merchant(db, ps_merchant.id)
|
||||
assert result is not None
|
||||
assert result.merchant_id == ps_merchant.id
|
||||
|
||||
def test_get_program_by_merchant_not_found(self, db):
|
||||
"""Returns None when merchant has no program."""
|
||||
result = self.service.get_program_by_merchant(db, 999999)
|
||||
assert result is None
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.loyalty
|
||||
class TestRequireProgram:
|
||||
"""Tests for require_program and require_program_by_merchant."""
|
||||
|
||||
def setup_method(self):
|
||||
self.service = ProgramService()
|
||||
|
||||
def test_require_program_found(self, db, ps_program):
|
||||
"""Returns program when it exists."""
|
||||
result = self.service.require_program(db, ps_program.id)
|
||||
assert result.id == ps_program.id
|
||||
|
||||
def test_require_program_raises_not_found(self, db):
|
||||
"""Raises exception when program doesn't exist."""
|
||||
with pytest.raises(LoyaltyProgramNotFoundException):
|
||||
self.service.require_program(db, 999999)
|
||||
|
||||
def test_require_program_by_merchant_found(self, db, ps_program, ps_merchant):
|
||||
"""Returns program for merchant."""
|
||||
result = self.service.require_program_by_merchant(db, ps_merchant.id)
|
||||
assert result.merchant_id == ps_merchant.id
|
||||
|
||||
def test_require_program_by_merchant_raises_not_found(self, db):
|
||||
"""Raises exception when merchant has no program."""
|
||||
with pytest.raises(LoyaltyProgramNotFoundException):
|
||||
self.service.require_program_by_merchant(db, 999999)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Create Operations
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.loyalty
|
||||
class TestCreateProgram:
|
||||
"""Tests for create_program."""
|
||||
|
||||
def setup_method(self):
|
||||
self.service = ProgramService()
|
||||
|
||||
def test_create_program_points(self, db, ps_merchant):
|
||||
"""Create a points-based program."""
|
||||
data = ProgramCreate(
|
||||
loyalty_type="points",
|
||||
points_per_euro=5,
|
||||
card_name="Unit Test Rewards",
|
||||
card_color="#FF0000",
|
||||
)
|
||||
program = self.service.create_program(db, ps_merchant.id, data)
|
||||
|
||||
assert program.id is not None
|
||||
assert program.merchant_id == ps_merchant.id
|
||||
assert program.loyalty_type == "points"
|
||||
assert program.points_per_euro == 5
|
||||
assert program.card_name == "Unit Test Rewards"
|
||||
assert program.is_active is True
|
||||
|
||||
def test_create_program_stamps(self, db, ps_merchant):
|
||||
"""Create a stamps-based program."""
|
||||
data = ProgramCreate(
|
||||
loyalty_type="stamps",
|
||||
stamps_target=8,
|
||||
stamps_reward_description="Free coffee",
|
||||
)
|
||||
program = self.service.create_program(db, ps_merchant.id, data)
|
||||
|
||||
assert program.loyalty_type == "stamps"
|
||||
assert program.stamps_target == 8
|
||||
assert program.stamps_reward_description == "Free coffee"
|
||||
|
||||
def test_create_program_with_rewards(self, db, ps_merchant):
|
||||
"""Create program with configured rewards."""
|
||||
from app.modules.loyalty.schemas.program import PointsRewardConfig
|
||||
|
||||
data = ProgramCreate(
|
||||
loyalty_type="points",
|
||||
points_per_euro=10,
|
||||
points_rewards=[
|
||||
PointsRewardConfig(
|
||||
id="r1",
|
||||
name="5 EUR off",
|
||||
points_required=100,
|
||||
is_active=True,
|
||||
),
|
||||
],
|
||||
)
|
||||
program = self.service.create_program(db, ps_merchant.id, data)
|
||||
|
||||
assert len(program.points_rewards) == 1
|
||||
assert program.points_rewards[0]["name"] == "5 EUR off"
|
||||
|
||||
def test_create_program_duplicate_raises(self, db, ps_program, ps_merchant):
|
||||
"""Cannot create two programs for the same merchant."""
|
||||
data = ProgramCreate(loyalty_type="points")
|
||||
with pytest.raises(LoyaltyProgramAlreadyExistsException):
|
||||
self.service.create_program(db, ps_merchant.id, data)
|
||||
|
||||
def test_create_program_creates_merchant_settings(self, db, ps_merchant):
|
||||
"""Creating a program also creates merchant loyalty settings."""
|
||||
data = ProgramCreate(loyalty_type="points")
|
||||
self.service.create_program(db, ps_merchant.id, data)
|
||||
|
||||
settings = self.service.get_merchant_settings(db, ps_merchant.id)
|
||||
assert settings is not None
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Update Operations
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.loyalty
|
||||
class TestUpdateProgram:
|
||||
"""Tests for update_program."""
|
||||
|
||||
def setup_method(self):
|
||||
self.service = ProgramService()
|
||||
|
||||
def test_update_program_fields(self, db, ps_program):
|
||||
"""Update specific fields."""
|
||||
data = ProgramUpdate(points_per_euro=20, card_name="Updated Name")
|
||||
result = self.service.update_program(db, ps_program.id, data)
|
||||
|
||||
assert result.points_per_euro == 20
|
||||
assert result.card_name == "Updated Name"
|
||||
|
||||
def test_update_program_partial(self, db, ps_program):
|
||||
"""Partial update preserves unchanged fields."""
|
||||
original_points = ps_program.points_per_euro
|
||||
data = ProgramUpdate(card_name="Only Name")
|
||||
result = self.service.update_program(db, ps_program.id, data)
|
||||
|
||||
assert result.card_name == "Only Name"
|
||||
assert result.points_per_euro == original_points
|
||||
|
||||
def test_update_nonexistent_raises(self, db):
|
||||
"""Updating non-existent program raises exception."""
|
||||
data = ProgramUpdate(card_name="Ghost")
|
||||
with pytest.raises(LoyaltyProgramNotFoundException):
|
||||
self.service.update_program(db, 999999, data)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Activate / Deactivate
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.loyalty
|
||||
class TestActivateDeactivate:
|
||||
"""Tests for activate_program and deactivate_program."""
|
||||
|
||||
def setup_method(self):
|
||||
self.service = ProgramService()
|
||||
|
||||
def test_deactivate_program(self, db, ps_program):
|
||||
"""Deactivate an active program."""
|
||||
assert ps_program.is_active is True
|
||||
result = self.service.deactivate_program(db, ps_program.id)
|
||||
assert result.is_active is False
|
||||
|
||||
def test_activate_program(self, db, ps_program):
|
||||
"""Activate an inactive program."""
|
||||
ps_program.is_active = False
|
||||
db.commit()
|
||||
|
||||
result = self.service.activate_program(db, ps_program.id)
|
||||
assert result.is_active is True
|
||||
|
||||
def test_activate_nonexistent_raises(self, db):
|
||||
"""Activating non-existent program raises exception."""
|
||||
with pytest.raises(LoyaltyProgramNotFoundException):
|
||||
self.service.activate_program(db, 999999)
|
||||
|
||||
def test_deactivate_nonexistent_raises(self, db):
|
||||
"""Deactivating non-existent program raises exception."""
|
||||
with pytest.raises(LoyaltyProgramNotFoundException):
|
||||
self.service.deactivate_program(db, 999999)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Delete Operations
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.loyalty
|
||||
class TestDeleteProgram:
|
||||
"""Tests for delete_program."""
|
||||
|
||||
def setup_method(self):
|
||||
self.service = ProgramService()
|
||||
|
||||
def test_delete_program(self, db, ps_program):
|
||||
"""Delete a program removes it from DB."""
|
||||
program_id = ps_program.id
|
||||
self.service.delete_program(db, program_id)
|
||||
|
||||
result = self.service.get_program(db, program_id)
|
||||
assert result is None
|
||||
|
||||
def test_delete_program_also_deletes_settings(self, db, ps_merchant):
|
||||
"""Deleting a program also removes merchant settings."""
|
||||
data = ProgramCreate(loyalty_type="points")
|
||||
program = self.service.create_program(db, ps_merchant.id, data)
|
||||
|
||||
# Verify settings exist
|
||||
settings = self.service.get_merchant_settings(db, ps_merchant.id)
|
||||
assert settings is not None
|
||||
|
||||
self.service.delete_program(db, program.id)
|
||||
|
||||
# Settings should be gone too
|
||||
settings = self.service.get_merchant_settings(db, ps_merchant.id)
|
||||
assert settings is None
|
||||
|
||||
def test_delete_nonexistent_raises(self, db):
|
||||
"""Deleting non-existent program raises exception."""
|
||||
with pytest.raises(LoyaltyProgramNotFoundException):
|
||||
self.service.delete_program(db, 999999)
|
||||
|
||||
Reference in New Issue
Block a user