Custom emoji support

tests/factories.py

@@ -204,3 +204,8 @@ class InboxObjectFactory(factory.alchemy.SQLAlchemyModelFactory):
             # Hide replies from the stream
             is_hidden_from_stream=True if ro.in_reply_to else False,
         )
+
+
+class FollowerFactory(factory.alchemy.SQLAlchemyModelFactory):
+    class Meta(BaseModelMeta):
+        model = models.Follower

tests/test_emoji.py

@@ -0,0 +1,61 @@
+from fastapi.testclient import TestClient
+
+from app import activitypub as ap
+from app import models
+from app.config import generate_csrf_token
+from app.database import Session
+from app.utils.emoji import EMOJIS_BY_NAME
+from tests.utils import generate_admin_session_cookies
+
+
+def test_emoji_are_loaded() -> None:
+    assert len(EMOJIS_BY_NAME) >= 1
+
+
+def test_emoji_ap_endpoint(db: Session, client: TestClient) -> None:
+    response = client.get("/e/goose_hacker", headers={"Accept": ap.AP_CONTENT_TYPE})
+    assert response.status_code == 200
+    assert response.headers["content-type"] == ap.AP_CONTENT_TYPE
+    emoji_resp = response.json()
+    assert emoji_resp["type"] == "Emoji"
+
+
+def test_emoji_ap_endpoint__not_found(db: Session, client: TestClient) -> None:
+    response = client.get("/e/goose_hacker2", headers={"Accept": ap.AP_CONTENT_TYPE})
+    assert response.status_code == 404
+
+
+def test_emoji_note_with_emoji(db: Session, client: TestClient) -> None:
+    # Call admin endpoint to create a note with
+    note_content = "😺 :goose_hacker:"
+
+    response = client.post(
+        "/admin/actions/new",
+        data={
+            "redirect_url": "http://testserver/",
+            "content": note_content,
+            "visibility": ap.VisibilityEnum.PUBLIC.name,
+            "csrf_token": generate_csrf_token(),
+        },
+        cookies=generate_admin_session_cookies(),
+    )
+
+    # Then the server returns a 302
+    assert response.status_code == 302
+
+    # And the Follow activity was created in the outbox
+    outbox_object = db.query(models.OutboxObject).one()
+    assert outbox_object.ap_type == "Note"
+    assert len(outbox_object.tags) == 1
+    emoji_tag = outbox_object.tags[0]
+    assert emoji_tag["type"] == "Emoji"
+    assert emoji_tag["name"] == ":goose_hacker:"
+    url = emoji_tag["icon"]["url"]
+
+    # And the custom emoji is rendered in the HTML version
+    html_resp = client.get("/o/" + outbox_object.public_id)
+    html_resp.raise_for_status()
+    assert html_resp.status_code == 200
+    assert url in html_resp.text
+    # And the unicode emoji is rendered with twemoji
+    assert f'/static/twemoji/{hex(ord("😺"))[2:]}.svg' in html_resp.text

tests/test_process_outgoing_activities.py

@@ -120,32 +120,6 @@ def test_process_next_outgoing_activity__error_500(
     assert outgoing_activity.tries == 1
 
 
-def test_process_next_outgoing_activity__connect_error(
-    db: Session,
-    respx_mock: respx.MockRouter,
-) -> None:
-    outbox_object = _setup_outbox_object()
-    recipient_inbox_url = "https://example.com/inbox"
-    respx_mock.post(recipient_inbox_url).mock(side_effect=httpx.ConnectError)
-
-    # And an outgoing activity
-    outgoing_activity = factories.OutgoingActivityFactory(
-        recipient=recipient_inbox_url,
-        outbox_object_id=outbox_object.id,
-    )
-
-    # When processing the next outgoing activity
-    # Then it is processed
-    assert process_next_outgoing_activity(db) is True
-
-    assert respx_mock.calls.call_count == 1
-
-    outgoing_activity = db.query(models.OutgoingActivity).one()
-    assert outgoing_activity.is_sent is False
-    assert outgoing_activity.error is not None
-    assert outgoing_activity.tries == 1
-
-
 def test_process_next_outgoing_activity__errored(
     db: Session,
     respx_mock: respx.MockRouter,
@@ -179,5 +153,31 @@ def test_process_next_outgoing_activity__errored(
     assert process_next_outgoing_activity(db) is False
 
 
+def test_process_next_outgoing_activity__connect_error(
+    db: Session,
+    respx_mock: respx.MockRouter,
+) -> None:
+    outbox_object = _setup_outbox_object()
+    recipient_inbox_url = "https://example.com/inbox"
+    respx_mock.post(recipient_inbox_url).mock(side_effect=httpx.ConnectError)
+
+    # And an outgoing activity
+    outgoing_activity = factories.OutgoingActivityFactory(
+        recipient=recipient_inbox_url,
+        outbox_object_id=outbox_object.id,
+    )
+
+    # When processing the next outgoing activity
+    # Then it is processed
+    assert process_next_outgoing_activity(db) is True
+
+    assert respx_mock.calls.call_count == 1
+
+    outgoing_activity = db.query(models.OutgoingActivity).one()
+    assert outgoing_activity.is_sent is False
+    assert outgoing_activity.error is not None
+    assert outgoing_activity.tries == 1
+
+
 # TODO(ts):
 # - parse retry after