chore/backend/tests/test_web_push.py

"""Tests for web push notifications triggered by child actions.

Patches `utils.push_sender.webpush` so no real HTTP requests are made.
A push subscription is seeded in push_subscriptions_db so send_push_to_user
actually calls webpush (rather than short-circuiting at "no subscriptions").
"""
import json
import pytest
from unittest.mock import patch, MagicMock
from flask import Flask
from werkzeug.security import generate_password_hash
from tinydb import Query

from api.child_api import child_api
from api.auth_api import auth_api
from db.db import (
    child_db, task_db, reward_db, users_db,
    pending_confirmations_db, push_subscriptions_db,
)
from tests.conftest import TEST_SECRET_KEY, TEST_REFRESH_TOKEN_EXPIRY_DAYS

TEST_USER_ID = "wp_test_user_id"
TEST_EMAIL = "wptest@example.com"
TEST_PASSWORD = "wptestpass"
TEST_CHILD_ID = "wp_child_id"
TEST_TASK_ID = "wp_task_id"
TEST_REWARD_ID_AFFORD = "wp_reward_afford"
TEST_REWARD_ID_CANT = "wp_reward_cant"
TEST_ENDPOINT = "https://push.example.com/wp_endpoint"


def seed_data():
    users_db.remove(Query().id == TEST_USER_ID)
    child_db.remove(Query().id == TEST_CHILD_ID)
    task_db.remove(Query().id == TEST_TASK_ID)
    reward_db.remove((Query().id == TEST_REWARD_ID_AFFORD) | (Query().id == TEST_REWARD_ID_CANT))
    pending_confirmations_db.remove(Query().user_id == TEST_USER_ID)
    push_subscriptions_db.remove(Query().user_id == TEST_USER_ID)

    users_db.insert({
        "id": TEST_USER_ID,
        "first_name": "Push",
        "last_name": "Tester",
        "email": TEST_EMAIL,
        "password": generate_password_hash(TEST_PASSWORD),
        "verified": True,
        "role": "user",
        "image_id": None,
        "marked_for_deletion": False,
        "marked_for_deletion_at": None,
        "timezone": "America/New_York",
        "email_digest_enabled": True,
    })
    child_db.insert({
        "id": TEST_CHILD_ID,
        "user_id": TEST_USER_ID,
        "name": "Push Child",
        "age": 8,
        "points": 50,
        "tasks": [TEST_TASK_ID],
        "rewards": [TEST_REWARD_ID_AFFORD, TEST_REWARD_ID_CANT],
        "image_id": None,
    })
    task_db.insert({
        "id": TEST_TASK_ID,
        "user_id": TEST_USER_ID,
        "name": "Clean Room",
        "points": 10,
        "type": "chore",
        "image_id": None,
    })
    # Affordable reward (cost <= child.points = 50)
    reward_db.insert({
        "id": TEST_REWARD_ID_AFFORD,
        "user_id": TEST_USER_ID,
        "name": "Movie Night",
        "cost": 20,
        "image_id": None,
    })
    # Unaffordable reward (cost > child.points = 50)
    reward_db.insert({
        "id": TEST_REWARD_ID_CANT,
        "user_id": TEST_USER_ID,
        "name": "New Bike",
        "cost": 200,
        "image_id": None,
    })


def seed_subscription():
    push_subscriptions_db.insert({
        "id": "wp_sub_id",
        "user_id": TEST_USER_ID,
        "endpoint": TEST_ENDPOINT,
        "keys": {"p256dh": "BNgz3test", "auth": "authtest"},
        "created_at": "2024-01-01T00:00:00+00:00",
    })


def cleanup():
    users_db.remove(Query().id == TEST_USER_ID)
    child_db.remove(Query().id == TEST_CHILD_ID)
    task_db.remove(Query().id == TEST_TASK_ID)
    reward_db.remove((Query().id == TEST_REWARD_ID_AFFORD) | (Query().id == TEST_REWARD_ID_CANT))
    pending_confirmations_db.remove(Query().user_id == TEST_USER_ID)
    push_subscriptions_db.remove(Query().user_id == TEST_USER_ID)


