from __future__ import annotations import json from unittest.mock import AsyncMock, Mock, patch from asgiref.sync import async_to_sync from django.conf import settings from django.test import TestCase, TransactionTestCase from core.clients import transport from core.clients.signal import SignalClient from core.models import ChatSession, Message, Person, PersonIdentifier, User from core.views.compose import _build_signal_reply_metadata class SignalReplyMetadataTests(TestCase): def setUp(self): self.user = User.objects.create_user( username="signal-reply-meta-user", email="signal-reply-meta@example.com", password="x", ) self.person = Person.objects.create(user=self.user, name="Signal Reply") self.identifier = PersonIdentifier.objects.create( user=self.user, person=self.person, service="signal", identifier="+15550001000", ) self.session = ChatSession.objects.create( user=self.user, identifier=self.identifier, ) def test_build_signal_reply_metadata_uses_signal_source(self): incoming = Message.objects.create( user=self.user, session=self.session, sender_uuid="+15550001000", text="quoted body", ts=1772538353497, source_service="signal", source_message_id="1772538353497", source_chat_id="+15550001000", ) payload = _build_signal_reply_metadata(incoming, "+15550001000") self.assertEqual(1772538353497, payload.get("quote_timestamp")) self.assertEqual("+15550001000", payload.get("quote_author")) self.assertEqual("quoted body", payload.get("quote_text")) def test_build_signal_reply_metadata_uses_chat_number_when_sender_is_uuid(self): incoming = Message.objects.create( user=self.user, session=self.session, sender_uuid="756078fd-d447-426d-a620-581a86d64f51", text="quoted body", ts=1772538353497, source_service="signal", source_message_id="1772538353497", source_chat_id="+15550001000", ) payload = _build_signal_reply_metadata(incoming, "+15550001000") self.assertEqual("+15550001000", payload.get("quote_author")) def test_build_signal_reply_metadata_uses_local_sender_for_own_messages(self): outgoing = Message.objects.create( user=self.user, session=self.session, sender_uuid="", custom_author="USER", text="my previous message", ts=1772538353900, source_service="web", source_message_id="1772538353900", source_chat_id="+15550001000", ) payload = _build_signal_reply_metadata(outgoing, "+15550001000") expected_author = str(getattr(settings, "SIGNAL_NUMBER", "") or "").strip() if expected_author: self.assertEqual(expected_author, payload.get("quote_author")) else: self.assertEqual("+15550001000", payload.get("quote_author")) class SignalTransportSendTests(TestCase): def test_transport_passes_reply_metadata_to_signal_api(self): with patch( "core.clients.transport.prepare_outbound_attachments", new=AsyncMock(return_value=[]), ), patch( "core.clients.transport.signalapi.send_message_raw", new=AsyncMock(return_value=1772538354000), ) as mocked_send: result = async_to_sync(transport.send_message_raw)( "signal", "+15550001000", text="reply payload", attachments=[], metadata={ "quote_timestamp": 1772538353497, "quote_author": "+15550001000", "quote_text": "quoted body", }, ) self.assertEqual(1772538354000, result) mocked_send.assert_awaited_once_with( "+15550001000", "reply payload", [], metadata={ "quote_timestamp": 1772538353497, "quote_author": "+15550001000", "quote_text": "quoted body", }, ) class SignalInboundReplyLinkTests(TransactionTestCase): def setUp(self): self.user = User.objects.create_user( username="signal-inbound-user", email="signal-inbound@example.com", password="x", ) self.person = Person.objects.create(user=self.user, name="Signal Inbound") self.identifier = PersonIdentifier.objects.create( user=self.user, person=self.person, service="signal", identifier="+15550002000", ) self.session = ChatSession.objects.create( user=self.user, identifier=self.identifier, ) self.anchor = Message.objects.create( user=self.user, session=self.session, sender_uuid="+15550002000", text="anchor inbound", ts=1772545458187, source_service="signal", source_message_id="1772545458187", source_chat_id="+15550002000", ) def test_process_raw_inbound_event_links_signal_reply(self): fake_ur = Mock() fake_ur.message_received = AsyncMock(return_value=None) client = SignalClient.__new__(SignalClient) client.service = "signal" client.ur = fake_ur client.log = Mock() client.client = Mock() client.client.bot_uuid = "" client.client.phone_number = "" client._resolve_signal_identifiers = AsyncMock(return_value=[self.identifier]) client._auto_link_single_user_signal_identifier = AsyncMock(return_value=[]) payload = { "envelope": { "sourceNumber": "+15550002000", "sourceUuid": "756078fd-d447-426d-a620-581a86d64f51", "timestamp": 1772545462051, "dataMessage": { "message": "reply inbound s3", "quote": { "targetSentTimestamp": 1772545458187, "authorNumber": "+15550002000", }, }, } } async_to_sync(client._process_raw_inbound_event)(json.dumps(payload)) created = ( Message.objects.filter( user=self.user, session=self.session, text="reply inbound s3", ) .order_by("-ts") .first() ) self.assertIsNotNone(created) self.assertEqual(self.anchor.id, created.reply_to_id) self.assertEqual("1772545458187", created.reply_source_message_id) def test_process_raw_inbound_event_applies_reaction(self): fake_ur = Mock() fake_ur.message_received = AsyncMock(return_value=None) fake_ur.xmpp = Mock() fake_ur.xmpp.client = Mock() fake_ur.xmpp.client.apply_external_reaction = AsyncMock(return_value=None) client = SignalClient.__new__(SignalClient) client.service = "signal" client.ur = fake_ur client.log = Mock() client.client = Mock() client.client.bot_uuid = "" client.client.phone_number = "" client._resolve_signal_identifiers = AsyncMock(return_value=[self.identifier]) client._auto_link_single_user_signal_identifier = AsyncMock(return_value=[]) payload = { "envelope": { "sourceNumber": "+15550002000", "sourceUuid": "756078fd-d447-426d-a620-581a86d64f51", "timestamp": 1772545463000, "dataMessage": { "reaction": { "emoji": "❤️", "targetSentTimestamp": 1772545458187, } }, } } async_to_sync(client._process_raw_inbound_event)(json.dumps(payload)) self.anchor.refresh_from_db() reactions = list((self.anchor.receipt_payload or {}).get("reactions") or []) self.assertTrue( any(str(row.get("emoji") or "") == "❤️" for row in reactions), "Expected Signal heart reaction to be applied to anchor receipt payload.", ) def test_process_raw_inbound_event_applies_sync_reaction_using_destination_fallback( self, ): fake_ur = Mock() fake_ur.message_received = AsyncMock(return_value=None) fake_ur.xmpp = Mock() fake_ur.xmpp.client = Mock() fake_ur.xmpp.client.apply_external_reaction = AsyncMock(return_value=None) client = SignalClient.__new__(SignalClient) client.service = "signal" client.ur = fake_ur client.log = Mock() client.client = Mock() # Emulate sync event from our own linked device. client.client.bot_uuid = "f5f53a90-8b8c-4f0c-9520-8f13aab0b219" client.client.phone_number = "+15550009999" client._resolve_signal_identifiers = AsyncMock(return_value=[self.identifier]) client._auto_link_single_user_signal_identifier = AsyncMock(return_value=[]) payload = { "envelope": { "sourceNumber": "+15550009999", "sourceUuid": "f5f53a90-8b8c-4f0c-9520-8f13aab0b219", "destinationNumber": "+15550002000", "timestamp": 1772545465000, "syncMessage": { "sentMessage": { "destinationNumber": "+15550002000", "message": { "reaction": { "emoji": "🔥", "targetSentTimestamp": 1772545458187, } }, } }, } } async_to_sync(client._process_raw_inbound_event)(json.dumps(payload)) self.anchor.refresh_from_db() reactions = list((self.anchor.receipt_payload or {}).get("reactions") or []) self.assertTrue( any(str(row.get("emoji") or "") == "🔥" for row in reactions), "Expected sync reaction to be applied via destination-number fallback resolution.", ) def test_process_raw_inbound_event_applies_edit(self): fake_ur = Mock() fake_ur.message_received = AsyncMock(return_value=None) fake_ur.xmpp = Mock() fake_ur.xmpp.client = Mock() fake_ur.xmpp.client.apply_external_reaction = AsyncMock(return_value=None) client = SignalClient.__new__(SignalClient) client.service = "signal" client.ur = fake_ur client.log = Mock() client.client = Mock() client.client.bot_uuid = "" client.client.phone_number = "" client._resolve_signal_identifiers = AsyncMock(return_value=[self.identifier]) client._auto_link_single_user_signal_identifier = AsyncMock(return_value=[]) payload = { "envelope": { "sourceNumber": "+15550002000", "sourceUuid": "756078fd-d447-426d-a620-581a86d64f51", "timestamp": 1772545466000, "dataMessage": { "editMessage": { "targetSentTimestamp": 1772545458187, "dataMessage": {"message": "anchor edited"}, } }, } } async_to_sync(client._process_raw_inbound_event)(json.dumps(payload)) self.anchor.refresh_from_db() self.assertEqual("anchor edited", str(self.anchor.text or "")) edits = list((self.anchor.receipt_payload or {}).get("edit_history") or []) self.assertEqual(1, len(edits)) def test_process_raw_inbound_event_applies_delete_tombstone_flag(self): fake_ur = Mock() fake_ur.message_received = AsyncMock(return_value=None) fake_ur.xmpp = Mock() fake_ur.xmpp.client = Mock() fake_ur.xmpp.client.apply_external_reaction = AsyncMock(return_value=None) client = SignalClient.__new__(SignalClient) client.service = "signal" client.ur = fake_ur client.log = Mock() client.client = Mock() client.client.bot_uuid = "" client.client.phone_number = "" client._resolve_signal_identifiers = AsyncMock(return_value=[self.identifier]) client._auto_link_single_user_signal_identifier = AsyncMock(return_value=[]) payload = { "envelope": { "sourceNumber": "+15550002000", "sourceUuid": "756078fd-d447-426d-a620-581a86d64f51", "timestamp": 1772545467000, "dataMessage": { "delete": { "targetSentTimestamp": 1772545458187, } }, } } async_to_sync(client._process_raw_inbound_event)(json.dumps(payload)) self.anchor.refresh_from_db() self.assertTrue(bool((self.anchor.receipt_payload or {}).get("is_deleted"))) self.assertTrue(bool((self.anchor.receipt_payload or {}).get("deleted") or {})) class SignalRuntimeCommandWritebackTests(TestCase): def setUp(self): self.user = User.objects.create_user( username="signal-runtime-writeback-user", email="signal-runtime-writeback@example.com", password="x", ) self.person = Person.objects.create(user=self.user, name="Signal Runtime") self.identifier = PersonIdentifier.objects.create( user=self.user, person=self.person, service="signal", identifier="+15550003000", ) self.session = ChatSession.objects.create( user=self.user, identifier=self.identifier ) self.message = Message.objects.create( user=self.user, session=self.session, sender_uuid="", custom_author="USER", text="queued signal send", ts=1772545467000, delivered_ts=None, source_service="web", source_message_id="1772545467000", source_chat_id="+15550003000", ) def test_execute_runtime_send_updates_legacy_message_ids(self): client = SignalClient.__new__(SignalClient) client.service = "signal" client.log = Mock() command = { "id": "cmd-1", "action": "send_message_raw", "payload": { "recipient": "+15550003000", "text": "queued signal send", "attachments": [], "metadata": {"legacy_message_id": str(self.message.id)}, }, } with patch( "core.clients.signal.signalapi.send_message_raw", new=AsyncMock(return_value=1772545467999), ): async_to_sync(client._execute_runtime_command)(command) self.message.refresh_from_db() self.assertEqual(1772545467999, int(self.message.delivered_ts or 0)) self.assertEqual("1772545467999", str(self.message.source_message_id or ""))