Ryan Kegel
ebaef16daf
All checks were successful
Chore App Build, Test, and Push Docker Images / build-and-push (push) Successful in 3m23s
- Introduced a dual-token system for user authentication: a short-lived access token and a long-lived rotating refresh token. - Created a new RefreshToken model to manage refresh tokens securely. - Updated auth_api.py to handle login, refresh, and logout processes with the new token system. - Enhanced security measures including token rotation and theft detection. - Updated frontend to handle token refresh on 401 errors and adjusted SSE authentication. - Removed CORS middleware as it's unnecessary behind the nginx proxy. - Added tests to ensure functionality and security of the new token system.
322 lines
11 KiB
Python
322 lines
11 KiB
Python
import pytest
|
||
from tests.conftest import TEST_SECRET_KEY, TEST_REFRESH_TOKEN_EXPIRY_DAYS
|
||
import os
|
||
from werkzeug.security import generate_password_hash
|
||
|
||
from flask import Flask
|
||
from api.chore_schedule_api import chore_schedule_api
|
||
from api.auth_api import auth_api
|
||
from db.db import users_db, child_db, chore_schedules_db, task_extensions_db
|
||
from tinydb import Query
|
||
|
||
|
||
TEST_EMAIL = "sched_test@example.com"
|
||
TEST_PASSWORD = "testpass"
|
||
TEST_CHILD_ID = "sched-child-1"
|
||
TEST_TASK_ID = "sched-task-1"
|
||
TEST_USER_ID = "sched-user-1"
|
||
|
||
|
||
def add_test_user():
|
||
users_db.remove(Query().email == TEST_EMAIL)
|
||
users_db.insert({
|
||
"id": TEST_USER_ID,
|
||
"first_name": "Sched",
|
||
"last_name": "Tester",
|
||
"email": TEST_EMAIL,
|
||
"password": generate_password_hash(TEST_PASSWORD),
|
||
"verified": True,
|
||
"image_id": "",
|
||
})
|
||
|
||
|
||
def add_test_child():
|
||
child_db.remove(Query().id == TEST_CHILD_ID)
|
||
child_db.insert({
|
||
"id": TEST_CHILD_ID,
|
||
"user_id": TEST_USER_ID,
|
||
"name": "Test Child",
|
||
"points": 0,
|
||
"image_id": "",
|
||
"tasks": [TEST_TASK_ID],
|
||
"rewards": [],
|
||
})
|
||
|
||
|
||
def login_and_set_cookie(client):
|
||
resp = client.post('/auth/login', json={"email": TEST_EMAIL, "password": TEST_PASSWORD})
|
||
assert resp.status_code == 200
|
||
|
||
|
||
@pytest.fixture
|
||
def client():
|
||
app = Flask(__name__)
|
||
app.register_blueprint(chore_schedule_api)
|
||
app.register_blueprint(auth_api, url_prefix='/auth')
|
||
app.config['TESTING'] = True
|
||
app.config['SECRET_KEY'] = TEST_SECRET_KEY
|
||
app.config['REFRESH_TOKEN_EXPIRY_DAYS'] = TEST_REFRESH_TOKEN_EXPIRY_DAYS
|
||
with app.test_client() as client:
|
||
add_test_user()
|
||
add_test_child()
|
||
chore_schedules_db.truncate()
|
||
task_extensions_db.truncate()
|
||
login_and_set_cookie(client)
|
||
yield client
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# GET schedule
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_get_schedule_not_found(client):
|
||
resp = client.get(f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/schedule')
|
||
assert resp.status_code == 404
|
||
|
||
|
||
def test_get_schedule_returns_404_for_unknown_child(client):
|
||
resp = client.get('/child/bad-child/task/bad-task/schedule')
|
||
assert resp.status_code == 404
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# PUT (set) schedule – days mode
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_set_schedule_days_mode(client):
|
||
payload = {
|
||
"mode": "days",
|
||
"day_configs": [
|
||
{"day": 1, "hour": 8, "minute": 0},
|
||
{"day": 3, "hour": 9, "minute": 30},
|
||
],
|
||
}
|
||
resp = client.put(f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/schedule', json=payload)
|
||
assert resp.status_code == 200
|
||
data = resp.get_json()
|
||
assert data["mode"] == "days"
|
||
assert len(data["day_configs"]) == 2
|
||
assert data["child_id"] == TEST_CHILD_ID
|
||
assert data["task_id"] == TEST_TASK_ID
|
||
|
||
|
||
def test_get_schedule_after_set(client):
|
||
payload = {"mode": "days", "day_configs": [{"day": 0, "hour": 7, "minute": 0}]}
|
||
client.put(f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/schedule', json=payload)
|
||
|
||
resp = client.get(f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/schedule')
|
||
assert resp.status_code == 200
|
||
data = resp.get_json()
|
||
assert data["mode"] == "days"
|
||
assert data["day_configs"][0]["day"] == 0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# PUT (set) schedule – interval mode
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_set_schedule_interval_mode(client):
|
||
payload = {
|
||
"mode": "interval",
|
||
"interval_days": 3,
|
||
"anchor_date": "2026-03-01",
|
||
"interval_has_deadline": True,
|
||
"interval_hour": 14,
|
||
"interval_minute": 30,
|
||
}
|
||
resp = client.put(f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/schedule', json=payload)
|
||
assert resp.status_code == 200
|
||
data = resp.get_json()
|
||
assert data["mode"] == "interval"
|
||
assert data["interval_days"] == 3
|
||
assert data["anchor_date"] == "2026-03-01"
|
||
assert data["interval_has_deadline"] is True
|
||
assert data["interval_hour"] == 14
|
||
assert data["interval_minute"] == 30
|
||
|
||
|
||
def test_set_schedule_interval_days_1_valid(client):
|
||
"""interval_days=1 is now valid (range changed to [1, 7])."""
|
||
payload = {"mode": "interval", "interval_days": 1, "anchor_date": ""}
|
||
resp = client.put(f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/schedule', json=payload)
|
||
assert resp.status_code == 200
|
||
assert resp.get_json()["interval_days"] == 1
|
||
|
||
|
||
def test_set_schedule_interval_days_0_invalid(client):
|
||
"""interval_days=0 is still out of range."""
|
||
payload = {"mode": "interval", "interval_days": 0, "anchor_date": ""}
|
||
resp = client.put(f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/schedule', json=payload)
|
||
assert resp.status_code == 400
|
||
|
||
|
||
def test_set_schedule_interval_days_8_invalid(client):
|
||
"""interval_days=8 is still out of range."""
|
||
payload = {"mode": "interval", "interval_days": 8, "anchor_date": ""}
|
||
resp = client.put(f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/schedule', json=payload)
|
||
assert resp.status_code == 400
|
||
|
||
|
||
def test_set_schedule_interval_has_deadline_false(client):
|
||
"""interval_has_deadline=False is accepted and persisted."""
|
||
payload = {
|
||
"mode": "interval",
|
||
"interval_days": 2,
|
||
"anchor_date": "",
|
||
"interval_has_deadline": False,
|
||
"interval_hour": 0,
|
||
"interval_minute": 0,
|
||
}
|
||
resp = client.put(f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/schedule', json=payload)
|
||
assert resp.status_code == 200
|
||
assert resp.get_json()["interval_has_deadline"] is False
|
||
|
||
|
||
def test_set_schedule_interval_anchor_date_empty_string(client):
|
||
"""anchor_date empty string is valid (means use today)."""
|
||
payload = {"mode": "interval", "interval_days": 2, "anchor_date": ""}
|
||
resp = client.put(f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/schedule', json=payload)
|
||
assert resp.status_code == 200
|
||
assert resp.get_json()["anchor_date"] == ""
|
||
|
||
|
||
def test_set_schedule_invalid_mode(client):
|
||
payload = {"mode": "weekly", "day_configs": []}
|
||
resp = client.put(f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/schedule', json=payload)
|
||
assert resp.status_code == 400
|
||
|
||
|
||
def test_set_schedule_upserts_existing(client):
|
||
# Set once with days mode
|
||
client.put(
|
||
f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/schedule',
|
||
json={"mode": "days", "day_configs": [{"day": 1, "hour": 8, "minute": 0}]},
|
||
)
|
||
# Overwrite with interval mode
|
||
client.put(
|
||
f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/schedule',
|
||
json={"mode": "interval", "interval_days": 2, "anchor_date": "", "interval_hour": 9, "interval_minute": 0},
|
||
)
|
||
resp = client.get(f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/schedule')
|
||
assert resp.status_code == 200
|
||
assert resp.get_json()["mode"] == "interval"
|
||
|
||
|
||
def test_old_record_missing_new_fields_loads_with_defaults(client):
|
||
"""Old DB records without anchor_date/interval_has_deadline load with correct defaults."""
|
||
from db.chore_schedules import upsert_schedule
|
||
from models.chore_schedule import ChoreSchedule
|
||
|
||
# Insert a schedule as if it was created before phase 2 (missing new fields)
|
||
old_style = ChoreSchedule(
|
||
child_id=TEST_CHILD_ID,
|
||
task_id=TEST_TASK_ID,
|
||
mode='interval',
|
||
interval_days=3,
|
||
anchor_date='', # default value
|
||
interval_has_deadline=True, # default value
|
||
interval_hour=8,
|
||
interval_minute=0,
|
||
)
|
||
upsert_schedule(old_style)
|
||
|
||
# Manually wipe the new fields from the raw stored record to simulate a pre-phase-2 record
|
||
from db.db import chore_schedules_db
|
||
from tinydb import Query
|
||
ScheduleQ = Query()
|
||
record = chore_schedules_db.search(
|
||
(ScheduleQ.child_id == TEST_CHILD_ID) & (ScheduleQ.task_id == TEST_TASK_ID)
|
||
)[0]
|
||
doc_id = chore_schedules_db.get(
|
||
(ScheduleQ.child_id == TEST_CHILD_ID) & (ScheduleQ.task_id == TEST_TASK_ID)
|
||
).doc_id
|
||
chore_schedules_db.update(
|
||
lambda rec: (
|
||
rec.pop('anchor_date', None),
|
||
rec.pop('interval_has_deadline', None),
|
||
),
|
||
doc_ids=[doc_id],
|
||
)
|
||
|
||
resp = client.get(f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/schedule')
|
||
assert resp.status_code == 200
|
||
data = resp.get_json()
|
||
assert data['anchor_date'] == ''
|
||
assert data['interval_has_deadline'] is True
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# DELETE schedule
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_delete_schedule(client):
|
||
client.put(
|
||
f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/schedule',
|
||
json={"mode": "days", "day_configs": []},
|
||
)
|
||
resp = client.delete(f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/schedule')
|
||
assert resp.status_code == 200
|
||
|
||
# Verify gone
|
||
resp2 = client.get(f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/schedule')
|
||
assert resp2.status_code == 404
|
||
|
||
|
||
def test_delete_schedule_not_found(client):
|
||
chore_schedules_db.truncate()
|
||
resp = client.delete(f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/schedule')
|
||
assert resp.status_code == 404
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# POST extend
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_extend_chore_time(client):
|
||
resp = client.post(
|
||
f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/extend',
|
||
json={"date": "2025-01-15"},
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.get_json()
|
||
assert data["child_id"] == TEST_CHILD_ID
|
||
assert data["task_id"] == TEST_TASK_ID
|
||
assert data["date"] == "2025-01-15"
|
||
|
||
|
||
def test_extend_chore_time_duplicate_returns_409(client):
|
||
task_extensions_db.truncate()
|
||
client.post(
|
||
f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/extend',
|
||
json={"date": "2025-01-15"},
|
||
)
|
||
resp = client.post(
|
||
f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/extend',
|
||
json={"date": "2025-01-15"},
|
||
)
|
||
assert resp.status_code == 409
|
||
|
||
|
||
def test_extend_chore_time_different_dates_allowed(client):
|
||
task_extensions_db.truncate()
|
||
r1 = client.post(
|
||
f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/extend',
|
||
json={"date": "2025-01-15"},
|
||
)
|
||
r2 = client.post(
|
||
f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/extend',
|
||
json={"date": "2025-01-16"},
|
||
)
|
||
assert r1.status_code == 200
|
||
assert r2.status_code == 200
|
||
|
||
|
||
def test_extend_chore_time_missing_date(client):
|
||
resp = client.post(f'/child/{TEST_CHILD_ID}/task/{TEST_TASK_ID}/extend', json={})
|
||
assert resp.status_code == 400
|
||
|
||
|
||
def test_extend_chore_time_bad_child(client):
|
||
resp = client.post('/child/bad-child/task/bad-task/extend', json={"date": "2025-01-15"})
|
||
assert resp.status_code == 404
|