104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
from unittest.mock import AsyncMock, patch
|
|
|
|
from asgiref.sync import async_to_sync
|
|
from django.test import SimpleTestCase, TestCase, override_settings
|
|
|
|
from core.clients import transport
|
|
from core.clients.xmpp import XMPPComponent, _resolve_person_from_xmpp_localpart
|
|
from core.models import Person, PersonIdentifier, User
|
|
|
|
|
|
class SignalAttachmentFetchTests(SimpleTestCase):
|
|
def test_signal_service_allows_direct_url_fetch(self):
|
|
response = AsyncMock()
|
|
response.status = 200
|
|
response.headers = {"Content-Type": "image/png"}
|
|
response.read = AsyncMock(return_value=b"png-bytes")
|
|
|
|
request_ctx = AsyncMock()
|
|
request_ctx.__aenter__.return_value = response
|
|
request_ctx.__aexit__.return_value = False
|
|
|
|
session = AsyncMock()
|
|
session.get.return_value = request_ctx
|
|
|
|
session_ctx = AsyncMock()
|
|
session_ctx.__aenter__.return_value = session
|
|
session_ctx.__aexit__.return_value = False
|
|
|
|
with patch(
|
|
"core.clients.transport.aiohttp.ClientSession",
|
|
return_value=session_ctx,
|
|
):
|
|
fetched = async_to_sync(transport.fetch_attachment)(
|
|
"signal",
|
|
{
|
|
"url": "https://example.com/file_share/demo.png",
|
|
"filename": "demo.png",
|
|
"content_type": "image/png",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(b"png-bytes", fetched["content"])
|
|
self.assertEqual("image/png", fetched["content_type"])
|
|
self.assertEqual("demo.png", fetched["filename"])
|
|
self.assertEqual(9, fetched["size"])
|
|
|
|
|
|
@override_settings(
|
|
XMPP_JID="component.example.test",
|
|
XMPP_USER_DOMAIN="example.test",
|
|
)
|
|
class XMPPContactJidTests(TestCase):
|
|
def _component(self):
|
|
return XMPPComponent(
|
|
ur=AsyncMock(),
|
|
jid="component.example.test",
|
|
secret="secret",
|
|
server="localhost",
|
|
port=5347,
|
|
)
|
|
|
|
def test_resolve_person_from_escaped_localpart(self):
|
|
user = User.objects.create_user(username="user", password="pw")
|
|
person = Person.objects.create(user=user, name="Misinformation Club")
|
|
|
|
resolved = _resolve_person_from_xmpp_localpart(
|
|
user=user,
|
|
localpart_value=r"misinformation\20club",
|
|
)
|
|
|
|
self.assertEqual(person.id, resolved.id)
|
|
|
|
def test_send_from_external_escapes_contact_jid(self):
|
|
user = User.objects.create_user(username="user2", password="pw")
|
|
person = Person.objects.create(user=user, name="Misinformation Club")
|
|
identifier = PersonIdentifier.objects.create(
|
|
user=user,
|
|
person=person,
|
|
service="signal",
|
|
identifier="group.example",
|
|
)
|
|
component = self._component()
|
|
component.send_xmpp_message = AsyncMock(return_value="xmpp-id")
|
|
|
|
with (
|
|
patch("core.clients.xmpp.transport.record_bridge_mapping"),
|
|
patch("core.clients.xmpp.history.save_bridge_ref", new=AsyncMock()),
|
|
):
|
|
async_to_sync(component.send_from_external)(
|
|
user,
|
|
identifier,
|
|
"hello",
|
|
False,
|
|
attachments=[],
|
|
source_ref={},
|
|
)
|
|
|
|
component.send_xmpp_message.assert_awaited_once_with(
|
|
"user2@example.test",
|
|
r"misinformation\20club|signal@component.example.test",
|
|
"hello",
|
|
use_omemo_encryption=True,
|
|
)
|