# 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)
Reference in New Issue View Git Blame Copy Permalink