127 lines
4.3 KiB
Python
127 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
from asgiref.sync import async_to_sync
|
|
from django.test import TestCase
|
|
|
|
from core.messaging import history
|
|
from core.models import ChatSession, Message, Person, PersonIdentifier, User
|
|
from core.views.compose import _serialize_message
|
|
|
|
|
|
class ReactionNormalizationTests(TestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(
|
|
username="react-normalize",
|
|
email="react-normalize@example.com",
|
|
password="pw",
|
|
)
|
|
self.person = Person.objects.create(user=self.user, name="Reaction Person")
|
|
self.identifier = PersonIdentifier.objects.create(
|
|
user=self.user,
|
|
person=self.person,
|
|
service="signal",
|
|
identifier="+15551239999",
|
|
)
|
|
self.session = ChatSession.objects.create(
|
|
user=self.user,
|
|
identifier=self.identifier,
|
|
)
|
|
|
|
def test_apply_reaction_prefers_exact_source_timestamp_match(self):
|
|
near_message = Message.objects.create(
|
|
user=self.user,
|
|
session=self.session,
|
|
ts=1700000000100,
|
|
sender_uuid="author-near",
|
|
text="near",
|
|
source_service="signal",
|
|
source_message_id="1700000000100",
|
|
)
|
|
exact_message = Message.objects.create(
|
|
user=self.user,
|
|
session=self.session,
|
|
ts=1700000000000,
|
|
sender_uuid="author-exact",
|
|
text="exact",
|
|
source_service="signal",
|
|
source_message_id="1700000000000",
|
|
)
|
|
|
|
updated = async_to_sync(history.apply_reaction)(
|
|
self.user,
|
|
self.identifier,
|
|
target_ts=1700000000000,
|
|
emoji="❤️",
|
|
source_service="signal",
|
|
actor="reactor-1",
|
|
target_author="author-exact",
|
|
remove=False,
|
|
payload={"origin": "test"},
|
|
)
|
|
|
|
self.assertEqual(str(exact_message.id), str(updated.id))
|
|
exact_message.refresh_from_db()
|
|
near_message.refresh_from_db()
|
|
self.assertEqual(1, len((exact_message.receipt_payload or {}).get("reactions") or []))
|
|
self.assertEqual(
|
|
"exact_source_message_id_ts",
|
|
str((exact_message.receipt_payload or {}).get("reaction_last_match_strategy") or ""),
|
|
)
|
|
self.assertEqual(0, len((near_message.receipt_payload or {}).get("reactions") or []))
|
|
|
|
def test_remove_without_emoji_is_audited_not_active(self):
|
|
message = Message.objects.create(
|
|
user=self.user,
|
|
session=self.session,
|
|
ts=1700000001000,
|
|
sender_uuid="author-1",
|
|
text="msg",
|
|
source_service="signal",
|
|
source_message_id="1700000001000",
|
|
)
|
|
|
|
async_to_sync(history.apply_reaction)(
|
|
self.user,
|
|
self.identifier,
|
|
target_ts=1700000001000,
|
|
emoji="",
|
|
source_service="whatsapp",
|
|
actor="actor-1",
|
|
remove=True,
|
|
payload={"origin": "test"},
|
|
)
|
|
|
|
message.refresh_from_db()
|
|
payload = dict(message.receipt_payload or {})
|
|
self.assertEqual([], list(payload.get("reactions") or []))
|
|
self.assertEqual(1, len(list(payload.get("reaction_events") or [])))
|
|
|
|
def test_emoji_only_reply_text_is_not_reaction(self):
|
|
anchor = Message.objects.create(
|
|
user=self.user,
|
|
session=self.session,
|
|
ts=1700000002000,
|
|
sender_uuid="author-1",
|
|
text="anchor",
|
|
source_service="signal",
|
|
source_message_id="1700000002000",
|
|
)
|
|
heart_reply = Message.objects.create(
|
|
user=self.user,
|
|
session=self.session,
|
|
ts=1700000003000,
|
|
sender_uuid="author-2",
|
|
text="❤️",
|
|
source_service="signal",
|
|
source_message_id="1700000003000",
|
|
reply_to=anchor,
|
|
reply_source_service="signal",
|
|
reply_source_message_id="1700000002000",
|
|
receipt_payload={},
|
|
)
|
|
|
|
serialized = _serialize_message(heart_reply)
|
|
self.assertEqual("❤️", serialized["text"])
|
|
self.assertEqual([], list(serialized.get("reactions") or []))
|
|
self.assertEqual(str(anchor.id), serialized["reply_to_id"])
|