108 lines
3.4 KiB
Python
108 lines
3.4 KiB
Python
"""Reminder API ownership and transactional cancellation coverage."""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
import uuid
|
|
|
|
|
|
def _bearer(token):
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def test_reminders_are_owned_and_cancel_related_work(
|
|
client, createAuthenticatedUser
|
|
):
|
|
from core import outbox, postgres
|
|
|
|
owner = createAuthenticatedUser("reminder-owner")
|
|
stranger = createAuthenticatedUser("reminder-stranger")
|
|
runAt = datetime.now(timezone.utc) + timedelta(days=1)
|
|
|
|
created = client.post(
|
|
"/api/reminders",
|
|
headers=_bearer(owner["token"]),
|
|
json={
|
|
"message": "renew the certificate",
|
|
"run_at": runAt.isoformat(),
|
|
"timezone": "America/Chicago",
|
|
"recurrence": {"frequency": "weekly", "interval": 1},
|
|
},
|
|
)
|
|
assert created.status_code == 201, created.get_json()
|
|
reminder = created.get_json()
|
|
|
|
ownerList = client.get(
|
|
"/api/reminders", headers=_bearer(owner["token"])
|
|
).get_json()["reminders"]
|
|
assert [item["id"] for item in ownerList] == [reminder["id"]]
|
|
assert client.get(
|
|
"/api/reminders", headers=_bearer(stranger["token"])
|
|
).get_json() == {"reminders": []}
|
|
|
|
queued = outbox.enqueue_message(
|
|
owner["user_uuid"],
|
|
"discord_dm",
|
|
{"content": "not delivered yet", "reminder_id": reminder["id"]},
|
|
idempotency_key=f"pending-reminder-{uuid.uuid4()}",
|
|
)
|
|
|
|
strangerCancel = client.delete(
|
|
f"/api/reminders/{reminder['id']}",
|
|
headers=_bearer(stranger["token"]),
|
|
)
|
|
assert strangerCancel.status_code == 404
|
|
storedReminder = postgres.select_one("reminders", {"id": reminder["id"]})
|
|
assert storedReminder["status"] == "active"
|
|
assert outbox.get_message(queued["id"])["status"] == "pending"
|
|
|
|
ownerCancel = client.delete(
|
|
f"/api/reminders/{reminder['id']}", headers=_bearer(owner["token"])
|
|
)
|
|
assert ownerCancel.status_code == 200
|
|
assert ownerCancel.get_json()["status"] == "cancelled"
|
|
|
|
job = postgres.execute(
|
|
"""
|
|
SELECT * FROM scheduled_jobs
|
|
WHERE payload->>'reminder_id' = %(reminder_id)s
|
|
""",
|
|
{"reminder_id": reminder["id"]},
|
|
)[0]
|
|
assert job["user_uuid"] == uuid.UUID(owner["user_uuid"])
|
|
assert job["status"] == "cancelled"
|
|
assert outbox.get_message(queued["id"])["status"] == "cancelled"
|
|
|
|
assert client.delete(
|
|
f"/api/reminders/{reminder['id']}", headers=_bearer(owner["token"])
|
|
).status_code == 404
|
|
|
|
|
|
def test_reminder_routes_validate_time_and_authentication(
|
|
client, createAuthenticatedUser
|
|
):
|
|
user = createAuthenticatedUser("reminder-validation")
|
|
headers = _bearer(user["token"])
|
|
|
|
assert client.get("/api/reminders").status_code == 401
|
|
past = client.post(
|
|
"/api/reminders",
|
|
headers=headers,
|
|
json={
|
|
"message": "too late",
|
|
"run_at": (datetime.now(timezone.utc) - timedelta(minutes=1)).isoformat(),
|
|
"timezone": "UTC",
|
|
},
|
|
)
|
|
assert past.status_code == 400
|
|
assert "future" in past.get_json()["error"]
|
|
|
|
badTimezone = client.post(
|
|
"/api/reminders",
|
|
headers=headers,
|
|
json={
|
|
"message": "bad timezone",
|
|
"run_at": (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat(),
|
|
"timezone": "UTC'; DROP TABLE reminders; --",
|
|
},
|
|
)
|
|
assert badTimezone.status_code == 400
|