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 2025-01-03 09:49:21 +01:00
Code Issues Projects Releases Packages Wiki Activity

Handle Filepaths via the Pathlib Module (#2688)

Browse source
This commit is contained in:
eldbud 2021-10-05 20:50:11 +03:00 committed by Hinrich Mahler
parent 267d4e8191
commit 1edfa1504c
27 changed files with 328 additions and 306 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
AUTHORS.rst
View file

@ -43,6 +43,7 @@ The following wonderful people contributed directly or indirectly to this projec
- `DonalDuck004 <https://github.com/DonalDuck004>`_
- `Eana Hufwe <https://github.com/blueset>`_
- `Ehsan Online <https://github.com/ehsanonline>`_
- `Eldad Carin <https://github.com/eldbud>`_
- `Eli Gao <https://github.com/eligao>`_
- `Emilio Molinari <https://github.com/xates>`_
- `ErgoZ Riftbit Vaper <https://github.com/ergoz>`_

2
examples/arbitrarycallbackdatabot.py
View file

@ -84,7 +84,7 @@ def handle_invalid_button(update: Update, context: CallbackContext) -> None:
def main() -> None:
"""Run the bot."""
# We use persistence to demonstrate how buttons can still work after the bot was restarted
persistence = PicklePersistence(filename='arbitrarycallbackdatabot.pickle')
persistence = PicklePersistence(filepath='arbitrarycallbackdatabot')
# Create the Updater and pass it your bot's token.
updater = Updater("TOKEN", persistence=persistence, arbitrary_callback_data=True)

4
examples/passportbot.py
View file

@ -11,6 +11,7 @@ See https://git.io/fAvYd for how to use Telegram Passport properly with python-t
"""
import logging
from pathlib import Path
from telegram import Update
from telegram.ext import Updater, MessageHandler, Filters, CallbackContext
@ -101,8 +102,7 @@ def msg(update: Update, context: CallbackContext) -> None:
def main() -> None:
"""Start the bot."""
# Create the Updater and pass it your token and private key
with open('private.key', 'rb') as private_key:
updater = Updater("TOKEN", private_key=private_key.read())
updater = Updater("TOKEN", private_key=Path('private.key').read_bytes())
# Get the dispatcher to register handlers
dispatcher = updater.dispatcher

2
examples/persistentconversationbot.py
View file

@ -132,7 +132,7 @@ def done(update: Update, context: CallbackContext) -> int:
def main() -> None:
"""Run the bot."""
# Create the Updater and pass it your bot's token.
persistence = PicklePersistence(filename='conversationbot')
persistence = PicklePersistence(filepath='conversationbot')
updater = Updater("TOKEN", persistence=persistence)
# Get the dispatcher to register handlers

13
setup.py
View file

@ -1,8 +1,8 @@
#!/usr/bin/env python
"""The setup and build script for the python-telegram-bot library."""
import os
import subprocess
import sys
from pathlib import Path
from setuptools import setup, find_packages
@ -13,7 +13,7 @@ def get_requirements(raw=False):
"""Build the requirements list for this project"""
requirements_list = []
with open('requirements.txt') as reqs:
with Path('requirements.txt').open() as reqs:
for install in reqs:
if install.startswith('# only telegram.ext:'):
if raw:
@ -47,16 +47,13 @@ def get_setup_kwargs(raw=False):
packages, requirements = get_packages_requirements(raw=raw)
raw_ext = "-raw" if raw else ""
readme = f'README{"_RAW" if raw else ""}.rst'
readme = Path(f'README{"_RAW" if raw else ""}.rst')
fn = os.path.join('telegram', 'version.py')
with open(fn) as fh:
with Path('telegram/version.py').open() as fh:
for line in fh.readlines():
if line.startswith('__version__'):
exec(line)
with open(readme, 'r', encoding='utf-8') as fd:
kwargs = dict(
script_name=f'setup{raw_ext}.py',
name=f'python-telegram-bot{raw_ext}',
@ -76,7 +73,7 @@ def get_setup_kwargs(raw=False):
download_url=f'https://pypi.org/project/python-telegram-bot{raw_ext}/',
keywords='python telegram bot api wrapper',
description="We have made you a wrapper you can't refuse",
long_description=fd.read(),
long_description=readme.read_text(),
long_description_content_type='text/x-rst',
packages=packages,

85
telegram/ext/picklepersistence.py
View file

@ -19,6 +19,7 @@
"""This module contains the PicklePersistence class."""
import pickle
from collections import defaultdict
from pathlib import Path
from typing import (
Any,
Dict,
@ -27,6 +28,7 @@ from typing import (
overload,
cast,
DefaultDict,
Union,
)
from telegram.ext import BasePersistence, PersistenceInput
@ -47,11 +49,14 @@ class PicklePersistence(BasePersistence[UD, CD, BD]):
:meth:`telegram.ext.BasePersistence.insert_bot`.
.. versionchanged:: 14.0
The parameters and attributes ``store_*_data`` were replaced by :attr:`store_data`.
* The parameters and attributes ``store_*_data`` were replaced by :attr:`store_data`.
* The parameter and attribute ``filename`` were replaced by :attr:`filepath`.
* :attr:`filepath` now also accepts :obj:`pathlib.Path` as argument.
Args:
filename (:obj:`str`): The filename for storing the pickle files. When :attr:`single_file`
is :obj:`False` this will be used as a prefix.
filepath (:obj:`str` | :obj:`pathlib.Path`): The filepath for storing the pickle files.
When :attr:`single_file` is :obj:`False` this will be used as a prefix.
store_data (:class:`PersistenceInput`, optional): Specifies which kinds of data will be
saved by this persistence instance. By default, all available kinds of data will be
saved.
@ -70,8 +75,8 @@ class PicklePersistence(BasePersistence[UD, CD, BD]):
.. versionadded:: 13.6
Attributes:
filename (:obj:`str`): The filename for storing the pickle files. When :attr:`single_file`
is :obj:`False` this will be used as a prefix.
filepath (:obj:`str` | :obj:`pathlib.Path`): The filepath for storing the pickle files.
When :attr:`single_file` is :obj:`False` this will be used as a prefix.
store_data (:class:`PersistenceInput`): Specifies which kinds of data will be saved by this
persistence instance.
single_file (:obj:`bool`): Optional. When :obj:`False` will store 5 separate files of
@ -88,7 +93,7 @@ class PicklePersistence(BasePersistence[UD, CD, BD]):
"""
__slots__ = (
'filename',
'filepath',
'single_file',
'on_flush',
'user_data',
@ -102,7 +107,7 @@ class PicklePersistence(BasePersistence[UD, CD, BD]):
@overload
def __init__(
self: 'PicklePersistence[Dict, Dict, Dict]',
filename: str,
filepath: Union[Path, str],
store_data: PersistenceInput = None,
single_file: bool = True,
on_flush: bool = False,
@ -112,7 +117,7 @@ class PicklePersistence(BasePersistence[UD, CD, BD]):
@overload
def __init__(
self: 'PicklePersistence[UD, CD, BD]',
filename: str,
filepath: Union[Path, str],
store_data: PersistenceInput = None,
single_file: bool = True,
on_flush: bool = False,
@ -122,14 +127,14 @@ class PicklePersistence(BasePersistence[UD, CD, BD]):
def __init__(
self,
filename: str,
filepath: Union[Path, str],
store_data: PersistenceInput = None,
single_file: bool = True,
on_flush: bool = False,
context_types: ContextTypes[Any, UD, CD, BD] = None,
):
super().__init__(store_data=store_data)
self.filename = filename
self.filepath = Path(filepath)
self.single_file = single_file
self.on_flush = on_flush
self.user_data: Optional[DefaultDict[int, UD]] = None
@ -141,8 +146,7 @@ class PicklePersistence(BasePersistence[UD, CD, BD]):
def _load_singlefile(self) -> None:
try:
filename = self.filename
with open(self.filename, "rb") as file:
with self.filepath.open("rb") as file:
data = pickle.load(file)
self.user_data = defaultdict(self.context_types.user_data, data['user_data'])
self.chat_data = defaultdict(self.context_types.chat_data, data['chat_data'])
@ -157,24 +161,24 @@ class PicklePersistence(BasePersistence[UD, CD, BD]):
self.bot_data = self.context_types.bot_data()
self.callback_data = None
except pickle.UnpicklingError as exc:
filename = self.filepath.name
raise TypeError(f"File {filename} does not contain valid pickle data") from exc
except Exception as exc:
raise TypeError(f"Something went wrong unpickling {filename}") from exc
raise TypeError(f"Something went wrong unpickling {self.filepath.name}") from exc
@staticmethod
def _load_file(filename: str) -> Any:
def _load_file(filepath: Path) -> Any:
try:
with open(filename, "rb") as file:
with filepath.open("rb") as file:
return pickle.load(file)
except OSError:
return None
except pickle.UnpicklingError as exc:
raise TypeError(f"File {filename} does not contain valid pickle data") from exc
raise TypeError(f"File {filepath.name} does not contain valid pickle data") from exc
except Exception as exc:
raise TypeError(f"Something went wrong unpickling {filename}") from exc
raise TypeError(f"Something went wrong unpickling {filepath.name}") from exc
def _dump_singlefile(self) -> None:
with open(self.filename, "wb") as file:
data = {
'conversations': self.conversations,
'user_data': self.user_data,
@ -182,11 +186,12 @@ class PicklePersistence(BasePersistence[UD, CD, BD]):
'bot_data': self.bot_data,
'callback_data': self.callback_data,
}
with self.filepath.open("wb") as file:
pickle.dump(data, file)
@staticmethod
def _dump_file(filename: str, data: object) -> None:
with open(filename, "wb") as file:
def _dump_file(filepath: Path, data: object) -> None:
with filepath.open("wb") as file:
pickle.dump(data, file)
def get_user_data(self) -> DefaultDict[int, UD]:
@ -198,8 +203,7 @@ class PicklePersistence(BasePersistence[UD, CD, BD]):
if self.user_data:
pass
elif not self.single_file:
filename = f"{self.filename}_user_data"
data = self._load_file(filename)
data = self._load_file(Path(f"{self.filepath}_user_data"))
if not data:
data = defaultdict(self.context_types.user_data)
else:
@ -218,8 +222,7 @@ class PicklePersistence(BasePersistence[UD, CD, BD]):
if self.chat_data:
pass
elif not self.single_file:
filename = f"{self.filename}_chat_data"
data = self._load_file(filename)
data = self._load_file(Path(f"{self.filepath}_chat_data"))
if not data:
data = defaultdict(self.context_types.chat_data)
else:
@ -239,8 +242,7 @@ class PicklePersistence(BasePersistence[UD, CD, BD]):
if self.bot_data:
pass
elif not self.single_file:
filename = f"{self.filename}_bot_data"
data = self._load_file(filename)
data = self._load_file(Path(f"{self.filepath}_bot_data"))
if not data:
data = self.context_types.bot_data()
self.bot_data = data
@ -260,8 +262,7 @@ class PicklePersistence(BasePersistence[UD, CD, BD]):
if self.callback_data:
pass
elif not self.single_file:
filename = f"{self.filename}_callback_data"
data = self._load_file(filename)
data = self._load_file(Path(f"{self.filepath}_callback_data"))
if not data:
data = None
self.callback_data = data
@ -283,8 +284,7 @@ class PicklePersistence(BasePersistence[UD, CD, BD]):
if self.conversations:
pass
elif not self.single_file:
filename = f"{self.filename}_conversations"
data = self._load_file(filename)
data = self._load_file(Path(f"{self.filepath}_conversations"))
if not data:
data = {name: {}}
self.conversations = data
@ -310,8 +310,7 @@ class PicklePersistence(BasePersistence[UD, CD, BD]):
self.conversations[name][key] = new_state
if not self.on_flush:
if not self.single_file:
filename = f"{self.filename}_conversations"
self._dump_file(filename, self.conversations)
self._dump_file(Path(f"{self.filepath}_conversations"), self.conversations)
else:
self._dump_singlefile()
@ -330,8 +329,7 @@ class PicklePersistence(BasePersistence[UD, CD, BD]):
self.user_data[user_id] = data
if not self.on_flush:
if not self.single_file:
filename = f"{self.filename}_user_data"
self._dump_file(filename, self.user_data)
self._dump_file(Path(f"{self.filepath}_user_data"), self.user_data)
else:
self._dump_singlefile()
@ -350,8 +348,7 @@ class PicklePersistence(BasePersistence[UD, CD, BD]):
self.chat_data[chat_id] = data
if not self.on_flush:
if not self.single_file:
filename = f"{self.filename}_chat_data"
self._dump_file(filename, self.chat_data)
self._dump_file(Path(f"{self.filepath}_chat_data"), self.chat_data)
else:
self._dump_singlefile()
@ -367,8 +364,7 @@ class PicklePersistence(BasePersistence[UD, CD, BD]):
self.bot_data = data
if not self.on_flush:
if not self.single_file:
filename = f"{self.filename}_bot_data"
self._dump_file(filename, self.bot_data)
self._dump_file(Path(f"{self.filepath}_bot_data"), self.bot_data)
else:
self._dump_singlefile()
@ -387,8 +383,7 @@ class PicklePersistence(BasePersistence[UD, CD, BD]):
self.callback_data = (data[0], data[1].copy())
if not self.on_flush:
if not self.single_file:
filename = f"{self.filename}_callback_data"
self._dump_file(filename, self.callback_data)
self._dump_file(Path(f"{self.filepath}_callback_data"), self.callback_data)
else:
self._dump_singlefile()
@ -426,12 +421,12 @@ class PicklePersistence(BasePersistence[UD, CD, BD]):
self._dump_singlefile()
else:
if self.user_data:
self._dump_file(f"{self.filename}_user_data", self.user_data)
self._dump_file(Path(f"{self.filepath}_user_data"), self.user_data)
if self.chat_data:
self._dump_file(f"{self.filename}_chat_data", self.chat_data)
self._dump_file(Path(f"{self.filepath}_chat_data"), self.chat_data)
if self.bot_data:
self._dump_file(f"{self.filename}_bot_data", self.bot_data)
self._dump_file(Path(f"{self.filepath}_bot_data"), self.bot_data)
if self.callback_data:
self._dump_file(f"{self.filename}_callback_data", self.callback_data)
self._dump_file(Path(f"{self.filepath}_callback_data"), self.callback_data)
if self.conversations:
self._dump_file(f"{self.filename}_conversations", self.conversations)
self._dump_file(Path(f"{self.filepath}_conversations"), self.conversations)

51
telegram/files/file.py
View file

@ -17,11 +17,10 @@
# You should have received a copy of the GNU Lesser Public License
# along with this program. If not, see [http://www.gnu.org/licenses/].
"""This module contains an object that represents a Telegram File."""
import os
import shutil
import urllib.parse as urllib_parse
from base64 import b64decode
from os.path import basename
from pathlib import Path
from typing import IO, TYPE_CHECKING, Any, Optional, Union
from telegram import TelegramObject
@ -97,8 +96,8 @@ class File(TelegramObject):
self._id_attrs = (self.file_unique_id,)
def download(
self, custom_path: str = None, out: IO = None, timeout: int = None
) -> Union[str, IO]:
self, custom_path: Union[Path, str] = None, out: IO = None, timeout: int = None
) -> Union[Path, IO]:
"""
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
@ -112,8 +111,12 @@ class File(TelegramObject):
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.
.. versionchanged:: 14.0
* ``custom_path`` parameter now also accepts :obj:`pathlib.Path` as argument.
* Returns :obj:`pathlib.Path` object in cases where previously returned `str` object.
Args:
custom_path (:obj:`str`, optional): Custom path.
custom_path (:obj:`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.
timeout (:obj:`int` | :obj:`float`, optional): If this value is specified, use it as
@ -121,7 +124,8 @@ class File(TelegramObject):
the connection pool).
Returns:
:obj:`str` | :obj:`io.BufferedWriter`: The same object as :attr:`out` if specified.
:obj:`pathlib.Path` | :obj:`io.BufferedWriter`: The same object as :attr:`out` if
specified.
Otherwise, returns the filename downloaded to or the file path of the local file.
Raises:
@ -129,20 +133,15 @@ class File(TelegramObject):
"""
if custom_path is not None and out is not None:
raise ValueError('custom_path and out are mutually exclusive')
raise ValueError('`custom_path` and `out` are mutually exclusive')
local_file = is_local_file(self.file_path)
if local_file:
url = self.file_path
else:
# Convert any UTF-8 char into a url encoded ASCII string.
url = self._get_encoded_url()
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:
with open(url, 'rb') as file:
buf = file.read()
buf = path.read_bytes()
else:
buf = self.bot.request.retrieve(url)
if self._credentials:
@ -152,31 +151,30 @@ class File(TelegramObject):
out.write(buf)
return out
if custom_path and local_file:
shutil.copyfile(self.file_path, custom_path)
return custom_path
if custom_path is not None and local_file:
shutil.copyfile(self.file_path, str(custom_path))
return Path(custom_path)
if custom_path:
filename = custom_path
filename = Path(custom_path)
elif local_file:
return self.file_path
return Path(self.file_path)
elif self.file_path:
filename = basename(self.file_path)
filename = Path(Path(self.file_path).name)
else:
filename = os.path.join(os.getcwd(), self.file_id)
filename = Path.cwd() / self.file_id
buf = self.bot.request.retrieve(url, timeout=timeout)
if self._credentials:
buf = decrypt(
b64decode(self._credentials.secret), b64decode(self._credentials.hash), buf
)
with open(filename, 'wb') as fobj:
fobj.write(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(self.file_path)
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
@ -197,8 +195,7 @@ class File(TelegramObject):
if buf is None:
buf = bytearray()
if is_local_file(self.file_path):
with open(self.file_path, "rb") as file:
buf.extend(file.read())
buf.extend(Path(self.file_path).read_bytes())
else:
buf.extend(self.bot.request.retrieve(self._get_encoded_url()))
return buf

4
telegram/files/inputfile.py
View file

@ -22,7 +22,7 @@
import imghdr
import logging
import mimetypes
import os
from pathlib import Path
from typing import IO, Optional, Tuple, Union
from uuid import uuid4
@ -64,7 +64,7 @@ class InputFile:
if filename:
self.filename = filename
elif hasattr(obj, 'name') and not isinstance(obj.name, int): # type: ignore[union-attr]
self.filename = os.path.basename(obj.name) # type: ignore[union-attr]
self.filename = Path(obj.name).name # type: ignore[union-attr]
image_mime_type = self.is_image(self.input_file_content)
if image_mime_type:

13
telegram/request.py
View file

@ -24,6 +24,7 @@ import os
import socket
import sys
import warnings
from pathlib import Path
try:
import ujson as json
@ -80,6 +81,7 @@ def _render_part(self: RequestField, name: str, value: str) -> str: # pylint: d
Monkey patch urllib3.urllib3.fields.RequestField to make it *not* support RFC2231 compliant
Content-Disposition headers since telegram servers don't understand it. Instead just escape
\\ and " and replace any \n and \r with a space.
"""
value = value.replace('\\', '\\\\').replace('"', '\\"')
value = value.replace('\r', ' ').replace('\n', ' ')
@ -382,17 +384,18 @@ class Request:
return self._request_wrapper('GET', url, **urlopen_kwargs)
def download(self, url: str, filename: str, timeout: float = None) -> None:
def download(self, url: str, filepath: Union[Path, str], timeout: float = None) -> None:
"""Download a file by its URL.
Args:
url (:obj:`str`): The web location we want to retrieve.
filepath (:obj:`pathlib.Path` | :obj:`str`): The filepath to download the file to.
timeout (:obj:`int` | :obj:`float`, optional): If this value is specified, use it as
the read timeout from the server (instead of the one specified during creation of
the connection pool).
filename (:obj:`str`): The filename within the path to download the file.
.. versionchanged:: 14.0
The ``filepath`` parameter now also accepts :obj:`pathlib.Path` objects as argument.
"""
buf = self.retrieve(url, timeout=timeout)
with open(filename, 'wb') as fobj:
fobj.write(buf)
Path(filepath).write_bytes(self.retrieve(url, timeout))

2
tests/conftest.py
View file

@ -120,7 +120,7 @@ def bot(bot_info):
@pytest.fixture(scope='session')
def raw_bot(bot_info):
return DictBot(bot_info['token'], private_key=PRIVATE_KEY, request=DictRequest())
return DictBot(bot_info['token'], private_key=PRIVATE_KEY, request=DictRequest(8))
DEFAULT_BOTS = {}

8
tests/test_animation.py
View file

@ -30,14 +30,14 @@ from tests.conftest import check_shortcut_call, check_shortcut_signature, check_
@pytest.fixture(scope='function')
def animation_file():
f = open('tests/data/game.gif', 'rb')
f = Path('tests/data/game.gif').open('rb')
yield f
f.close()
@pytest.fixture(scope='class')
def animation(bot, chat_id):
with open('tests/data/game.gif', 'rb') as f:
with Path('tests/data/game.gif').open('rb') as f:
return bot.send_animation(
chat_id, animation=f, timeout=50, thumb=open('tests/data/thumb.jpg', 'rb')
).animation
@ -120,9 +120,9 @@ class TestAnimation:
assert new_file.file_id == animation.file_id
assert new_file.file_path.startswith('https://')
new_file.download('game.gif')
new_filepath: Path = new_file.download('game.gif')
assert os.path.isfile('game.gif')
assert new_filepath.is_file()
@flaky(3, 1)
def test_send_animation_url_file(self, bot, chat_id, animation):

10
tests/test_audio.py
View file

@ -30,16 +30,16 @@ from tests.conftest import check_shortcut_call, check_shortcut_signature, check_
@pytest.fixture(scope='function')
def audio_file():
f = open('tests/data/telegram.mp3', 'rb')
f = Path('tests/data/telegram.mp3').open('rb')
yield f
f.close()
@pytest.fixture(scope='class')
def audio(bot, chat_id):
with open('tests/data/telegram.mp3', 'rb') as f:
with Path('tests/data/telegram.mp3').open('rb') as f:
return bot.send_audio(
chat_id, audio=f, timeout=50, thumb=open('tests/data/thumb.jpg', 'rb')
chat_id, audio=f, timeout=50, thumb=Path('tests/data/thumb.jpg').open('rb')
).audio
@ -132,11 +132,11 @@ class TestAudio:
assert new_file.file_size == self.file_size
assert new_file.file_id == audio.file_id
assert new_file.file_unique_id == audio.file_unique_id
assert new_file.file_path.startswith('https://')
assert str(new_file.file_path).startswith('https://')
new_file.download('telegram.mp3')
assert os.path.isfile('telegram.mp3')
assert Path('telegram.mp3').is_file()
@flaky(3, 1)
def test_send_mp3_url_file(self, bot, chat_id, audio):

4
tests/test_bot.py
View file

@ -100,7 +100,7 @@ def message(bot, chat_id):
@pytest.fixture(scope='class')
def media_message(bot, chat_id):
with open('tests/data/telegram.ogg', 'rb') as f:
with Path('tests/data/telegram.ogg').open('rb') as f:
return bot.send_voice(chat_id, voice=f, caption='my caption', timeout=10)
@ -1925,7 +1925,7 @@ class TestBot:
def func():
assert bot.set_chat_photo(channel_id, f)
with open('tests/data/telegram_test_channel.jpg', 'rb') as f:
with Path('tests/data/telegram_test_channel.jpg').open('rb') as f:
expect_bad_request(func, 'Type of file mismatch', 'Telegram did not accept the file.')
def test_set_chat_photo_local_files(self, monkeypatch, bot, chat_id):

6
tests/test_chatphoto.py
View file

@ -17,6 +17,8 @@
# You should have received a copy of the GNU Lesser Public License
# along with this program. If not, see [http://www.gnu.org/licenses/].
import os
from pathlib import Path
import pytest
from flaky import flaky
@ -73,7 +75,7 @@ class TestChatPhoto:
new_file.download('telegram.jpg')
assert os.path.isfile('telegram.jpg')
assert Path('telegram.jpg').is_file()
new_file = bot.get_file(chat_photo.big_file_id)
@ -82,7 +84,7 @@ class TestChatPhoto:
new_file.download('telegram.jpg')
assert os.path.isfile('telegram.jpg')
assert Path('telegram.jpg').is_file()
def test_send_with_chat_photo(self, monkeypatch, bot, super_group_id, chat_photo):
def test(url, data, **kwargs):

4
tests/test_constants.py
View file

@ -16,6 +16,8 @@
#
# You should have received a copy of the GNU Lesser Public License
# along with this program. If not, see [http://www.gnu.org/licenses/].
from pathlib import Path
import pytest
from flaky import flaky
@ -37,7 +39,7 @@ class TestConstants:
@flaky(3, 1)
def test_max_caption_length(self, bot, chat_id):
good_caption = 'a' * constants.MAX_CAPTION_LENGTH
with open('tests/data/telegram.png', 'rb') as f:
with Path('tests/data/telegram.png').open('rb') as f:
good_msg = bot.send_photo(photo=f, caption=good_caption, chat_id=chat_id)
assert good_msg.caption == good_caption

6
tests/test_document.py
View file

@ -37,7 +37,7 @@ def document_file():
@pytest.fixture(scope='class')
def document(bot, chat_id):
with open('tests/data/telegram.png', 'rb') as f:
with Path('tests/data/telegram.png').open('rb') as f:
return bot.send_document(chat_id, document=f, timeout=50).document
@ -111,7 +111,7 @@ class TestDocument:
new_file.download('telegram.png')
assert os.path.isfile('telegram.png')
assert Path('telegram.png').is_file()
@flaky(3, 1)
def test_send_url_gif_file(self, bot, chat_id):
@ -281,7 +281,7 @@ class TestDocument:
@flaky(3, 1)
def test_error_send_empty_file(self, bot, chat_id):
with open(os.devnull, 'rb') as f, pytest.raises(TelegramError):
with Path(os.devnull).open('rb') as f, pytest.raises(TelegramError):
bot.send_document(chat_id=chat_id, document=f)
@flaky(3, 1)

44
tests/test_file.py
View file

@ -92,7 +92,7 @@ class TestFile:
bot.get_file(file_id='')
def test_download_mutuall_exclusive(self, file):
with pytest.raises(ValueError, match='custom_path and out are mutually exclusive'):
with pytest.raises(ValueError, match='`custom_path` and `out` are mutually exclusive'):
file.download('custom_path', 'out')
def test_download(self, monkeypatch, file):
@ -103,41 +103,44 @@ class TestFile:
out_file = file.download()
try:
with open(out_file, 'rb') as fobj:
assert fobj.read() == self.file_content
assert out_file.read_bytes() == self.file_content
finally:
os.unlink(out_file)
out_file.unlink()
def test_download_local_file(self, local_file):
assert local_file.download() == local_file.file_path
assert local_file.download() == Path(local_file.file_path)
def test_download_custom_path(self, monkeypatch, file):
@pytest.mark.parametrize(
'custom_path_type', [str, Path], ids=['str custom_path', 'pathlib.Path custom_path']
)
def test_download_custom_path(self, monkeypatch, file, custom_path_type):
def test(*args, **kwargs):
return self.file_content
monkeypatch.setattr('telegram.request.Request.retrieve', test)
file_handle, custom_path = mkstemp()
custom_path = Path(custom_path)
try:
out_file = file.download(custom_path)
out_file = file.download(custom_path_type(custom_path))
assert out_file == custom_path
with open(out_file, 'rb') as fobj:
assert fobj.read() == self.file_content
assert out_file.read_bytes() == self.file_content
finally:
os.close(file_handle)
os.unlink(custom_path)
custom_path.unlink()
def test_download_custom_path_local_file(self, local_file):
@pytest.mark.parametrize(
'custom_path_type', [str, Path], ids=['str custom_path', 'pathlib.Path custom_path']
)
def test_download_custom_path_local_file(self, local_file, custom_path_type):
file_handle, custom_path = mkstemp()
custom_path = Path(custom_path)
try:
out_file = local_file.download(custom_path)
out_file = local_file.download(custom_path_type(custom_path))
assert out_file == custom_path
with open(out_file, 'rb') as fobj:
assert fobj.read() == self.file_content
assert out_file.read_bytes() == self.file_content
finally:
os.close(file_handle)
os.unlink(custom_path)
custom_path.unlink()
def test_download_no_filename(self, monkeypatch, file):
def test(*args, **kwargs):
@ -148,12 +151,11 @@ class TestFile:
monkeypatch.setattr('telegram.request.Request.retrieve', test)
out_file = file.download()
assert out_file[-len(file.file_id) :] == file.file_id
assert str(out_file)[-len(file.file_id) :] == file.file_id
try:
with open(out_file, 'rb') as fobj:
assert fobj.read() == self.file_content
assert out_file.read_bytes() == self.file_content
finally:
os.unlink(out_file)
out_file.unlink()
def test_download_file_obj(self, monkeypatch, file):
def test(*args, **kwargs):

12
tests/test_files.py
View file

@ -68,14 +68,15 @@ class TestFiles:
assert telegram.utils.files.parse_file_input(string) == expected
def test_parse_file_input_file_like(self):
with open('tests/data/game.gif', 'rb') as file:
source_file = Path('tests/data/game.gif')
with source_file.open('rb') as file:
parsed = telegram.utils.files.parse_file_input(file)
assert isinstance(parsed, InputFile)
assert not parsed.attach
assert parsed.filename == 'game.gif'
with open('tests/data/game.gif', 'rb') as file:
with source_file.open('rb') as file:
parsed = telegram.utils.files.parse_file_input(file, attach=True, filename='test_file')
assert isinstance(parsed, InputFile)
@ -83,16 +84,15 @@ class TestFiles:
assert parsed.filename == 'test_file'
def test_parse_file_input_bytes(self):
with open('tests/data/text_file.txt', 'rb') as file:
parsed = telegram.utils.files.parse_file_input(file.read())
source_file = Path('tests/data/text_file.txt')
parsed = telegram.utils.files.parse_file_input(source_file.read_bytes())
assert isinstance(parsed, InputFile)
assert not parsed.attach
assert parsed.filename == 'application.octet-stream'
with open('tests/data/text_file.txt', 'rb') as file:
parsed = telegram.utils.files.parse_file_input(
file.read(), attach=True, filename='test_file'
source_file.read_bytes(), attach=True, filename='test_file'
)
assert isinstance(parsed, InputFile)

15
tests/test_inputfile.py
View file

@ -17,16 +17,16 @@
# You should have received a copy of the GNU Lesser Public License
# along with this program. If not, see [http://www.gnu.org/licenses/].
import logging
import os
import subprocess
import sys
from io import BytesIO
from pathlib import Path
from telegram import InputFile
class TestInputFile:
png = os.path.join('tests', 'data', 'game.png')
png = Path('tests/data/game.png')
def test_slot_behaviour(self, mro_slots):
inst = InputFile(BytesIO(b'blah'), filename='tg.jpg')
@ -35,15 +35,12 @@ class TestInputFile:
assert len(mro_slots(inst)) == len(set(mro_slots(inst))), "duplicate slot"
def test_subprocess_pipe(self):
if sys.platform == 'win32':
cmd = ['type', self.png]
else:
cmd = ['cat', self.png]
cmd_str = 'type' if sys.platform == 'win32' else 'cat'
cmd = [cmd_str, str(self.png)]
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, shell=(sys.platform == 'win32'))
in_file = InputFile(proc.stdout)
assert in_file.input_file_content == open(self.png, 'rb').read()
assert in_file.input_file_content == self.png.read_bytes()
assert in_file.mimetype == 'image/png'
assert in_file.filename == 'image.png'
@ -124,7 +121,7 @@ class TestInputFile:
def test_send_bytes(self, bot, chat_id):
# We test this here and not at the respective test modules because it's not worth
# duplicating the test for the different methods
with open('tests/data/text_file.txt', 'rb') as file:
with Path('tests/data/text_file.txt').open('rb') as file:
message = bot.send_document(chat_id, file.read())
out = BytesIO()

3
tests/test_inputmedia.py
View file

@ -510,13 +510,12 @@ class TestSendMediaGroup:
self, bot, chat_id, video_file, photo_file, animation_file # noqa: F811
): # noqa: F811
def func():
with open('tests/data/telegram.jpg', 'rb') as file:
return bot.send_media_group(
chat_id,
[
InputMediaVideo(video_file),
InputMediaPhoto(photo_file),
InputMediaPhoto(file.read()),
InputMediaPhoto(Path('tests/data/telegram.jpg').read_bytes()),
],
)

131
tests/test_persistence.py
View file

@ -19,6 +19,7 @@
import gzip
import signal
import uuid
from pathlib import Path
from threading import Lock
from telegram.ext import PersistenceInput
@ -55,7 +56,7 @@ from telegram.ext import (
@pytest.fixture(autouse=True)
def change_directory(tmp_path):
orig_dir = os.getcwd()
orig_dir = Path.cwd()
# Switch to a temporary directory so we don't have to worry about cleaning up files
# (str() for py<3.6)
os.chdir(str(tmp_path))
@ -871,7 +872,7 @@ class TestBasePersistence:
@pytest.fixture(scope='function')
def pickle_persistence():
return PicklePersistence(
filename='pickletest',
filepath='pickletest',
single_file=False,
on_flush=False,
)
@ -880,7 +881,7 @@ def pickle_persistence():
@pytest.fixture(scope='function')
def pickle_persistence_only_bot():
return PicklePersistence(
filename='pickletest',
filepath='pickletest',
store_data=PersistenceInput(callback_data=False, user_data=False, chat_data=False),
single_file=False,
on_flush=False,
@ -890,7 +891,7 @@ def pickle_persistence_only_bot():
@pytest.fixture(scope='function')
def pickle_persistence_only_chat():
return PicklePersistence(
filename='pickletest',
filepath='pickletest',
store_data=PersistenceInput(callback_data=False, user_data=False, bot_data=False),
single_file=False,
on_flush=False,
@ -900,7 +901,7 @@ def pickle_persistence_only_chat():
@pytest.fixture(scope='function')
def pickle_persistence_only_user():
return PicklePersistence(
filename='pickletest',
filepath='pickletest',
store_data=PersistenceInput(callback_data=False, chat_data=False, bot_data=False),
single_file=False,
on_flush=False,
@ -910,7 +911,7 @@ def pickle_persistence_only_user():
@pytest.fixture(scope='function')
def pickle_persistence_only_callback():
return PicklePersistence(
filename='pickletest',
filepath='pickletest',
store_data=PersistenceInput(user_data=False, chat_data=False, bot_data=False),
single_file=False,
on_flush=False,
@ -927,8 +928,7 @@ def bad_pickle_files():
'pickletest_conversations',
'pickletest',
]:
with open(name, 'w') as f:
f.write('(())')
Path(name).write_text('(())')
yield True
@ -958,17 +958,17 @@ def good_pickle_files(user_data, chat_data, bot_data, callback_data, conversatio
'callback_data': callback_data,
'conversations': conversations,
}
with open('pickletest_user_data', 'wb') as f:
with Path('pickletest_user_data').open('wb') as f:
pickle.dump(user_data, f)
with open('pickletest_chat_data', 'wb') as f:
with Path('pickletest_chat_data').open('wb') as f:
pickle.dump(chat_data, f)
with open('pickletest_bot_data', 'wb') as f:
with Path('pickletest_bot_data').open('wb') as f:
pickle.dump(bot_data, f)
with open('pickletest_callback_data', 'wb') as f:
with Path('pickletest_callback_data').open('wb') as f:
pickle.dump(callback_data, f)
with open('pickletest_conversations', 'wb') as f:
with Path('pickletest_conversations').open('wb') as f:
pickle.dump(conversations, f)
with open('pickletest', 'wb') as f:
with Path('pickletest').open('wb') as f:
pickle.dump(data, f)
yield True
@ -981,15 +981,15 @@ def pickle_files_wo_bot_data(user_data, chat_data, callback_data, conversations)
'conversations': conversations,
'callback_data': callback_data,
}
with open('pickletest_user_data', 'wb') as f:
with Path('pickletest_user_data').open('wb') as f:
pickle.dump(user_data, f)
with open('pickletest_chat_data', 'wb') as f:
with Path('pickletest_chat_data').open('wb') as f:
pickle.dump(chat_data, f)
with open('pickletest_callback_data', 'wb') as f:
with Path('pickletest_callback_data').open('wb') as f:
pickle.dump(callback_data, f)
with open('pickletest_conversations', 'wb') as f:
with Path('pickletest_conversations').open('wb') as f:
pickle.dump(conversations, f)
with open('pickletest', 'wb') as f:
with Path('pickletest').open('wb') as f:
pickle.dump(data, f)
yield True
@ -1002,15 +1002,15 @@ def pickle_files_wo_callback_data(user_data, chat_data, bot_data, conversations)
'bot_data': bot_data,
'conversations': conversations,
}
with open('pickletest_user_data', 'wb') as f:
with Path('pickletest_user_data').open('wb') as f:
pickle.dump(user_data, f)
with open('pickletest_chat_data', 'wb') as f:
with Path('pickletest_chat_data').open('wb') as f:
pickle.dump(chat_data, f)
with open('pickletest_bot_data', 'wb') as f:
with Path('pickletest_bot_data').open('wb') as f:
pickle.dump(bot_data, f)
with open('pickletest_conversations', 'wb') as f:
with Path('pickletest_conversations').open('wb') as f:
pickle.dump(conversations, f)
with open('pickletest', 'wb') as f:
with Path('pickletest').open('wb') as f:
pickle.dump(data, f)
yield True
@ -1339,7 +1339,7 @@ class TestPicklePersistence:
assert not pickle_persistence.user_data == user_data
pickle_persistence.update_user_data(12345, user_data[12345])
assert pickle_persistence.user_data == user_data
with open('pickletest_user_data', 'rb') as f:
with Path('pickletest_user_data').open('rb') as f:
user_data_test = defaultdict(dict, pickle.load(f))
assert user_data_test == user_data
@ -1351,7 +1351,7 @@ class TestPicklePersistence:
assert not pickle_persistence.chat_data == chat_data
pickle_persistence.update_chat_data(-12345, chat_data[-12345])
assert pickle_persistence.chat_data == chat_data
with open('pickletest_chat_data', 'rb') as f:
with Path('pickletest_chat_data').open('rb') as f:
chat_data_test = defaultdict(dict, pickle.load(f))
assert chat_data_test == chat_data
@ -1363,7 +1363,7 @@ class TestPicklePersistence:
assert not pickle_persistence.bot_data == bot_data
pickle_persistence.update_bot_data(bot_data)
assert pickle_persistence.bot_data == bot_data
with open('pickletest_bot_data', 'rb') as f:
with Path('pickletest_bot_data').open('rb') as f:
bot_data_test = pickle.load(f)
assert bot_data_test == bot_data
@ -1375,7 +1375,7 @@ class TestPicklePersistence:
assert not pickle_persistence.callback_data == callback_data
pickle_persistence.update_callback_data(callback_data)
assert pickle_persistence.callback_data == callback_data
with open('pickletest_callback_data', 'rb') as f:
with Path('pickletest_callback_data').open('rb') as f:
callback_data_test = pickle.load(f)
assert callback_data_test == callback_data
@ -1385,7 +1385,7 @@ class TestPicklePersistence:
pickle_persistence.update_conversation('name1', (123, 123), 5)
assert pickle_persistence.conversations['name1'] == conversation1
assert pickle_persistence.get_conversations('name1') == conversation1
with open('pickletest_conversations', 'rb') as f:
with Path('pickletest_conversations').open('rb') as f:
conversations_test = defaultdict(dict, pickle.load(f))
assert conversations_test['name1'] == conversation1
@ -1405,7 +1405,7 @@ class TestPicklePersistence:
assert not pickle_persistence.user_data == user_data
pickle_persistence.update_user_data(12345, user_data[12345])
assert pickle_persistence.user_data == user_data
with open('pickletest', 'rb') as f:
with Path('pickletest').open('rb') as f:
user_data_test = defaultdict(dict, pickle.load(f)['user_data'])
assert user_data_test == user_data
@ -1417,7 +1417,7 @@ class TestPicklePersistence:
assert not pickle_persistence.chat_data == chat_data
pickle_persistence.update_chat_data(-12345, chat_data[-12345])
assert pickle_persistence.chat_data == chat_data
with open('pickletest', 'rb') as f:
with Path('pickletest').open('rb') as f:
chat_data_test = defaultdict(dict, pickle.load(f)['chat_data'])
assert chat_data_test == chat_data
@ -1429,7 +1429,7 @@ class TestPicklePersistence:
assert not pickle_persistence.bot_data == bot_data
pickle_persistence.update_bot_data(bot_data)
assert pickle_persistence.bot_data == bot_data
with open('pickletest', 'rb') as f:
with Path('pickletest').open('rb') as f:
bot_data_test = pickle.load(f)['bot_data']
assert bot_data_test == bot_data
@ -1441,7 +1441,7 @@ class TestPicklePersistence:
assert not pickle_persistence.callback_data == callback_data
pickle_persistence.update_callback_data(callback_data)
assert pickle_persistence.callback_data == callback_data
with open('pickletest', 'rb') as f:
with Path('pickletest').open('rb') as f:
callback_data_test = pickle.load(f)['callback_data']
assert callback_data_test == callback_data
@ -1451,7 +1451,7 @@ class TestPicklePersistence:
pickle_persistence.update_conversation('name1', (123, 123), 5)
assert pickle_persistence.conversations['name1'] == conversation1
assert pickle_persistence.get_conversations('name1') == conversation1
with open('pickletest', 'rb') as f:
with Path('pickletest').open('rb') as f:
conversations_test = defaultdict(dict, pickle.load(f)['conversations'])
assert conversations_test['name1'] == conversation1
@ -1487,7 +1487,7 @@ class TestPicklePersistence:
pickle_persistence.update_user_data(54321, user_data[54321])
assert pickle_persistence.user_data == user_data
with open('pickletest_user_data', 'rb') as f:
with Path('pickletest_user_data').open('rb') as f:
user_data_test = defaultdict(dict, pickle.load(f))
assert not user_data_test == user_data
@ -1498,7 +1498,7 @@ class TestPicklePersistence:
pickle_persistence.update_chat_data(54321, chat_data[54321])
assert pickle_persistence.chat_data == chat_data
with open('pickletest_chat_data', 'rb') as f:
with Path('pickletest_chat_data').open('rb') as f:
chat_data_test = defaultdict(dict, pickle.load(f))
assert not chat_data_test == chat_data
@ -1509,7 +1509,7 @@ class TestPicklePersistence:
pickle_persistence.update_bot_data(bot_data)
assert pickle_persistence.bot_data == bot_data
with open('pickletest_bot_data', 'rb') as f:
with Path('pickletest_bot_data').open('rb') as f:
bot_data_test = pickle.load(f)
assert not bot_data_test == bot_data
@ -1520,7 +1520,7 @@ class TestPicklePersistence:
pickle_persistence.update_callback_data(callback_data)
assert pickle_persistence.callback_data == callback_data
with open('pickletest_callback_data', 'rb') as f:
with Path('pickletest_callback_data').open('rb') as f:
callback_data_test = pickle.load(f)
assert not callback_data_test == callback_data
@ -1531,24 +1531,24 @@ class TestPicklePersistence:
pickle_persistence.update_conversation('name1', (123, 123), 5)
assert pickle_persistence.conversations['name1'] == conversation1
with open('pickletest_conversations', 'rb') as f:
with Path('pickletest_conversations').open('rb') as f:
conversations_test = defaultdict(dict, pickle.load(f))
assert not conversations_test['name1'] == conversation1
pickle_persistence.flush()
with open('pickletest_user_data', 'rb') as f:
with Path('pickletest_user_data').open('rb') as f:
user_data_test = defaultdict(dict, pickle.load(f))
assert user_data_test == user_data
with open('pickletest_chat_data', 'rb') as f:
with Path('pickletest_chat_data').open('rb') as f:
chat_data_test = defaultdict(dict, pickle.load(f))
assert chat_data_test == chat_data
with open('pickletest_bot_data', 'rb') as f:
with Path('pickletest_bot_data').open('rb') as f:
bot_data_test = pickle.load(f)
assert bot_data_test == bot_data
with open('pickletest_conversations', 'rb') as f:
with Path('pickletest_conversations').open('rb') as f:
conversations_test = defaultdict(dict, pickle.load(f))
assert conversations_test['name1'] == conversation1
@ -1564,7 +1564,7 @@ class TestPicklePersistence:
assert not pickle_persistence.user_data == user_data
pickle_persistence.update_user_data(54321, user_data[54321])
assert pickle_persistence.user_data == user_data
with open('pickletest', 'rb') as f:
with Path('pickletest').open('rb') as f:
user_data_test = defaultdict(dict, pickle.load(f)['user_data'])
assert not user_data_test == user_data
@ -1573,7 +1573,7 @@ class TestPicklePersistence:
assert not pickle_persistence.chat_data == chat_data
pickle_persistence.update_chat_data(54321, chat_data[54321])
assert pickle_persistence.chat_data == chat_data
with open('pickletest', 'rb') as f:
with Path('pickletest').open('rb') as f:
chat_data_test = defaultdict(dict, pickle.load(f)['chat_data'])
assert not chat_data_test == chat_data
@ -1582,7 +1582,7 @@ class TestPicklePersistence:
assert not pickle_persistence.bot_data == bot_data
pickle_persistence.update_bot_data(bot_data)
assert pickle_persistence.bot_data == bot_data
with open('pickletest', 'rb') as f:
with Path('pickletest').open('rb') as f:
bot_data_test = pickle.load(f)['bot_data']
assert not bot_data_test == bot_data
@ -1591,7 +1591,7 @@ class TestPicklePersistence:
assert not pickle_persistence.callback_data == callback_data
pickle_persistence.update_callback_data(callback_data)
assert pickle_persistence.callback_data == callback_data
with open('pickletest', 'rb') as f:
with Path('pickletest').open('rb') as f:
callback_data_test = pickle.load(f)['callback_data']
assert not callback_data_test == callback_data
@ -1600,24 +1600,24 @@ class TestPicklePersistence:
assert not pickle_persistence.conversations['name1'] == conversation1
pickle_persistence.update_conversation('name1', (123, 123), 5)
assert pickle_persistence.conversations['name1'] == conversation1
with open('pickletest', 'rb') as f:
with Path('pickletest').open('rb') as f:
conversations_test = defaultdict(dict, pickle.load(f)['conversations'])
assert not conversations_test['name1'] == conversation1
pickle_persistence.flush()
with open('pickletest', 'rb') as f:
with Path('pickletest').open('rb') as f:
user_data_test = defaultdict(dict, pickle.load(f)['user_data'])
assert user_data_test == user_data
with open('pickletest', 'rb') as f:
with Path('pickletest').open('rb') as f:
chat_data_test = defaultdict(dict, pickle.load(f)['chat_data'])
assert chat_data_test == chat_data
with open('pickletest', 'rb') as f:
with Path('pickletest').open('rb') as f:
bot_data_test = pickle.load(f)['bot_data']
assert bot_data_test == bot_data
with open('pickletest', 'rb') as f:
with Path('pickletest').open('rb') as f:
conversations_test = defaultdict(dict, pickle.load(f)['conversations'])
assert conversations_test['name1'] == conversation1
@ -1656,7 +1656,7 @@ class TestPicklePersistence:
dp.add_handler(h1)
dp.process_update(update)
pickle_persistence_2 = PicklePersistence(
filename='pickletest',
filepath='pickletest',
single_file=False,
on_flush=False,
)
@ -1675,7 +1675,7 @@ class TestPicklePersistence:
dp.bot.callback_data_cache._callback_queries['test'] = 'Working4!'
u._signal_handler(signal.SIGINT, None)
pickle_persistence_2 = PicklePersistence(
filename='pickletest',
filepath='pickletest',
single_file=False,
on_flush=False,
)
@ -1695,7 +1695,7 @@ class TestPicklePersistence:
dp.bot.callback_data_cache._callback_queries['test'] = 'Working4!'
u._signal_handler(signal.SIGINT, None)
pickle_persistence_2 = PicklePersistence(
filename='pickletest',
filepath='pickletest',
store_data=PersistenceInput(callback_data=False, chat_data=False, user_data=False),
single_file=False,
on_flush=False,
@ -1715,7 +1715,7 @@ class TestPicklePersistence:
dp.bot.callback_data_cache._callback_queries['test'] = 'Working4!'
u._signal_handler(signal.SIGINT, None)
pickle_persistence_2 = PicklePersistence(
filename='pickletest',
filepath='pickletest',
store_data=PersistenceInput(callback_data=False, user_data=False, bot_data=False),
single_file=False,
on_flush=False,
@ -1735,7 +1735,7 @@ class TestPicklePersistence:
dp.bot.callback_data_cache._callback_queries['test'] = 'Working4!'
u._signal_handler(signal.SIGINT, None)
pickle_persistence_2 = PicklePersistence(
filename='pickletest',
filepath='pickletest',
store_data=PersistenceInput(callback_data=False, chat_data=False, bot_data=False),
single_file=False,
on_flush=False,
@ -1758,7 +1758,7 @@ class TestPicklePersistence:
del u
del pickle_persistence_only_callback
pickle_persistence_2 = PicklePersistence(
filename='pickletest',
filepath='pickletest',
store_data=PersistenceInput(user_data=False, chat_data=False, bot_data=False),
single_file=False,
on_flush=False,
@ -1852,6 +1852,21 @@ class TestPicklePersistence:
assert nested_ch.conversations[nested_ch._get_key(update)] == 1
assert nested_ch.conversations == pickle_persistence.conversations['name3']
@pytest.mark.parametrize(
'filepath',
['pickletest', Path('pickletest')],
ids=['str filepath', 'pathlib.Path filepath'],
)
def test_filepath_argument_types(self, filepath):
pick_persist = PicklePersistence(
filepath=filepath,
on_flush=False,
)
pick_persist.update_user_data(1, 1)
assert pick_persist.get_user_data()[1] == 1
assert Path(filepath).is_file()
def test_with_job(self, job_queue, dp, pickle_persistence):
dp.bot.arbitrary_callback_data = True

16
tests/test_photo.py
View file

@ -43,7 +43,7 @@ def photo_file():
@pytest.fixture(scope='class')
def _photo(bot, chat_id):
def func():
with open('tests/data/telegram.jpg', 'rb') as f:
with Path('tests/data/telegram.jpg').open('rb') as f:
return bot.send_photo(chat_id, photo=f, timeout=50).photo
return expect_bad_request(func, 'Type of file mismatch', 'Telegram did not accept the file.')
@ -288,7 +288,7 @@ class TestPhoto:
new_file.download('telegram.jpg')
assert os.path.isfile('telegram.jpg') is True
assert Path('telegram.jpg').is_file()
@flaky(3, 1)
def test_send_url_jpg_file(self, bot, chat_id, thumb, photo):
@ -343,7 +343,7 @@ class TestPhoto:
"""
Regression test for https://github.com/python-telegram-bot/python-telegram-bot/issues/1202
"""
with open('tests/data/测试.png', 'rb') as f:
with Path('tests/data/测试.png').open('rb') as f:
message = bot.send_photo(photo=f, chat_id=chat_id)
photo = message.photo[-1]
@ -356,21 +356,21 @@ class TestPhoto:
@flaky(3, 1)
def test_send_bytesio_jpg_file(self, bot, chat_id):
file_name = 'tests/data/telegram_no_standard_header.jpg'
filepath: Path = Path('tests/data/telegram_no_standard_header.jpg')
# raw image bytes
raw_bytes = BytesIO(open(file_name, 'rb').read())
raw_bytes = BytesIO(filepath.read_bytes())
input_file = InputFile(raw_bytes)
assert input_file.mimetype == 'application/octet-stream'
# raw image bytes with name info
raw_bytes = BytesIO(open(file_name, 'rb').read())
raw_bytes.name = file_name
raw_bytes = BytesIO(filepath.read_bytes())
raw_bytes.name = str(filepath)
input_file = InputFile(raw_bytes)
assert input_file.mimetype == 'image/jpeg'
# send raw photo
raw_bytes = BytesIO(open(file_name, 'rb').read())
raw_bytes = BytesIO(filepath.read_bytes())
message = bot.send_photo(chat_id, photo=raw_bytes)
photo = message.photo[-1]
assert isinstance(photo.file_id, str)

15
tests/test_request.py
View file

@ -16,6 +16,8 @@
#
# You should have received a copy of the GNU Lesser Public License
# along with this program. If not, see [http://www.gnu.org/licenses/].
from pathlib import Path
import pytest
from telegram.error import TelegramError
@ -48,3 +50,16 @@ def test_parse_illegal_json():
with pytest.raises(TelegramError, match='Invalid server response'):
Request._parse(server_response)
@pytest.mark.parametrize(
"destination_path_type",
[str, Path],
ids=['str destination_path', 'pathlib.Path destination_path'],
)
def test_download(destination_path_type):
destination_filepath = Path.cwd() / 'tests' / 'data' / 'downloaded_request.txt'
request = Request()
request.download("http://google.com", destination_path_type(destination_filepath))
assert destination_filepath.is_file()
destination_filepath.unlink()

29
tests/test_sticker.py
View file

@ -30,39 +30,37 @@ from tests.conftest import check_shortcut_call, check_shortcut_signature, check_
@pytest.fixture(scope='function')
def sticker_file():
f = open('tests/data/telegram.webp', 'rb')
yield f
f.close()
with Path('tests/data/telegram.webp').open('rb') as file:
yield file
@pytest.fixture(scope='class')
def sticker(bot, chat_id):
with open('tests/data/telegram.webp', 'rb') as f:
with Path('tests/data/telegram.webp').open('rb') as f:
return bot.send_sticker(chat_id, sticker=f, timeout=50).sticker
@pytest.fixture(scope='function')
def animated_sticker_file():
f = open('tests/data/telegram_animated_sticker.tgs', 'rb')
with Path('tests/data/telegram_animated_sticker.tgs').open('rb') as f:
yield f
f.close()
@pytest.fixture(scope='class')
def animated_sticker(bot, chat_id):
with open('tests/data/telegram_animated_sticker.tgs', 'rb') as f:
with Path('tests/data/telegram_animated_sticker.tgs').open('rb') as f:
return bot.send_sticker(chat_id, sticker=f, timeout=50).sticker
@pytest.fixture(scope='function')
def video_sticker_file():
with open('tests/data/telegram_video_sticker.webm', 'rb') as f:
with Path('tests/data/telegram_video_sticker.webm').open('rb') as f:
yield f
@pytest.fixture(scope='class')
def video_sticker(bot, chat_id):
with open('tests/data/telegram_video_sticker.webm', 'rb') as f:
with Path('tests/data/telegram_video_sticker.webm').open('rb') as f:
return bot.send_sticker(chat_id, sticker=f, timeout=50).sticker
@ -153,7 +151,7 @@ class TestSticker:
new_file.download('telegram.webp')
assert os.path.isfile('telegram.webp')
assert Path('telegram.webp').is_file()
@flaky(3, 1)
def test_resend(self, bot, chat_id, sticker):
@ -377,9 +375,8 @@ def video_sticker_set(bot):
@pytest.fixture(scope='function')
def sticker_set_thumb_file():
f = open('tests/data/sticker_set_thumb.png', 'rb')
yield f
f.close()
with Path('tests/data/sticker_set_thumb.png').open('rb') as file:
yield file
class TestStickerSet:
@ -455,7 +452,7 @@ class TestStickerSet:
@flaky(3, 1)
def test_bot_methods_1_png(self, bot, chat_id, sticker_file):
with open('tests/data/telegram_sticker.png', 'rb') as f:
with Path('tests/data/telegram_sticker.png').open('rb') as f:
# chat_id was hardcoded as 95205500 but it stopped working for some reason
file = bot.upload_sticker_file(chat_id, f)
assert file
@ -476,13 +473,13 @@ class TestStickerSet:
assert bot.add_sticker_to_set(
chat_id,
f'animated_test_by_{bot.username}',
tgs_sticker=open('tests/data/telegram_animated_sticker.tgs', 'rb'),
tgs_sticker=Path('tests/data/telegram_animated_sticker.tgs').open('rb'),
emojis='😄',
)
@flaky(3, 1)
def test_bot_methods_1_webm(self, bot, chat_id):
with open('tests/data/telegram_video_sticker.webm', 'rb') as f:
with Path('tests/data/telegram_video_sticker.webm').open('rb') as f:
assert bot.add_sticker_to_set(
chat_id, f'video_test_by_{bot.username}', webm_sticker=f, emojis='🤔'
)

6
tests/test_video.py
View file

@ -30,14 +30,14 @@ from tests.conftest import check_shortcut_call, check_shortcut_signature, check_
@pytest.fixture(scope='function')
def video_file():
f = open('tests/data/telegram.mp4', 'rb')
f = Path('tests/data/telegram.mp4').open('rb')
yield f
f.close()
@pytest.fixture(scope='class')
def video(bot, chat_id):
with open('tests/data/telegram.mp4', 'rb') as f:
with Path('tests/data/telegram.mp4').open('rb') as f:
return bot.send_video(chat_id, video=f, timeout=50).video
@ -141,7 +141,7 @@ class TestVideo:
new_file.download('telegram.mp4')
assert os.path.isfile('telegram.mp4')
assert Path('telegram.mp4').is_file()
@flaky(3, 1)
def test_send_mp4_file_url(self, bot, chat_id, video):

4
tests/test_videonote.py
View file

@ -36,7 +36,7 @@ def video_note_file():
@pytest.fixture(scope='class')
def video_note(bot, chat_id):
with open('tests/data/telegram2.mp4', 'rb') as f:
with Path('tests/data/telegram2.mp4').open('rb') as f:
return bot.send_video_note(chat_id, video_note=f, timeout=50).video_note
@ -123,7 +123,7 @@ class TestVideoNote:
new_file.download('telegram2.mp4')
assert os.path.isfile('telegram2.mp4')
assert Path('telegram2.mp4').is_file()
@flaky(3, 1)
def test_resend(self, bot, chat_id, video_note):

8
tests/test_voice.py
View file

@ -30,14 +30,14 @@ from tests.conftest import check_shortcut_call, check_shortcut_signature, check_
@pytest.fixture(scope='function')
def voice_file():
f = open('tests/data/telegram.ogg', 'rb')
f = Path('tests/data/telegram.ogg').open('rb')
yield f
f.close()
@pytest.fixture(scope='class')
def voice(bot, chat_id):
with open('tests/data/telegram.ogg', 'rb') as f:
with Path('tests/data/telegram.ogg').open('rb') as f:
return bot.send_voice(chat_id, voice=f, timeout=50).voice
@ -111,9 +111,9 @@ class TestVoice:
assert new_file.file_unique_id == voice.file_unique_id
assert new_file.file_path.startswith('https://')
new_file.download('telegram.ogg')
new_filepath = new_file.download('telegram.ogg')
assert os.path.isfile('telegram.ogg')
assert new_filepath.is_file()
@flaky(3, 1)
def test_send_ogg_url_file(self, bot, chat_id, voice):