mirror of
https://github.com/python-telegram-bot/python-telegram-bot.git
synced 2024-12-23 06:50:29 +01:00
138 lines
5.3 KiB
Python
138 lines
5.3 KiB
Python
#!/usr/bin/env python
|
|
#
|
|
# A library that provides a Python interface to the Telegram Bot API
|
|
# Copyright (C) 2015-2018
|
|
# Leandro Toledo de Souza <devs@python-telegram-bot.org>
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Lesser Public License as published by
|
|
# 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
|
|
# GNU Lesser Public License for more details.
|
|
#
|
|
# 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
|
|
import sys
|
|
from collections import defaultdict
|
|
from queue import Queue
|
|
from threading import Thread, Event
|
|
from time import sleep
|
|
|
|
import pytest
|
|
|
|
from telegram import Bot
|
|
from telegram.ext import Dispatcher, JobQueue, Updater
|
|
from tests.bots import get_bot
|
|
|
|
TRAVIS = os.getenv('TRAVIS', False)
|
|
|
|
if TRAVIS:
|
|
pytest_plugins = ['tests.travis_fold']
|
|
|
|
# THIS KEY IS OBVIOUSLY COMPROMISED
|
|
# DO NOT USE IN PRODUCTION!
|
|
PRIVATE_KEY = b"-----BEGIN RSA PRIVATE KEY-----\r\nMIIEowIBAAKCAQEA0AvEbNaOnfIL3GjB8VI4M5IaWe+GcK8eSPHkLkXREIsaddum\r\nwPBm/+w8lFYdnY+O06OEJrsaDtwGdU//8cbGJ/H/9cJH3dh0tNbfszP7nTrQD+88\r\nydlcYHzClaG8G+oTe9uEZSVdDXj5IUqR0y6rDXXb9tC9l+oSz+ShYg6+C4grAb3E\r\nSTv5khZ9Zsi/JEPWStqNdpoNuRh7qEYc3t4B/a5BH7bsQENyJSc8AWrfv+drPAEe\r\njQ8xm1ygzWvJp8yZPwOIYuL+obtANcoVT2G2150Wy6qLC0bD88Bm40GqLbSazueC\r\nRHZRug0B9rMUKvKc4FhG4AlNzBCaKgIcCWEqKwIDAQABAoIBACcIjin9d3Sa3S7V\r\nWM32JyVF3DvTfN3XfU8iUzV7U+ZOswA53eeFM04A/Ly4C4ZsUNfUbg72O8Vd8rg/\r\n8j1ilfsYpHVvphwxaHQlfIMa1bKCPlc/A6C7b2GLBtccKTbzjARJA2YWxIaqk9Nz\r\nMjj1IJK98i80qt29xRnMQ5sqOO3gn2SxTErvNchtBiwOH8NirqERXig8VCY6fr3n\r\nz7ZImPU3G/4qpD0+9ULrt9x/VkjqVvNdK1l7CyAuve3D7ha3jPMfVHFtVH5gqbyp\r\nKotyIHAyD+Ex3FQ1JV+H7DkP0cPctQiss7OiO9Zd9C1G2OrfQz9el7ewAPqOmZtC\r\nKjB3hUECgYEA/4MfKa1cvaCqzd3yUprp1JhvssVkhM1HyucIxB5xmBcVLX2/Kdhn\r\nhiDApZXARK0O9IRpFF6QVeMEX7TzFwB6dfkyIePsGxputA5SPbtBlHOvjZa8omMl\r\nEYfNa8x/mJkvSEpzvkWPascuHJWv1cEypqphu/70DxubWB5UKo/8o6cCgYEA0HFy\r\ncgwPMB//nltHGrmaQZPFT7/Qgl9ErZT3G9S8teWY4o4CXnkdU75tBoKAaJnpSfX3\r\nq8VuRerF45AFhqCKhlG4l51oW7TUH50qE3GM+4ivaH5YZB3biwQ9Wqw+QyNLAh/Q\r\nnS4/Wwb8qC9QuyEgcCju5lsCaPEXZiZqtPVxZd0CgYEAshBG31yZjO0zG1TZUwfy\r\nfN3euc8mRgZpSdXIHiS5NSyg7Zr8ZcUSID8jAkJiQ3n3OiAsuq1MGQ6kNa582kLT\r\nFPQdI9Ea8ahyDbkNR0gAY9xbM2kg/Gnro1PorH9PTKE0ekSodKk1UUyNrg4DBAwn\r\nqE6E3ebHXt/2WmqIbUD653ECgYBQCC8EAQNX3AFegPd1GGxU33Lz4tchJ4kMCNU0\r\nN2NZh9VCr3nTYjdTbxsXU8YP44CCKFG2/zAO4kymyiaFAWEOn5P7irGF/JExrjt4\r\nibGy5lFLEq/HiPtBjhgsl1O0nXlwUFzd7OLghXc+8CPUJaz5w42unqT3PBJa40c3\r\nQcIPdQKBgBnSb7BcDAAQ/Qx9juo/RKpvhyeqlnp0GzPSQjvtWi9dQRIu9Pe7luHc\r\nm1Img1EO1OyE3dis/rLaDsAa2AKu1Yx6h85EmNjavBqP9wqmFa0NIQQH8fvzKY3/\r\nP8IHY6009aoamLqYaexvrkHVq7fFKiI6k8myMJ6qblVNFv14+KXU\r\n-----END RSA PRIVATE KEY-----" # noqa: E501
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def bot_info():
|
|
return get_bot()
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def bot(bot_info):
|
|
return Bot(bot_info['token'], private_key=PRIVATE_KEY)
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def chat_id(bot_info):
|
|
return bot_info['chat_id']
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def group_id(bot_info):
|
|
return bot_info['group_id']
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def channel_id(bot_info):
|
|
return bot_info['channel_id']
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def provider_token(bot_info):
|
|
return bot_info['payment_provider_token']
|
|
|
|
|
|
def create_dp(bot):
|
|
# Dispatcher is heavy to init (due to many threads and such) so we have a single session
|
|
# scoped one here, but before each test, reset it (dp fixture below)
|
|
dispatcher = Dispatcher(bot, Queue(), job_queue=JobQueue(bot), workers=2)
|
|
thr = Thread(target=dispatcher.start)
|
|
thr.start()
|
|
sleep(2)
|
|
yield dispatcher
|
|
sleep(1)
|
|
if dispatcher.running:
|
|
dispatcher.stop()
|
|
thr.join()
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def _dp(bot):
|
|
for dp in create_dp(bot):
|
|
yield dp
|
|
|
|
|
|
@pytest.fixture(scope='function')
|
|
def dp(_dp):
|
|
# Reset the dispatcher first
|
|
while not _dp.update_queue.empty():
|
|
_dp.update_queue.get(False)
|
|
_dp.chat_data = defaultdict(dict)
|
|
_dp.user_data = defaultdict(dict)
|
|
_dp.persistence = None
|
|
_dp.handlers = {}
|
|
_dp.groups = []
|
|
_dp.error_handlers = []
|
|
_dp.__stop_event = Event()
|
|
_dp.__exception_event = Event()
|
|
_dp.__async_queue = Queue()
|
|
_dp.__async_threads = set()
|
|
if _dp._Dispatcher__singleton_semaphore.acquire(blocking=0):
|
|
Dispatcher._set_singleton(_dp)
|
|
yield _dp
|
|
Dispatcher._Dispatcher__singleton_semaphore.release()
|
|
|
|
|
|
@pytest.fixture(scope='function')
|
|
def updater(bot):
|
|
up = Updater(bot=bot, workers=2)
|
|
yield up
|
|
if up.running:
|
|
up.stop()
|
|
|
|
|
|
@pytest.fixture(scope='function')
|
|
def thumb_file():
|
|
f = open(u'tests/data/thumb.jpg', 'rb')
|
|
yield f
|
|
f.close()
|
|
|
|
|
|
@pytest.fixture(scope='class')
|
|
def class_thumb_file():
|
|
f = open(u'tests/data/thumb.jpg', 'rb')
|
|
yield f
|
|
f.close()
|
|
|
|
|
|
def pytest_configure(config):
|
|
if sys.version_info >= (3,):
|
|
config.addinivalue_line('filterwarnings', 'ignore::ResourceWarning')
|
|
# TODO: Write so good code that we don't need to ignore ResourceWarnings anymore
|