Ryan Kegel
ad2bdf4c4f
All checks were successful
Chore App Build, Test, and Push Docker Images / build-and-push (push) Successful in 2m14s
- Implemented push subscription API with tests for subscribing and unsubscribing users. - Created web push notification tests triggered by child actions. - Added digest scheduler to send email digests to users at 9 PM local time. - Developed utility functions for creating and validating digest action tokens. - Integrated web push sender to handle sending notifications to users. - Added service worker for handling push notifications in the frontend. - Created a push opt-in component for user notification preferences. - Implemented tests for the push opt-in component to ensure correct behavior. - Updated frontend services to manage push subscriptions and permissions.
134 lines
4.8 KiB
Python
134 lines
4.8 KiB
Python
import os
|
|
import time
|
|
import pytest
|
|
from tinydb import Query
|
|
|
|
from db.db import digest_action_tokens_db
|
|
from utils.digest_token import (
|
|
create_action_token,
|
|
validate_and_consume_token,
|
|
create_unsubscribe_token,
|
|
validate_unsubscribe_token,
|
|
)
|
|
|
|
|
|
def cleanup():
|
|
digest_action_tokens_db.truncate()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_tokens():
|
|
cleanup()
|
|
yield
|
|
cleanup()
|
|
|
|
|
|
class TestCreateActionToken:
|
|
def test_returns_token_with_correct_fields(self):
|
|
token = create_action_token('u1', 'c1', 'e1', 'chore', 'approve')
|
|
assert token.user_id == 'u1'
|
|
assert token.child_id == 'c1'
|
|
assert token.entity_id == 'e1'
|
|
assert token.entity_type == 'chore'
|
|
assert token.action == 'approve'
|
|
assert token.used is False
|
|
assert token.signature
|
|
|
|
def test_persists_to_db(self):
|
|
token = create_action_token('u1', 'c1', 'e1', 'chore', 'approve')
|
|
stored = digest_action_tokens_db.get(Query().id == token.id)
|
|
assert stored is not None
|
|
assert stored['entity_type'] == 'chore'
|
|
|
|
def test_different_tokens_have_unique_ids(self):
|
|
t1 = create_action_token('u1', 'c1', 'e1', 'chore', 'approve')
|
|
t2 = create_action_token('u1', 'c1', 'e1', 'reward', 'deny')
|
|
assert t1.id != t2.id
|
|
|
|
|
|
class TestValidateAndConsumeToken:
|
|
def test_valid_token_is_returned_and_marked_used(self):
|
|
token = create_action_token('u1', 'c1', 'e1', 'chore', 'approve')
|
|
result = validate_and_consume_token(token.id)
|
|
assert result is not None
|
|
assert result.id == token.id
|
|
|
|
stored = digest_action_tokens_db.get(Query().id == token.id)
|
|
assert stored['used'] is True
|
|
|
|
def test_nonexistent_token_returns_none(self):
|
|
assert validate_and_consume_token('nonexistent-id') is None
|
|
|
|
def test_already_used_token_returns_none(self):
|
|
token = create_action_token('u1', 'c1', 'e1', 'chore', 'approve')
|
|
validate_and_consume_token(token.id) # first use
|
|
result = validate_and_consume_token(token.id) # second use
|
|
assert result is None
|
|
|
|
def test_expired_token_returns_none(self):
|
|
# Create a token that is already expired
|
|
from datetime import datetime, timedelta, timezone
|
|
from db.digest_action_tokens import insert_token
|
|
from models.digest_action_token import DigestActionToken
|
|
import uuid, json, hmac, hashlib
|
|
|
|
token_id = str(uuid.uuid4())
|
|
expires_at = (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat()
|
|
payload = {
|
|
'id': token_id,
|
|
'user_id': 'u1',
|
|
'child_id': 'c1',
|
|
'entity_id': 'e1',
|
|
'entity_type': 'chore',
|
|
'action': 'approve',
|
|
'expires_at': expires_at,
|
|
}
|
|
secret = os.environ.get('DIGEST_TOKEN_SECRET')
|
|
canonical = json.dumps(payload, sort_keys=True, separators=(',', ':'))
|
|
sig = hmac.new(secret.encode(), canonical.encode(), hashlib.sha256).hexdigest()
|
|
token = DigestActionToken(
|
|
id=token_id, user_id='u1', child_id='c1', entity_id='e1',
|
|
entity_type='chore', action='approve', expires_at=expires_at,
|
|
used=False, signature=sig,
|
|
)
|
|
insert_token(token)
|
|
assert validate_and_consume_token(token_id) is None
|
|
|
|
def test_tampered_signature_returns_none(self):
|
|
token = create_action_token('u1', 'c1', 'e1', 'chore', 'approve')
|
|
# Tamper the signature in DB
|
|
digest_action_tokens_db.update(
|
|
{'signature': 'deadbeef' * 8},
|
|
Query().id == token.id
|
|
)
|
|
assert validate_and_consume_token(token.id) is None
|
|
|
|
|
|
class TestUnsubscribeToken:
|
|
def test_create_and_validate(self):
|
|
token = create_unsubscribe_token('user123')
|
|
assert token
|
|
result = validate_unsubscribe_token(token)
|
|
assert result == 'user123'
|
|
|
|
def test_invalid_token_returns_none(self):
|
|
assert validate_unsubscribe_token('garbage-token') is None
|
|
|
|
def test_tampered_token_returns_none(self):
|
|
token = create_unsubscribe_token('user123')
|
|
# Change one character
|
|
tampered = token[:-1] + ('A' if token[-1] != 'A' else 'B')
|
|
assert validate_unsubscribe_token(tampered) is None
|
|
|
|
def test_expired_token_returns_none(self):
|
|
"""Build a token with a past expiry timestamp by mocking time."""
|
|
import base64, hmac, hashlib
|
|
user_id = 'user_expired'
|
|
expiry_ts = int(time.time()) - 1 # already expired
|
|
payload_str = f"{user_id}:{expiry_ts}"
|
|
secret = os.environ.get('DIGEST_TOKEN_SECRET')
|
|
sig = hmac.new(secret.encode(), payload_str.encode(), hashlib.sha256).hexdigest()
|
|
raw = f"{payload_str}:{sig}"
|
|
token = base64.urlsafe_b64encode(raw.encode()).decode()
|
|
assert validate_unsubscribe_token(token) is None
|