How to use the opsdroid.events.File function in opsdroid

To help you get started, we’ve selected a few opsdroid examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github opsdroid / opsdroid / tests / test_connector_matrix.py View on Github external
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(
github opsdroid / opsdroid / tests / test_events.py View on Github external
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
github opsdroid / opsdroid / tests / test_events.py View on Github external
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")
github opsdroid / opsdroid / tests / test_events.py View on Github external
def test_error_on_construct(self):
        with self.assertRaises(ValueError):
            events.File()

        with self.assertRaises(ValueError):
            events.File(b"a", "https://localhost")
github opsdroid / opsdroid / opsdroid / connector / matrix / create_events.py View on Github external
async def create_file(self, event, roomid):
        """Send a File event."""
        kwargs = await self._file_kwargs(event, roomid)
        return events.File(**kwargs)
github opsdroid / opsdroid / opsdroid / events.py View on Github external
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 {}
github opsdroid / opsdroid / opsdroid / connector / matrix / connector.py View on Github external
    @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"