OpenSource/python-telegram-bot
OpenSource/python-telegram-bot
1
0
Fork 0
mirror of https://github.com/python-telegram-bot/python-telegram-bot.git synced 2024-12-22 14:35:00 +01:00
Code Issues Projects Releases Packages Wiki Activity

Split File.download Into File.download_to_drive And File.download_to_memory (#3223)

Browse source
This commit is contained in:
Poolitzer 2022-11-02 08:28:41 +01:00 committed by GitHub
parent 55106d6d57
commit 636654cb71
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
17 changed files with 261 additions and 98 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
examples/conversationbot.py
View file

@ -23,7 +23,7 @@ try:
except ImportError:
__version_info__ = (0, 0, 0, 0, 0) # type: ignore[assignment]
if __version_info__ < (20, 0, 0, "alpha", 1):
if __version_info__ < (20, 0, 0, "beta", 0):
raise RuntimeError(
f"This example is not compatible with your current PTB version {TG_VER}. To view the "
f"{TG_VER} version of this example, "
@ -81,7 +81,7 @@ async def photo(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
"""Stores the photo and asks for a location."""
user = update.message.from_user
photo_file = await update.message.photo[-1].get_file()
await photo_file.download("user_photo.jpg")
await photo_file.download_to_memory("user_photo.jpg")
logger.info("Photo of %s: %s", user.first_name, "user_photo.jpg")
await update.message.reply_text(
"Gorgeous! Now, send me your location please, or send /skip if you don't want to."

12
examples/passportbot.py
View file

@ -21,7 +21,7 @@ try:
except ImportError:
__version_info__ = (0, 0, 0, 0, 0) # type: ignore[assignment]
if __version_info__ < (20, 0, 0, "alpha", 1):
if __version_info__ < (20, 0, 0, "beta", 0):
raise RuntimeError(
f"This example is not compatible with your current PTB version {TG_VER}. To view the "
f"{TG_VER} version of this example, "
@ -77,25 +77,25 @@ async def msg(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
for file in data.files:
actual_file = await file.get_file()
print(actual_file)
await actual_file.download()
await actual_file.download_to_memory()
if (
data.type in ("passport", "driver_license", "identity_card", "internal_passport")
and data.front_side
):
front_file = await data.front_side.get_file()
print(data.type, front_file)
await front_file.download()
await front_file.download_to_memory()
if data.type in ("driver_license" and "identity_card") and data.reverse_side:
reverse_file = await data.reverse_side.get_file()
print(data.type, reverse_file)
await reverse_file.download()
await reverse_file.download_to_memory()
if (
data.type in ("passport", "driver_license", "identity_card", "internal_passport")
and data.selfie
):
selfie_file = await data.selfie.get_file()
print(data.type, selfie_file)
await selfie_file.download()
await selfie_file.download_to_memory()
if data.translation and data.type in (
"passport",
"driver_license",
@ -111,7 +111,7 @@ async def msg(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
for file in data.translation:
actual_file = await file.get_file()
print(actual_file)
await actual_file.download()
await actual_file.download_to_memory()
def main() -> None:

7
telegram/_bot.py
View file

@ -3100,10 +3100,9 @@ class Bot(TelegramObject, AbstractAsyncContextManager):
Use this method to get basic info about a file and prepare it for downloading. For the
moment, bots can download files of up to
:tg-const:`telegram.constants.FileSizeLimit.FILESIZE_DOWNLOAD` in size. The file can then
be downloaded
with :meth:`telegram.File.download`. It is guaranteed that the link will be
valid for at least 1 hour. When the link expires, a new one can be requested by
calling get_file again.
be e.g. downloaded with :meth:`telegram.File.download_to_memory`. It is guaranteed that
the link will be valid for at least 1 hour. When the link expires, a new one can be
requested by calling get_file again.
Note:
This function may not preserve the original file name and MIME type.

163
telegram/_files/file.py
View file

@ -21,7 +21,7 @@ import shutil
import urllib.parse as urllib_parse
from base64 import b64decode
from pathlib import Path
from typing import IO, TYPE_CHECKING, Optional, Union
from typing import TYPE_CHECKING, BinaryIO, Optional
from telegram._passport.credentials import decrypt
from telegram._telegramobject import TelegramObject
@ -35,18 +35,22 @@ if TYPE_CHECKING:
class File(TelegramObject):
"""
This object represents a file ready to be downloaded. The file can be downloaded with
:attr:`download`. It is guaranteed that the link will be valid for at least 1 hour. When the
link expires, a new one can be requested by calling :meth:`telegram.Bot.get_file`.
This object represents a file ready to be downloaded. The file can be e.g. downloaded with
:attr:`download_to_memory`. It is guaranteed that the link will be valid for at least 1 hour.
When the link expires, a new one can be requested by calling :meth:`telegram.Bot.get_file`.
Objects of this class are comparable in terms of equality. Two objects of this class are
considered equal, if their :attr:`file_unique_id` is equal.
.. versionchanged:: 20.0:
``download`` was split into :meth:`download_to_memory` and :meth:`download_to_object`.
Note:
* Maximum file size to download is
:tg-const:`telegram.constants.FileSizeLimit.FILESIZE_DOWNLOAD`.
* If you obtain an instance of this class from :attr:`telegram.PassportFile.get_file`,
then it will automatically be decrypted as it downloads when you call :meth:`download()`.
then it will automatically be decrypted as it downloads when you call e.g.
:meth:`download_to_memory`.
Args:
file_id (:obj:`str`): Identifier for this file, which can be used to download
@ -55,7 +59,8 @@ class File(TelegramObject):
is supposed to be the same over time and for different bots.
Can't be used to download or reuse the file.
file_size (:obj:`int`, optional): Optional. File size in bytes, if known.
file_path (:obj:`str`, optional): File path. Use :attr:`download` to get the file.
file_path (:obj:`str`, optional): File path. Use e.g. :meth:`download_to_memory` to get the
file.
Attributes:
file_id (:obj:`str`): Identifier for this file.
@ -63,8 +68,8 @@ class File(TelegramObject):
is supposed to be the same over time and for different bots.
Can't be used to download or reuse the file.
file_size (:obj:`str`): Optional. File size in bytes.
file_path (:obj:`str`): Optional. File path. Use :meth:`download` to get the file.
file_path (:obj:`str`): Optional. File path. Use e.g. :meth:`download_to_memory` to get
the file.
"""
__slots__ = (
@ -97,38 +102,55 @@ class File(TelegramObject):
self._id_attrs = (self.file_unique_id,)
async def download(
def _get_encoded_url(self) -> str:
"""Convert any UTF-8 char in :obj:`File.file_path` into a url encoded ASCII string."""
sres = urllib_parse.urlsplit(str(self.file_path))
return urllib_parse.urlunsplit(
urllib_parse.SplitResult(
sres.scheme, sres.netloc, urllib_parse.quote(sres.path), sres.query, sres.fragment
)
)
def _prepare_decrypt(self, buf: bytes) -> bytes:
return decrypt(b64decode(self._credentials.secret), b64decode(self._credentials.hash), buf)
async def download_to_memory(
self,
custom_path: FilePathInput = None,
out: IO = None,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,
connect_timeout: ODVInput[float] = DEFAULT_NONE,
pool_timeout: ODVInput[float] = DEFAULT_NONE,
) -> Union[Path, IO]:
) -> Path:
"""
Download this file. By default, the file is saved in the current working directory with its
original filename as reported by Telegram. If the file has no filename, it the file ID will
be used as filename. If a :paramref:`custom_path` is supplied, it will be saved to that
path instead. If :paramref:`out` is defined, the file contents will be saved to that object
using the :obj:`out.write<io.BufferedWriter.write>` method.
original filename as reported by Telegram. If the file has no filename, the file ID will
be used as filename. If :paramref:`custom_path` is supplied as a :obj:`str` or
:obj:`pathlib.Path`, it will be saved to that path.
Note:
* :paramref:`custom_path` and :paramref:`out` are mutually exclusive.
* If neither :paramref:`custom_path` nor :paramref:`out` is provided and
:attr:`file_path` is the path of a local file (which is the case when a Bot API
Server is running in local mode), this method will just return the path.
If :paramref:`custom_path` isn't provided and :attr:`file_path` is the path of a
local file (which is the case when a Bot API Server is running in local mode), this
method will just return the path.
The only exception to this are encrypted files (e.g. a passport file). For these, a
file with the prefix `decrypted_` will be created in the same directory as the
original file in order to decrypt the file without changing the existing one
in-place.
.. versionchanged:: 20.0
* :paramref:`custom_path` parameter now also accepts :class:`pathlib.Path` as argument.
* Returns :class:`pathlib.Path` object in cases where previously a :obj:`str` was
returned.
* This method was previously called ``download``. It was split into
:meth:`download_to_memory` and :meth:`download_to_object`.
Args:
custom_path (:class:`pathlib.Path` | :obj:`str`, optional): Custom path.
out (:obj:`io.BufferedWriter`, optional): A file-like object. Must be opened for
writing in binary mode, if applicable.
custom_path (:class:`pathlib.Path` | :obj:`str` , optional): The path where the file
will be saved to. If not specified, will be saved in the current working directory.
read_timeout (:obj:`float` | :obj:`None`, optional): Value to pass to
:paramref:`telegram.request.BaseRequest.post.read_timeout`. Defaults to
:attr:`~telegram.request.BaseRequest.DEFAULT_NONE`.
@ -143,32 +165,22 @@ class File(TelegramObject):
:attr:`~telegram.request.BaseRequest.DEFAULT_NONE`.
Returns:
:class:`pathlib.Path` | :obj:`io.BufferedWriter`: The same object as :paramref:`out` if
specified. Otherwise, returns the filename downloaded to or the file path of the
local file.
Raises:
ValueError: If both :paramref:`custom_path` and :paramref:`out` are passed.
:class:`pathlib.Path`: Returns the Path object the file was downloaded to.
"""
if custom_path is not None and out is not None:
raise ValueError("`custom_path` and `out` are mutually exclusive")
local_file = is_local_file(self.file_path)
url = None if local_file else self._get_encoded_url()
path = Path(self.file_path) if local_file else None
if out:
if local_file:
buf = path.read_bytes()
# if _credentials exists we want to decrypt the file
if local_file and self._credentials:
file_to_decrypt = Path(self.file_path)
buf = self._prepare_decrypt(file_to_decrypt.read_bytes())
if custom_path is not None:
path = Path(custom_path)
else:
buf = await self.get_bot().request.retrieve(url)
if self._credentials:
buf = decrypt(
b64decode(self._credentials.secret), b64decode(self._credentials.hash), buf
)
out.write(buf)
return out
path = Path(str(file_to_decrypt.parent) + "/decrypted_" + file_to_decrypt.name)
path.write_bytes(buf)
return path
if custom_path is not None and local_file:
shutil.copyfile(self.file_path, str(custom_path))
@ -191,20 +203,58 @@ class File(TelegramObject):
pool_timeout=pool_timeout,
)
if self._credentials:
buf = decrypt(
b64decode(self._credentials.secret), b64decode(self._credentials.hash), buf
)
buf = self._prepare_decrypt(buf)
filename.write_bytes(buf)
return filename
def _get_encoded_url(self) -> str:
"""Convert any UTF-8 char in :obj:`File.file_path` into a url encoded ASCII string."""
sres = urllib_parse.urlsplit(str(self.file_path))
return urllib_parse.urlunsplit(
urllib_parse.SplitResult(
sres.scheme, sres.netloc, urllib_parse.quote(sres.path), sres.query, sres.fragment
async def download_to_object(
self,
out: BinaryIO,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,
connect_timeout: ODVInput[float] = DEFAULT_NONE,
pool_timeout: ODVInput[float] = DEFAULT_NONE,
) -> None:
"""
Download this file into memory. :paramref:`out` needs to be supplied with a
:obj:`io.BufferedIOBase`, the file contents will be saved to that object using the
:obj:`out.write<io.BufferedIOBase.write>` method.
.. versionadded:: 20.0
Args:
out (:obj:`io.BufferedIOBase`): A file-like object. Must be opened for writing in
binary mode.
read_timeout (:obj:`float` | :obj:`None`, optional): Value to pass to
:paramref:`telegram.request.BaseRequest.post.read_timeout`. Defaults to
:attr:`~telegram.request.BaseRequest.DEFAULT_NONE`.
write_timeout (:obj:`float` | :obj:`None`, optional): Value to pass to
:paramref:`telegram.request.BaseRequest.post.write_timeout`. Defaults to
:attr:`~telegram.request.BaseRequest.DEFAULT_NONE`.
connect_timeout (:obj:`float` | :obj:`None`, optional): Value to pass to
:paramref:`telegram.request.BaseRequest.post.connect_timeout`. Defaults to
:attr:`~telegram.request.BaseRequest.DEFAULT_NONE`.
pool_timeout (:obj:`float` | :obj:`None`, optional): Value to pass to
:paramref:`telegram.request.BaseRequest.post.pool_timeout`. Defaults to
:attr:`~telegram.request.BaseRequest.DEFAULT_NONE`.
"""
local_file = is_local_file(self.file_path)
url = None if local_file else self._get_encoded_url()
path = Path(self.file_path) if local_file else None
if local_file:
buf = path.read_bytes()
else:
buf = await self.get_bot().request.retrieve(
url,
read_timeout=read_timeout,
write_timeout=write_timeout,
connect_timeout=connect_timeout,
pool_timeout=pool_timeout,
)
)
if self._credentials:
buf = self._prepare_decrypt(buf)
out.write(buf)
async def download_as_bytearray(self, buf: bytearray = None) -> bytearray:
"""Download this file and return it as a bytearray.
@ -219,10 +269,15 @@ class File(TelegramObject):
"""
if buf is None:
buf = bytearray()
if is_local_file(self.file_path):
buf.extend(Path(self.file_path).read_bytes())
bytes_data = Path(self.file_path).read_bytes()
else:
buf.extend(await self.get_bot().request.retrieve(self._get_encoded_url()))
bytes_data = await self.get_bot().request.retrieve(self._get_encoded_url())
if self._credentials:
buf.extend(self._prepare_decrypt(bytes_data))
else:
buf.extend(bytes_data)
return buf
def set_credentials(self, credentials: "FileCredentials") -> None:

BIN
tests/data/image_decrypted.jpg Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 24 KiB

BIN
tests/data/image_encrypted.jpg Normal file
View file

Binary file not shown.

2
tests/test_animation.py
View file

@ -129,7 +129,7 @@ class TestAnimation:
assert new_file.file_id == animation.file_id
assert new_file.file_path.startswith("https://")
new_filepath = await new_file.download("game.gif")
new_filepath = await new_file.download_to_memory("game.gif")
assert new_filepath.is_file()

2
tests/test_audio.py
View file

@ -144,7 +144,7 @@ class TestAudio:
assert new_file.file_unique_id == audio.file_unique_id
assert str(new_file.file_path).startswith("https://")
await new_file.download("telegram.mp3")
await new_file.download_to_memory("telegram.mp3")
assert path.is_file()

4
tests/test_chatphoto.py
View file

@ -86,7 +86,7 @@ class TestChatPhoto:
assert new_file.file_unique_id == chat_photo.small_file_unique_id
assert new_file.file_path.startswith("https://")
await new_file.download(jpg_file)
await new_file.download_to_memory(jpg_file)
assert jpg_file.is_file()
@ -95,7 +95,7 @@ class TestChatPhoto:
assert new_file.file_unique_id == chat_photo.big_file_unique_id
assert new_file.file_path.startswith("https://")
await new_file.download(jpg_file)
await new_file.download_to_memory(jpg_file)
assert jpg_file.is_file()

2
tests/test_document.py
View file

@ -119,7 +119,7 @@ class TestDocument:
assert new_file.file_unique_id == document.file_unique_id
assert new_file.file_path.startswith("https://")
await new_file.download("telegram.png")
await new_file.download_to_memory("telegram.png")
assert path.is_file()

149
tests/test_file.py
View file

@ -23,7 +23,7 @@ from tempfile import TemporaryFile, mkstemp
import pytest
from flaky import flaky
from telegram import File, Voice
from telegram import File, FileCredentials, Voice
from telegram.error import TelegramError
from tests.conftest import data_file
@ -40,6 +40,39 @@ def file(bot):
return file
@pytest.fixture(scope="class")
def encrypted_file(bot):
# check https://github.com/python-telegram-bot/python-telegram-bot/wiki/\
# PTB-test-writing-knowledge-base#how-to-generate-encrypted-passport-files
# if you want to know the source of these values
fc = FileCredentials(
"Oq3G4sX+bKZthoyms1YlPqvWou9esb+z0Bi/KqQUG8s=",
"Pt7fKPgYWKA/7a8E64Ea1X8C+Wf7Ky1tF4ANBl63vl4=",
)
ef = File(TestFile.file_id, TestFile.file_unique_id, TestFile.file_size, TestFile.file_path)
ef.set_bot(bot)
ef.set_credentials(fc)
return ef
@pytest.fixture(scope="class")
def encrypted_local_file(bot):
# check encrypted_file() for the source of the fc values
fc = FileCredentials(
"Oq3G4sX+bKZthoyms1YlPqvWou9esb+z0Bi/KqQUG8s=",
"Pt7fKPgYWKA/7a8E64Ea1X8C+Wf7Ky1tF4ANBl63vl4=",
)
ef = File(
TestFile.file_id,
TestFile.file_unique_id,
TestFile.file_size,
file_path=str(data_file("image_encrypted.jpg")),
)
ef.set_bot(bot)
ef.set_credentials(fc)
return ef
@pytest.fixture(scope="class")
def local_file(bot):
file = File(
@ -95,16 +128,12 @@ class TestFile:
with pytest.raises(TelegramError):
await bot.get_file(file_id="")
async def test_download_mutually_exclusive(self, file):
with pytest.raises(ValueError, match="`custom_path` and `out` are mutually exclusive"):
await file.download("custom_path", "out")
async def test_download(self, monkeypatch, file):
async def test(*args, **kwargs):
return self.file_content
monkeypatch.setattr(file.get_bot().request, "retrieve", test)
out_file = await file.download()
out_file = await file.download_to_memory()
try:
assert out_file.read_bytes() == self.file_content
@ -112,7 +141,7 @@ class TestFile:
out_file.unlink()
async def test_download_local_file(self, local_file):
assert await local_file.download() == Path(local_file.file_path)
assert await local_file.download_to_memory() == Path(local_file.file_path)
@pytest.mark.parametrize(
"custom_path_type", [str, Path], ids=["str custom_path", "pathlib.Path custom_path"]
@ -125,7 +154,7 @@ class TestFile:
file_handle, custom_path = mkstemp()
custom_path = Path(custom_path)
try:
out_file = await file.download(custom_path_type(custom_path))
out_file = await file.download_to_memory(custom_path_type(custom_path))
assert out_file == custom_path
assert out_file.read_bytes() == self.file_content
finally:
@ -139,7 +168,7 @@ class TestFile:
file_handle, custom_path = mkstemp()
custom_path = Path(custom_path)
try:
out_file = await local_file.download(custom_path_type(custom_path))
out_file = await local_file.download_to_memory(custom_path_type(custom_path))
assert out_file == custom_path
assert out_file.read_bytes() == self.file_content
finally:
@ -153,7 +182,7 @@ class TestFile:
file.file_path = None
monkeypatch.setattr(file.get_bot().request, "retrieve", test)
out_file = await file.download()
out_file = await file.download_to_memory()
assert str(out_file)[-len(file.file_id) :] == file.file_id
try:
@ -167,19 +196,15 @@ class TestFile:
monkeypatch.setattr(file.get_bot().request, "retrieve", test)
with TemporaryFile() as custom_fobj:
out_fobj = await file.download(out=custom_fobj)
assert out_fobj is custom_fobj
out_fobj.seek(0)
assert out_fobj.read() == self.file_content
await file.download_to_object(out=custom_fobj)
custom_fobj.seek(0)
assert custom_fobj.read() == self.file_content
async def test_download_file_obj_local_file(self, local_file):
with TemporaryFile() as custom_fobj:
out_fobj = await local_file.download(out=custom_fobj)
assert out_fobj is custom_fobj
out_fobj.seek(0)
assert out_fobj.read() == self.file_content
await local_file.download_to_object(out=custom_fobj)
custom_fobj.seek(0)
assert custom_fobj.read() == self.file_content
async def test_download_bytearray(self, monkeypatch, file):
async def test(*args, **kwargs):
@ -210,6 +235,90 @@ class TestFile:
assert buf2[len(buf) :] == buf
assert buf2[: len(buf)] == buf
async def test_download_encrypted(self, monkeypatch, bot, encrypted_file):
async def test(*args, **kwargs):
return data_file("image_encrypted.jpg").read_bytes()
monkeypatch.setattr(encrypted_file.get_bot().request, "retrieve", test)
out_file = await encrypted_file.download_to_memory()
try:
assert out_file.read_bytes() == data_file("image_decrypted.jpg").read_bytes()
finally:
out_file.unlink()
async def test_download_file_obj_encrypted(self, monkeypatch, encrypted_file):
async def test(*args, **kwargs):
return data_file("image_encrypted.jpg").read_bytes()
monkeypatch.setattr(encrypted_file.get_bot().request, "retrieve", test)
with TemporaryFile() as custom_fobj:
await encrypted_file.download_to_object(out=custom_fobj)
custom_fobj.seek(0)
assert custom_fobj.read() == data_file("image_decrypted.jpg").read_bytes()
async def test_download_local_file_encrypted(self, encrypted_local_file):
out_file = await encrypted_local_file.download_to_memory()
try:
assert out_file.read_bytes() == data_file("image_decrypted.jpg").read_bytes()
finally:
out_file.unlink()
@pytest.mark.parametrize(
"custom_path_type", [str, Path], ids=["str custom_path", "pathlib.Path custom_path"]
)
async def test_download_custom_path_local_file_encrypted(
self, encrypted_local_file, custom_path_type
):
file_handle, custom_path = mkstemp()
custom_path = Path(custom_path)
try:
out_file = await encrypted_local_file.download_to_memory(custom_path_type(custom_path))
assert out_file == custom_path
assert out_file.read_bytes() == data_file("image_decrypted.jpg").read_bytes()
finally:
os.close(file_handle)
custom_path.unlink()
async def test_download_file_obj_local_file_encrypted(self, monkeypatch, encrypted_local_file):
async def test(*args, **kwargs):
return data_file("image_encrypted.jpg").read_bytes()
monkeypatch.setattr(encrypted_local_file.get_bot().request, "retrieve", test)
with TemporaryFile() as custom_fobj:
await encrypted_local_file.download_to_object(out=custom_fobj)
custom_fobj.seek(0)
assert custom_fobj.read() == data_file("image_decrypted.jpg").read_bytes()
async def test_download_bytearray_encrypted(self, monkeypatch, encrypted_file):
async def test(*args, **kwargs):
return data_file("image_encrypted.jpg").read_bytes()
monkeypatch.setattr(encrypted_file.get_bot().request, "retrieve", test)
# Check that a download to a newly allocated bytearray works.
buf = await encrypted_file.download_as_bytearray()
assert buf == bytearray(data_file("image_decrypted.jpg").read_bytes())
# Check that a download to a given bytearray works (extends the bytearray).
buf2 = buf[:]
buf3 = await encrypted_file.download_as_bytearray(buf=buf2)
assert buf3 is buf2
assert buf2[len(buf) :] == buf
assert buf2[: len(buf)] == buf
async def test_download_bytearray_local_file_encrypted(self, encrypted_local_file):
# Check that a download to a newly allocated bytearray works.
buf = await encrypted_local_file.download_as_bytearray()
assert buf == bytearray(data_file("image_decrypted.jpg").read_bytes())
# Check that a download to a given bytearray works (extends the bytearray).
buf2 = buf[:]
buf3 = await encrypted_local_file.download_as_bytearray(buf=buf2)
assert buf3 is buf2
assert buf2[len(buf) :] == buf
assert buf2[: len(buf)] == buf
def test_equality(self, bot):
a = File(self.file_id, self.file_unique_id, bot)
b = File("", self.file_unique_id, bot)

4
tests/test_inputfile.py
View file

@ -145,7 +145,7 @@ class TestInputFile:
message = await bot.send_document(chat_id, data_file("text_file.txt").read_bytes())
out = BytesIO()
assert await (await message.document.get_file()).download(out=out)
await (await message.document.get_file()).download_to_object(out=out)
out.seek(0)
assert out.read().decode("utf-8") == "PTB Rocks! ⅞"
@ -158,7 +158,7 @@ class TestInputFile:
)
out = BytesIO()
assert await (await message.document.get_file()).download(out=out)
await (await message.document.get_file()).download_to_object(out=out)
out.seek(0)
assert out.read().decode("utf-8") == "PTB Rocks! ⅞"

2
tests/test_photo.py
View file

@ -304,7 +304,7 @@ class TestPhoto:
assert new_file.file_unique_id == photo.file_unique_id
assert new_file.file_path.startswith("https://") is True
await new_file.download("telegram.jpg")
await new_file.download_to_memory("telegram.jpg")
assert path.is_file()

2
tests/test_sticker.py
View file

@ -169,7 +169,7 @@ class TestSticker:
assert new_file.file_unique_id == sticker.file_unique_id
assert new_file.file_path.startswith("https://")
await new_file.download("telegram.webp")
await new_file.download_to_memory("telegram.webp")
assert path.is_file()

2
tests/test_video.py
View file

@ -149,7 +149,7 @@ class TestVideo:
assert new_file.file_unique_id == video.file_unique_id
assert new_file.file_path.startswith("https://")
await new_file.download("telegram.mp4")
await new_file.download_to_memory("telegram.mp4")
assert path.is_file()

2
tests/test_videonote.py
View file

@ -133,7 +133,7 @@ class TestVideoNote:
assert new_file.file_unique_id == video_note.file_unique_id
assert new_file.file_path.startswith("https://")
await new_file.download("telegram2.mp4")
await new_file.download_to_memory("telegram2.mp4")
assert path.is_file()

2
tests/test_voice.py
View file

@ -121,7 +121,7 @@ class TestVoice:
assert new_file.file_unique_id == voice.file_unique_id
assert new_file.file_path.startswith("https://")
await new_file.download("telegram.ogg")
await new_file.download_to_memory("telegram.ogg")
assert path.is_file()