Cover the push providers with tests and document the subsystem

Unit tests for the provider registry, both providers' registration parsing, the APNs payload translation, provider token signing and caching, header and status mapping against a mock transport, provider grouping and the delivery timeout clamp, plus service tests for the configuration surface, the retention sweep and the per-provider metrics. Api tests cover the provider listing on GET /push.json, registration with and without an explicit provider, idempotency and the rejection of an unknown or unconfigured provider. Provider settings in unit tests are supplied by monkeypatching the provider's setting reader rather than writing site_settings, because the unit tier shares its database with the running api-tier server. devplacepy/push/CLAUDE.md documents the protocol, how to add a provider, the invariants and the APNs specifics; the root, routers and services files point at it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-31 22:47:20 +02:00
parent 53ddf4f233
commit 7674dac628
7 changed files with 520 additions and 2 deletions
@@ -48,3 +48,298 @@ def test_create_notification_authorization_is_jwt():

    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)