devplacepy/tests/unit/push.py

# retoor <retoor@molodetz.nl>

import uuid
import requests
from tests.conftest import BASE_URL
def _session_push():
    s = requests.Session()
    name = f"push_{uuid.uuid4().hex[:10]}"
    s.post(
        f"{BASE_URL}/auth/signup",
        data={
            "username": name,
            "email": f"{name}@test.dev",
            "password": "secret123",
            "confirm_password": "secret123",
        },
        allow_redirects=True,
    )
    return s


def test_browser_base64_is_url_safe_unpadded():
    import base64
    from devplacepy import push

    encoded = push.browser_base64(b"\xff\xfe\x00 hello")
    assert "=" not in encoded
    assert "+" not in encoded and "/" not in encoded
    assert base64.urlsafe_b64decode(encoded + "==") == b"\xff\xfe\x00 hello"


def test_hkdf_returns_requested_length():
    from devplacepy import push

    derived = push.hkdf(b"input-key-material", b"salt", b"info", 16)
    assert isinstance(derived, bytes)
    assert len(derived) == 16


def test_public_key_standard_b64_non_empty():
    from devplacepy import push

    assert push.public_key_standard_b64()


def test_create_notification_authorization_is_jwt():
    from devplacepy import push

    token = push.create_notification_authorization("https://push.example.com/endpoint")
    assert token.count(".") == 2


def test_provider_registry_resolves_default_for_missing_name():
    from devplacepy.push import providers

    assert providers.get(None) is providers.PROVIDERS["webpush"]
    assert providers.get("") is providers.PROVIDERS["webpush"]
    assert providers.get("  APNS ") is providers.PROVIDERS["apns"]
    assert providers.get("nope") is None


def test_webpush_parse_registration_accepts_subscription_shape():
    from devplacepy.push import providers

    webpush = providers.PROVIDERS["webpush"]
    fields = webpush.parse_registration(
        {
            "endpoint": "https://push.example.com/sub",
            "expirationTime": None,
            "keys": {"p256dh": "p", "auth": "a"},
        }
    )
    assert fields == {
        "endpoint": "https://push.example.com/sub",
        "key_auth": "a",
        "key_p256dh": "p",
    }


def test_webpush_parse_registration_rejects_incomplete_bodies():
    from devplacepy.push import providers

    webpush = providers.PROVIDERS["webpush"]
    assert webpush.parse_registration({"endpoint": "https://push.example.com/x"}) is None
    assert webpush.parse_registration({"keys": {"p256dh": "p", "auth": "a"}}) is None
    assert (
        webpush.parse_registration(
            {"endpoint": "https://push.example.com/x", "keys": {"p256dh": "p"}}
        )
        is None
    )


def test_apns_parse_registration_validates_device_token():
    from devplacepy.push import providers

    apns = providers.PROVIDERS["apns"]
    token = "a1b2c3d4" * 8
    assert apns.parse_registration({"token": f" {token} "}) == {"token": token}
    assert apns.parse_registration({"token": "abc"}) is None
    assert apns.parse_registration({"token": "z" * 64}) is None
    assert apns.parse_registration({"token": "a" * 500}) is None
    assert apns.parse_registration({"token": None}) is None
    assert apns.parse_registration({}) is None


def test_apns_prepare_translates_the_shared_payload():
    import json
    from devplacepy.push import providers

    body = json.loads(
        providers.PROVIDERS["apns"].prepare(
            {
                "title": "DevPlace",
                "message": "You have a new notification.",
                "icon": "/static/apple-touch-icon.png",
                "url": "/notifications",
            }
        )
    )
    assert body["aps"]["alert"] == {
        "title": "DevPlace",
        "body": "You have a new notification.",
    }
    assert body["aps"]["thread-id"] == "devplace-notification"
    assert body["url"] == "/notifications"
    assert body["icon"] == "/static/apple-touch-icon.png"


