2015-09-20 17:28:10 +02:00
|
|
|
#!/usr/bin/env python
|
|
|
|
#
|
|
|
|
# A library that provides a Python interface to the Telegram Bot API
|
2022-01-03 08:15:18 +01:00
|
|
|
# Copyright (C) 2015-2022
|
2016-01-05 14:12:03 +01:00
|
|
|
# Leandro Toledo de Souza <devs@python-telegram-bot.org>
|
2015-09-20 17:28:10 +02:00
|
|
|
#
|
|
|
|
# This program is free software: you can redistribute it and/or modify
|
2017-08-11 23:58:41 +02:00
|
|
|
# it under the terms of the GNU Lesser Public License as published by
|
2015-09-20 17:28:10 +02:00
|
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
|
|
# (at your option) any later version.
|
|
|
|
#
|
|
|
|
# This program is distributed in the hope that it will be useful,
|
|
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
2017-08-11 23:58:41 +02:00
|
|
|
# GNU Lesser Public License for more details.
|
2015-09-20 17:28:10 +02:00
|
|
|
#
|
2017-08-11 23:58:41 +02:00
|
|
|
# You should have received a copy of the GNU Lesser Public License
|
2015-09-20 17:28:10 +02:00
|
|
|
# along with this program. If not, see [http://www.gnu.org/licenses/].
|
2018-03-01 09:10:04 +01:00
|
|
|
import os
|
2020-11-29 16:20:46 +01:00
|
|
|
from pathlib import Path
|
2018-03-01 09:10:04 +01:00
|
|
|
from tempfile import TemporaryFile, mkstemp
|
2015-09-20 17:28:10 +02:00
|
|
|
|
2017-08-11 23:58:41 +02:00
|
|
|
import pytest
|
2016-04-24 04:11:25 +02:00
|
|
|
|
2022-11-02 08:28:41 +01:00
|
|
|
from telegram import File, FileCredentials, Voice
|
2021-09-22 16:49:10 +02:00
|
|
|
from telegram.error import TelegramError
|
2021-10-13 08:12:48 +02:00
|
|
|
from tests.conftest import data_file
|
2015-09-20 17:28:10 +02:00
|
|
|
|
|
|
|
|
2017-08-11 23:58:41 +02:00
|
|
|
@pytest.fixture(scope="class")
|
|
|
|
def file(bot):
|
2022-10-07 11:51:53 +02:00
|
|
|
file = File(
|
2017-08-11 23:58:41 +02:00
|
|
|
TestFile.file_id,
|
2020-03-28 16:37:26 +01:00
|
|
|
TestFile.file_unique_id,
|
2017-08-11 23:58:41 +02:00
|
|
|
file_path=TestFile.file_path,
|
|
|
|
file_size=TestFile.file_size,
|
|
|
|
)
|
2022-10-07 11:51:53 +02:00
|
|
|
file.set_bot(bot)
|
|
|
|
return file
|
2015-09-20 17:28:10 +02:00
|
|
|
|
|
|
|
|
2022-11-02 08:28:41 +01:00
|
|
|
@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
|
|
|
|
|
|
|
|
|
2020-11-29 16:20:46 +01:00
|
|
|
@pytest.fixture(scope="class")
|
|
|
|
def local_file(bot):
|
2022-10-07 11:51:53 +02:00
|
|
|
file = File(
|
2020-11-29 16:20:46 +01:00
|
|
|
TestFile.file_id,
|
|
|
|
TestFile.file_unique_id,
|
2021-10-13 08:12:48 +02:00
|
|
|
file_path=str(data_file("local_file.txt")),
|
2020-11-29 16:20:46 +01:00
|
|
|
file_size=TestFile.file_size,
|
|
|
|
)
|
2022-10-07 11:51:53 +02:00
|
|
|
file.set_bot(bot)
|
|
|
|
return file
|
2020-11-29 16:20:46 +01:00
|
|
|
|
|
|
|
|
2020-06-15 18:20:51 +02:00
|
|
|
class TestFile:
|
2017-08-11 23:58:41 +02:00
|
|
|
file_id = "NOTVALIDDOESNOTMATTER"
|
2020-03-28 16:37:26 +01:00
|
|
|
file_unique_id = "adc3145fd2e84d95b64d68eaa22aa33e"
|
2017-08-11 23:58:41 +02:00
|
|
|
file_path = (
|
2021-01-17 23:24:20 +01:00
|
|
|
"https://api.org/file/bot133505823:AAHZFMHno3mzVLErU5b5jJvaeG--qUyLyG0/document/file_3"
|
2017-08-11 23:58:41 +02:00
|
|
|
)
|
|
|
|
file_size = 28232
|
2021-01-17 23:24:20 +01:00
|
|
|
file_content = "Saint-Saëns".encode() # Intentionally contains unicode chars.
|
2015-09-20 17:28:10 +02:00
|
|
|
|
2021-08-19 22:01:10 +02:00
|
|
|
def test_slot_behaviour(self, file, mro_slots):
|
2021-05-29 16:18:16 +02:00
|
|
|
for attr in file.__slots__:
|
|
|
|
assert getattr(file, attr, "err") != "err", f"got extra slot '{attr}'"
|
|
|
|
assert len(mro_slots(file)) == len(set(mro_slots(file))), "duplicate slot"
|
|
|
|
|
2017-08-11 23:58:41 +02:00
|
|
|
def test_de_json(self, bot):
|
|
|
|
json_dict = {
|
|
|
|
"file_id": self.file_id,
|
2020-03-28 16:37:26 +01:00
|
|
|
"file_unique_id": self.file_unique_id,
|
2017-08-11 23:58:41 +02:00
|
|
|
"file_path": self.file_path,
|
|
|
|
"file_size": self.file_size,
|
|
|
|
}
|
|
|
|
new_file = File.de_json(json_dict, bot)
|
2022-10-07 11:51:53 +02:00
|
|
|
assert new_file.api_kwargs == {}
|
2015-09-20 17:28:10 +02:00
|
|
|
|
2017-08-11 23:58:41 +02:00
|
|
|
assert new_file.file_id == self.file_id
|
2020-03-28 16:37:26 +01:00
|
|
|
assert new_file.file_unique_id == self.file_unique_id
|
2017-08-11 23:58:41 +02:00
|
|
|
assert new_file.file_path == self.file_path
|
|
|
|
assert new_file.file_size == self.file_size
|
2015-09-20 17:28:10 +02:00
|
|
|
|
2017-08-11 23:58:41 +02:00
|
|
|
def test_to_dict(self, file):
|
|
|
|
file_dict = file.to_dict()
|
2015-09-20 17:28:10 +02:00
|
|
|
|
2017-08-11 23:58:41 +02:00
|
|
|
assert isinstance(file_dict, dict)
|
|
|
|
assert file_dict["file_id"] == file.file_id
|
2020-03-28 16:37:26 +01:00
|
|
|
assert file_dict["file_unique_id"] == file.file_unique_id
|
2017-08-11 23:58:41 +02:00
|
|
|
assert file_dict["file_path"] == file.file_path
|
|
|
|
assert file_dict["file_size"] == file.file_size
|
2017-06-22 18:29:21 +02:00
|
|
|
|
2022-11-09 20:43:27 +01:00
|
|
|
@pytest.mark.flaky(3, 1)
|
2022-04-24 12:38:09 +02:00
|
|
|
async def test_error_get_empty_file_id(self, bot):
|
2017-08-11 23:58:41 +02:00
|
|
|
with pytest.raises(TelegramError):
|
2022-04-24 12:38:09 +02:00
|
|
|
await bot.get_file(file_id="")
|
2015-09-20 17:28:10 +02:00
|
|
|
|
2022-04-24 12:38:09 +02:00
|
|
|
async def test_download(self, monkeypatch, file):
|
|
|
|
async def test(*args, **kwargs):
|
2018-03-01 09:10:04 +01:00
|
|
|
return self.file_content
|
2017-05-14 23:29:31 +02:00
|
|
|
|
2022-04-24 12:38:09 +02:00
|
|
|
monkeypatch.setattr(file.get_bot().request, "retrieve", test)
|
2022-11-02 08:28:41 +01:00
|
|
|
out_file = await file.download_to_memory()
|
2018-03-01 09:10:04 +01:00
|
|
|
|
|
|
|
try:
|
2021-10-05 19:50:11 +02:00
|
|
|
assert out_file.read_bytes() == self.file_content
|
2018-03-01 09:10:04 +01:00
|
|
|
finally:
|
2021-10-05 19:50:11 +02:00
|
|
|
out_file.unlink()
|
2018-03-01 09:10:04 +01:00
|
|
|
|
2022-04-24 12:38:09 +02:00
|
|
|
async def test_download_local_file(self, local_file):
|
2022-11-02 08:28:41 +01:00
|
|
|
assert await local_file.download_to_memory() == Path(local_file.file_path)
|
2020-11-29 16:20:46 +01:00
|
|
|
|
2021-10-05 19:50:11 +02:00
|
|
|
@pytest.mark.parametrize(
|
|
|
|
"custom_path_type", [str, Path], ids=["str custom_path", "pathlib.Path custom_path"]
|
|
|
|
)
|
2022-04-24 12:38:09 +02:00
|
|
|
async def test_download_custom_path(self, monkeypatch, file, custom_path_type):
|
|
|
|
async def test(*args, **kwargs):
|
2018-03-01 09:10:04 +01:00
|
|
|
return self.file_content
|
|
|
|
|
2022-04-24 12:38:09 +02:00
|
|
|
monkeypatch.setattr(file.get_bot().request, "retrieve", test)
|
2018-03-01 09:10:04 +01:00
|
|
|
file_handle, custom_path = mkstemp()
|
2021-10-05 19:50:11 +02:00
|
|
|
custom_path = Path(custom_path)
|
2018-03-01 09:10:04 +01:00
|
|
|
try:
|
2022-11-02 08:28:41 +01:00
|
|
|
out_file = await file.download_to_memory(custom_path_type(custom_path))
|
2018-03-01 09:10:04 +01:00
|
|
|
assert out_file == custom_path
|
2021-10-05 19:50:11 +02:00
|
|
|
assert out_file.read_bytes() == self.file_content
|
2018-03-01 09:10:04 +01:00
|
|
|
finally:
|
|
|
|
os.close(file_handle)
|
2021-10-05 19:50:11 +02:00
|
|
|
custom_path.unlink()
|
2018-03-01 09:10:04 +01:00
|
|
|
|
2021-10-05 19:50:11 +02:00
|
|
|
@pytest.mark.parametrize(
|
|
|
|
"custom_path_type", [str, Path], ids=["str custom_path", "pathlib.Path custom_path"]
|
|
|
|
)
|
2022-04-24 12:38:09 +02:00
|
|
|
async def test_download_custom_path_local_file(self, local_file, custom_path_type):
|
2020-11-29 16:20:46 +01:00
|
|
|
file_handle, custom_path = mkstemp()
|
2021-10-05 19:50:11 +02:00
|
|
|
custom_path = Path(custom_path)
|
2020-11-29 16:20:46 +01:00
|
|
|
try:
|
2022-11-02 08:28:41 +01:00
|
|
|
out_file = await local_file.download_to_memory(custom_path_type(custom_path))
|
2020-11-29 16:20:46 +01:00
|
|
|
assert out_file == custom_path
|
2021-10-05 19:50:11 +02:00
|
|
|
assert out_file.read_bytes() == self.file_content
|
2020-11-29 16:20:46 +01:00
|
|
|
finally:
|
|
|
|
os.close(file_handle)
|
2021-10-05 19:50:11 +02:00
|
|
|
custom_path.unlink()
|
2020-11-29 16:20:46 +01:00
|
|
|
|
2022-04-24 12:38:09 +02:00
|
|
|
async def test_download_no_filename(self, monkeypatch, file):
|
|
|
|
async def test(*args, **kwargs):
|
2020-02-06 11:21:21 +01:00
|
|
|
return self.file_content
|
|
|
|
|
|
|
|
file.file_path = None
|
|
|
|
|
2022-04-24 12:38:09 +02:00
|
|
|
monkeypatch.setattr(file.get_bot().request, "retrieve", test)
|
2022-11-02 08:28:41 +01:00
|
|
|
out_file = await file.download_to_memory()
|
2020-02-06 11:21:21 +01:00
|
|
|
|
2021-10-05 19:50:11 +02:00
|
|
|
assert str(out_file)[-len(file.file_id) :] == file.file_id
|
2020-02-06 11:21:21 +01:00
|
|
|
try:
|
2021-10-05 19:50:11 +02:00
|
|
|
assert out_file.read_bytes() == self.file_content
|
2020-02-06 11:21:21 +01:00
|
|
|
finally:
|
2021-10-05 19:50:11 +02:00
|
|
|
out_file.unlink()
|
2020-02-06 11:21:21 +01:00
|
|
|
|
2022-04-24 12:38:09 +02:00
|
|
|
async def test_download_file_obj(self, monkeypatch, file):
|
|
|
|
async def test(*args, **kwargs):
|
2018-03-01 09:10:04 +01:00
|
|
|
return self.file_content
|
|
|
|
|
2022-04-24 12:38:09 +02:00
|
|
|
monkeypatch.setattr(file.get_bot().request, "retrieve", test)
|
2018-03-01 09:10:04 +01:00
|
|
|
with TemporaryFile() as custom_fobj:
|
2022-11-02 08:28:41 +01:00
|
|
|
await file.download_to_object(out=custom_fobj)
|
|
|
|
custom_fobj.seek(0)
|
|
|
|
assert custom_fobj.read() == self.file_content
|
2018-03-01 09:10:04 +01:00
|
|
|
|
2022-04-24 12:38:09 +02:00
|
|
|
async def test_download_file_obj_local_file(self, local_file):
|
2020-11-29 16:20:46 +01:00
|
|
|
with TemporaryFile() as custom_fobj:
|
2022-11-02 08:28:41 +01:00
|
|
|
await local_file.download_to_object(out=custom_fobj)
|
|
|
|
custom_fobj.seek(0)
|
|
|
|
assert custom_fobj.read() == self.file_content
|
2020-11-29 16:20:46 +01:00
|
|
|
|
2022-04-24 12:38:09 +02:00
|
|
|
async def test_download_bytearray(self, monkeypatch, file):
|
|
|
|
async def test(*args, **kwargs):
|
2018-03-01 09:10:04 +01:00
|
|
|
return self.file_content
|
|
|
|
|
2022-04-24 12:38:09 +02:00
|
|
|
monkeypatch.setattr(file.get_bot().request, "retrieve", test)
|
2018-03-01 09:10:04 +01:00
|
|
|
|
|
|
|
# Check that a download to a newly allocated bytearray works.
|
2022-04-24 12:38:09 +02:00
|
|
|
buf = await file.download_as_bytearray()
|
2018-03-01 09:10:04 +01:00
|
|
|
assert buf == bytearray(self.file_content)
|
|
|
|
|
|
|
|
# Check that a download to a given bytearray works (extends the bytearray).
|
|
|
|
buf2 = buf[:]
|
2022-04-24 12:38:09 +02:00
|
|
|
buf3 = await file.download_as_bytearray(buf=buf2)
|
2018-03-01 09:10:04 +01:00
|
|
|
assert buf3 is buf2
|
|
|
|
assert buf2[len(buf) :] == buf
|
|
|
|
assert buf2[: len(buf)] == buf
|
2017-05-14 23:29:31 +02:00
|
|
|
|
2022-04-24 12:38:09 +02:00
|
|
|
async def test_download_bytearray_local_file(self, local_file):
|
2020-11-29 16:20:46 +01:00
|
|
|
# Check that a download to a newly allocated bytearray works.
|
2022-04-24 12:38:09 +02:00
|
|
|
buf = await local_file.download_as_bytearray()
|
2020-11-29 16:20:46 +01:00
|
|
|
assert buf == bytearray(self.file_content)
|
|
|
|
|
|
|
|
# Check that a download to a given bytearray works (extends the bytearray).
|
|
|
|
buf2 = buf[:]
|
2022-04-24 12:38:09 +02:00
|
|
|
buf3 = await local_file.download_as_bytearray(buf=buf2)
|
2020-11-29 16:20:46 +01:00
|
|
|
assert buf3 is buf2
|
|
|
|
assert buf2[len(buf) :] == buf
|
|
|
|
assert buf2[: len(buf)] == buf
|
|
|
|
|
2022-11-02 08:28:41 +01:00
|
|
|
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
|
|
|
|
|
2017-08-11 23:58:41 +02:00
|
|
|
def test_equality(self, bot):
|
2020-03-28 16:37:26 +01:00
|
|
|
a = File(self.file_id, self.file_unique_id, bot)
|
|
|
|
b = File("", self.file_unique_id, bot)
|
|
|
|
c = File(self.file_id, self.file_unique_id, None)
|
|
|
|
d = File("", "", bot)
|
|
|
|
e = Voice(self.file_id, self.file_unique_id, 0)
|
2017-05-14 23:29:31 +02:00
|
|
|
|
2017-08-11 23:58:41 +02:00
|
|
|
assert a == b
|
|
|
|
assert hash(a) == hash(b)
|
|
|
|
assert a is not b
|
2017-05-14 23:29:31 +02:00
|
|
|
|
2017-08-11 23:58:41 +02:00
|
|
|
assert a == c
|
|
|
|
assert hash(a) == hash(c)
|
2017-05-14 23:29:31 +02:00
|
|
|
|
2017-08-11 23:58:41 +02:00
|
|
|
assert a != d
|
|
|
|
assert hash(a) != hash(d)
|
2016-04-24 04:11:25 +02:00
|
|
|
|
2017-08-11 23:58:41 +02:00
|
|
|
assert a != e
|
|
|
|
assert hash(a) != hash(e)
|