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_no_mime_type(self):
opsdroid = amock.CoroutineMock()
mock_connector = Connector({}, opsdroid=opsdroid)
event = events.Image(
b"aslkdjsalkdjlaj",
user_id="user_id",
user="user",
target="default",
connector=mock_connector,
)
self.assertEqual(await event.get_mimetype(), "")
async def test_respond_image(self):
gif_bytes = (
b"GIF89a\x01\x00\x01\x00\x00\xff\x00,"
b"\x00\x00\x00\x00\x01\x00\x01\x00\x00\x02\x00;"
)
image = events.Image(file_bytes=gif_bytes, 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(image)
patched_send.assert_called_once_with(
"!test:localhost",
"mxc://aurl",
async def test_explicit_mime_type(self):
opsdroid = amock.CoroutineMock()
mock_connector = Connector({}, opsdroid=opsdroid)
event = events.Image(
self.gif_bytes,
user_id="user_id",
user="user",
target="default",
mimetype="image/jpeg",
connector=mock_connector,
)
self.assertEqual(await event.get_mimetype(), "image/jpeg")
async def test_image(self):
opsdroid = amock.CoroutineMock()
mock_connector = Connector({}, opsdroid=opsdroid)
event = events.Image(
self.gif_bytes,
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(), self.gif_bytes)
self.assertEqual(await event.get_mimetype(), "image/gif")
self.assertEqual(await event.get_dimensions(), (1, 1))
async def test_respond_image_failure(self):
post_response = amock.Mock()
post_response.status = 400
gif_bytes = (
b"GIF89a\x01\x00\x01\x00\x00\xff\x00,"
b"\x00\x00\x00\x00\x01\x00\x01\x00\x00\x02\x00;"
)
image = Image(file_bytes=gif_bytes, target={"id": "123"})
with amock.patch.object(self.connector.session, "post") as patched_request:
patched_request.return_value = asyncio.Future()
patched_request.return_value.set_result(post_response)
await self.connector.send_image(image)
self.assertLogs("_LOOGER", "debug")
async def test_respond_mxc(self):
gif_bytes = (
b"GIF89a\x01\x00\x01\x00\x00\xff\x00,"
b"\x00\x00\x00\x00\x01\x00\x01\x00\x00\x02\x00;"
)
image = events.Image(url="mxc://aurl", target="!test:localhost")
with amock.patch(
api_string.format("send_content")
) as patched_send, amock.patch(
"opsdroid.events.Image.get_file_bytes"
) as patched_bytes:
patched_bytes.return_value = asyncio.Future()
patched_bytes.return_value.set_result(gif_bytes)
patched_send.return_value = asyncio.Future()
patched_send.return_value.set_result(None)
await self.connector.send(image)
patched_send.assert_called_once_with(
"!test:localhost", "mxc://aurl", "opsdroid_upload", "m.image", {}
)
@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"
await self.connection.send_content(
async def create_image(self, event, roomid):
"""Send a Image event."""
kwargs = await self._file_kwargs(event, roomid)
return events.Image(**kwargs)
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"
await self.connection.send_content(
file_event.target, mxc_url, name, msg_type, extra_info
)