PersonalPublic/gpt_chat_bot
1
0
Fork 0
mirror of https://github.com/Balshgit/gpt_chat_bot.git synced 2025-12-16 21:20:39 +03:00
Code Issues Packages Projects Releases Wiki Activity

add gpt model health check (#21)

Browse Source
This commit is contained in:
Dmitry Afanasyev
2023-09-29 20:30:58 +03:00
committed by GitHub
parent 52df4d338f
commit 42f5191042
9 changed files with 228 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

269
bot_microservice/tests/integration/bot/conftest.py
View File

@@ -1,269 +0,0 @@
"""This module contains subclasses of classes from the python-telegram-bot library that
modify behavior of the respective parent classes in order to make them easier to use in the
pytest framework. A common change is to allow monkeypatching of the class members by not
enforcing slots in the subclasses."""
import asyncio
from asyncio import AbstractEventLoop
from contextlib import contextmanager
from datetime import tzinfo
from typing import Any, AsyncGenerator, Iterator
import pytest
import pytest_asyncio
import respx
from httpx import AsyncClient, Response
from pytest_asyncio.plugin import SubRequest
from telegram import Bot, User
from telegram.ext import Application, ApplicationBuilder, Defaults, ExtBot
from constants import CHAT_GPT_BASE_URI
from core.bot import BotApplication
from core.handlers import bot_event_handlers
from main import Application as AppApplication
from settings.config import AppSettings, get_settings
from tests.integration.bot.networking import NonchalantHttpxRequest
from tests.integration.factories.bot import BotInfoFactory, BotUserFactory
@pytest.fixture(scope="session")
def test_settings() -> AppSettings:
return get_settings()
class PytestExtBot(ExtBot): # type: ignore
def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
# Makes it easier to work with the bot in tests
self._unfreeze()
# Here we override get_me for caching because we don't want to call the API repeatedly in tests
async def get_me(self, *args: Any, **kwargs: Any) -> User:
return await _mocked_get_me(self)
class PytestBot(Bot):
def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
# Makes it easier to work with the bot in tests
self._unfreeze()
# Here we override get_me for caching because we don't want to call the API repeatedly in tests
async def get_me(self, *args: Any, **kwargs: Any) -> User:
return await _mocked_get_me(self)
class PytestApplication(Application): # type: ignore
pass
def make_bot(bot_info: dict[str, Any] | None = None, **kwargs: Any) -> PytestExtBot:
"""
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"))
kwargs.pop("token", None)
return PytestExtBot(
token=token,
private_key=None,
request=NonchalantHttpxRequest(connection_pool_size=8),
get_updates_request=NonchalantHttpxRequest(connection_pool_size=1),
**kwargs,
)
async def _mocked_get_me(bot: Bot) -> User:
if bot._bot_user is None:
bot._bot_user = _get_bot_user(bot.token)
return bot._bot_user
def _get_bot_user(token: str) -> User:
"""Used to return a mock user in bot.get_me(). This saves API calls on every init."""
bot_info = BotInfoFactory()
# We don't take token from bot_info, because we need to make a bot with a specific ID. So we
# generate the correct user_id from the token (token from bot_info is random each test run).
# This is important in e.g. bot equality tests. The other parameters like first_name don't
# matter as much. In the future we may provide a way to get all the correct info from the token
user_id = int(token.split(":")[0])
first_name = bot_info.get(
"name",
)
username = bot_info.get(
"username",
).strip("@")
return User(
user_id,
first_name,
is_bot=True,
username=username,
can_join_groups=True,
can_read_all_group_messages=False,
supports_inline_queries=True,
)
# Redefine the event_loop fixture to have a session scope. Otherwise `bot` fixture can't be
# session. See https://github.com/pytest-dev/pytest-asyncio/issues/68 for more details.
@pytest.fixture(scope="session")
def event_loop(request: SubRequest) -> AbstractEventLoop:
"""
Пересоздаем луп для изоляции тестов. В основном нужно для запуска юнит тестов
в связке с интеграционными, т.к. без этого pytest зависает.
Для интеграционных тестов фикстура определяется дополнительная фикстура на всю сессию.
"""
loop = asyncio.get_event_loop_policy().new_event_loop()
asyncio.set_event_loop(loop)
return loop
@pytest.fixture(scope="session")
def bot_info() -> dict[str, Any]:
return BotInfoFactory()
@pytest_asyncio.fixture(scope="session")
async def bot_application(bot_info: dict[str, Any]) -> AsyncGenerator[Any, None]:
# 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()
await application.initialize()
yield application
if application.running:
await application.stop()
await application.shutdown()
@pytest_asyncio.fixture(scope="session")
async def bot(bot_info: dict[str, Any], bot_application: Any) -> AsyncGenerator[PytestExtBot, None]:
"""Makes an ExtBot instance with the given bot_info"""
async with make_bot(bot_info) as _bot:
_bot.application = bot_application
yield _bot
@pytest.fixture()
def one_time_bot(bot_info: dict[str, Any], bot_application: Any) -> PytestExtBot:
"""A function scoped bot since the session bot would shutdown when `async with app` finishes"""
bot = make_bot(bot_info)
bot.application = bot_application
return bot
@pytest_asyncio.fixture(scope="session")
async def cdc_bot(bot_info: dict[str, Any], bot_application: Any) -> AsyncGenerator[PytestExtBot, None]:
"""Makes an ExtBot instance with the given bot_info that uses arbitrary callback_data"""
async with make_bot(bot_info, arbitrary_callback_data=True) as _bot:
_bot.application = bot_application
yield _bot
@pytest_asyncio.fixture(scope="session")
async def raw_bot(bot_info: dict[str, Any], bot_application: Any) -> AsyncGenerator[PytestBot, None]:
"""Makes an regular Bot instance with the given bot_info"""
async with PytestBot(
bot_info["token"],
private_key=None,
request=NonchalantHttpxRequest(8),
get_updates_request=NonchalantHttpxRequest(1),
) as _bot:
_bot.application = bot_application
yield _bot
# Here we store the default bots so that we don't have to create them again and again.
# They are initialized but not shutdown on pytest_sessionfinish because it is causing
# problems with the event loop (Event loop is closed).
_default_bots: dict[Defaults, PytestExtBot] = {}
@pytest_asyncio.fixture(scope="session")
async def default_bot(request: SubRequest, bot_info: dict[str, Any]) -> PytestExtBot:
param = request.param if hasattr(request, "param") else {}
defaults = Defaults(**param)
# If the bot is already created, return it. Else make a new one.
default_bot = _default_bots.get(defaults)
if default_bot is None:
default_bot = make_bot(bot_info, defaults=defaults)
await default_bot.initialize()
_default_bots[defaults] = default_bot # Defaults object is hashable
return default_bot
@pytest_asyncio.fixture(scope="session")
async def tz_bot(timezone: tzinfo, bot_info: dict[str, Any]) -> PytestExtBot:
defaults = Defaults(tzinfo=timezone)
try: # If the bot is already created, return it. Saves time since get_me is not called again.
return _default_bots[defaults]
except KeyError:
default_bot = make_bot(bot_info, defaults=defaults)
await default_bot.initialize()
_default_bots[defaults] = default_bot
return default_bot
@pytest.fixture(scope="session")
def chat_id(bot_info: dict[str, Any]) -> int:
return bot_info["chat_id"]
@pytest.fixture(scope="session")
def super_group_id(bot_info: dict[str, Any]) -> int:
return bot_info["super_group_id"]
@pytest.fixture(scope="session")
def forum_group_id(bot_info: dict[str, Any]) -> int:
return int(bot_info["forum_group_id"])
@pytest.fixture(scope="session")
def channel_id(bot_info: dict[str, Any]) -> int:
return bot_info["channel_id"]
@pytest.fixture(scope="session")
def provider_token(bot_info: dict[str, Any]) -> str:
return bot_info["payment_provider_token"]
@pytest_asyncio.fixture(scope="session")
async def main_application(
bot_application: PytestApplication, test_settings: AppSettings
) -> AsyncGenerator[AppApplication, None]:
bot_app = BotApplication(
settings=test_settings,
handlers=bot_event_handlers.handlers,
)
bot_app.application._initialized = True
bot_app.application.bot = make_bot(BotInfoFactory())
bot_app.application.bot._bot_user = BotUserFactory()
fast_api_app = AppApplication(settings=test_settings, bot_app=bot_app)
yield fast_api_app
@pytest_asyncio.fixture()
async def rest_client(
main_application: AppApplication,
) -> AsyncGenerator[AsyncClient, None]:
"""
Default http client. Use to test unauthorized requests, public endpoints
or special authorization methods.
"""
async with AsyncClient(
app=main_application.fastapi_app,
base_url="http://test",
headers={"Content-Type": "application/json"},
) as client:
yield client
@contextmanager
def mocked_ask_question_api(host: str) -> Iterator[respx.MockRouter]:
with respx.mock(
assert_all_mocked=True,
assert_all_called=True,
base_url=host,
) as respx_mock:
ask_question_route = respx_mock.post(url=CHAT_GPT_BASE_URI, name="ask_question")
ask_question_route.return_value = Response(status_code=200, text="Привет! Как я могу помочь вам сегодня?")
yield respx_mock

68
bot_microservice/tests/integration/bot/test_bot_updates.py
View File

@@ -2,18 +2,18 @@ import asyncio
from asyncio import AbstractEventLoop
from unittest import mock
import httpx
import pytest
import telegram
from assertpy import assert_that
from faker import Faker
from httpx import AsyncClient
from httpx import AsyncClient, Response
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
from constants import BotStagesEnum
from core.bot import BotApplication, BotQueue
from main import Application
from settings.config import AppSettings, settings
from tests.integration.bot.conftest import mocked_ask_question_api
from tests.integration.bot.networking import MockedRequest
from tests.integration.factories.bot import (
BotCallBackQueryFactory,
@@ -21,6 +21,7 @@ from tests.integration.factories.bot import (
BotUpdateFactory,
CallBackFactory,
)
from tests.integration.utils import mocked_ask_question_api
pytestmark = [
pytest.mark.asyncio,
@@ -31,11 +32,6 @@ pytestmark = [
faker = Faker()
async def test_bot_updates(rest_client: AsyncClient) -> None:
response = await rest_client.get("/api/healthcheck")
assert response.status_code == 200
async def test_bot_webhook_endpoint(
rest_client: AsyncClient,
main_application: Application,
@@ -169,8 +165,8 @@ async def test_about_bot_callback_action(
assert mocked_reply_text.call_args.args == (
f"Бот использует бесплатную модель {settings.GPT_MODEL} для ответов на вопросы. "
f"Принимает запросы на разных языках.\n\nБот так же умеет переводить русские голосовые сообщения в текст. "
f"Просто пришлите голосовуху и получите поток сознания в виде текста, но без знаков препинания",
f"\nПринимает запросы на разных языках.\n\nБот так же умеет переводить русские голосовые сообщения "
f"в текст. Просто пришлите голосовуху и получите поток сознания в виде текста, но без знаков препинания",
)
assert mocked_reply_text.call_args.kwargs == {"parse_mode": "Markdown"}
@@ -198,7 +194,10 @@ async def test_ask_question_action(
) -> None:
with mock.patch.object(
telegram._bot.Bot, "send_message", return_value=lambda *args, **kwargs: (args, kwargs)
) as mocked_send_message, mocked_ask_question_api(host=test_settings.GPT_BASE_HOST):
) as mocked_send_message, mocked_ask_question_api(
host=test_settings.GPT_BASE_HOST,
return_value=Response(status_code=httpx.codes.OK, text="Привет! Как я могу помочь вам сегодня?"),
):
bot_update = BotUpdateFactory(message=BotMessageFactory.create_instance(text="Привет!"))
bot_update["message"].pop("entities")
@@ -214,6 +213,55 @@ async def test_ask_question_action(
)
async def test_ask_question_action_not_success(
main_application: Application,
test_settings: AppSettings,
) -> None:
with mock.patch.object(
telegram._bot.Bot, "send_message", return_value=lambda *args, **kwargs: (args, kwargs)
) as mocked_send_message, mocked_ask_question_api(
host=test_settings.GPT_BASE_HOST, return_value=Response(status_code=httpx.codes.INTERNAL_SERVER_ERROR)
):
bot_update = BotUpdateFactory(message=BotMessageFactory.create_instance(text="Привет!"))
bot_update["message"].pop("entities")
await main_application.bot_app.application.process_update(
update=Update.de_json(data=bot_update, bot=main_application.bot_app.bot)
)
assert_that(mocked_send_message.call_args.kwargs).is_equal_to(
{
"text": "Что-то пошло не так, попробуйте еще раз или обратитесь к администратору",
"chat_id": bot_update["message"]["chat"]["id"],
},
include=["text", "chat_id"],
)
async def test_ask_question_action_critical_error(
main_application: Application,
test_settings: AppSettings,
) -> None:
with mock.patch.object(
telegram._bot.Bot, "send_message", return_value=lambda *args, **kwargs: (args, kwargs)
) as mocked_send_message, mocked_ask_question_api(
host=test_settings.GPT_BASE_HOST,
side_effect=Exception(),
):
bot_update = BotUpdateFactory(message=BotMessageFactory.create_instance(text="Привет!"))
bot_update["message"].pop("entities")
await main_application.bot_app.application.process_update(
update=Update.de_json(data=bot_update, bot=main_application.bot_app.bot)
)
assert_that(mocked_send_message.call_args.kwargs).is_equal_to(
{
"text": "Вообще всё сломалось :(",
"chat_id": bot_update["message"]["chat"]["id"],
},
include=["text", "chat_id"],
)
async def test_no_update_message(
main_application: Application,
test_settings: AppSettings,