Files
orion/app/modules/loyalty/schemas/points.py
Samir Boulahtit 7d652716bb
Some checks failed
CI / ruff (push) Successful in 12s
CI / validate (push) Successful in 27s
CI / dependency-scanning (push) Successful in 31s
CI / pytest (push) Failing after 3h14m58s
CI / docs (push) Has been cancelled
CI / deploy (push) Has been cancelled
feat(loyalty): production readiness round 2 — 12 security, integrity & correctness fixes
Security:
- Fix TOCTOU race conditions: move balance/limit checks after row lock in redeem_points, add_stamp, redeem_stamps
- Add PIN ownership verification to update/delete/unlock store routes
- Gate adjust_points endpoint to merchant_owner role only

Data integrity:
- Track total_points_voided in void_points
- Add order_reference idempotency guard in earn_points

Correctness:
- Fix LoyaltyProgramAlreadyExistsException to use merchant_id parameter
- Add StorefrontProgramResponse excluding wallet IDs from public API
- Add bounds (±100000) to PointsAdjustRequest.points_delta

Audit & config:
- Add CARD_REACTIVATED transaction type with audit record
- Improve admin audit logging with actor identity and old values
- Use merchant-specific PIN lockout settings with global fallback
- Guard MerchantLoyaltySettings creation with get_or_create pattern

Tests: 27 new tests (265 total) covering all 12 items — unit and integration.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-16 23:37:23 +01:00

240 lines
5.2 KiB
Python

# app/modules/loyalty/schemas/points.py
"""
Pydantic schemas for points operations.
Merchant-based points:
- Points earned at any store count toward merchant total
- Points can be redeemed at any store within the merchant
- Supports voiding points for returns
"""
from pydantic import BaseModel, Field
class PointsEarnRequest(BaseModel):
"""Schema for earning points from a purchase."""
card_id: int | None = Field(
None,
description="Card ID (use this or qr_code)",
)
qr_code: str | None = Field(
None,
description="QR code data from card scan",
)
card_number: str | None = Field(
None,
description="Card number (manual entry)",
)
# Purchase info
purchase_amount_cents: int = Field(
...,
gt=0,
description="Purchase amount in cents",
)
order_reference: str | None = Field(
None,
max_length=100,
description="Order reference for tracking",
)
# Authentication
staff_pin: str | None = Field(
None,
min_length=4,
max_length=6,
description="Staff PIN for verification",
)
# Optional metadata
notes: str | None = Field(
None,
max_length=500,
description="Optional note",
)
class PointsEarnResponse(BaseModel):
"""Schema for points earning response."""
success: bool = True
message: str = "Points earned successfully"
# Points info
points_earned: int
points_per_euro: int
purchase_amount_cents: int
# Card state after earning
card_id: int
card_number: str
points_balance: int
total_points_earned: int
# Location
store_id: int | None = None
class PointsRedeemRequest(BaseModel):
"""Schema for redeeming points for a reward."""
card_id: int | None = Field(
None,
description="Card ID (use this or qr_code)",
)
qr_code: str | None = Field(
None,
description="QR code data from card scan",
)
card_number: str | None = Field(
None,
description="Card number (manual entry)",
)
# Reward selection
reward_id: str = Field(
...,
description="ID of the reward to redeem",
)
# Authentication
staff_pin: str | None = Field(
None,
min_length=4,
max_length=6,
description="Staff PIN for verification",
)
# Optional metadata
notes: str | None = Field(
None,
max_length=500,
description="Optional note",
)
class PointsRedeemResponse(BaseModel):
"""Schema for points redemption response."""
success: bool = True
message: str = "Reward redeemed successfully"
# Reward info
reward_id: str
reward_name: str
points_spent: int
# Card state after redemption
card_id: int
card_number: str
points_balance: int
total_points_redeemed: int
# Location
store_id: int | None = None
class PointsVoidRequest(BaseModel):
"""Schema for voiding points (for returns)."""
card_id: int | None = Field(
None,
description="Card ID (use this or qr_code)",
)
qr_code: str | None = Field(
None,
description="QR code data from card scan",
)
card_number: str | None = Field(
None,
description="Card number (manual entry)",
)
# Points to void (use one method)
points_to_void: int | None = Field(
None,
gt=0,
description="Number of points to void",
)
original_transaction_id: int | None = Field(
None,
description="ID of original transaction to void",
)
order_reference: str | None = Field(
None,
max_length=100,
description="Order reference to find and void",
)
# Authentication
staff_pin: str | None = Field(
None,
min_length=4,
max_length=6,
description="Staff PIN for verification",
)
# Required metadata
notes: str | None = Field(
None,
max_length=500,
description="Reason for voiding",
)
class PointsVoidResponse(BaseModel):
"""Schema for points void response."""
success: bool = True
message: str = "Points voided successfully"
# Void info
points_voided: int
# Card state after void
card_id: int
card_number: str
points_balance: int
# Location
store_id: int | None = None
class PointsAdjustRequest(BaseModel):
"""Schema for manual points adjustment (admin)."""
points_delta: int = Field(
...,
ge=-100000,
le=100000,
description="Points to add (positive) or remove (negative)",
)
reason: str = Field(
...,
min_length=5,
max_length=500,
description="Reason for adjustment (required)",
)
staff_pin: str | None = Field(
None,
min_length=4,
max_length=6,
description="Staff PIN for verification",
)
class PointsAdjustResponse(BaseModel):
"""Schema for points adjustment response."""
success: bool = True
message: str = "Points adjusted successfully"
# Adjustment info
points_delta: int
# Card state after adjustment
card_id: int
card_number: str
points_balance: int