Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Make Tests fortelegram.ext Independent of Networking#4454

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to ourterms of service andprivacy statement. We’ll occasionally send you account related emails.

Already on GitHub?Sign in to your account

Merged
Bibo-Joshi merged 7 commits intomasterfromno-request-tests
Sep 9, 2024
Merged
Show file tree
Hide file tree
Changes fromall commits
Commits
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 29 additions & 0 deletionstests/auxil/monkeypatch.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
#!/usr/bin/env python
#
# A library that provides a Python interface to the Telegram Bot API
# Copyright (C) 2015-2024
# 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 asyncio


async def return_true(*args, **kwargs):
return True


async def empty_get_updates(*args, **kwargs):
# The `await` gives the event loop a chance to run other tasks
await asyncio.sleep(0)
return []
31 changes: 29 additions & 2 deletionstests/auxil/networking.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -17,7 +17,7 @@
# 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
from typing import Optional
from typing import Optional, Tuple

import pytest
from httpx import AsyncClient, AsyncHTTPTransport, Response
Expand All@@ -26,7 +26,7 @@
from telegram._utils.strings import TextEncoding
from telegram._utils.types import ODVInput
from telegram.error import BadRequest, RetryAfter, TimedOut
from telegram.request import HTTPXRequest, RequestData
from telegram.request importBaseRequest,HTTPXRequest, RequestData