def test_apns_prepare_survives_an_empty_payload():
    import json
    from devplacepy.push import providers

    body = json.loads(providers.PROVIDERS["apns"].prepare({}))
    assert body["aps"]["alert"]["title"] == "DevPlace"
    assert body["url"] == "/notifications"


def _apns_settings(monkeypatch, **values):
    from devplacepy.push.providers import apns

    defaults = {
        apns.TEAM_ID_KEY: "",
        apns.KEY_ID_KEY: "",
        apns.AUTH_KEY_KEY: "",
        apns.TOPIC_KEY: "",
        apns.ENVIRONMENT_KEY: "",
    }
    defaults.update(values)
    monkeypatch.setattr(apns, "_setting", lambda key: defaults.get(key, ""))
    apns._token_state.clear()
    return defaults


def _ec_private_key_pem():
    from cryptography.hazmat.primitives import serialization
    from cryptography.hazmat.primitives.asymmetric import ec

    key = ec.generate_private_key(ec.SECP256R1())
    return key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption(),
    ).decode("utf-8")


def test_apns_is_configured_requires_every_credential(monkeypatch):
    from devplacepy.push import providers
    from devplacepy.push.providers import apns

    provider = providers.PROVIDERS["apns"]
    _apns_settings(monkeypatch)
    assert provider.is_configured() is False
    assert providers.is_active(provider) is False

    _apns_settings(
        monkeypatch,
        **{
            apns.TEAM_ID_KEY: "TEAMID1234",
            apns.KEY_ID_KEY: "KEYID12345",
            apns.AUTH_KEY_KEY: "pem",
        },
    )
    assert provider.is_configured() is False

    _apns_settings(
        monkeypatch,
        **{
            apns.TEAM_ID_KEY: "TEAMID1234",
            apns.KEY_ID_KEY: "KEYID12345",
            apns.AUTH_KEY_KEY: "pem",
            apns.TOPIC_KEY: "nl.molodetz.devplace",
        },
    )
    assert provider.is_configured() is True


def test_apns_host_falls_back_to_production(monkeypatch):
    from devplacepy.push.providers import apns

    _apns_settings(monkeypatch)
    assert apns.host() == "api.push.apple.com"

    _apns_settings(monkeypatch, **{apns.ENVIRONMENT_KEY: "sandbox"})
    assert apns.host() == "api.sandbox.push.apple.com"

    _apns_settings(monkeypatch, **{apns.ENVIRONMENT_KEY: "nonsense"})
    assert apns.host() == "api.push.apple.com"


def test_apns_provider_token_is_signed_and_cached(monkeypatch):
    import jwt
    from devplacepy.push.providers import apns

    _apns_settings(monkeypatch)
    pem = _ec_private_key_pem()
    token = apns.provider_token("TEAMID1234", "KEYID12345", pem)
    assert apns.provider_token("TEAMID1234", "KEYID12345", pem) == token

    header = jwt.get_unverified_header(token)
    claims = jwt.decode(token, options={"verify_signature": False})
    assert header["alg"] == "ES256"
    assert header["kid"] == "KEYID12345"
    assert claims["iss"] == "TEAMID1234"
    assert isinstance(claims["iat"], int)

    other = apns.provider_token("TEAMID1234", "KEYID12345", _ec_private_key_pem())
    assert other != token


def test_apns_provider_token_rejects_a_broken_auth_key(monkeypatch):
    import pytest
    from devplacepy.push.providers import apns

    _apns_settings(monkeypatch)
    with pytest.raises(ValueError):
        apns.provider_token("TEAMID1234", "KEYID12345", "not-a-pem")
    with pytest.raises(ValueError):
        apns.provider_token("TEAMID1234", "KEYID12345", "not-a-pem")


