Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_respond_file(self):
file_event = events.File(
file_bytes=b"aslkdjlaksdjlkajdlk", target="!test:localhost"
)
with amock.patch(
api_string.format("send_content")
) as patched_send, amock.patch(
api_string.format("media_upload")
) as patched_upload:
patched_upload.return_value = asyncio.Future()
patched_upload.return_value.set_result({"content_uri": "mxc://aurl"})
patched_send.return_value = asyncio.Future()
patched_send.return_value.set_result(None)
await self.connector.send(file_event)
patched_send.assert_called_once_with(
async def test_repeat_file_bytes(self, mock_get):
f = events.File(url="http://spam.eggs/monty.jpg")
fut = asyncio.Future()
fut.set_result(b"bob")
mock_get.return_value.__aenter__.return_value.read = amock.CoroutineMock(
return_value=fut
)
assert await f.get_file_bytes() == b"bob"
assert mock_get.call_count == 1
# Now test we don't re-download the url
assert await f.get_file_bytes() == b"bob"
assert mock_get.call_count == 1
async def test_file(self):
opsdroid = amock.CoroutineMock()
mock_connector = Connector({}, opsdroid=opsdroid)
event = events.File(
bytes("some file contents", "utf-8"),
user_id="user_id",
user="user",
target="default",
connector=mock_connector,
)
self.assertEqual(event.user_id, "user_id")
self.assertEqual(event.user, "user")
self.assertEqual(event.target, "default")
self.assertEqual((await event.get_file_bytes()).decode(), "some file contents")
def test_error_on_construct(self):
with self.assertRaises(ValueError):
events.File()
with self.assertRaises(ValueError):
events.File(b"a", "https://localhost")
async def create_file(self, event, roomid):
"""Send a File event."""
kwargs = await self._file_kwargs(event, roomid)
return events.File(**kwargs)
if not results: # pragma: nocover
return ""
# If we only have one result use it.
if len(results) == 1: # pragma: nocover
return results[0].mime_type
# If we have multiple matches with the same confidence, pick one that
# actually has a mime_type.
confidence = results[0].confidence
results = filter(lambda x: x.confidence == confidence, results)
results = list(filter(lambda x: bool(x.mime_type), results))
return results[0].mime_type
class Image(File):
"""Event class specifically for image files."""
async def get_dimensions(self):
"""Return the image dimensions `(w,h)`."""
fbytes = await self.get_file_bytes()
return get_image_size_from_bytesio(io.BytesIO(fbytes), len(fbytes))
class NewRoom(Event):
"""Event class to represent the creation of a new room."""
def __init__(self, name=None, params=None, *args, **kwargs): # noqa: D107
super().__init__(*args, **kwargs)
self.name = name
self.room_params = params or {}
@register_event(events.File)
@register_event(events.Image)
@ensure_room_id_and_send
async def _send_file(self, file_event):
mxc_url, uploaded = await self._file_to_mxc_url(file_event)
if isinstance(file_event, events.Image):
if uploaded:
extra_info = await self._get_image_info(file_event)
else:
extra_info = {}
msg_type = "m.image"
else:
extra_info = {}
msg_type = "m.file"
name = file_event.name or "opsdroid_upload"