class NonchalantHttpxRequest(HTTPXRequest):
Expand DownExpand Up@@ -60,6 +60,33 @@ async def _request_wrapper(
pytest.xfail(f"Ignoring TimedOut error: {e}")


class OfflineRequest(BaseRequest):
"""This Request class disallows making requests to Telegram's servers.
Use this in tests that should not depend on the network.
"""

async def initialize(self) -> None:
pass

async def shutdown(self) -> None:
pass

def __init__(self, *args, **kwargs):
pass

async def do_request(
self,
url: str,
method: str,
request_data: Optional[RequestData] = None,
read_timeout: ODVInput[float] = BaseRequest.DEFAULT_NONE,
write_timeout: ODVInput[float] = BaseRequest.DEFAULT_NONE,
connect_timeout: ODVInput[float] = BaseRequest.DEFAULT_NONE,
pool_timeout: ODVInput[float] = BaseRequest.DEFAULT_NONE,
) -> Tuple[int, bytes]:
pytest.fail("OfflineRequest: Network access disallowed in this test")


async def expect_bad_request(func, message, reason):
"""
Wrapper for testing bot functions expected to result in an :class:`telegram.error.BadRequest`.
Expand Down
11 changes: 7 additions & 4 deletionstests/auxil/pytest_classes.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -25,7 +25,7 @@
from tests.auxil.ci_bots import BOT_INFO_PROVIDER
from tests.auxil.constants import PRIVATE_KEY
from tests.auxil.envvars import TEST_WITH_OPT_DEPS
from tests.auxil.networking import NonchalantHttpxRequest
from tests.auxil.networking import NonchalantHttpxRequest, OfflineRequest


def _get_bot_user(token: str) -> User:
Expand DownExpand Up@@ -93,17 +93,20 @@ class PytestUpdater(Updater):
pass


def make_bot(bot_info=None, **kwargs):
def make_bot(bot_info=None,offline: bool = False,**kwargs):
"""
Tests are executed on tg.ext.ExtBot, as that class only extends the functionality of tg.bot
"""
token = kwargs.pop("token", (bot_info or {}).get("token"))
private_key = kwargs.pop("private_key", PRIVATE_KEY)
kwargs.pop("token", None)

request_class = OfflineRequest if offline else NonchalantHttpxRequest

return PytestExtBot(
token=token,
private_key=private_key if TEST_WITH_OPT_DEPS else None,
request=NonchalantHttpxRequest(8),
get_updates_request=NonchalantHttpxRequest(1),
request=request_class(8),
get_updates_request=request_class(1),
**kwargs,
)
53 changes: 11 additions & 42 deletionstests/conftest.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -37,15 +37,14 @@
Update,
User,
)
from telegram.ext import ApplicationBuilder, Defaults, Updater
from telegram.ext.filters import MessageFilter, UpdateFilter
from telegram.ext import Defaults
from tests.auxil.build_messages import DATE
from tests.auxil.ci_bots import BOT_INFO_PROVIDER
from tests.auxil.constants import PRIVATE_KEY
from tests.auxil.envvars import RUN_TEST_OFFICIAL, TEST_WITH_OPT_DEPS
from tests.auxil.files import data_file
from tests.auxil.networking import NonchalantHttpxRequest
from tests.auxil.pytest_classes importPytestApplication,PytestBot, make_bot
from tests.auxil.pytest_classes import PytestBot, make_bot
from tests.auxil.timezones import BasicTimezone

if TEST_WITH_OPT_DEPS:
Expand DownExpand Up@@ -124,6 +123,15 @@ async def bot(bot_info):
yield _bot


@pytest.fixture(scope="session")
async def offline_bot(bot_info):
"""Makes an offline Bot instance with the given bot_info
Note that in tests/ext we also override the `bot` fixture to return the offline bot instead.
"""
async with make_bot(bot_info, offline=True) as _bot:
yield _bot


@pytest.fixture
def one_time_bot(bot_info):
"""A function scoped bot since the session bot would shutdown when `async with app` finishes"""
Expand DownExpand Up@@ -211,28 +219,6 @@ def subscription_channel_id(bot_info):
return bot_info["subscription_channel_id"]


@pytest.fixture
async def app(bot_info):
# We build a new bot each time so that we use `app` in a context manager without problems
application = (
ApplicationBuilder().bot(make_bot(bot_info)).application_class(PytestApplication).build()
)
yield application
if application.running:
await application.stop()
await application.shutdown()


@pytest.fixture
async def updater(bot_info):
# We build a new bot each time so that we use `updater` in a context manager without problems
up = Updater(bot=make_bot(bot_info), update_queue=asyncio.Queue())
yield up
if up.running:
await up.stop()
await up.shutdown()


@pytest.fixture
def thumb_file():
with data_file("thumb.jpg").open("rb") as f:
Expand All@@ -245,23 +231,6 @@ def class_thumb_file():
yield f


@pytest.fixture(
scope="class",
params=[{"class": MessageFilter}, {"class": UpdateFilter}],
ids=["MessageFilter", "UpdateFilter"],
)
def mock_filter(request):
class MockFilter(request.param["class"]):
def __init__(self):
super().__init__()
self.tested = False

def filter(self, _):
self.tested = True

return MockFilter()


def _get_false_update_fixture_decorator_params():
message = Message(1, DATE, Chat(1, ""), from_user=User(1, "", False), text="test")
params = [
Expand Down
102 changes: 102 additions & 0 deletionstests/ext/conftest.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,102 @@
import asyncio

import pytest

from telegram.ext import ApplicationBuilder, Updater
from telegram.ext.filters import MessageFilter, UpdateFilter
from tests.auxil.constants import PRIVATE_KEY
from tests.auxil.envvars import TEST_WITH_OPT_DEPS
from tests.auxil.monkeypatch import return_true
from tests.auxil.networking import OfflineRequest
from tests.auxil.pytest_classes import PytestApplication, PytestBot, make_bot

# This module overrides the bot fixtures defined in the global conftest.py to use the offline bot.
# We don't want the tests on telegram.ext to depend on the network, so we use the offline bot
# instead.


@pytest.fixture(scope="session")
async def bot(bot_info, offline_bot):
return offline_bot


@pytest.fixture
async def app(bot_info, monkeypatch):
# We build a new bot each time so that we use `app` in a context manager without problems
application = (
ApplicationBuilder()
.bot(make_bot(bot_info, offline=True))
.application_class(PytestApplication)
.build()
)
monkeypatch.setattr(application.bot, "delete_webhook", return_true)
monkeypatch.setattr(application.bot, "set_webhook", return_true)
yield application
if application.running:
await application.stop()
await application.shutdown()


@pytest.fixture
async def updater(bot_info, monkeypatch):
# We build a new bot each time so that we use `updater` in a context manager without problems
up = Updater(bot=make_bot(bot_info, offline=True), update_queue=asyncio.Queue())
monkeypatch.setattr(up.bot, "delete_webhook", return_true)
monkeypatch.setattr(up.bot, "set_webhook", return_true)
yield up
if up.running:
await up.stop()
await up.shutdown()


@pytest.fixture
def one_time_bot(bot_info):
"""A function scoped bot since the session bot would shutdown when `async with app` finishes"""
return make_bot(bot_info, offline=True)


@pytest.fixture(scope="session")
async def cdc_bot(bot_info):
"""Makes an ExtBot instance with the given bot_info that uses arbitrary callback_data"""
async with make_bot(bot_info, arbitrary_callback_data=True, offline=True) as _bot:
yield _bot


@pytest.fixture(scope="session")
async def raw_bot(bot_info):
"""Makes an regular Bot instance with the given bot_info"""
async with PytestBot(
bot_info["token"],
private_key=PRIVATE_KEY if TEST_WITH_OPT_DEPS else None,
request=OfflineRequest(),
get_updates_request=OfflineRequest(),
) as _bot:
yield _bot


@pytest.fixture
async def one_time_raw_bot(bot_info):
"""Makes an regular Bot instance with the given bot_info"""
return PytestBot(
bot_info["token"],
private_key=PRIVATE_KEY if TEST_WITH_OPT_DEPS else None,
request=OfflineRequest(),
get_updates_request=OfflineRequest(),
)


@pytest.fixture(
scope="class",
params=[{"class": MessageFilter}, {"class": UpdateFilter}],
ids=["MessageFilter", "UpdateFilter"],
)
def mock_filter(request):
class MockFilter(request.param["class"]):
def __init__(self):
super().__init__()
self.tested = False

def filter(self, _):
self.tested = True

return MockFilter()
Loading
Loading
[8]ページ先頭
©2009-2025 Movatter.jp