Обоснование выбора технологий

Test IT Test Management System [https://testit.software/]. Эта система управления тестированием программных продуктов. В компании широко используется и помогает оптимизировать работу тестировщиков. На нее никто не жалуется.

Python. Автоматизированные тесты у нас пишутся на этом языке программирования, так что другие языки не рассматривались. Telegram-бот становится более производительным, если использовать асинхронный подход к написанию кода, поэтому мы взяли следующие основные библиотеки:

• asyncio,

• aiogram,

• threading,

• motor.motor_asyncio.

MongoDB [https://www.mongodb.com/]. Это документоориентированная система управления базами данных. Данная СУБД была выбрана из-за возможности хранения сложных по структуре объектов в виде документов. Можно практически не обрабатывать тело ответа от Test IT, чтобы сохранить данные в коллекцию. Плюс поиск и выгрузка документов из БД производится быстро и легко.

@dp.message_handler(commands=["unapproved_autotests"])
async def unapproved_autotests_command_handler(message: types.Message):
    """
    Обработчик команды /unapproved_autotests. \n
    :param message: объект, представляющий сообщение пользователя.
    """
    localization = await data_base.get_user_localization(user_id=message.chat.id)
    wait_msg = await message.answer(text=MESSAGES.get('get_autotests').get(localization))
    
    # сохраняем тип состояния - unapproved (неподтвержденные автотесты)
    await data_base.save_selected_autotests_type(
        user_id=message.chat.id, 
        autotests_type='unapproved')
    
    # получает наименования модулей {“module_id": {“name”: module_name, “localization”:localization}}
    all_modules = await data_base.get_modules_names(report_mode="key_id")
    
    # получает автотесты, которые нужно актуализировать
    all_unapproved_autotests = await testit_client.get_autotests_for_actualize(
        modules=all_modules,
        message=message,
        localization=localization)
    
    # обновляет БД неактуализированными автотестами
    await data_base.update_unapproved_autotests(
        user_id=message.chat.id,
        autotests_dict=all_unapproved_autotests)
    
    await bot.delete_message(
          chat_id=message.chat.id, 
          message_id=wait_msg.message_id)
    
    # перенаправляет на метод создания меню
    await send_modules_menu(message=message)

for autotest in all_autotests:
  if mode == "unapproved" and autotest.get('mustBeApproved'):
      autotests_dict = await self.add_autotests_dict(
                   autotests_dict=autotests_dict,
                   modules_names=modules_names,
                   autotest=autotest)

async def send_modules_menu(message: types.Message):
    """
    Отправляет меню со списком модулей. \n
    :param message: объект, представляющий сообщение пользователя.
    """
   # получает наименования модулей {“module_id": {“name”: module_name, “localization”:localization}}
    modules_names = await data_base.get_modules_names(report_mode="key_id")
    
   # составляем блоки для меню
    markup = types.InlineKeyboardMarkup()

    buttons = []
    localization = await data_base.get_user_localization(user_id=message.chat.id)
    autotests_type = await data_base.get_selected_autotests_type(user_id=message.chat.id)
  
    for module_id in modules_names.keys():
        buttons.append(
            types.InlineKeyboardButton(
                text=modules_names[module_id].get("localization").get(localization),
                callback_data=f'module#{modules_names[module_id].get("
                                         name")}#{autotests_type}'))
        if len(buttons) == 2:
            markup.add(*buttons)
            buttons = []

    markup.add(
        types.InlineKeyboardButton(
            text=MESSAGES.get('exit_btn').get(localization),
            callback_data='exit_button'))

    await message.answer(text=MESSAGES.get('select_module').get(localization), reply_markup=markup)

async def schedule_subscription():
     """ Подписка по расписанию. """
     aioschedule.every().day.at("21:15").do(
            job_func=send_last_unapproved_autotests_on_a_schedule, 
            bot=bot)
    while True:
        await aioschedule.run_pending()
        await asyncio.sleep(1)

def run_bot():
    """ Запускает бота. """
    asyncio.run_coroutine_threadsafe(coro=main(), loop=loop)
    loop.run_forever()

def run_scheduled_delivery():
    """ Запускает подписку. """
    asyncio.run_coroutine_threadsafe(coro=schedule_subscription(), loop=loop)

if __name__ == "__main__":
    try:
        thread1 = Thread(target=run_bot)
        thread2 = Thread(target=run_scheduled_delivery)
        thread1.start()
        thread2.start()
    except (KeyboardInterrupt, SystemExit) as ex:
        logger.error(f'Stop session: {ex}')

async def send_last_unapproved_autotests_on_a_schedule(bot: Bot):
    """
    Отправляет список неподтвержденных автотестов всем подписчикам по расписанию
    за прошлый день. \n
    :param bot: объект, представляющий бота.
    """
    # получаем список пользователей, которые подписались на рассылку
    subscribers_ids = await data_base.get_subscribers_ids()
    if subscribers_ids is None:
        return

    # получаем все автотесты, у которых изменился тест-кейс за прошедший день
    last_unapproved_autotests = await testit_client.get_last_unapproved_autotests()

    modules = await data_base.get_modules_names(report_mode="key_id")

    # оповещаем всех подписчиков
    for id_ in subscribers_ids:
        localization = await data_base.get_user_localization(user_id=id_)
        # подсчет для вывода статистики
        autotests_count = {modules[module_id]["localization"][localization]: len(autotests)
                           for module_id, autotests in last_unapproved_autotests.items()
                           if len(autotests) > 0}
        
        # меню с модулями, если имеются автотесты за прошлый день
        if autotests_count:
            keyboard = InlineKeyboardMarkup()
            keyboard.add(
                types.InlineKeyboardButton(
                    text=MESSAGES.get("open_modules_with_last_unapproved_autotests").get(
                               localization),
                    callback_data='last_unapproved_autotests'))
            await bot.send_message(
                chat_id=id_,
                text=MESSAGES.get("last_unapproved_description_with_autotests").get(
                          localization).format(autotests_count),
                reply_markup=keyboard,
                parse_mode='Markdown')
        else:
            await bot.send_message(
                chat_id=id_,
                text=MESSAGES.get("last_unapproved_description_without_autotests").get(
                          localization))

GET/api/v2/autoTests/{autotest_id}/workItems/changed/id — получить список идентификаторов тест-кейсов, которые необходимо подтвердить.

GET/api/v2/workItems/{id_} — получить тест-кейс по идентификатору.

POST/api/v2/autoTests/search — получить список всех автотестов.

No result - "-"

Passed - "100%"

Passed Passed - "100%" (0/1), где 0 = M, 1 = N

Passed Failed - "0%" (1/1)

Passed Failed Passed - "0%" (2/2)

Passed Failed Passed Failed - "0%" (3/3)

Passed Failed Passed Failed Failed - "25%" (3/4)

@dp.message_handler(lambda message: message.text.isdigit(), state=user_states.min_percent)
async def input_min_percent_handler(message: types.Message):
    """
    Обработчик ввода минимального процента. \n
    :param message: объект, представляющий сообщение пользователя.
    """
    localization = await data_base.get_user_localization(user_id=message.chat.id)
    # считываем текст ввода
    _min_percent = int(message.text)
    
    if 0 <= _min_percent <= 100:
        # переходим к следующему состоянию FSM
        await user_states.next()
        
        await data_base.save_min_percent(
                       user_id=message.chat.id, 
                       min_percent=_min_percent)
        
        await message.answer(text=MESSAGES.get('input_max_percent').get(localization))
    else:
        return await message.answer(text=MESSAGES.get('retry_input').get(localization))

@dp.message_handler(lambda message: message.text.isdigit(), state=user_states.max_percent)
async def input_max_percent_handler(message: types.Message):
    """
    Обработчик ввода максимального процента. \n
    :param message: объект, представляющий сообщение пользователя.
    """
    localization = await data_base.get_user_localization(user_id=message.chat.id)
    _max_percent = int(message.text)
    
    if 0 <= _max_percent <= 100:
        # выгружаем из БД значение минимального процента
        _min_percent = await data_base.get_min_percent(user_id=message.chat.id)

        if _max_percent >= _min_percent:
            # заканчиваем режим состояний FSM (всего два состояния)
            await user_states.next()
            
            await data_base.save_max_percent(
                                   user_id=message.chat.id, 
                                   max_percent=_max_percent)
 
            wait_msg = await message.answer(text=MESSAGES.get(
                                              'get_autotests').get(localization))
            
            # сохраняем в БД тип состояния - 'stability' 
            await data_base.save_selected_autotests_type(
                                  user_id=message.chat.id, 
                                  autotests_type='stability')

            all_modules = await data_base.get_modules_names(report_mode="key_id")
           
            # получаем автотесты, которые нужно актуализировать            
            all_stability_autotests = await testit_client.get_autotests_for_actualize(
                modules=all_modules,
                mode="stability",
                min_percent=_min_percent,
                max_percent=_max_percent,
                message=message,
                localization=localization)
            
            await data_base.update_stability_autotests(
                          user_id=message.chat.id, 
                          autotests_dict=all_stability_autotests)
            
            await bot.delete_message(
                          chat_id=message.chat.id,
                          message_id=wait_msg.message_id)

            await send_modules_menu(message=message)
        else:
            return await message.answer(text=MESSAGES.get(
                                    'max_percent_error').get(localization))
    else:
        return await message.answer(text=MESSAGES.get('retry_input').get(localization))

async def add_entities_for_autotest_to_data_base_by_batch(
               func,
               entities: list = None,
               batch: int = 5,
               **kwargs):
    """
    Добавляет сущности в БД партиями для определенного автотеста.\n
    :param func: метод добавления сущностей в БД.
    :param entities: список сущностей.
    :param batch: группа сущностей, чтобы разбить большой массив на части.
    :param kwargs: остальные параметры.
    """
    flag = True
    
    while flag:
        if not entities:
            flag = False
            
        # проходимся по партиям
        data = entities[:batch]
        entities = entities[batch:]
        
        # создаем таски и ждем их выполнения
        data_ = []
        for entity in data:
            data_.append(asyncio.create_task(func(entity, **kwargs)))
        await asyncio.gather(*data_)

@dp.callback_query_handler(lambda call: call.data == "test_plan_search_btn")
async def test_plan_search_callback(call: CallbackQuery):
    """
    Обработчик нажатия на кнопку поиска тест-плана. \n
    :param call: объект, представляющий входящий callback-запрос.
    """
    localization = await data_base.get_user_localization(user_id=call.message.chat.id)

    await bot.delete_message(chat_id=call.message.chat.id, message_id=call.message.message_id)
    
    # устанавливаем машину состояний FSM
    await search_state.search.set()
    await call.message.answer(MESSAGES.get('input_name').get(localization).format("тест-плана"))

@dp.message_handler(state=search_state.search)
async def test_plan_search_fsm(message: types.Message):
    """
    Обработчик ввода имени тест-плана. \n
    :param message: объект, представляющий сообщение пользователя.
    """
    test_plan_name = message.text
    
    # останавливаем машину состояний (одно состояние)
    await search_state.next()
    
    # сохраняем состояние тест-плана в режиме поиска
    await data_base.save_test_plan_action(
                    user_id=message.chat.id,
                    action='test_plans_results',
                    text_search=test_plan_name)
    
   # передаем в меню тест-плана параметр page=1 – пагинация начинается со страницы 1
    await send_test_plans(message=message, page=1)

async def run_test_plan(message: types.Message, test_plan_id: str):
    """
    Запускает тестовый план. \n
    :param message: объект, представляющий сообщение пользователя.
    :param test_plan_id: идентификатор тестового плана.
    """
    localization = await data_base.get_user_localization(user_id=message.chat.id)
    test_suites_ids = []
    points_for_run = []
    
    # получаем тестовый набор
    test_suites = await testit_client.get_test_suites(test_plan_id=test_plan_id)
    for test_suite in test_suites:
        test_suites_ids.append(test_suite.get('id'))
    
    # получаем идентификаторы автотестов.
    points_data = await testit_client.post_points_with_last_result(
                              test_suites_ids=test_suites_ids)
    for _, points in points_data.items():
        for point in points:
            points_for_run.append(point.get('id'))

    test_run_data = {
        'autoTests': [],
        'build': '1',
        'entityTypeName': 'TestRuns',
        'isAutomated': True,
        'name': None,
        'projectId': Config.testit_project_id,
        'runByUserId': None,
        'stateName': 'NotStarted',
        'stoppedByUserId': None,
        'testPlanId': test_plan_id,
        'testPointIds': points_for_run
    }
    try:
        await testit_client.post_test_runs(data=test_run_data)
        
        await message.answer(text=MESSAGES.get('successfully_launched_test_plan').get(
                                    localization))
    except Exception:
        await message.answer(text=MESSAGES.get('unsuccessfully_launched_test_plan').get(
                                   localization))