andros
d5377e94eb
Django backend that listens to the relay SSE global stream and dispatches APNs push notifications to subscribed iOS devices (TestFlight/sandbox).
102 lines
3.7 KiB
Python
102 lines
3.7 KiB
Python
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from app.push.apns import _author_from_post, _build_payload
|
|
|
|
|
|
class TestAuthorFromPost:
|
|
def test_extracts_netloc(self):
|
|
post = "https://alice.org/social.org#2025-02-05T11:20:00+0100"
|
|
assert _author_from_post(post) == "alice.org"
|
|
|
|
def test_post_without_fragment(self):
|
|
post = "https://alice.org/social.org"
|
|
assert _author_from_post(post) == "alice.org"
|
|
|
|
|
|
class TestBuildPayload:
|
|
def test_mention(self):
|
|
notification = {"type": "mention", "post": "https://alice.org/social.org#ts"}
|
|
payload = _build_payload(notification)
|
|
assert "alice.org mentioned you" in payload["aps"]["alert"]["body"]
|
|
|
|
def test_reaction(self):
|
|
notification = {
|
|
"type": "reaction",
|
|
"post": "https://alice.org/social.org#ts",
|
|
"emoji": "❤",
|
|
"parent": "https://example.com/social.org#ts2",
|
|
}
|
|
payload = _build_payload(notification)
|
|
assert "❤" in payload["aps"]["alert"]["body"]
|
|
|
|
def test_reply(self):
|
|
notification = {
|
|
"type": "reply",
|
|
"post": "https://bob.org/social.org#ts",
|
|
"parent": "https://example.com/social.org#ts2",
|
|
}
|
|
payload = _build_payload(notification)
|
|
assert "replied" in payload["aps"]["alert"]["body"]
|
|
|
|
def test_boost(self):
|
|
notification = {
|
|
"type": "boost",
|
|
"post": "https://bob.org/social.org#ts",
|
|
"boosted": "https://example.com/social.org#ts2",
|
|
}
|
|
payload = _build_payload(notification)
|
|
assert "boosted" in payload["aps"]["alert"]["body"]
|
|
|
|
def test_notification_included_in_payload(self):
|
|
notification = {"type": "mention", "post": "https://alice.org/social.org#ts"}
|
|
payload = _build_payload(notification)
|
|
assert payload["notification"] == notification
|
|
|
|
|
|
@pytest.mark.django_db
|
|
class TestSendPush:
|
|
@patch("app.push.apns.httpx.Client")
|
|
@patch("app.push.apns._get_token", return_value="fake-token")
|
|
def test_send_push_success(self, mock_token, mock_client_class):
|
|
from app.push.apns import send_push
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_client = MagicMock()
|
|
mock_client.__enter__ = MagicMock(return_value=mock_client)
|
|
mock_client.__exit__ = MagicMock(return_value=False)
|
|
mock_client.post.return_value = mock_response
|
|
mock_client_class.return_value = mock_client
|
|
|
|
result = send_push(
|
|
"device_token_abc", {"type": "mention", "post": "https://a.org/social.org#ts"}
|
|
)
|
|
assert result is True
|
|
|
|
@patch("app.push.apns.httpx.Client")
|
|
@patch("app.push.apns._get_token", return_value="fake-token")
|
|
def test_send_push_removes_stale_token(self, mock_token, mock_client_class):
|
|
from app.push.apns import send_push
|
|
from app.subscriptions.models import Device
|
|
|
|
device = Device.objects.create(
|
|
feed="https://example.com/social.org", device_token="stale_token_xyz"
|
|
)
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 410
|
|
mock_response.text = '{"reason": "Unregistered"}'
|
|
mock_client = MagicMock()
|
|
mock_client.__enter__ = MagicMock(return_value=mock_client)
|
|
mock_client.__exit__ = MagicMock(return_value=False)
|
|
mock_client.post.return_value = mock_response
|
|
mock_client_class.return_value = mock_client
|
|
|
|
result = send_push(
|
|
device.device_token, {"type": "mention", "post": "https://a.org/social.org#ts"}
|
|
)
|
|
assert result is False
|
|
assert not Device.objects.filter(device_token="stale_token_xyz").exists()
|