add more tests (#79)

* add more tests * add more tests & update poetry.lock
2025-12-16 21:20:39 +03:00 · 2024-01-08 15:38:56 +03:00
parent de55d873f9
commit 7cbe7b7c50
7 changed files with 368 additions and 38 deletions
--- a/bot_microservice/tests/integration/bot/test_bot_api.py
+++ b/bot_microservice/tests/integration/bot/test_bot_api.py
@@ -75,6 +75,38 @@ async def test_change_chatgpt_model_priority(
    assert upd_model2.priority == priority


+@pytest.mark.parametrize(
+    "bot_api_key_header",
+    [
+        pytest.param({"BOT-API-KEY": ""}, id="empty key in header"),
+        pytest.param({"BOT-API-KEY": "Hello World"}, id="incorrect token"),
+        pytest.param({"Hello-World": "lasdkj3-wer-weqwe_34"}, id="correct token but wrong api key"),
+        pytest.param({}, id="no api key header"),
+    ],
+)
+async def test_cant_change_chatgpt_model_priority_with_wrong_api_key(
+    dbsession: Session,
+    rest_client: AsyncClient,
+    faker: Faker,
+    test_settings: AppSettings,
+    bot_api_key_header: dict[str, str],
+) -> None:
+    ChatGptModelFactory(priority=0)
+    model2 = ChatGptModelFactory(priority=1)
+    priority = faker.random_int(min=2, max=7)
+    user = UserFactory(username=test_settings.SUPERUSER)
+    AccessTokenFactory(user_id=user.id, token="lasdkj3-wer-weqwe_34")  # noqa: S106
+    response = await rest_client.put(
+        url=f"/api/chatgpt/models/{model2.id}/priority",
+        json={"priority": priority},
+        headers=bot_api_key_header,
+    )
+    assert response.status_code == 403
+
+    changed_model = dbsession.query(ChatGptModels).filter_by(id=model2.id).one()
+    assert changed_model.priority == model2.priority
+
+
 async def test_reset_chatgpt_models_priority(
    dbsession: Session,
    rest_client: AsyncClient,
@@ -101,6 +133,37 @@ async def test_reset_chatgpt_models_priority(
        assert model.priority == 0


+@pytest.mark.parametrize(
+    "bot_api_key_header",
+    [
+        pytest.param({"BOT-API-KEY": ""}, id="empty key in header"),
+        pytest.param({"BOT-API-KEY": "Hello World"}, id="incorrect token"),
+    ],
+)
+async def test_cant_reset_chatgpt_models_priority_with_wrong_api_key(
+    dbsession: Session, rest_client: AsyncClient, test_settings: AppSettings, bot_api_key_header: dict[str, str]
+) -> None:
+    chat_gpt_models = ChatGptModelFactory.create_batch(size=2)
+    model_with_highest_priority = ChatGptModelFactory(priority=42)
+
+    priorities = sorted([model.priority for model in chat_gpt_models] + [model_with_highest_priority.priority])
+
+    user = UserFactory(username=test_settings.SUPERUSER)
+    AccessTokenFactory(user_id=user.id)
+
+    response = await rest_client.put(
+        url="/api/chatgpt/models/priority/reset",
+        headers=bot_api_key_header,
+    )
+    assert response.status_code == 403
+
+    models = dbsession.query(ChatGptModels).all()
+
+    changed_priorities = sorted([model.priority for model in models])
+
+    assert changed_priorities == priorities
+
+
 async def test_create_new_chatgpt_model(
    dbsession: Session,
    rest_client: AsyncClient,
@@ -142,6 +205,37 @@ async def test_create_new_chatgpt_model(
    }


+async def test_cant_create_new_chatgpt_model_with_wrong_api_key(
+    dbsession: Session,
+    rest_client: AsyncClient,
+    faker: Faker,
+    test_settings: AppSettings,
+) -> None:
+    ChatGptModelFactory.create_batch(size=2)
+    ChatGptModelFactory(priority=42)
+
+    user = UserFactory(username=test_settings.SUPERUSER)
+    AccessTokenFactory(user_id=user.id)
+
+    model_name = "new-gpt-model"
+    model_priority = faker.random_int(min=1, max=5)
+
+    models = dbsession.query(ChatGptModels).all()
+    assert len(models) == 3
+
+    response = await rest_client.post(
+        url="/api/chatgpt/models",
+        json={
+            "model": model_name,
+            "priority": model_priority,
+        },
+    )
+    assert response.status_code == 403
+
+    models = dbsession.query(ChatGptModels).all()
+    assert len(models) == 3
+
+
 async def test_add_existing_chatgpt_model(
    dbsession: Session,
    rest_client: AsyncClient,
@@ -197,3 +291,27 @@ async def test_delete_chatgpt_model(
    assert len(models) == 2

    assert model not in models
+
+
+async def test_cant_delete_chatgpt_model_with_wrong_api_key(
+    dbsession: Session,
+    rest_client: AsyncClient,
+    test_settings: AppSettings,
+) -> None:
+    ChatGptModelFactory.create_batch(size=2)
+    model = ChatGptModelFactory(priority=42)
+
+    user = UserFactory(username=test_settings.SUPERUSER)
+    access_token = AccessTokenFactory(user_id=user.id)
+
+    models = dbsession.query(ChatGptModels).all()
+    assert len(models) == 3
+
+    response = await rest_client.delete(
+        url=f"/api/chatgpt/models/{model.id}",
+        headers={"ROOT-ACCESS": access_token.token},
+    )
+    assert response.status_code == 403
+
+    models = dbsession.query(ChatGptModels).all()
+    assert len(models) == 3
--- a/bot_microservice/tests/integration/bot/test_bot_updates.py
+++ b/bot_microservice/tests/integration/bot/test_bot_updates.py
@@ -11,6 +11,7 @@ from sqlalchemy.orm import Session
 from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update

 from constants import BotStagesEnum
+from core.auth.models.users import User, UserQuestionCount
 from core.bot.app import BotApplication, BotQueue
 from main import Application
 from settings.config import AppSettings
@@ -22,6 +23,7 @@ from tests.integration.factories.bot import (
    CallBackFactory,
    ChatGptModelFactory,
 )
+from tests.integration.factories.user import UserFactory, UserQuestionCountFactory
 from tests.integration.utils import mocked_ask_question_api

 pytestmark = [
@@ -113,6 +115,33 @@ async def test_help_command(
        )


+async def test_help_command_user_is_banned(
+    main_application: Application,
+    test_settings: AppSettings,
+) -> None:
+    message = BotMessageFactory.create_instance(text="/help")
+    user = message["from"]
+    UserFactory(
+        id=user["id"],
+        first_name=user["first_name"],
+        last_name=user["last_name"],
+        username=user["username"],
+        is_active=False,
+        ban_reason="test reason",
+    )
+    with mock.patch.object(
+        telegram._bot.Bot, "send_message", return_value=lambda *args, **kwargs: (args, kwargs)
+    ) as mocked_send_message:
+        bot_update = BotUpdateFactory(message=message)
+
+        await main_application.bot_app.application.process_update(
+            update=Update.de_json(data=bot_update, bot=main_application.bot_app.bot)
+        )
+    assert mocked_send_message.call_args.kwargs["text"] == (
+        "You have banned for reason: *test reason*.\nPlease contact the /developer"
+    )
+
+
 async def test_start_entry(
    main_application: Application,
    test_settings: AppSettings,
@@ -232,6 +261,35 @@ async def test_website_callback_action(
        assert mocked_reply_text.call_args.args == ("Веб версия: http://localhost/chat/",)


+async def test_website_callback_action_user_is_banned(
+    main_application: Application,
+    test_settings: AppSettings,
+) -> None:
+    message = BotMessageFactory.create_instance(text="Список основных команд:")
+    user = message["from"]
+    UserFactory(
+        id=user["id"],
+        first_name=user["first_name"],
+        last_name=user["last_name"],
+        username=user["username"],
+        is_active=False,
+        ban_reason="test reason",
+    )
+    with mock.patch.object(telegram._message.Message, "reply_text") as mocked_reply_text:
+        bot_update = BotCallBackQueryFactory(
+            message=message,
+            callback_query=CallBackFactory(data=BotStagesEnum.website),
+        )
+
+        await main_application.bot_app.application.process_update(
+            update=Update.de_json(data=bot_update, bot=main_application.bot_app.bot)
+        )
+
+        assert mocked_reply_text.call_args.kwargs["text"] == (
+            "You have banned for reason: *test reason*.\nPlease contact the /developer"
+        )
+
+
 async def test_bug_report_action(
    main_application: Application,
    test_settings: AppSettings,
@@ -261,6 +319,35 @@ async def test_bug_report_action(
        )


+async def test_bug_report_action_user_is_banned(
+    main_application: Application,
+    test_settings: AppSettings,
+) -> None:
+    message = BotMessageFactory.create_instance(text="/bug_report")
+    user = message["from"]
+    UserFactory(
+        id=user["id"],
+        first_name=user["first_name"],
+        last_name=user["last_name"],
+        username=user["username"],
+        is_active=False,
+        ban_reason="test reason",
+    )
+    with (
+        mock.patch.object(telegram._message.Message, "reply_text") as mocked_reply_text,
+        mock.patch.object(telegram._bot.Bot, "send_message", return_value=lambda *args, **kwargs: (args, kwargs)),
+    ):
+        bot_update = BotUpdateFactory(message=message)
+
+        await main_application.bot_app.application.process_update(
+            update=Update.de_json(data=bot_update, bot=main_application.bot_app.bot)
+        )
+
+        assert mocked_reply_text.call_args.kwargs["text"] == (
+            "You have banned for reason: *test reason*.\nPlease contact the /developer"
+        )
+
+
 async def test_get_developer_action(
    main_application: Application,
    test_settings: AppSettings,
@@ -278,19 +365,28 @@ async def test_get_developer_action(
        assert mocked_reply_text.call_args.args == ("Автор бота: *Дмитрий Афанасьев*\n\nTg nickname: *Balshtg*",)


-async def test_ask_question_action(
+async def test_ask_question_action_bot_user_not_exists(
    dbsession: Session,
    main_application: Application,
    test_settings: AppSettings,
 ) -> None:
    ChatGptModelFactory.create_batch(size=3)
+    users = dbsession.query(User).all()
+    users_question_count = dbsession.query(UserQuestionCount).all()
+
+    assert len(users) == 0
+    assert len(users_question_count) == 0
+
+    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="Привет! Как я могу помочь вам сегодня?"),
    ):
-        bot_update = BotUpdateFactory(message=BotMessageFactory.create_instance(text="Привет!"))
+        bot_update = BotUpdateFactory(message=message)
        bot_update["message"].pop("entities")

        await main_application.bot_app.application.process_update(
@@ -313,6 +409,114 @@ async def test_ask_question_action(
            include=["text", "chat_id"],
        )

+    created_user = dbsession.query(User).filter_by(id=user["id"]).one()
+    assert created_user.username == user["username"]
+
+    created_user_question_count = dbsession.query(UserQuestionCount).filter_by(user_id=user["id"]).one()
+    assert created_user_question_count.question_count == 1
+
+
+async def test_ask_question_action_bot_user_already_exists(
+    dbsession: Session,
+    main_application: Application,
+    test_settings: AppSettings,
+) -> None:
+    ChatGptModelFactory.create_batch(size=3)
+    message = BotMessageFactory.create_instance(text="Привет!")
+    user = message["from"]
+    existing_user = UserFactory(
+        id=user["id"], first_name=user["first_name"], last_name=user["last_name"], username=user["username"]
+    )
+    existing_user_question_count = UserQuestionCountFactory(user_id=existing_user.id).question_count
+
+    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="Привет! Как я могу помочь вам сегодня?"),
+    ):
+        bot_update = BotUpdateFactory(message=message)
+        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_list[0].kwargs).is_equal_to(
+            {
+                "text": (
+                    "Ответ в среднем занимает 10-15 секунд.\n- Список команд: /help\n- Сообщить об ошибке: /bug_report"
+                ),
+                "chat_id": bot_update["message"]["chat"]["id"],
+            },
+            include=["text", "chat_id"],
+        )
+        assert_that(mocked_send_message.call_args_list[1].kwargs).is_equal_to(
+            {
+                "text": "Привет! Как я могу помочь вам сегодня?",
+                "chat_id": bot_update["message"]["chat"]["id"],
+            },
+            include=["text", "chat_id"],
+        )
+
+    users = dbsession.query(User).all()
+    assert len(users) == 1
+
+    updated_user_question_count = dbsession.query(UserQuestionCount).filter_by(user_id=user["id"]).one()
+    assert updated_user_question_count.question_count == existing_user_question_count + 1
+
+
+async def test_ask_question_action_user_is_banned(
+    dbsession: Session,
+    main_application: Application,
+    test_settings: AppSettings,
+) -> None:
+    ChatGptModelFactory.create_batch(size=3)
+
+    users_question_count = dbsession.query(UserQuestionCount).all()
+    assert len(users_question_count) == 0
+
+    message = BotMessageFactory.create_instance(text="Привет!")
+    user = message["from"]
+    UserFactory(
+        id=user["id"],
+        first_name=user["first_name"],
+        last_name=user["last_name"],
+        username=user["username"],
+        is_active=False,
+        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,
+    ):
+        bot_update = BotUpdateFactory(message=message)
+        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_list[0].kwargs).is_equal_to(
+            {
+                "text": ("You have banned for reason: *test reason*.\nPlease contact the /developer"),
+                "chat_id": bot_update["message"]["chat"]["id"],
+            },
+            include=["text", "chat_id"],
+        )
+
+    created_user = dbsession.query(User).filter_by(id=user["id"]).one()
+    assert created_user.username == user["username"]
+    assert created_user.is_active is False
+
+    created_user_question_count = dbsession.query(UserQuestionCount).filter_by(user_id=user["id"]).scalar()
+    assert created_user_question_count is None
+

 async def test_ask_question_action_not_success(
    dbsession: Session,