403 lines
14 KiB
Python
403 lines
14 KiB
Python
"""Phase 7.4 — Campaign Engine tests."""
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
|
|
import pytest
|
|
|
|
from app.events.publisher import get_event_publisher
|
|
from app.events.types import LoyaltyEventType
|
|
from app.tests.conftest import TENANT_A, tenant_headers
|
|
|
|
|
|
async def _setup(
|
|
client,
|
|
*,
|
|
suffix: str,
|
|
rules: dict | None = None,
|
|
starts_on: str | None = None,
|
|
ends_on: str | None = None,
|
|
) -> dict:
|
|
program = await client.post(
|
|
"/api/v1/programs",
|
|
json={"code": f"cmp{suffix}", "name": f"Campaign {suffix}"},
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert program.status_code == 201, program.text
|
|
program_id = program.json()["id"]
|
|
|
|
member = await client.post(
|
|
"/api/v1/members",
|
|
json={"program_id": program_id, "display_name": "Member", "enroll": True},
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert member.status_code == 201, member.text
|
|
member_id = member.json()["id"]
|
|
|
|
account = await client.post(
|
|
"/api/v1/point-accounts",
|
|
json={"program_id": program_id, "member_id": member_id},
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert account.status_code == 201, account.text
|
|
account_id = account.json()["id"]
|
|
|
|
payload = {
|
|
"program_id": program_id,
|
|
"code": f"camp{suffix}",
|
|
"name": f"Campaign {suffix}",
|
|
"rules": rules if rules is not None else {"earn_bonus_points": 50},
|
|
}
|
|
if starts_on is not None:
|
|
payload["starts_on"] = starts_on
|
|
if ends_on is not None:
|
|
payload["ends_on"] = ends_on
|
|
campaign = await client.post(
|
|
"/api/v1/campaigns", json=payload, headers=tenant_headers(TENANT_A)
|
|
)
|
|
assert campaign.status_code == 201, campaign.text
|
|
|
|
return {
|
|
"program_id": program_id,
|
|
"member_id": member_id,
|
|
"account_id": account_id,
|
|
"campaign_id": campaign.json()["id"],
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_full_lifecycle_schedule_activate_pause_resume_expire(client):
|
|
ctx = await _setup(client, suffix="l1", starts_on="2026-01-01")
|
|
campaign_id = ctx["campaign_id"]
|
|
|
|
scheduled = await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/schedule",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert scheduled.status_code == 200, scheduled.text
|
|
assert scheduled.json()["status"] == "scheduled"
|
|
|
|
activated = await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/activate",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert activated.status_code == 200, activated.text
|
|
assert activated.json()["status"] == "active"
|
|
|
|
paused = await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/pause",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert paused.status_code == 200, paused.text
|
|
assert paused.json()["status"] == "paused"
|
|
|
|
resumed = await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/resume",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert resumed.status_code == 200, resumed.text
|
|
assert resumed.json()["status"] == "active"
|
|
|
|
expired = await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/expire",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert expired.status_code == 200, expired.text
|
|
assert expired.json()["status"] == "expired"
|
|
|
|
events = {e.event_type for e in get_event_publisher().published}
|
|
assert LoyaltyEventType.CAMPAIGN_SCHEDULED.value in events
|
|
assert LoyaltyEventType.CAMPAIGN_ACTIVATED.value in events
|
|
assert LoyaltyEventType.CAMPAIGN_PAUSED.value in events
|
|
assert LoyaltyEventType.CAMPAIGN_RESUMED.value in events
|
|
assert LoyaltyEventType.CAMPAIGN_EXPIRED.value in events
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schedule_requires_starts_on(client):
|
|
ctx = await _setup(client, suffix="l2")
|
|
bad = await client.post(
|
|
f"/api/v1/campaigns/{ctx['campaign_id']}/schedule",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert bad.status_code == 422
|
|
assert bad.json()["error"]["code"] == "campaign_starts_on_required"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_transition_rejected(client):
|
|
ctx = await _setup(client, suffix="l3")
|
|
# draft campaign cannot pause directly
|
|
bad = await client.post(
|
|
f"/api/v1/campaigns/{ctx['campaign_id']}/pause",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert bad.status_code == 409
|
|
assert bad.json()["error"]["code"] == "invalid_campaign_transition"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_active_campaign_cannot_be_soft_deleted(client):
|
|
ctx = await _setup(client, suffix="l4")
|
|
activated = await client.post(
|
|
f"/api/v1/campaigns/{ctx['campaign_id']}/activate",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert activated.status_code == 200, activated.text
|
|
|
|
deleted = await client.post(
|
|
f"/api/v1/campaigns/{ctx['campaign_id']}/delete",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert deleted.status_code == 409
|
|
assert deleted.json()["error"]["code"] == "campaign_active_not_deletable"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rules_update_bumps_version_and_emits_event(client):
|
|
ctx = await _setup(client, suffix="l5")
|
|
campaign_id = ctx["campaign_id"]
|
|
|
|
updated = await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/rules",
|
|
json={"rules": {"earn_bonus_points": 75, "earn_multiplier": 2.0}},
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert updated.status_code == 200, updated.text
|
|
assert updated.json()["rule_version"] == 2
|
|
assert updated.json()["rules"]["earn_bonus_points"] == 75
|
|
|
|
updated_again = await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/rules",
|
|
json={"rules": {"earn_bonus_points": 90}},
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert updated_again.status_code == 200
|
|
assert updated_again.json()["rule_version"] == 3
|
|
|
|
events = {e.event_type for e in get_event_publisher().published}
|
|
assert LoyaltyEventType.CAMPAIGN_RULES_UPDATED.value in events
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_grants_bonus_points_via_ledger(client):
|
|
ctx = await _setup(client, suffix="a1", rules={"earn_bonus_points": 50})
|
|
campaign_id = ctx["campaign_id"]
|
|
await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/activate",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
|
|
apply_resp = await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/apply",
|
|
json={"member_id": ctx["member_id"]},
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert apply_resp.status_code == 201, apply_resp.text
|
|
body = apply_resp.json()
|
|
assert body["points_granted"] == 50
|
|
assert body["status"] == "applied"
|
|
assert body["ledger_entry_id"] is not None
|
|
assert body["rule_version"] == 1
|
|
|
|
balance = await client.get(
|
|
f"/api/v1/point-accounts/{ctx['account_id']}/balance",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert balance.json()["balance"] == 50
|
|
|
|
events = {e.event_type for e in get_event_publisher().published}
|
|
assert LoyaltyEventType.CAMPAIGN_APPLIED.value in events
|
|
assert LoyaltyEventType.POINTS_EARNED.value in events
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_with_base_amount_uses_multiplier(client):
|
|
ctx = await _setup(
|
|
client,
|
|
suffix="a2",
|
|
rules={"earn_bonus_points": 10, "earn_multiplier": 1.5},
|
|
)
|
|
campaign_id = ctx["campaign_id"]
|
|
await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/activate",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
|
|
apply_resp = await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/apply",
|
|
json={"member_id": ctx["member_id"], "base_amount": 100},
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert apply_resp.status_code == 201, apply_resp.text
|
|
# floor(100 * 1.5) + 10 = 160
|
|
assert apply_resp.json()["points_granted"] == 160
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_rejects_inactive_campaign(client):
|
|
ctx = await _setup(client, suffix="a3")
|
|
apply_resp = await client.post(
|
|
f"/api/v1/campaigns/{ctx['campaign_id']}/apply",
|
|
json={"member_id": ctx["member_id"]},
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert apply_resp.status_code == 409
|
|
assert apply_resp.json()["error"]["code"] == "campaign_not_active"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_rejects_before_start_window(client):
|
|
future = (dt.date.today() + dt.timedelta(days=30)).isoformat()
|
|
ctx = await _setup(client, suffix="a4", starts_on=future)
|
|
await client.post(
|
|
f"/api/v1/campaigns/{ctx['campaign_id']}/activate",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
apply_resp = await client.post(
|
|
f"/api/v1/campaigns/{ctx['campaign_id']}/apply",
|
|
json={"member_id": ctx["member_id"]},
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert apply_resp.status_code == 409
|
|
assert apply_resp.json()["error"]["code"] == "campaign_not_started"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_rejects_after_end_window(client):
|
|
past_start = (dt.date.today() - dt.timedelta(days=60)).isoformat()
|
|
past_end = (dt.date.today() - dt.timedelta(days=1)).isoformat()
|
|
ctx = await _setup(client, suffix="a5", starts_on=past_start, ends_on=past_end)
|
|
await client.post(
|
|
f"/api/v1/campaigns/{ctx['campaign_id']}/activate",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
apply_resp = await client.post(
|
|
f"/api/v1/campaigns/{ctx['campaign_id']}/apply",
|
|
json={"member_id": ctx["member_id"]},
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert apply_resp.status_code == 409
|
|
assert apply_resp.json()["error"]["code"] == "campaign_ended"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_respects_max_applications_per_member(client):
|
|
ctx = await _setup(
|
|
client,
|
|
suffix="a6",
|
|
rules={"earn_bonus_points": 10, "max_applications_per_member": 1},
|
|
)
|
|
campaign_id = ctx["campaign_id"]
|
|
await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/activate",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
|
|
first = await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/apply",
|
|
json={"member_id": ctx["member_id"]},
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert first.status_code == 201, first.text
|
|
|
|
second = await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/apply",
|
|
json={"member_id": ctx["member_id"]},
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert second.status_code == 409
|
|
assert second.json()["error"]["code"] == "campaign_max_applications_exceeded"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_idempotency_returns_same_application_and_single_grant(client):
|
|
ctx = await _setup(client, suffix="a7", rules={"earn_bonus_points": 20})
|
|
campaign_id = ctx["campaign_id"]
|
|
await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/activate",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
|
|
first = await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/apply",
|
|
json={"member_id": ctx["member_id"], "idempotency_key": "camp-idem-1"},
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert first.status_code == 201, first.text
|
|
|
|
second = await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/apply",
|
|
json={"member_id": ctx["member_id"], "idempotency_key": "camp-idem-1"},
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert second.status_code == 201
|
|
assert second.json()["id"] == first.json()["id"]
|
|
|
|
balance = await client.get(
|
|
f"/api/v1/point-accounts/{ctx['account_id']}/balance",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert balance.json()["balance"] == 20
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_zero_grant_rejected(client):
|
|
ctx = await _setup(client, suffix="a8", rules={"earn_bonus_points": 0})
|
|
campaign_id = ctx["campaign_id"]
|
|
await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/activate",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
apply_resp = await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/apply",
|
|
json={"member_id": ctx["member_id"]},
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert apply_resp.status_code == 422
|
|
assert apply_resp.json()["error"]["code"] == "campaign_grant_invalid"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_campaign_applications(client):
|
|
ctx = await _setup(client, suffix="a9", rules={"earn_bonus_points": 15})
|
|
campaign_id = ctx["campaign_id"]
|
|
await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/activate",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
applied = await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/apply",
|
|
json={"member_id": ctx["member_id"]},
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert applied.status_code == 201, applied.text
|
|
|
|
listed = await client.get(
|
|
f"/api/v1/campaigns/{campaign_id}/applications",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert listed.status_code == 200, listed.text
|
|
assert listed.json()["total"] >= 1
|
|
assert listed.json()["items"][0]["campaign_id"] == campaign_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_explicit_point_account_id(client):
|
|
ctx = await _setup(client, suffix="a10", rules={"earn_bonus_points": 30})
|
|
campaign_id = ctx["campaign_id"]
|
|
await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/activate",
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
apply_resp = await client.post(
|
|
f"/api/v1/campaigns/{campaign_id}/apply",
|
|
json={
|
|
"member_id": ctx["member_id"],
|
|
"point_account_id": ctx["account_id"],
|
|
},
|
|
headers=tenant_headers(TENANT_A),
|
|
)
|
|
assert apply_resp.status_code == 201, apply_resp.text
|
|
assert apply_resp.json()["point_account_id"] == ctx["account_id"]
|