From 6e82ac7902ef16bdff9e7bea8d473671b84bbdf7 Mon Sep 17 00:00:00 2001 From: Lev Chechulin Date: Thu, 19 Jun 2025 15:28:44 +0200 Subject: [PATCH 1/8] feat: interpret a number as a modification request --- src/tg/handlers/user_message_handler.py | 236 ++++++++++++------------ 1 file changed, 122 insertions(+), 114 deletions(-) diff --git a/src/tg/handlers/user_message_handler.py b/src/tg/handlers/user_message_handler.py index 129ce808..91f1b158 100644 --- a/src/tg/handlers/user_message_handler.py +++ b/src/tg/handlers/user_message_handler.py @@ -7,10 +7,11 @@ from ... import consts from ...consts import ButtonValues, PlainTextUserAction from ...db.db_client import DBClient +from ...db.db_objects import Reminder +from ...focalboard.focalboard_client import FocalboardClient from ...strings import load from ...tg.handlers import get_tasks_report_handler from ...trello.trello_client import TrelloClient -from ...focalboard.focalboard_client import FocalboardClient from .utils import get_chat_id, get_chat_name, get_sender_id, reply logger = logging.getLogger(__name__) @@ -25,6 +26,82 @@ def handle_callback_query( update.callback_query.answer() handle_user_message(update, tg_context, ButtonValues(update.callback_query.data)) +# helper to avoid code duplication +def _show_reminder_edit_options( + reminder: Reminder, update: telegram.Update, command_data: dict +): + """ + Shows the menu with options to edit a specific reminder. + """ + # keyboard for edit + button_list = [ + [ + telegram.InlineKeyboardButton( + load("manage_reminders_handler__edit_text_btn"), + callback_data=ButtonValues.MANAGE_REMINDERS__EDIT__TEXT.value, + ) + ], + [ + telegram.InlineKeyboardButton( + load("manage_reminders_handler__edit_datetime_btn"), + callback_data=ButtonValues.MANAGE_REMINDERS__EDIT__DATETIME.value, + ) + ], + [ + telegram.InlineKeyboardButton( + load("manage_reminders_handler__edit_title_btn"), + callback_data=ButtonValues.MANAGE_REMINDERS__EDIT__TITLE.value, + ) + ], + [ + telegram.InlineKeyboardButton( + load("manage_reminders_handler__edit_chat_btn"), + callback_data=ButtonValues.MANAGE_REMINDERS__EDIT__CHAT.value, + ) + ], + [ + ( + telegram.InlineKeyboardButton( + load("manage_reminders_handler__edit_suspend_btn"), + callback_data=ButtonValues.MANAGE_REMINDERS__EDIT__SUSPEND.value, + ) + if reminder.is_active + else telegram.InlineKeyboardButton( + load("manage_reminders_handler__edit_resume_btn"), + callback_data=ButtonValues.MANAGE_REMINDERS__EDIT__RESUME.value, + ) + ) + ], + [ + ( + telegram.InlineKeyboardButton( + load("manage_reminders_handler__edit_poll_active_btn"), + callback_data=ButtonValues.MANAGE_REMINDERS__DISABLE_POLL.value, + ) + if reminder.send_poll + else telegram.InlineKeyboardButton( + load("manage_reminders_handler__edit_poll_inactive_btn"), + callback_data=ButtonValues.MANAGE_REMINDERS__ENABLE_POLL.value, + ) + ) + ], + ] + reply_markup = telegram.InlineKeyboardMarkup(button_list) + weekday_str = calendar.TextCalendar().formatweekday(int(reminder.weekday), 15).strip() + reply( + load( + "manage_reminders_handler__weekly_reminder", + weekday=weekday_str, + time=reminder.time, + text=reminder.text, + ), + update, + reply_markup=reply_markup, + ) + set_next_action( + command_data, PlainTextUserAction.MANAGE_REMINDERS__CHOOSE_EDIT_ACTION + ) + def handle_user_message( update: telegram.Update, @@ -210,9 +287,30 @@ def handle_user_message( add_labels = button == ButtonValues.GET_TASKS_REPORT__LABELS__YES handle_task_report(command_data, add_labels, update) elif next_action == PlainTextUserAction.MANAGE_REMINDERS__CHOOSE_ACTION: + # If user sends a number, interpret it as a shortcut to edit that reminder + if user_input and user_input.isdigit(): + reminder_ids = command_data.get( + consts.ManageRemindersData.EXISTING_REMINDERS, [] + ) + try: + assert 0 < int(user_input) <= len(reminder_ids) + reminder_id, _, _ = reminder_ids[int(user_input) - 1] + except Exception: + reply(load("manage_reminders_handler__reminder_number_bad"), update) + return + + command_data[ + consts.ManageRemindersData.ACTION_TYPE + ] = ButtonValues.MANAGE_REMINDERS__ACTIONS__EDIT + command_data[consts.ManageRemindersData.CHOSEN_REMINDER_ID] = reminder_id + reminder = DBClient().get_reminder_by_id(reminder_id) + _show_reminder_edit_options(reminder, update, command_data) + return + if button is None: reply(load("user_message_handler__press_button_please"), update) return + if button == ButtonValues.MANAGE_REMINDERS__ACTIONS__NEW: reply(load("manager_reminders_handler__enter_chat_id"), update) set_next_action( @@ -277,76 +375,7 @@ def handle_user_message( ) elif action == ButtonValues.MANAGE_REMINDERS__ACTIONS__EDIT: reminder = DBClient().get_reminder_by_id(reminder_id) - # keyboard for edit - button_list = [ - [ - telegram.InlineKeyboardButton( - load("manage_reminders_handler__edit_text_btn"), - callback_data=ButtonValues.MANAGE_REMINDERS__EDIT__TEXT.value, - ) - ], - [ - telegram.InlineKeyboardButton( - load("manage_reminders_handler__edit_datetime_btn"), - callback_data=ButtonValues.MANAGE_REMINDERS__EDIT__DATETIME.value, - ) - ], - [ - telegram.InlineKeyboardButton( - load("manage_reminders_handler__edit_title_btn"), - callback_data=ButtonValues.MANAGE_REMINDERS__EDIT__TITLE.value, - ) - ], - [ - telegram.InlineKeyboardButton( - load("manage_reminders_handler__edit_chat_btn"), - callback_data=ButtonValues.MANAGE_REMINDERS__EDIT__CHAT.value, - ) - ], - [ - ( - telegram.InlineKeyboardButton( - load("manage_reminders_handler__edit_suspend_btn"), - callback_data=ButtonValues.MANAGE_REMINDERS__EDIT__SUSPEND.value, - ) - if reminder.is_active - else telegram.InlineKeyboardButton( - load("manage_reminders_handler__edit_resume_btn"), - callback_data=ButtonValues.MANAGE_REMINDERS__EDIT__RESUME.value, - ) - ) - ], - [ - ( - telegram.InlineKeyboardButton( - load("manage_reminders_handler__edit_poll_active_btn"), - callback_data=ButtonValues.MANAGE_REMINDERS__DISABLE_POLL.value, - ) - if reminder.send_poll - else telegram.InlineKeyboardButton( - load("manage_reminders_handler__edit_poll_inactive_btn"), - callback_data=ButtonValues.MANAGE_REMINDERS__ENABLE_POLL.value, - ) - ) - ], - ] - reply_markup = telegram.InlineKeyboardMarkup(button_list) - weekday_str = ( - calendar.TextCalendar().formatweekday(int(reminder.weekday), 15).strip() - ) - reply( - load( - "manage_reminders_handler__weekly_reminder", - weekday=weekday_str, - time=reminder.time, - text=reminder.text, - ), - update, - reply_markup=reply_markup, - ) - set_next_action( - command_data, PlainTextUserAction.MANAGE_REMINDERS__CHOOSE_EDIT_ACTION - ) + _show_reminder_edit_options(reminder, update, command_data) else: logger.error(f'Bad reminder action "{action}"') elif next_action == PlainTextUserAction.MANAGE_REMINDERS__CHOOSE_EDIT_ACTION: @@ -581,6 +610,8 @@ def handle_user_message( return command_data[consts.ManageRemindersData.TIME] = user_input action = command_data[consts.ManageRemindersData.ACTION_TYPE] + weekday_num = command_data[consts.ManageRemindersData.WEEKDAY_NUM] + time = command_data[consts.ManageRemindersData.TIME] if action == ButtonValues.MANAGE_REMINDERS__ACTIONS__EDIT: reminder_id = int( @@ -602,54 +633,31 @@ def handle_user_message( update, reply_markup=telegram.InlineKeyboardMarkup([buttons]), ) - set_next_action(command_data, PlainTextUserAction.MANAGE_REMINDERS__TOGGLE_POLL) - return - elif next_action == PlainTextUserAction.MANAGE_REMINDERS__TOGGLE_POLL: - if button == ButtonValues.MANAGE_REMINDERS__POLL__YES: - poll_options = { - "question": load("manage_reminders_handler__poll_question"), - "options": [ - load("manage_reminders_handler__poll_option_yes_btn"), - load("manage_reminders_handler__poll_option_no_btn"), - ], - "is_anonymous": False, - } - button_yes = telegram.InlineKeyboardButton( - load("manage_reminders_handler__toggle_poll_yes_btn"), - callback_data=consts.ButtonValues.MANAGE_REMINDERS__TOGGLE_POLL__YES.value, - ) - button_no = telegram.InlineKeyboardButton( - load("manage_reminders_handler__toggle_poll_no_btn"), - callback_data=consts.ButtonValues.MANAGE_REMINDERS__TOGGLE_POLL__NO.value, - ) - buttons = [button_yes, button_no] - reply("", update, poll_options=poll_options) - reply( - "Добавить?", - update, - reply_markup=telegram.InlineKeyboardMarkup([buttons]), - ) - set_next_action( - command_data, consts.PlainTextUserAction.MANAGE_REMINDERS__SUCCESS - ) + set_next_action(command_data, PlainTextUserAction.MANAGE_REMINDERS__SUCCESS) return elif next_action == PlainTextUserAction.MANAGE_REMINDERS__SUCCESS: - text = command_data[consts.ManageRemindersData.REMINDER_TEXT] - group_chat_id = command_data[consts.ManageRemindersData.GROUP_CHAT_ID] - name = command_data[consts.ManageRemindersData.REMINDER_NAME] + text = command_data.get(consts.ManageRemindersData.REMINDER_TEXT) + group_chat_id = command_data.get(consts.ManageRemindersData.GROUP_CHAT_ID) + name = command_data.get(consts.ManageRemindersData.REMINDER_NAME) weekday_num = command_data[consts.ManageRemindersData.WEEKDAY_NUM] weekday_name = command_data[consts.ManageRemindersData.WEEKDAY_NAME] time = command_data[consts.ManageRemindersData.TIME] if button == consts.ButtonValues.MANAGE_REMINDERS__TOGGLE_POLL__YES: - DBClient().add_reminder( - creator_chat_id=get_sender_id(update), - group_chat_id=group_chat_id, - name=name, - text=text, - weekday_num=weekday_num, - time=time, - send_poll=True, - ) + if text is None: + reminder_id = int( + command_data[consts.ManageRemindersData.CHOSEN_REMINDER_ID] + ) + DBClient().update_reminder(reminder_id, weekday=weekday_num, time=time) + else: + DBClient().add_reminder( + creator_chat_id=get_sender_id(update), + group_chat_id=group_chat_id, + name=name, + text=text, + weekday_num=weekday_num, + time=time, + send_poll=True, + ) else: DBClient().add_reminder( creator_chat_id=get_sender_id(update), @@ -700,4 +708,4 @@ def handle_task_report(command_data, add_labels, update): reply(message, update) # finished with last action for /trello_client_get_lists set_next_action(command_data, None) - return + return \ No newline at end of file From 575d992b8ac9e20be2a0f03220f3f9d6370b7a48 Mon Sep 17 00:00:00 2001 From: houndlord <45179481+houndlord@users.noreply.github.com> Date: Thu, 19 Jun 2025 19:25:58 +0200 Subject: [PATCH 2/8] Update user_message_handler.py --- src/tg/handlers/user_message_handler.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/tg/handlers/user_message_handler.py b/src/tg/handlers/user_message_handler.py index 35c7c6fa..33ba8a85 100644 --- a/src/tg/handlers/user_message_handler.py +++ b/src/tg/handlers/user_message_handler.py @@ -24,8 +24,6 @@ logger = logging.getLogger(__name__) -logger = logging.getLogger(__name__) - SECTIONS = [ ("Идеи для статей", TrelloListAlias.TOPIC_SUGGESTION), ("Готовая тема", TrelloListAlias.TOPIC_READY), @@ -873,4 +871,4 @@ def handle_task_report(command_data, add_labels, update): reply(message, update) # finished with last action for /trello_client_get_lists set_next_action(command_data, None) - return \ No newline at end of file + return From 508766c2331959e7b22e63a094310bbe0b746b5d Mon Sep 17 00:00:00 2001 From: Lev Chechulin Date: Thu, 19 Jun 2025 19:34:38 +0200 Subject: [PATCH 3/8] fix: import Reminder for type hints --- src/tg/handlers/user_message_handler.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/tg/handlers/user_message_handler.py b/src/tg/handlers/user_message_handler.py index 33ba8a85..5e15199e 100644 --- a/src/tg/handlers/user_message_handler.py +++ b/src/tg/handlers/user_message_handler.py @@ -15,6 +15,7 @@ TrelloListAlias, ) from ...db.db_client import DBClient +from ...db.db_objects import Reminder from ...focalboard.focalboard_client import FocalboardClient from ...strings import load from ...tg.handlers import get_tasks_report_handler From a09fa7a4ed1ea8e2a7428fe5f72fa6244beceea4 Mon Sep 17 00:00:00 2001 From: Lev Chechulin Date: Sat, 12 Jul 2025 22:39:07 +0200 Subject: [PATCH 4/8] feat: new integration testing strategy --- src/config_manager.py | 4 +- tests/integration/conftest.py | 127 ++++++++++----------- tests/integration/test_telegram_bot.py | 148 +++++++++++++++---------- 3 files changed, 154 insertions(+), 125 deletions(-) diff --git a/src/config_manager.py b/src/config_manager.py index edd1552e..f6535f49 100644 --- a/src/config_manager.py +++ b/src/config_manager.py @@ -151,10 +151,10 @@ def _load_config(self, config_path: str) -> dict: try: return json.loads(fin.read()) except json.JSONDecodeError as e: - logger.error(exc_info=e) + logger.exception(f"Failed to decode JSON from '{config_path}'") except IOError: logger.warning(f"Config file at {config_path} not found") def _write_config_override(self, config_override: dict): with open(self.config_override_path, "w") as fout: - fout.write(json.dumps(config_override, indent=4)) + fout.write(json.dumps(config_override, indent=4)) \ No newline at end of file diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index c9320abb..f950c5f1 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -1,19 +1,21 @@ import asyncio import json import os +import logging import pytest from pytest_report import PytestReport, PytestTestStatus from telethon import TelegramClient from telethon.sessions import StringSession -from src.utils.singleton import Singleton +from src.config_manager import ConfigManager -if os.path.exists("config_override_integration_tests.json"): - with open("config_override_integration_tests.json") as config_override: - config = json.load(config_override)["telegram"] -else: - config = json.loads(os.environ["CONFIG_OVERRIDE"])["telegram"] +logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') +logger = logging.getLogger(__name__) + +config_manager = ConfigManager("./config.json", "./config_override_integration_tests.json") +config_manager.load_config_with_override() +config = config_manager.get_telegram_config() api_id = int(config["api_id"]) api_hash = config["api_hash"] @@ -21,66 +23,67 @@ telegram_chat_id = int(config["error_logs_recipients"][0]) telegram_bot_name = config.get("handle", "") - -class WrappedTelegramClientAsync(Singleton): - def __init__(self): - self.client = TelegramClient( - StringSession(api_session), api_id, api_hash, sequential_updates=True - ) - - async def __aenter__(self): - await self.client.connect() - await self.client.get_me() - return self.client - - async def __aexit__(self, exc_t, exc_v, exc_tb): - await self.client.disconnect() - await self.client.disconnected - - -@pytest.fixture(scope="session") -async def telegram_client() -> TelegramClient: - async with WrappedTelegramClientAsync() as client: - yield client - - -@pytest.fixture(scope="session") -async def conversation(telegram_client): - async with telegram_client.conversation(telegram_bot_name) as conv: - yield conv - +@pytest.fixture(scope="function") +async def conversation(): + """ + Provides a completely fresh Telegram client and conversation for each test function. + """ + client = TelegramClient(StringSession(api_session), api_id, api_hash, sequential_updates=True) + await client.connect() + try: + bot_entity = await client.get_entity(telegram_bot_name) + async with client.conversation(bot_entity, timeout=10) as conv: + yield conv + finally: + await client.disconnect() def pytest_sessionfinish(session, exitstatus): - passed = exitstatus == pytest.ExitCode.OK - print("\nrun status code:", exitstatus) + passed = exitstatus == 0 + logger.info(f"Pytest session finished with status code: {exitstatus}") PytestReport().mark_finish() - asyncio.run(report_test_result(passed)) + try: + asyncio.run(report_test_result(passed)) + except Exception: + logger.error("FATAL: Could not send test report to Telegram.", exc_info=True) async def report_test_result(passed: bool): - async with WrappedTelegramClientAsync() as client: - async with client.conversation(telegram_chat_id, timeout=30) as conv: - telegram_bot_mention = ( - f"@{telegram_bot_name}" if telegram_bot_name else "Бот" + """ + Sends the test report using a new, completely isolated client. + """ + report_client = TelegramClient(StringSession(api_session), api_id, api_hash) + try: + await report_client.connect() + report_chat_entity = await report_client.get_entity(telegram_chat_id) + + telegram_bot_mention = f"@{telegram_bot_name}" + + report_path = "./integration_test_report.txt" + with open(report_path, "w", encoding="utf-8") as f: + json.dump(PytestReport().data, f, indent=4, ensure_ascii=False) + + if passed: + caption = f"{telegram_bot_mention} протестирован. Все тесты пройдены успешно." + await report_client.send_file(report_chat_entity, report_path, caption=caption) + else: + caption = f"{telegram_bot_mention} разломан. Подробности в файле и в сообщении ниже." + + failure_details = "\n".join( + ["Сломались тесты:"] + + [ + f'\n--- FAIL: {test["cmd"]}\n' + f'-> {test["exception_class"]}\n' + f'-> {test["exception_message"]}' + for test in PytestReport().data.get("tests", []) + if test.get("status") == PytestTestStatus.FAILED + ] ) - if passed: - message = f"{telegram_bot_mention} протестирован." - else: - message = "\n".join( - [f"{telegram_bot_mention} разломан.", "Сломались команды:"] - + [ - f"{test['cmd']}{telegram_bot_mention}\n" - f"{test['exception_class']}\n{test['exception_message']}" - for test in PytestReport().data["tests"] - if test["status"] == PytestTestStatus.FAILED - ] - ) - with open("./integration_test_report.txt", "w") as integration_test_report: - json.dump( - PytestReport().data, - integration_test_report, - indent=4, - sort_keys=True, - ensure_ascii=False, - ) - await conv.send_file("./integration_test_report.txt", caption=message) + if not failure_details.strip() or len(PytestReport().data.get("tests", [])) == 0: + failure_details = "Детали в логах. Вероятно, тесты не были собраны, или ошибка произошла в фикстуре." + + await report_client.send_file(report_chat_entity, report_path, caption=caption) + if failure_details: + await report_client.send_message(report_chat_entity, failure_details) + finally: + if report_client.is_connected(): + await report_client.disconnect() \ No newline at end of file diff --git a/tests/integration/test_telegram_bot.py b/tests/integration/test_telegram_bot.py index 35627391..3daf98d3 100644 --- a/tests/integration/test_telegram_bot.py +++ b/tests/integration/test_telegram_bot.py @@ -1,85 +1,111 @@ import asyncio +import nest_asyncio import time +from typing import List +import re -import nest_asyncio import pytest -from pytest_report import PytestReport, PytestTestStatus from telethon.tl.custom.message import Message +from telethon.errors import TimeoutError +from telethon.tl.custom.conversation import Conversation + +from pytest_report import PytestReport, PytestTestStatus +from src.config_manager import ConfigManager +from src.sheets.sheets_client import GoogleSheetsClient +from src.strings import StringsDBClient, load + +def strip_html(text: str) -> str: + return re.sub('<[^<]+?>', '', text) + +def setup_strings_for_test_run(): + + ConfigManager.drop_instance() + StringsDBClient.drop_instance() + GoogleSheetsClient.drop_instance() + config_manager = ConfigManager("./config.json", "./config_override_integration_tests.json") + config_manager.load_config_with_override() + + strings_db_config = config_manager.get_strings_db_config() + sheets_config = config_manager.get_sheets_config() -async def _test_command(report_state, conversation, command: str, timeout=120): - test_report = {"cmd": command} + if not (strings_db_config and sheets_config): + pytest.skip("Skipping test: strings_db_config or sheets_config is missing.") + + strings_db_client = StringsDBClient(strings_db_config) + sheets_client = GoogleSheetsClient(sheets_config) + strings_db_client.fetch_strings_sheet(sheets_client) + +async def _test_conversation_step(conversation: Conversation, message_to_send: str, expected_response_id: str, timeout: int): + await conversation.send_message(message_to_send) + resp = await conversation.get_response(timeout=timeout) + + expected_html_response = load(expected_response_id) + expected_plain_text_response = strip_html(expected_html_response) + + actual_response = resp.raw_text.strip() + + assert expected_plain_text_response in actual_response, \ + f"Expected response containing '{expected_plain_text_response}' but got '{actual_response}'" + return resp + +async def _test_command_flow(report_state: PytestReport, conversation: Conversation, command_flow: list, timeout=120): + setup_strings_for_test_run() + + command_str = " -> ".join([item[0] for item in command_flow]) + test_report = {"cmd": command_str} start_time = time.time() + try: - await conversation.send_message(command) - resp: Message = await conversation.get_response(timeout=timeout) - await asyncio.sleep(1) - test_report["response"] = "\\n".join(resp.raw_text.splitlines()) - assert resp.raw_text + await conversation.send_message("/clean_chat_data") + await conversation.get_response() + + for message_to_send, expected_response_id in command_flow: + await _test_conversation_step(conversation, message_to_send, expected_response_id, timeout) + test_report["status"] = PytestTestStatus.OK except BaseException as e: test_report["status"] = PytestTestStatus.FAILED test_report["exception_class"] = str(e.__class__) test_report["exception_message"] = str(e) - raise finally: test_report["time_elapsed"] = time.time() - start_time report_state.data["tests"].append(test_report) -class Test: +class TestTelegramBot: report_state = PytestReport() - loop = asyncio.get_event_loop() - nest_asyncio.apply(loop) - - @pytest.mark.parametrize("command", ("/mute_errors",)) - def test_mute(self, conversation, command: str): - Test.loop.run_until_complete( - _test_command(Test.report_state, conversation, command) - ) - - @pytest.mark.parametrize( - "command", - ( - "/start", - "/help", - ), - ) - def test_start_help(self, conversation, command: str): - Test.loop.run_until_complete( - _test_command(Test.report_state, conversation, command) - ) - - @pytest.mark.parametrize( - "command", - ( - "/get_sheets_report", - "/get_tasks_report_focalboard", - ), - ) - def test_not_failing_reports(self, conversation, command: str): - Test.loop.run_until_complete( - _test_command(Test.report_state, conversation, command) - ) - + @pytest.mark.parametrize( - "command", - ("/get_tg_analytics_report",), + "command_flow", + [ + ([("/start", "start_handler__message")]), + ([("/help", "help__commands_list")]), + ( + [ + ("/manage_reminders", "manage_reminders_handler__no_reminders"), + ("1", "manage_reminders_handler__reminder_number_bad"), + ] + ), + ( + [ + ("/manage_all_reminders", "manage_reminders_handler__no_reminders"), + ] + ), + ], ) - def test_not_failing_analytics(self, conversation, command: str): - Test.loop.run_until_complete( - _test_command(Test.report_state, conversation, command) - ) + @pytest.mark.asyncio + async def test_command_flows(self, conversation: Conversation, command_flow: List[tuple]): + await _test_command_flow(self.report_state, conversation, command_flow) - @pytest.mark.parametrize( - "command", - ( - "/manage_reminders", - "/manage_all_reminders", - ), - ) - def test_reminder(self, conversation, command: str): - Test.loop.run_until_complete( - _test_command(Test.report_state, conversation, command) - ) + @pytest.mark.xfail + @pytest.mark.parametrize("command", ("/bad_cmd",)) + @pytest.mark.asyncio + async def test_failing_command(self, conversation: Conversation, command: str): + try: + await _test_conversation_step(conversation, command, "should_not_matter", timeout=10) + except TimeoutError: + pass + except Exception as e: + pytest.fail(f"Test for failing command failed with unexpected exception: {e}") \ No newline at end of file From 165bc60e60a72c082bb9118d709f9078c2745567 Mon Sep 17 00:00:00 2001 From: Lev Chechulin Date: Sat, 12 Jul 2025 22:42:43 +0200 Subject: [PATCH 5/8] fix formatting in config manager --- src/config_manager.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/config_manager.py b/src/config_manager.py index f6535f49..4d62a065 100644 --- a/src/config_manager.py +++ b/src/config_manager.py @@ -157,4 +157,4 @@ def _load_config(self, config_path: str) -> dict: def _write_config_override(self, config_override: dict): with open(self.config_override_path, "w") as fout: - fout.write(json.dumps(config_override, indent=4)) \ No newline at end of file + fout.write(json.dumps(config_override, indent=4)) From 074b68c18aa048a1b36c4e354a65d67949c07cf9 Mon Sep 17 00:00:00 2001 From: Lev Chechulin Date: Thu, 31 Jul 2025 18:35:41 +0200 Subject: [PATCH 6/8] feat: new tests definitions scheme --- tests/integration/test_telegram_bot.py | 87 ++++++++++++++++++-------- 1 file changed, 61 insertions(+), 26 deletions(-) diff --git a/tests/integration/test_telegram_bot.py b/tests/integration/test_telegram_bot.py index 3daf98d3..050752e5 100644 --- a/tests/integration/test_telegram_bot.py +++ b/tests/integration/test_telegram_bot.py @@ -1,13 +1,15 @@ import asyncio import nest_asyncio import time -from typing import List +from typing import List, Dict import re +import os import pytest from telethon.tl.custom.message import Message from telethon.errors import TimeoutError from telethon.tl.custom.conversation import Conversation +from telethon import events from pytest_report import PytestReport, PytestTestStatus from src.config_manager import ConfigManager @@ -15,10 +17,11 @@ from src.strings import StringsDBClient, load def strip_html(text: str) -> str: + """A simple helper to remove HTML tags for plain text comparison.""" return re.sub('<[^<]+?>', '', text) def setup_strings_for_test_run(): - + """Initializes and populates the string DB right before a test.""" ConfigManager.drop_instance() StringsDBClient.drop_instance() GoogleSheetsClient.drop_instance() @@ -36,32 +39,63 @@ def setup_strings_for_test_run(): sheets_client = GoogleSheetsClient(sheets_config) strings_db_client.fetch_strings_sheet(sheets_client) -async def _test_conversation_step(conversation: Conversation, message_to_send: str, expected_response_id: str, timeout: int): - await conversation.send_message(message_to_send) - resp = await conversation.get_response(timeout=timeout) - - expected_html_response = load(expected_response_id) - expected_plain_text_response = strip_html(expected_html_response) - - actual_response = resp.raw_text.strip() - - assert expected_plain_text_response in actual_response, \ - f"Expected response containing '{expected_plain_text_response}' but got '{actual_response}'" - return resp - -async def _test_command_flow(report_state: PytestReport, conversation: Conversation, command_flow: list, timeout=120): +async def _test_command_flow(report_state: PytestReport, conversation: Conversation, command_flow: List[Dict], timeout=120): + """ + Tests a sequence of user actions using the dictionary-based schema. + """ setup_strings_for_test_run() - command_str = " -> ".join([item[0] for item in command_flow]) + command_str = " -> ".join([f"{step['type']}: '{step['input']}'" for step in command_flow]) test_report = {"cmd": command_str} start_time = time.time() + last_bot_message: Message = None + try: await conversation.send_message("/clean_chat_data") await conversation.get_response() - for message_to_send, expected_response_id in command_flow: - await _test_conversation_step(conversation, message_to_send, expected_response_id, timeout) + for step in command_flow: + action_type = step['type'] + action_input = step['input'] + expected_response_id = step['expected'] + + if action_type == 'message': + await conversation.send_message(action_input) + last_bot_message = await conversation.get_response(timeout=timeout) + + elif action_type == 'click': + if not last_bot_message or not last_bot_message.buttons: + pytest.fail(f"Action failed: Tried to click '{action_input}', but the last bot message had no buttons.") + + new_message_task = asyncio.create_task( + conversation.wait_event(events.NewMessage(incoming=True), timeout=timeout) + ) + edited_message_task = asyncio.create_task( + conversation.wait_event( + events.MessageEdited(incoming=True, func=lambda e: e.message.id == last_bot_message.id), + timeout=timeout + ) + ) + + await last_bot_message.click(text=action_input) + + done, pending = await asyncio.wait([new_message_task, edited_message_task], return_when=asyncio.FIRST_COMPLETED) + for task in pending: + task.cancel() + + event = done.pop().result() + last_bot_message = event.message + else: + pytest.fail(f"Unknown action type in test flow: '{action_type}'") + + # Assertion logic for both action types + expected_html = load(expected_response_id) + expected_plain = strip_html(expected_html) + actual_plain = last_bot_message.raw_text.strip() + + assert expected_plain in actual_plain, \ + f"Action {step} failed. Expected response containing '{expected_plain}' but got '{actual_plain}'" test_report["status"] = PytestTestStatus.OK except BaseException as e: @@ -80,23 +114,23 @@ class TestTelegramBot: @pytest.mark.parametrize( "command_flow", [ - ([("/start", "start_handler__message")]), - ([("/help", "help__commands_list")]), + ([{'type': 'message', 'input': "/start", 'expected': "start_handler__message"}]), + ([{'type': 'message', 'input': "/help", 'expected': "help__commands_list"}]), ( [ - ("/manage_reminders", "manage_reminders_handler__no_reminders"), - ("1", "manage_reminders_handler__reminder_number_bad"), + {'type': 'message', 'input': "/manage_reminders", 'expected': "manage_reminders_handler__no_reminders"}, + {'type': 'click', 'input': "Создать новое", 'expected': "manager_reminders_handler__enter_chat_id"}, ] ), ( [ - ("/manage_all_reminders", "manage_reminders_handler__no_reminders"), + {'type': 'message', 'input': "/manage_all_reminders", 'expected': "manage_reminders_handler__no_reminders"}, ] ), ], ) @pytest.mark.asyncio - async def test_command_flows(self, conversation: Conversation, command_flow: List[tuple]): + async def test_command_flows(self, conversation: Conversation, command_flow: List[Dict]): await _test_command_flow(self.report_state, conversation, command_flow) @pytest.mark.xfail @@ -104,7 +138,8 @@ async def test_command_flows(self, conversation: Conversation, command_flow: Lis @pytest.mark.asyncio async def test_failing_command(self, conversation: Conversation, command: str): try: - await _test_conversation_step(conversation, command, "should_not_matter", timeout=10) + await conversation.send_message(command) + await conversation.get_response(timeout=10) except TimeoutError: pass except Exception as e: From d6ed95cba34957fa1ef3eb34fe5ffbb3070d1663 Mon Sep 17 00:00:00 2001 From: Lev Chechulin Date: Wed, 13 Aug 2025 17:36:24 +0200 Subject: [PATCH 7/8] fix: change scope to session in conftest --- tests/integration/conftest.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index f950c5f1..568e344e 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -23,7 +23,7 @@ telegram_chat_id = int(config["error_logs_recipients"][0]) telegram_bot_name = config.get("handle", "") -@pytest.fixture(scope="function") +@pytest.fixture(scope="session") async def conversation(): """ Provides a completely fresh Telegram client and conversation for each test function. From fbfcffa7fcbc69a70ea34de6fd29d16514030950 Mon Sep 17 00:00:00 2001 From: Lev Chechulin Date: Wed, 13 Aug 2025 17:51:40 +0200 Subject: [PATCH 8/8] fix: revert old error message --- src/config_manager.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/config_manager.py b/src/config_manager.py index 4d62a065..edd1552e 100644 --- a/src/config_manager.py +++ b/src/config_manager.py @@ -151,7 +151,7 @@ def _load_config(self, config_path: str) -> dict: try: return json.loads(fin.read()) except json.JSONDecodeError as e: - logger.exception(f"Failed to decode JSON from '{config_path}'") + logger.error(exc_info=e) except IOError: logger.warning(f"Config file at {config_path} not found")