feat(loyalty): wallet debug page, Google Wallet fixes, and module config env_file standardization
Some checks failed
CI / ruff (push) Successful in 12s
CI / validate (push) Successful in 27s
CI / dependency-scanning (push) Successful in 32s
CI / pytest (push) Failing after 1h13m39s
CI / docs (push) Has been cancelled
CI / deploy (push) Has been cancelled

- Add wallet diagnostics page at /admin/loyalty/wallet-debug (super admin only)
  with explorer-sidebar pattern: config validation, class status, card inspector,
  save URL tester, recent enrollments, and Apple Wallet status panels
- Fix Google Wallet fat JWT: include both loyaltyClasses and loyaltyObjects in
  payload, use UNDER_REVIEW instead of DRAFT for class reviewStatus
- Fix StorefrontProgramResponse schema: accept google_class_id values while
  keeping exclude=True (was rejecting non-None values)
- Standardize all module configs to read from .env file directly
  (env_file=".env", extra="ignore") matching core Settings pattern
- Add MOD-026 architecture rule enforcing env_file in module configs
- Add SVC-005 noqa support in architecture validator
- Add test files for dev_tools domain_health and isolation_audit services
- Add google_wallet_status.py script for querying Google Wallet API
- Use table_wrapper macro in wallet-debug.html (FE-005 compliance)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-19 22:18:39 +01:00
parent 11b8e31a29
commit f89c0382f0
31 changed files with 1721 additions and 64 deletions

View File

