mirror of
https://github.com/Balshgit/gpt_chat_bot.git
synced 2026-02-03 11:40:39 +03:00
@@ -70,11 +70,11 @@ def check_user_is_banned(func: Any) -> Any:
|
||||
@wraps(func)
|
||||
async def wrapper(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
if not update.effective_message:
|
||||
logger.error('no effective message', update=update, context=context)
|
||||
logger.error("no effective message", update=update, context=context)
|
||||
return
|
||||
|
||||
if not update.effective_user:
|
||||
logger.error('no effective user', update=update, context=context)
|
||||
logger.error("no effective user", update=update, context=context)
|
||||
await update.effective_message.reply_text(
|
||||
"Бот не смог определить пользователя. :(\nОб ошибке уже сообщено."
|
||||
)
|
||||
|
||||
@@ -5,6 +5,7 @@ Revises:
|
||||
Create Date: 2023-10-05 18:28:30.915361
|
||||
|
||||
"""
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
|
||||
@@ -5,6 +5,7 @@ Revises: 0001_create_chatgpt_table
|
||||
Create Date: 2023-11-28 00:58:01.984654
|
||||
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
import sqlalchemy as sa
|
||||
|
||||
@@ -5,6 +5,7 @@ Revises: 0002_create_auth_tables
|
||||
Create Date: 2023-12-28 13:24:42.667724
|
||||
|
||||
"""
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
|
||||
@@ -5,6 +5,7 @@ Revises: 0003_create_user_question_count_table
|
||||
Create Date: 2024-01-08 20:56:34.815976
|
||||
|
||||
"""
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
|
||||
@@ -90,7 +90,7 @@ class AppSettings(SentrySettings, LoggingSettings, BaseSettings):
|
||||
# ==== gpt settings ====
|
||||
GPT_BASE_HOST: str = "http://chathpt_chat_service:8858"
|
||||
|
||||
@model_validator(mode="before") # type: ignore[arg-type]
|
||||
@model_validator(mode="before")
|
||||
def validate_boolean_fields(self) -> Any:
|
||||
values_dict: dict[str, Any] = self # type: ignore[assignment]
|
||||
for value in (
|
||||
|
||||
@@ -381,11 +381,14 @@ async def test_ask_question_action_bot_user_not_exists(
|
||||
message = BotMessageFactory.create_instance(text="Привет!")
|
||||
user = message["from"]
|
||||
|
||||
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.OK, text="Привет! Как я могу помочь вам сегодня?"),
|
||||
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.OK, text="Привет! Как я могу помочь вам сегодня?"),
|
||||
),
|
||||
):
|
||||
bot_update = BotUpdateFactory(message=message)
|
||||
bot_update["message"].pop("entities")
|
||||
@@ -436,11 +439,14 @@ async def test_ask_question_action_bot_user_already_exists(
|
||||
users = dbsession.query(User).all()
|
||||
assert len(users) == 1
|
||||
|
||||
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.OK, text="Привет! Как я могу помочь вам сегодня?"),
|
||||
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.OK, text="Привет! Как я могу помочь вам сегодня?"),
|
||||
),
|
||||
):
|
||||
bot_update = BotUpdateFactory(message=message)
|
||||
bot_update["message"].pop("entities")
|
||||
@@ -496,12 +502,15 @@ async def test_ask_question_action_user_is_banned(
|
||||
ban_reason="test reason",
|
||||
)
|
||||
|
||||
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.OK, text="Привет! Как я могу помочь вам сегодня?"),
|
||||
assert_all_called=False,
|
||||
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.OK, text="Привет! Как я могу помочь вам сегодня?"),
|
||||
assert_all_called=False,
|
||||
),
|
||||
):
|
||||
bot_update = BotUpdateFactory(message=message)
|
||||
bot_update["message"].pop("entities")
|
||||
@@ -531,10 +540,13 @@ async def test_ask_question_action_not_success(
|
||||
test_settings: AppSettings,
|
||||
) -> None:
|
||||
ChatGptModelFactory.create_batch(size=3)
|
||||
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)
|
||||
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")
|
||||
@@ -557,11 +569,14 @@ async def test_ask_question_action_critical_error(
|
||||
test_settings: AppSettings,
|
||||
) -> None:
|
||||
ChatGptModelFactory.create_batch(size=3)
|
||||
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(),
|
||||
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")
|
||||
|
||||
Reference in New Issue
Block a user