@pytest.fixture
def client():
    app = Flask(__name__)
    app.register_blueprint(child_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
    app.config['FRONTEND_URL'] = 'http://localhost:5173'
    cleanup()
    seed_data()
    with app.test_client() as c:
        c.post('/auth/login', json={"email": TEST_EMAIL, "password": TEST_PASSWORD})
        yield c
    cleanup()


class TestWebPushOnChoreConfirm:
    def test_push_fired_to_all_subscriptions_on_chore_confirm(self, client):
        """Web push is sent to all stored subscriptions when a chore is confirmed."""
        seed_subscription()
        with patch('utils.push_sender.webpush') as mock_webpush:
            resp = client.post(
                f'/child/{TEST_CHILD_ID}/confirm-chore',
                json={'task_id': TEST_TASK_ID},
            )
        assert resp.status_code == 200
        assert mock_webpush.called

    def test_push_payload_includes_required_fields(self, client):
        """Push payload includes user_id, approve_token, and deny_token."""
        seed_subscription()
        with patch('utils.push_sender.webpush') as mock_webpush:
            client.post(
                f'/child/{TEST_CHILD_ID}/confirm-chore',
                json={'task_id': TEST_TASK_ID},
            )
        call_kwargs = mock_webpush.call_args[1]
        payload = json.loads(call_kwargs['data'])
        assert payload['user_id'] == TEST_USER_ID
        assert 'approve_token' in payload
        assert 'deny_token' in payload

    def test_push_not_fired_when_no_subscriptions(self, client):
        """No error is raised when the parent has no push subscriptions."""
        push_subscriptions_db.remove(Query().user_id == TEST_USER_ID)
        with patch('utils.push_sender.webpush') as mock_webpush:
            resp = client.post(
                f'/child/{TEST_CHILD_ID}/confirm-chore',
                json={'task_id': TEST_TASK_ID},
            )
        assert resp.status_code == 200
        assert not mock_webpush.called


class TestWebPushOnRewardRequest:
    def test_push_fired_for_affordable_reward_request(self, client):
        """Web push is fired when a child requests a reward they can afford."""
        seed_subscription()
        with patch('utils.push_sender.webpush') as mock_webpush:
            resp = client.post(
                f'/child/{TEST_CHILD_ID}/request-reward',
                json={'reward_id': TEST_REWARD_ID_AFFORD},
            )
        assert resp.status_code == 200
        assert mock_webpush.called
        # Cleanup pending
        pending_confirmations_db.remove(Query().child_id == TEST_CHILD_ID)

    def test_push_not_fired_for_unaffordable_reward(self, client):
        """Web push is NOT fired when a child requests a reward they cannot afford."""
        seed_subscription()
        with patch('utils.push_sender.webpush') as mock_webpush:
            resp = client.post(
                f'/child/{TEST_CHILD_ID}/request-reward',
                json={'reward_id': TEST_REWARD_ID_CANT},
            )
        # request-reward rejects unaffordable requests with 400
        assert resp.status_code == 400
        assert not mock_webpush.called

    def test_push_not_fired_for_reward_when_no_subscriptions(self, client):
        """No error is raised for reward requests when the parent has no subscriptions."""
        push_subscriptions_db.remove(Query().user_id == TEST_USER_ID)
        with patch('utils.push_sender.webpush') as mock_webpush:
            resp = client.post(
                f'/child/{TEST_CHILD_ID}/request-reward',
                json={'reward_id': TEST_REWARD_ID_AFFORD},
            )
        assert resp.status_code == 200
        assert not mock_webpush.called
        # Cleanup pending
        pending_confirmations_db.remove(Query().child_id == TEST_CHILD_ID)