@@ -13,7 +13,11 @@ import logging
from fastapi import APIRouter, Depends, Path, Query
from sqlalchemy.orm import Session
from app.api.deps import get_current_admin_api, require_module_access
from app.api.deps import (
get_current_admin_api,
get_current_super_admin_api,
require_module_access,
)
from app.core.database import get_db
from app.modules.enums import FrontendType
from app.modules.loyalty.schemas import (
@@ -282,3 +286,262 @@ def get_wallet_status(
):
"""Get wallet integration status for the platform."""
return program_service.get_wallet_integration_status(db)
# =============================================================================
# Wallet Debug (super admin only)
# =============================================================================
@router.get("/debug/config") # noqa: API001
def debug_wallet_config(
current_user: User = Depends(get_current_super_admin_api),
db: Session = Depends(get_db),
):
"""Validate Google Wallet configuration (super admin only)."""
from app.modules.loyalty.config import config as loyalty_config
from app.modules.loyalty.services.google_wallet_service import google_wallet_service
result = google_wallet_service.validate_config()
result["origins"] = loyalty_config.google_wallet_origins or []
result["default_logo_url"] = loyalty_config.default_logo_url
# Check Apple Wallet config too
from app.modules.loyalty.services.apple_wallet_service import apple_wallet_service
apple_config = apple_wallet_service.validate_config()
result["apple"] = apple_config
return result
@router.get("/debug/classes") # noqa: API001
def debug_wallet_classes(
current_user: User = Depends(get_current_super_admin_api),
db: Session = Depends(get_db),
):
"""List all programs with their Google Wallet class status (super admin only)."""
from app.modules.loyalty.services.google_wallet_service import google_wallet_service
programs, _ = program_service.list_programs(db, skip=0, limit=1000)
results = []
for program in programs:
entry = {
"program_id": program.id,
"merchant_name": program.merchant.name if program.merchant else None,
"program_name": program.display_name,
"google_class_id": program.google_class_id,
"review_status": "NOT_CREATED",
"class_metadata": None,
}
if program.google_class_id:
status = google_wallet_service.get_class_status(program.google_class_id)
if status:
entry["review_status"] = status.get("review_status", "UNKNOWN")
entry["class_metadata"] = status
else:
entry["review_status"] = "UNKNOWN"
results.append(entry)
return {"programs": results}
@router.post("/debug/classes/{program_id}/create") # noqa: API001
def debug_create_wallet_class(
program_id: int = Path(..., gt=0),
current_user: User = Depends(get_current_super_admin_api),
db: Session = Depends(get_db),
):
"""Manually create a Google Wallet class for a program (super admin only)."""
from app.modules.loyalty.services.google_wallet_service import google_wallet_service
program = program_service.require_program(db, program_id)
try:
class_id = google_wallet_service.create_class(db, program)
return {
"success": True,
"class_id": class_id,
"program_id": program_id,
}
except Exception as exc:
return {
"success": False,
"error": str(exc),
"program_id": program_id,
}
@router.get("/debug/cards/{card_id}") # noqa: API001
def debug_inspect_card(
card_id: int = Path(..., ge=0),
card_number: str | None = Query(None, description="Search by card number instead"),
current_user: User = Depends(get_current_super_admin_api),
db: Session = Depends(get_db),
):
"""Inspect a card's wallet state (super admin only)."""
from app.modules.loyalty.services.card_service import card_service
from app.modules.loyalty.services.google_wallet_service import google_wallet_service
card = None
if card_number:
card = card_service.get_card_by_number(db, card_number)
elif card_id > 0:
card = card_service.get_card(db, card_id)
if not card:
return {"error": "Card not found"}
# Decode JWT preview if present
jwt_info = None
if card.google_object_jwt:
try:
import base64
import json
parts = card.google_object_jwt.split(".")
if len(parts) == 3:
# Decode header and payload (without verification)
payload_b64 = parts[1] + "=" * (4 - len(parts[1]) % 4)
payload = json.loads(base64.urlsafe_b64decode(payload_b64))
jwt_type = "fat" if "payload" in payload and any(
obj.get("classId") or obj.get("id")
for obj in payload.get("payload", {}).get("loyaltyObjects", [])
if isinstance(obj, dict) and ("classId" in obj or "state" in obj)
) else "reference"
jwt_info = {
"present": True,
"type": jwt_type,
"iss": payload.get("iss"),
"aud": payload.get("aud"),
"exp": payload.get("exp"),
"truncated_token": card.google_object_jwt[:80] + "...",
}
except Exception:
jwt_info = {"present": True, "type": "unknown", "decode_error": True}
# Check if object exists in Google
google_object_exists = None
if card.google_object_id and google_wallet_service.is_configured:
try:
http = google_wallet_service._get_http_client()
resp = http.get(
f"https://walletobjects.googleapis.com/walletobjects/v1/loyaltyObject/{card.google_object_id}"
)
google_object_exists = resp.status_code == 200
except Exception:
google_object_exists = None
return {
"card_id": card.id,
"card_number": card.card_number,
"customer_email": card.customer.email if card.customer else None,
"program_name": card.program.display_name if card.program else None,
"program_id": card.program_id,
"is_active": card.is_active,
"google_object_id": card.google_object_id,
"google_object_jwt": jwt_info,
"has_google_wallet": bool(card.google_object_id),
"google_object_exists_in_api": google_object_exists,
"apple_serial_number": card.apple_serial_number,
"has_apple_wallet": bool(card.apple_serial_number),
"created_at": str(card.created_at) if card.created_at else None,
}
@router.post("/debug/cards/{card_id}/generate-url") # noqa: API001
def debug_generate_save_url(
card_id: int = Path(..., gt=0),
current_user: User = Depends(get_current_super_admin_api),
db: Session = Depends(get_db),
):
"""Generate a fresh Google Wallet save URL for a card (super admin only)."""
from app.modules.loyalty.services.card_service import card_service
from app.modules.loyalty.services.google_wallet_service import google_wallet_service
card = card_service.get_card(db, card_id)
if not card:
return {"error": "Card not found"}
try:
url = google_wallet_service.get_save_url(db, card)
# Decode JWT to show preview
jwt_preview = None
if card.google_object_jwt:
try:
import base64
import json
parts = card.google_object_jwt.split(".")
if len(parts) == 3:
payload_b64 = parts[1] + "=" * (4 - len(parts[1]) % 4)
payload = json.loads(base64.urlsafe_b64decode(payload_b64))
# Determine if fat or reference JWT
objects = payload.get("payload", {}).get("loyaltyObjects", [])
is_fat = any(
isinstance(obj, dict) and ("classId" in obj or "state" in obj)
for obj in objects
)
jwt_preview = {
"type": "fat" if is_fat else "reference",
"iss": payload.get("iss"),
"aud": payload.get("aud"),
"exp": payload.get("exp"),
}
except Exception:
jwt_preview = {"type": "unknown"}
return {
"success": True,
"url": url,
"card_id": card_id,
"jwt_preview": jwt_preview,
}
except Exception as exc:
return {
"success": False,
"error": str(exc),
"card_id": card_id,
}
@router.get("/debug/recent-enrollments") # noqa: API001
def debug_recent_enrollments(
current_user: User = Depends(get_current_super_admin_api),
db: Session = Depends(get_db),
):
"""Show the last 20 enrollments with wallet status (super admin only)."""
from app.modules.loyalty.services.card_service import card_service
cards = card_service.get_recent_cards(db, limit=20)
results = []
for card in cards:
has_object = bool(card.google_object_id)
has_jwt = bool(card.google_object_jwt)
if has_object:
status = "wallet_ready"
elif has_jwt:
status = "jwt_only"
else:
status = "no_wallet"
results.append({
"card_id": card.id,
"card_number": card.card_number,
"customer_email": card.customer.email if card.customer else None,
"program_name": card.program.display_name if card.program else None,
"enrolled_at": str(card.created_at) if card.created_at else None,
"google_object_id": card.google_object_id,
"has_google_jwt": has_jwt,
"has_google_wallet": has_object,
"has_apple_wallet": bool(card.apple_serial_number),
"status": status,
})
return {"enrollments": results}