Web Push Testing Guide: VAPID Keys, pywebpush, and Notification Payload Validation
Web Push notifications rely on three moving parts: VAPID authentication, encrypted payloads using the Web Cryptography API, and browser subscription endpoints from push services like FCM or Mozilla's autopush. Testing all three layers without a real browser is entirely possible — this guide shows how using pywebpush in Python and the web-push library in Node.js, with Jest mocking the service worker registration API.
Key Takeaways
Generate real VAPID key pairs in tests using py_vapid. Hardcoding VAPID keys in tests is a security risk and breaks when keys rotate; generating ephemeral key pairs in fixtures gives you realistic test behavior without committing private keys.
Subscription endpoints are URLs — test that you validate them. A subscription object with a missing endpoint or an endpoint that doesn't start with https:// should be rejected before any network call is made.
Payload encryption is not optional — test the encrypted output structure. Web Push payloads must be encrypted with the subscription's p256dh key and auth secret; test that your service passes these to the webpush library rather than sending plaintext.
Test VAPID key rotation without breaking active subscriptions. When you rotate VAPID keys, all existing subscriptions become invalid — your tests must cover the transition period where old subscriptions return 410 Gone.
Mock service worker registration in Jest tests with a structured stub. Browser ServiceWorkerRegistration has specific methods (showNotification, getNotifications) that your frontend code depends on; a typed mock prevents runtime errors that unit tests would otherwise miss.
Web Push is the only push notification system that works across all major browsers without a vendor-specific SDK on the sender side. The downside: it's more complex to test because it involves asymmetric encryption, subscription objects with per-user keys, and browser-side service worker APIs. Let's break it down layer by layer.
The Web Push Flow
Before writing tests, understand what you're testing:
- The browser generates a push subscription (endpoint URL + public key + auth secret)
- The subscription is sent to your server
- Your server uses the subscription's public key and auth secret to encrypt the payload
- Your server authenticates the push request using VAPID (Voluntary Application Server Identification)
- The push service delivers the encrypted payload to the browser
- The service worker decrypts and displays the notification
Your server code owns steps 2–4. That's what we test.
Python Setup with pywebpush
pip install pywebpush py-vapid pytest pytest-mock cryptographyCreate a thin wrapper around pywebpush that your tests can target:
# myapp/web_push_service.py
import json
from py_vapid import Vapid01
from pywebpush import webpush, WebPushException
class WebPushService:
def __init__(self, vapid_private_key: str, vapid_claims: dict):
self.vapid_private_key = vapid_private_key
self.vapid_claims = vapid_claims
def validate_subscription(self, subscription: dict) -> None:
if not subscription.get("endpoint"):
raise ValueError("Subscription missing endpoint")
if not subscription["endpoint"].startswith("https://"):
raise ValueError("Endpoint must use HTTPS")
if not subscription.get("keys", {}).get("p256dh"):
raise ValueError("Subscription missing p256dh key")
if not subscription.get("keys", {}).get("auth"):
raise ValueError("Subscription missing auth secret")
def send(self, subscription: dict, title: str, body: str, data: dict = None) -> bool:
self.validate_subscription(subscription)
payload = json.dumps({
"title": title,
"body": body,
"data": data or {}
})
try:
webpush(
subscription_info=subscription,
data=payload,
vapid_private_key=self.vapid_private_key,
vapid_claims=self.vapid_claims
)
return True
except WebPushException as e:
if e.response and e.response.status_code == 410:
raise SubscriptionExpiredError(subscription["endpoint"])
raiseTesting VAPID Key Generation and Validation
Generate real ephemeral key pairs in test fixtures instead of hardcoding keys:
# test_vapid_keys.py
import pytest
from py_vapid import Vapid01
import base64
@pytest.fixture
def vapid_keypair():
"""Generate a fresh VAPID key pair for each test."""
vapid = Vapid01()
vapid.generate_keys()
return {
"private_key": vapid.private_key,
"public_key": vapid.public_key,
"private_pem": vapid.private_pem().decode("utf-8"),
"public_b64": vapid.public_key_urlsafe_base64
}
def test_vapid_public_key_is_valid_base64url(vapid_keypair):
pub = vapid_keypair["public_b64"]
# URL-safe base64 — no +, / characters
assert "+" not in pub
assert "/" not in pub
# Should decode to 65 bytes (uncompressed EC point)
decoded = base64.urlsafe_b64decode(pub + "==")
assert len(decoded) == 65
def test_vapid_private_key_pem_format(vapid_keypair):
pem = vapid_keypair["private_pem"]
assert pem.startswith("-----BEGIN EC PRIVATE KEY-----") or \
pem.startswith("-----BEGIN PRIVATE KEY-----")
def test_vapid_claims_require_subject(vapid_keypair):
from myapp.web_push_service import WebPushService
# Claims without 'sub' should fail when building auth headers
with pytest.raises(ValueError, match="sub"):
service = WebPushService(
vapid_private_key=vapid_keypair["private_pem"],
vapid_claims={"aud": "https://push.example.com"} # missing 'sub'
)
service._validate_claims()Testing Subscription Validation
# test_subscription_validation.py
import pytest
from myapp.web_push_service import WebPushService
@pytest.fixture
def service():
return WebPushService(
vapid_private_key="fake-key",
vapid_claims={"sub": "mailto:admin@example.com"}
)
VALID_SUBSCRIPTION = {
"endpoint": "https://fcm.googleapis.com/fcm/send/abc123",
"keys": {
"p256dh": "BBc3MV8hBqtAHv7JXYZ",
"auth": "HqABC123XYZ"
}
}
def test_valid_subscription_passes_validation(service):
# Should not raise
service.validate_subscription(VALID_SUBSCRIPTION)
def test_missing_endpoint_raises(service):
sub = {**VALID_SUBSCRIPTION, "endpoint": ""}
with pytest.raises(ValueError, match="missing endpoint"):
service.validate_subscription(sub)
def test_http_endpoint_rejected(service):
sub = {**VALID_SUBSCRIPTION, "endpoint": "http://insecure.example.com/push/abc"}
with pytest.raises(ValueError, match="must use HTTPS"):
service.validate_subscription(sub)
def test_missing_p256dh_raises(service):
sub = {**VALID_SUBSCRIPTION, "keys": {"auth": "HqABC123"}}
with pytest.raises(ValueError, match="p256dh"):
service.validate_subscription(sub)
def test_missing_auth_raises(service):
sub = {**VALID_SUBSCRIPTION, "keys": {"p256dh": "BBc3MV8hBqt"}}
with pytest.raises(ValueError, match="auth secret"):
service.validate_subscription(sub)Testing the Send Path with Mocked pywebpush
# test_web_push_send.py
from unittest.mock import patch, MagicMock
import pytest
from pywebpush import WebPushException
from myapp.web_push_service import WebPushService, SubscriptionExpiredError
SUBSCRIPTION = {
"endpoint": "https://fcm.googleapis.com/fcm/send/token123",
"keys": {"p256dh": "BBc3MV8hBqtA", "auth": "HqABC123"}
}
@pytest.fixture
def service():
return WebPushService(
vapid_private_key="fake-private-key",
vapid_claims={"sub": "mailto:admin@example.com"}
)
@patch("myapp.web_push_service.webpush")
def test_send_calls_webpush_with_subscription(mock_webpush, service):
mock_webpush.return_value = MagicMock(status_code=201)
result = service.send(
subscription=SUBSCRIPTION,
title="Flash Sale",
body="60% off for the next hour",
data={"url": "/sale"}
)
assert result is True
mock_webpush.assert_called_once()
call_kwargs = mock_webpush.call_args.kwargs
assert call_kwargs["subscription_info"] == SUBSCRIPTION
assert call_kwargs["vapid_private_key"] == "fake-private-key"
assert "Flash Sale" in call_kwargs["data"]
@patch("myapp.web_push_service.webpush")
def test_send_raises_on_410_gone(mock_webpush, service):
"""410 means the subscription is expired — must signal caller to remove it."""
mock_response = MagicMock(status_code=410)
mock_webpush.side_effect = WebPushException("Gone", response=mock_response)
with pytest.raises(SubscriptionExpiredError):
service.send(subscription=SUBSCRIPTION, title="Hi", body="There")
@patch("myapp.web_push_service.webpush")
def test_payload_is_json_with_title_body_data(mock_webpush, service):
import json
mock_webpush.return_value = MagicMock(status_code=201)
service.send(
subscription=SUBSCRIPTION,
title="Update available",
body="Version 2.1 is ready",
data={"version": "2.1", "release_notes_url": "/changelog"}
)
payload_str = mock_webpush.call_args.kwargs["data"]
payload = json.loads(payload_str)
assert payload["title"] == "Update available"
assert payload["data"]["version"] == "2.1"Node.js: Testing with web-push and Jest
// __tests__/webPush.test.js
const webpush = require('web-push');
const { sendWebPush, validateSubscription } = require('../src/webPushService');
jest.mock('web-push');
const MOCK_SUBSCRIPTION = {
endpoint: 'https://fcm.googleapis.com/fcm/send/abc123',
keys: {
p256dh: 'BBc3MV8hBqtAHvXYZ',
auth: 'HqABC123XYZ',
},
};
beforeEach(() => {
jest.clearAllMocks();
webpush.sendNotification.mockResolvedValue({ statusCode: 201 });
});
describe('sendWebPush', () => {
it('calls webpush.sendNotification with encrypted payload', async () => {
await sendWebPush(MOCK_SUBSCRIPTION, {
title: 'New order',
body: 'Your coffee is ready',
data: { orderId: '42' },
});
expect(webpush.sendNotification).toHaveBeenCalledTimes(1);
const [sub, payload] = webpush.sendNotification.mock.calls[0];
expect(sub).toEqual(MOCK_SUBSCRIPTION);
const parsed = JSON.parse(payload);
expect(parsed.title).toBe('New order');
expect(parsed.data.orderId).toBe('42');
});
it('removes subscription from db on 410 Gone', async () => {
const { removeSubscription } = require('../src/subscriptionStore');
jest.mock('../src/subscriptionStore');
webpush.sendNotification.mockRejectedValue({ statusCode: 410 });
await sendWebPush(MOCK_SUBSCRIPTION, { title: 'Hi', body: 'There' });
expect(removeSubscription).toHaveBeenCalledWith(MOCK_SUBSCRIPTION.endpoint);
});
});Service Worker Mock Testing
// __tests__/serviceWorker.test.js
// Mock the browser service worker API for unit testing notification display logic
const mockRegistration = {
showNotification: jest.fn().mockResolvedValue(undefined),
getNotifications: jest.fn().mockResolvedValue([]),
};
global.self = {
registration: mockRegistration,
addEventListener: jest.fn(),
};
const { handlePushEvent, handleNotificationClick } = require('../src/sw-handlers');
describe('push event handler', () => {
beforeEach(() => jest.clearAllMocks());
it('shows notification from push data', async () => {
const pushEvent = {
data: {
json: () => ({
title: 'Shipment update',
body: 'Package arrived at hub',
data: { trackingId: 'TRK-9901' },
}),
},
waitUntil: jest.fn(),
};
handlePushEvent(pushEvent);
await Promise.resolve(); // flush microtasks
expect(mockRegistration.showNotification).toHaveBeenCalledWith(
'Shipment update',
expect.objectContaining({ body: 'Package arrived at hub' })
);
});
});Testing Web Push thoroughly means covering the JavaScript frontend, the server-side encryption and VAPID signing, and the failure modes like subscription expiry. The layered approach here — validate inputs, mock the encryption library, test error codes — gives you confidence without needing a real browser or push service endpoint.