def _apns_response_status(monkeypatch, status, body):
    import httpx
    from tests.conftest import run_async
    from devplacepy.push import providers
    from devplacepy.push.providers import apns

    _apns_settings(
        monkeypatch,
        **{
            apns.TEAM_ID_KEY: "TEAMID1234",
            apns.KEY_ID_KEY: "KEYID12345",
            apns.AUTH_KEY_KEY: _ec_private_key_pem(),
            apns.TOPIC_KEY: "nl.molodetz.devplace",
        },
    )
    provider = providers.PROVIDERS["apns"]
    seen = {}

    def handler(request):
        seen["url"] = str(request.url)
        seen["headers"] = dict(request.headers)
        return httpx.Response(status, json=body)

    async def run():
        transport = httpx.MockTransport(handler)
        async with httpx.AsyncClient(transport=transport) as client:
            return await provider.deliver(
                client, {"token": "a" * 64}, provider.prepare({"message": "hi"})
            )

    return run_async(run()), seen


def test_apns_delivery_maps_statuses(monkeypatch):
    from devplacepy.push import providers

    accepted, seen = _apns_response_status(monkeypatch, 200, {})
    assert accepted.status == providers.ACCEPTED
    assert seen["url"] == f"https://api.push.apple.com/3/device/{'a' * 64}"
    assert seen["headers"]["apns-topic"] == "nl.molodetz.devplace"
    assert seen["headers"]["apns-push-type"] == "alert"
    assert seen["headers"]["apns-priority"] == "10"
    assert seen["headers"]["authorization"].startswith("bearer ")
    assert int(seen["headers"]["apns-expiration"]) > 0
    assert seen["headers"]["apns-id"]

    gone, _ = _apns_response_status(monkeypatch, 410, {"reason": "Unregistered"})
    assert gone.status == providers.DEAD

    bad_token, _ = _apns_response_status(monkeypatch, 400, {"reason": "BadDeviceToken"})
    assert bad_token.status == providers.DEAD

    payload_error, _ = _apns_response_status(
        monkeypatch, 400, {"reason": "PayloadTooLarge"}
    )
    assert payload_error.status == providers.REJECTED

    throttled, _ = _apns_response_status(monkeypatch, 429, {"reason": "TooManyRequests"})
    assert throttled.status == providers.REJECTED


def test_apns_delivery_without_configuration_never_raises(monkeypatch):
    import httpx
    from tests.conftest import run_async
    from devplacepy.push import providers

    _apns_settings(monkeypatch)
    provider = providers.PROVIDERS["apns"]

    async def run():
        transport = httpx.MockTransport(lambda request: httpx.Response(200, json={}))
        async with httpx.AsyncClient(transport=transport) as client:
            return await provider.deliver(client, {"token": "a" * 64}, "{}")

    assert run_async(run()).status == providers.REJECTED


def test_group_by_provider_treats_a_missing_provider_as_webpush():
    from devplacepy.push.delivery import group_by_provider

    grouped = group_by_provider(
        [
            {"id": 1, "provider": None},
            {"id": 2, "provider": ""},
            {"id": 3, "provider": "webpush"},
            {"id": 4, "provider": "apns"},
        ]
    )
    assert sorted(grouped) == ["apns", "webpush"]
    assert len(grouped["webpush"]) == 3
    assert len(grouped["apns"]) == 1


def test_delivery_timeout_is_clamped(monkeypatch):
    from devplacepy.push import delivery

    monkeypatch.setattr(delivery, "get_int_setting", lambda key, default: default)
    assert delivery.timeout_seconds() == float(delivery.DEFAULT_TIMEOUT_SECONDS)

    monkeypatch.setattr(delivery, "get_int_setting", lambda key, default: 0)
    assert delivery.timeout_seconds() == float(delivery.MIN_TIMEOUT_SECONDS)

    monkeypatch.setattr(delivery, "get_int_setting", lambda key, default: 100000)
    assert delivery.timeout_seconds() == float(delivery.MAX_TIMEOUT_SECONDS)