Samir Boulahtit
fd0de714a4
Some checks failed
Delete program tests now verify soft-delete (deleted_at set, record hidden from normal queries) instead of expecting hard deletion. Uses db.query() instead of db.get() since the soft-delete filter only applies to ORM queries, not identity map lookups. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
665 lines
22 KiB
Python
665 lines
22 KiB
Python
# 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 soft-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 soft-deleted (hidden from normal queries)
|
|
db.expire_all()
|
|
deleted = db.query(LoyaltyProgram).filter(LoyaltyProgram.id == program_id).first()
|
|
assert deleted is None
|
|
|
|
# But visible with include_deleted
|
|
visible = (
|
|
db.query(LoyaltyProgram)
|
|
.execution_options(include_deleted=True)
|
|
.filter(LoyaltyProgram.id == program_id)
|
|
.first()
|
|
)
|
|
assert visible is not None
|
|
assert visible.deleted_at is not 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
|
|
|
|
|
|
# ============================================================================
|
|
# LIST PROGRAMS — Search & Filters
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.api
|
|
@pytest.mark.loyalty
|
|
class TestAdminListProgramsSearch:
|
|
"""Tests for GET /api/v1/admin/loyalty/programs search and filter params."""
|
|
|
|
def test_search_by_merchant_name(
|
|
self, client, super_admin_headers, admin_program, admin_merchant
|
|
):
|
|
"""Search query filters programs by merchant name."""
|
|
# Use a substring of the merchant name
|
|
search_term = admin_merchant.name[:10]
|
|
response = client.get(
|
|
f"{BASE}/programs",
|
|
params={"search": search_term},
|
|
headers=super_admin_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total"] >= 1
|
|
# Our program should be in results
|
|
program_ids = [p["id"] for p in data["programs"]]
|
|
assert admin_program.id in program_ids
|
|
|
|
def test_search_no_results(
|
|
self, client, super_admin_headers, admin_program
|
|
):
|
|
"""Search with non-matching term returns empty."""
|
|
response = client.get(
|
|
f"{BASE}/programs",
|
|
params={"search": "zzz_no_such_merchant_999"},
|
|
headers=super_admin_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total"] == 0
|
|
assert data["programs"] == []
|
|
|
|
def test_search_case_insensitive(
|
|
self, client, super_admin_headers, admin_program, admin_merchant
|
|
):
|
|
"""Search is case-insensitive (ilike)."""
|
|
search_term = admin_merchant.name.upper()
|
|
response = client.get(
|
|
f"{BASE}/programs",
|
|
params={"search": search_term},
|
|
headers=super_admin_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total"] >= 1
|
|
|
|
def test_filter_by_active_status(
|
|
self, client, super_admin_headers, admin_program
|
|
):
|
|
"""is_active filter returns only matching programs."""
|
|
response = client.get(
|
|
f"{BASE}/programs",
|
|
params={"is_active": True},
|
|
headers=super_admin_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
# All returned programs should be active
|
|
for p in data["programs"]:
|
|
assert p["is_active"] is True
|
|
|
|
def test_filter_inactive_excludes_active(
|
|
self, client, super_admin_headers, admin_program
|
|
):
|
|
"""is_active=false excludes active programs."""
|
|
response = client.get(
|
|
f"{BASE}/programs",
|
|
params={"is_active": False},
|
|
headers=super_admin_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
# admin_program is active, should NOT be in results
|
|
program_ids = [p["id"] for p in data["programs"]]
|
|
assert admin_program.id not in program_ids
|
|
|
|
def test_pagination_skip_limit(
|
|
self, client, super_admin_headers, admin_program
|
|
):
|
|
"""Pagination params control results."""
|
|
response = client.get(
|
|
f"{BASE}/programs",
|
|
params={"skip": 0, "limit": 1},
|
|
headers=super_admin_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["programs"]) <= 1
|
|
|
|
def test_search_combined_with_active_filter(
|
|
self, client, super_admin_headers, admin_program, admin_merchant
|
|
):
|
|
"""Search and is_active filter work together."""
|
|
search_term = admin_merchant.name[:10]
|
|
response = client.get(
|
|
f"{BASE}/programs",
|
|
params={"search": search_term, "is_active": True},
|
|
headers=super_admin_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total"] >= 1
|
|
for p in data["programs"]:
|
|
assert p["is_active"] is True
|
|
|
|
def test_list_programs_requires_auth(self, client):
|
|
"""Unauthenticated request is rejected."""
|
|
response = client.get(f"{BASE}/programs")
|
|
assert response.status_code == 401
|
|
|
|
|
|
# ============================================================================
|
|
# CREATE — Duplicate Prevention
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.api
|
|
@pytest.mark.loyalty
|
|
class TestAdminCreateProgramDuplicate:
|
|
"""Tests for duplicate program creation prevention."""
|
|
|
|
def test_create_duplicate_program_rejected(
|
|
self, client, super_admin_headers, admin_program, admin_merchant
|
|
):
|
|
"""Cannot create a second program for a merchant that already has one."""
|
|
response = client.post(
|
|
f"{BASE}/merchants/{admin_merchant.id}/program",
|
|
json={
|
|
"loyalty_type": "stamps",
|
|
"stamps_target": 8,
|
|
},
|
|
headers=super_admin_headers,
|
|
)
|
|
# Should fail — merchant already has admin_program
|
|
assert response.status_code in [409, 422]
|
|
|
|
|
|
# ============================================================================
|
|
# GET /merchants/{merchant_id}/cards (On Behalf)
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.api
|
|
@pytest.mark.loyalty
|
|
class TestAdminListMerchantCards:
|
|
"""Tests for GET /api/v1/admin/loyalty/merchants/{merchant_id}/cards."""
|
|
|
|
def test_list_merchant_cards(
|
|
self, client, super_admin_headers, admin_merchant
|
|
):
|
|
"""Returns cards list for a merchant."""
|
|
response = client.get(
|
|
f"{BASE}/merchants/{admin_merchant.id}/cards",
|
|
headers=super_admin_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "cards" in data
|
|
assert "total" in data
|
|
|
|
def test_list_merchant_cards_requires_auth(self, client, admin_merchant):
|
|
"""Unauthenticated request is rejected."""
|
|
response = client.get(f"{BASE}/merchants/{admin_merchant.id}/cards")
|
|
assert response.status_code in [401, 403]
|
|
|
|
|
|
# ============================================================================
|
|
# GET /merchants/{merchant_id}/transactions (On Behalf)
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.api
|
|
@pytest.mark.loyalty
|
|
class TestAdminListMerchantTransactions:
|
|
"""Tests for GET /api/v1/admin/loyalty/merchants/{merchant_id}/transactions."""
|
|
|
|
def test_list_merchant_transactions(
|
|
self, client, super_admin_headers, admin_merchant
|
|
):
|
|
"""Returns transactions for a merchant."""
|
|
response = client.get(
|
|
f"{BASE}/merchants/{admin_merchant.id}/transactions",
|
|
headers=super_admin_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "transactions" in data
|
|
assert "total" in data
|
|
|
|
def test_list_merchant_transactions_requires_auth(self, client, admin_merchant):
|
|
"""Unauthenticated request is rejected."""
|
|
response = client.get(f"{BASE}/merchants/{admin_merchant.id}/transactions")
|
|
assert response.status_code in [401, 403]
|
|
|
|
|
|
# ============================================================================
|
|
# GET /merchants/{merchant_id}/pins (On Behalf, Read-Only)
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.api
|
|
@pytest.mark.loyalty
|
|
class TestAdminListMerchantPins:
|
|
"""Tests for GET /api/v1/admin/loyalty/merchants/{merchant_id}/pins."""
|
|
|
|
def test_list_merchant_pins(
|
|
self, client, super_admin_headers, admin_merchant
|
|
):
|
|
"""Returns PINs for a merchant."""
|
|
response = client.get(
|
|
f"{BASE}/merchants/{admin_merchant.id}/pins",
|
|
headers=super_admin_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "pins" in data
|
|
assert "total" in data
|
|
|
|
def test_list_merchant_pins_requires_auth(self, client, admin_merchant):
|
|
"""Unauthenticated request is rejected."""
|
|
response = client.get(f"{BASE}/merchants/{admin_merchant.id}/pins")
|
|
assert response.status_code in [401, 403]
|
|
|
|
|
|
# ============================================================================
|
|
# GET /merchants/{merchant_id}/locations (On Behalf)
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.api
|
|
@pytest.mark.loyalty
|
|
class TestAdminListMerchantLocations:
|
|
"""Tests for GET /api/v1/admin/loyalty/merchants/{merchant_id}/locations."""
|
|
|
|
def test_list_merchant_locations(
|
|
self, client, super_admin_headers, admin_merchant
|
|
):
|
|
"""Returns store locations for a merchant."""
|
|
response = client.get(
|
|
f"{BASE}/merchants/{admin_merchant.id}/locations",
|
|
headers=super_admin_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
def test_list_merchant_locations_requires_auth(self, client, admin_merchant):
|
|
"""Unauthenticated request is rejected."""
|
|
response = client.get(f"{BASE}/merchants/{admin_merchant.id}/locations")
|
|
assert response.status_code in [401, 403]
|