chore/backend/tests/test_push_subscription_api.py

import os
import pytest
from flask import Flask
from werkzeug.security import generate_password_hash
from tinydb import Query

from api.push_subscription_api import push_subscription_api
from api.auth_api import auth_api
from db.db import users_db, push_subscriptions_db
from tests.conftest import TEST_SECRET_KEY, TEST_REFRESH_TOKEN_EXPIRY_DAYS

TEST_EMAIL = "pushtest@example.com"
TEST_PASSWORD = "pushpassword123"
TEST_USER_ID = "push_test_user_id"
TEST_ENDPOINT = "https://fcm.googleapis.com/fcm/send/test-endpoint-abc"
TEST_KEYS = {"p256dh": "BNgz3XcMv1", "auth": "abc123"}


def seed_user():
    users_db.remove(Query().email == TEST_EMAIL)
    users_db.insert({
        "id": TEST_USER_ID,
        "first_name": "Push",
        "last_name": "Tester",
        "email": TEST_EMAIL,
        "password": generate_password_hash(TEST_PASSWORD),
        "verified": True,
        "image_id": None,
        "marked_for_deletion": False,
        "marked_for_deletion_at": None,
        "role": "user",
        "timezone": None,
        "email_digest_enabled": True,
    })


def cleanup():
    users_db.remove(Query().email == TEST_EMAIL)
    push_subscriptions_db.remove(Query().user_id == TEST_USER_ID)


@pytest.fixture
def client():
    app = Flask(__name__)
    app.register_blueprint(push_subscription_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['VAPID_PUBLIC_KEY'] = 'test-vapid-public-key'
    app.config['FRONTEND_URL'] = 'http://localhost:5173'
    cleanup()
    seed_user()
    with app.test_client() as c:
        yield c
    cleanup()


@pytest.fixture
def auth_client(client):
    client.post('/auth/login', json={"email": TEST_EMAIL, "password": TEST_PASSWORD})
    return client


class TestVapidKey:
    def test_returns_public_key(self, client):
        res = client.get('/push-vapid-key')
        assert res.status_code == 200
        assert res.get_json()['public_key'] == 'test-vapid-public-key'

    def test_unauthenticated_ok(self, client):
        # VAPID key endpoint is public
        res = client.get('/push-vapid-key')
        assert res.status_code == 200


class TestSubscribe:
    def test_subscribe_stores_subscription(self, auth_client):
        res = auth_client.post('/push-subscription', json={
            "endpoint": TEST_ENDPOINT,
            "keys": TEST_KEYS,
        })
        assert res.status_code == 200
        data = res.get_json()
        assert 'id' in data

        saved = push_subscriptions_db.get(
            (Query().user_id == TEST_USER_ID) & (Query().endpoint == TEST_ENDPOINT)
        )
        assert saved is not None

    def test_subscribe_updates_timezone(self, auth_client):
        auth_client.post('/push-subscription', json={
            "endpoint": TEST_ENDPOINT,
            "keys": TEST_KEYS,
            "timezone": "America/New_York",
        })
        user = users_db.get(Query().id == TEST_USER_ID)
        assert user['timezone'] == 'America/New_York'

    def test_subscribe_upserts_same_endpoint(self, auth_client):
        auth_client.post('/push-subscription', json={"endpoint": TEST_ENDPOINT, "keys": TEST_KEYS})
        auth_client.post('/push-subscription', json={"endpoint": TEST_ENDPOINT, "keys": TEST_KEYS})
        count = len(push_subscriptions_db.search(
            (Query().user_id == TEST_USER_ID) & (Query().endpoint == TEST_ENDPOINT)
        ))
        assert count == 1

    def test_subscribe_requires_endpoint(self, auth_client):
        res = auth_client.post('/push-subscription', json={"keys": TEST_KEYS})
        assert res.status_code == 400

    def test_subscribe_requires_keys(self, auth_client):
        res = auth_client.post('/push-subscription', json={"endpoint": TEST_ENDPOINT})
        assert res.status_code == 400

    def test_subscribe_requires_p256dh_and_auth(self, auth_client):
        res = auth_client.post('/push-subscription', json={
            "endpoint": TEST_ENDPOINT,
            "keys": {"p256dh": "only-one-key"},
        })
        assert res.status_code == 400

    def test_subscribe_requires_auth(self, client):
        res = client.post('/push-subscription', json={
            "endpoint": TEST_ENDPOINT,
            "keys": TEST_KEYS,
        })
        assert res.status_code == 401

    def test_subscribe_multiple_endpoints_per_user(self, auth_client):
        """Multiple subscriptions can coexist for the same user (one per device)."""
        endpoint2 = "https://fcm.googleapis.com/fcm/send/second-device-endpoint"
        keys2 = {"p256dh": "BNgz3XcMv2", "auth": "def456"}

        auth_client.post('/push-subscription', json={"endpoint": TEST_ENDPOINT, "keys": TEST_KEYS})
        auth_client.post('/push-subscription', json={"endpoint": endpoint2, "keys": keys2})

        subs = push_subscriptions_db.search(Query().user_id == TEST_USER_ID)
        assert len(subs) == 2
        endpoints = {s['endpoint'] for s in subs}
        assert TEST_ENDPOINT in endpoints
        assert endpoint2 in endpoints


class TestUnsubscribe:
    def test_unsubscribe_removes_subscription(self, auth_client):
        auth_client.post('/push-subscription', json={"endpoint": TEST_ENDPOINT, "keys": TEST_KEYS})
        res = auth_client.delete('/push-subscription', json={"endpoint": TEST_ENDPOINT})
        assert res.status_code == 200
        data = res.get_json()
        assert data['removed'] >= 1

        saved = push_subscriptions_db.get(
            (Query().user_id == TEST_USER_ID) & (Query().endpoint == TEST_ENDPOINT)
        )
        assert saved is None

    def test_unsubscribe_nonexistent_endpoint_ok(self, auth_client):
        res = auth_client.delete('/push-subscription', json={"endpoint": "https://nonexistent"})
        assert res.status_code == 200
        assert res.get_json()['removed'] == 0

    def test_unsubscribe_requires_endpoint(self, auth_client):
        res = auth_client.delete('/push-subscription', json={})
        assert res.status_code == 400

    def test_unsubscribe_requires_auth(self, client):
        res = client.delete('/push-subscription', json={"endpoint": TEST_ENDPOINT})
        assert res.status_code == 401