From 50e341a9ed763a99c5a896bc7f178fd5182c233a Mon Sep 17 00:00:00 2001 From: kotolex Date: Tue, 26 Dec 2023 08:34:29 +0500 Subject: [PATCH] FEATURE 9 and 12: cache and refactoring MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - из админ_командс убрана работа с БД и шелве - добавлен кеш, он загружается при старте сервиса и обновляется при изменениях, все запросы (антиспам, цитаты) идут уже через него, в БД лишний раз не ходим. В дальнейшем работаем только с ним везде через функции-хелперы - созданы хелперы для создания/удаления цитат и чатов - добавлено выражение в спам-лист (вчера пропустило) - добавлены тесты на спам(выше) и админ_коммандс --- src/admin_commands.py | 55 ++++---- src/constants.py | 2 +- src/models.py | 66 ++++++++- src/report.py | 7 + src/stan.py | 4 +- tests/test_bot.py | 310 +++++++++++++++++++++++++++--------------- tests/test_helpers.py | 3 +- 7 files changed, 296 insertions(+), 151 deletions(-) diff --git a/src/admin_commands.py b/src/admin_commands.py index 2cd00a5..32530b6 100644 --- a/src/admin_commands.py +++ b/src/admin_commands.py @@ -1,15 +1,14 @@ import logging -import shelve from datetime import datetime as dt from telebot import types, logger -from .constants import ADMIN_ID, LOGGING_LEVEL_DEBUG, LOGGING_LEVEL_INFO, DATA +from .constants import ADMIN_ID, LOGGING_LEVEL_DEBUG, LOGGING_LEVEL_INFO from .filters import is_white_id from .helpers import me, my_ip, is_admin -from .models import Chat, Quote, session +from .models import is_chat_exists, add_chat, delete_chat, update_chat, chat_by_id, all_chat_quotes from .reminder import remind, print_get_jobs, create_report_text -from .report import reset_report_stats +from .report import reset_report_stats, current_data from .robot import bot @@ -34,7 +33,7 @@ async def remind_manually(message): except ValueError as ve: await bot.send_message(message.chat.id, f"Не удалось разобрать дату!\n{ve}") else: - await remind(message.chat.id, today) + remind(message.chat.id, today) else: await bot.send_message( message.chat.id, @@ -54,15 +53,14 @@ async def list_jobs(_message): @bot.message_handler(func=is_admin, commands=["stats"]) async def send_stats(message: types.Message): """Show group statistics.""" - chat_id = message.chat.id if len(message.text.split()) == 1 else message.text.split()[-1] + chat_id = message.chat.id if len(splitted := message.text.split()) == 1 else splitted[-1] report_text = create_report_text(chat_id) if report_text: await bot.send_message(message.chat.id, report_text) else: - with shelve.open(f"{DATA}{chat_id}") as shelve_db: - shelve_messages = {v['User'].first_name: v['Count'] for k, v in shelve_db['Messages'].items()} - logging.info(f"[STATS {chat_id}] Невозможно отправить отчёт, недостаточно данных. Счётчик сообщений: {shelve_messages}.\n" - f"Баны: {shelve_db['Banned']=}") + shelve_messages, banned = current_data(chat_id) + logging.info(f"[STATS {chat_id}] Невозможно отправить отчёт, недостаточно данных. " + f"Счётчик сообщений: {shelve_messages}.\nБаны: {banned}") @bot.message_handler(func=is_admin, commands=["reset_stats"], chat_types=["supergroup", "group"]) @@ -76,12 +74,11 @@ async def send_stats(message: types.Message): async def enable_stan(message: types.Message): """Add group to database.""" logging.info(f"[{message.chat.title}] [{message.from_user.id}] {message.from_user.username}: {message.text}") - if not session.query(Chat.id).filter(Chat.chat_id == message.chat.id).first(): - session.add(Chat(chat_id=message.chat.id, title=message.chat.title, antispam=1, report=0, reminder=1)) - session.commit() + if not is_chat_exists(message.chat.id): + add_chat(message.chat.id, message.chat.title) await bot.send_message( message.chat.id, - f"""Группа "{message.chat.title} добавлена в БД. + f"""Группа "{message.chat.title}" добавлена в БД. /get_group_info - узнать текущие настройки""", ) else: @@ -91,11 +88,9 @@ async def enable_stan(message: types.Message): @bot.message_handler(func=is_admin, commands=["disable_stan"], chat_types=["supergroup", "group"]) async def disable_stan(message: types.Message): logging.info(f"[{message.chat.title}] [{message.from_user.id}] {message.from_user.username}: {message.text}") - if session.query(Chat.id).filter(Chat.chat_id == message.chat.id).first(): - chat = session.query(Chat).filter_by(chat_id=message.chat.id).first() - session.delete(chat) - session.commit() - await bot.send_message(message.chat.id, f"Группа {message.chat.title} и все связанные с ней цитаты удалёны!") + if is_chat_exists(message.chat.id): + title = delete_chat(message.chat.id) + await bot.send_message(message.chat.id, f"Группа {title} и все связанные с ней цитаты удалены!") else: await bot.send_message(message.chat.id, "Отказ. Этой группы нет в БД.") @@ -115,36 +110,32 @@ async def set_antispam_report_reminder(message: types.Message): except ValueError: logging.info(f"[ERROR] Неверные аргументы {message.text}") else: - session.query(Chat).filter_by(chat_id=message.chat.id).update( - {"antispam": antispam, "report": rep, "reminder": rem} - ) - session.commit() - await bot.send_message(message.chat.id, f"""Настройки обновлены. Проверить: /get_group_info""") + update_chat(message.chat.id, antispam, rep, rem) + await bot.send_message(message.chat.id, "Настройки обновлены. Проверить: /get_group_info") @bot.message_handler(func=is_white_id, commands=["get_quotes"], chat_types=["supergroup", "group"]) async def get_quotes(message: types.Message): logging.info(f"[{message.chat.id}] {message.from_user.first_name} {message.text}.") - quotes = session.query(Quote.text).filter_by(chat_id=message.chat.id).all() + quotes = all_chat_quotes(message.chat.id) if quotes: - length = len(session.query(Quote).filter(Quote.chat_id == message.chat.id).all()) - text = f"Количество цитат: {length}\n\Последние добавленные\n\n" - text += "\n".join(f"· {quote[0]}" for quote in quotes[-10:]) + text = f"Количество цитат: {len(quotes)}\n\Последние добавленные\n\n" + text += "\n".join(f"· {quote}" for quote in quotes[-10:]) await bot.send_message(message.chat.id, f"{text}") else: - await bot.send_message(message.chat.id, f"Цитаты отсутствуют. Подробнее: /get_group_info") + await bot.send_message(message.chat.id, "Цитаты отсутствуют. Подробнее: /get_group_info") @bot.message_handler(func=is_white_id, commands=["get_group_info"], chat_types=["supergroup", "group"]) async def get_group_info(message: types.Message): - group = session.query(Chat).filter_by(chat_id=message.chat.id).first() + group = chat_by_id(message.chat.id) if group: await bot.send_message( message.chat.id, f"""Группа: {group.title} ID группы: {group.chat_id} -Количество цитат: {len(session.query(Quote).filter(Quote.chat_id == message.chat.id).all())} +Количество цитат: {len(group.quotes)} Последние добавленные: /get_quotes Текущие настройки: @@ -153,7 +144,7 @@ ID группы: {group.chat_id} Праздники: {group.reminder}""", ) else: - await bot.send_message(message.chat.id, f"Группа не включена. Включить: /enable_stan") + await bot.send_message(message.chat.id, "Группа не включена. Включить: /enable_stan") @bot.message_handler(func=is_white_id, commands=["set_logging_level"], chat_types=["supergroup", "group"]) diff --git a/src/constants.py b/src/constants.py index ed99bb2..690b882 100644 --- a/src/constants.py +++ b/src/constants.py @@ -59,7 +59,7 @@ SPAM = {"me.sv/", "tg.sv/", "goo.by/", "go.sv/", "intim.video/", "uclck.ru/", ' 'P2P', 'Идет набор', 'в лс', 'нaбop', 'комaндy', 'kpuпт', 'бupж', '💰', 'пишитe', '💸', '$ в неделю', 'обучаю с нуля', 'доходы для всех', 'по трейдингу', 'фальшивые рубли', '‼️', 'зароботку', 'Набuраю', 'цuфр', 'Обучу вас', 'прuб', 'остались курсы', 'Crypt', 'kр', 'рu', - 'в лc'} + 'в лc', "в личные сообщения"} NON_GRATA = {"дудар", "хауди", "dudar"} BAN_WORDS = {"GREEN"} diff --git a/src/models.py b/src/models.py index b801bbc..11a08c6 100644 --- a/src/models.py +++ b/src/models.py @@ -1,4 +1,7 @@ +import logging +import os from pathlib import Path +from typing import Type from sqlalchemy import Column, String, Integer, Boolean, ForeignKey from sqlalchemy import create_engine @@ -51,27 +54,78 @@ class Chat(Base): return f'Chat({self.chat_id=}, {self.title=})' +CACHE: dict[int, Type[Chat]] = {} + + +def _fill_cache(): + if os.environ.get("testing", "FALSE") == "FALSE": + for chat in session.query(Chat).all(): + CACHE[chat.chat_id] = chat + + def is_antispam_enabled(chat_id: int) -> bool: - try: - return session.query(Chat.antispam).filter_by(chat_id=chat_id).first()[0] - except TypeError: + if chat_id not in CACHE: return False + return CACHE[chat_id].antispam def is_quote_in_chat(text: str, chat_id: int) -> bool: - result = session.query(Quote.text).filter_by(text=text, chat_id=chat_id).first() - return bool(result) + return text in all_chat_quotes(chat_id) def delete_quote_in_chat(text: str, chat_id: int): session.execute(delete(Quote).filter_by(text=text, chat_id=chat_id)) session.commit() + _reload_chat(chat_id) def all_chat_quotes(chat_id: int) -> list: - return session.query(Quote.text).filter(Quote.chat_id == chat_id).all() + if chat_id not in CACHE: + return [] + return [quote.text for quote in CACHE[chat_id].quotes] + + +def is_chat_exists(chat_id: int) -> bool: + return chat_id in CACHE def add_quote(chat_id: int, text: str): session.add(Quote(chat_id=chat_id, text=text)) session.commit() + _reload_chat(chat_id) + + +def chat_by_id(chat_id: int) -> None | Chat: + return CACHE.get(chat_id) + + +def add_chat(chat_id: int, chat_title: str): + session.add(Chat(chat_id=chat_id, title=chat_title, antispam=1, report=0, reminder=1)) + session.commit() + _reload_chat(chat_id) + + +def update_chat(chat_id: int, antispam: int = 1, rep: int = 0, rem: int = 1): + session.query(Chat).filter_by(chat_id=chat_id).update({"antispam": antispam, "report": rep, "reminder": rem}) + session.commit() + _reload_chat(chat_id) + + +def delete_chat(chat_id: int) -> str: + """ + Удаляем чат из БД вместе с его цитатами, возвращаем его название для логирования + """ + chat = session.query(Chat).filter_by(chat_id=chat_id).first() + session.delete(chat) + session.commit() + del CACHE[chat_id] + return chat.title + + +def _reload_chat(chat_id: int): + chat = session.query(Chat).filter(Chat.chat_id == chat_id).first() + CACHE[chat_id] = chat + + +_fill_cache() +logging.warning("[START] Load data from db to cache") diff --git a/src/report.py b/src/report.py index f7c915b..bef2c75 100644 --- a/src/report.py +++ b/src/report.py @@ -18,6 +18,13 @@ def increment(chat_id: int, banned: bool = True): shelve_db[category] += 1 +def current_data(chat_id: int) -> tuple: + with shelve.open(f"{DATA}{chat_id}") as shelve_db: + shelve_messages = {v['User'].first_name: v['Count'] for k, v in shelve_db['Messages'].items()} + banned = shelve_db['Banned'] + return shelve_messages, banned + + def create_report_text(chat_id): report = "" top_users = "" diff --git a/src/stan.py b/src/stan.py index b38fbbf..611b535 100644 --- a/src/stan.py +++ b/src/stan.py @@ -17,7 +17,7 @@ TYPING_TIMEOUT = 0.13 / 4 # Reading time is quarter of the same text writing ti def speak(chance_of: int, chat_id: int) -> None | str: number = random.randint(0, chance_of) if number == 0: - return random.choice([i[0] for i in all_chat_quotes(chat_id)]) + return random.choice(all_chat_quotes(chat_id)) return None @@ -40,7 +40,7 @@ async def act(message: types.Message): async def add_stan_quote(message: types.Message): if message.reply_to_message and message.reply_to_message.text: quote = message.reply_to_message.text - if quote not in {i[0] for i in all_chat_quotes(message.chat.id)}: + if not is_quote_in_chat(quote, message.chat.id): add_quote(message.chat.id, quote.replace("\n", " ")) await bot.send_message(message.chat.id, "➕\n └ " + quote.replace("\n", " "), parse_mode='Markdown') await bot.delete_message(message.chat.id, message.id) diff --git a/tests/test_bot.py b/tests/test_bot.py index 50fcfe8..a5d61c7 100644 --- a/tests/test_bot.py +++ b/tests/test_bot.py @@ -10,6 +10,7 @@ os.environ["LUTZPYBOT"] = "00000:AAAAAAAAAAAA" os.environ["whiteids"] = "100,200,300" os.environ["rollback"] = "1,2,3" os.environ["use_reminder"] = "FALSE" +os.environ["testing"] = "TRUE" from src import helpers @@ -106,11 +107,8 @@ class TestBot(IsolatedAsyncioTestCase): RESULTS.clear() async def test_start(self): - # Нужно пропустить проверку на спам - with patch("src.filters.is_antispam_enabled") as mocked: - mocked.return_value = False - await self.bot.process_new_updates([get_update('/start')]) - self.assertEqual(RESULTS[0][0]['text'], "Начни с прочтения") + await self.bot.process_new_updates([get_update('/start')]) + self.assertEqual(RESULTS[0][0]['text'], "Начни с прочтения") async def test_non_grata(self): await self.bot.process_new_updates([get_update('как вам дударь?')]) @@ -121,117 +119,75 @@ class TestBot(IsolatedAsyncioTestCase): self.assertEqual(RESULTS[0][0]['text'], "у нас тут таких не любят") async def test_rules(self): - # Нужно пропустить проверку на спам - with patch("src.filters.is_antispam_enabled") as mocked: - mocked.return_value = False - await self.bot.process_new_updates([get_update('/rules')]) - self.assertEqual(RESULTS[0][0]['text'], "Читай...") - self.assertTrue("https://telegra.ph/pythonchatru-07-07" in RESULTS[0][0]['reply_markup']) + await self.bot.process_new_updates([get_update('/rules')]) + self.assertEqual(RESULTS[0][0]['text'], "Читай...") + self.assertTrue("https://telegra.ph/pythonchatru-07-07" in RESULTS[0][0]['reply_markup']) async def test_faq(self): - # Нужно пропустить проверку на спам - with patch("src.filters.is_antispam_enabled") as mocked: - mocked.return_value = False - await self.bot.process_new_updates([get_update('/faq')]) - self.assertEqual(RESULTS[0][0]['text'], "Читай...") - self.assertTrue("https://telegra.ph/faq-10-07-4" in RESULTS[0][0]['reply_markup']) + await self.bot.process_new_updates([get_update('/faq')]) + self.assertEqual(RESULTS[0][0]['text'], "Читай...") + self.assertTrue("https://telegra.ph/faq-10-07-4" in RESULTS[0][0]['reply_markup']) async def test_lib(self): - # Нужно пропустить проверку на спам - with patch("src.filters.is_antispam_enabled") as mocked: - mocked.return_value = False - await self.bot.process_new_updates([get_update('/lib')]) - self.assertEqual(RESULTS[0][0]['text'], "Читай...") - self.assertTrue("https://telegra.ph/what-to-read-10-06" in RESULTS[0][0]['reply_markup']) + await self.bot.process_new_updates([get_update('/lib')]) + self.assertEqual(RESULTS[0][0]['text'], "Читай...") + self.assertTrue("https://telegra.ph/what-to-read-10-06" in RESULTS[0][0]['reply_markup']) async def test_tr(self): - # Нужно пропустить проверку на спам - with patch("src.filters.is_antispam_enabled") as mocked: - mocked.return_value = False - await self.bot.process_new_updates([get_update('/tr', reply_to="ghbdtn")]) - self.assertEqual(RESULTS[0][0]['text'], "привет") + await self.bot.process_new_updates([get_update('/tr', reply_to="ghbdtn")]) + self.assertEqual(RESULTS[0][0]['text'], "привет") async def test_tr2(self): - # Нужно пропустить проверку на спам - with patch("src.filters.is_antispam_enabled") as mocked: - mocked.return_value = False - await self.bot.process_new_updates([get_update('/tr', reply_to="руддщ")]) - self.assertEqual(RESULTS[0][0]['text'], "hello") + await self.bot.process_new_updates([get_update('/tr', reply_to="руддщ")]) + self.assertEqual(RESULTS[0][0]['text'], "hello") async def test_tsya(self): - # Нужно пропустить проверку на спам - with patch("src.filters.is_antispam_enabled") as mocked: - mocked.return_value = False - await self.bot.process_new_updates([get_update('/tsya')]) - self.assertEqual(RESULTS[0][0]['text'], "-тся и -ться в глаголах") - self.assertTrue("https://tsya.ru/" in RESULTS[0][0]['reply_markup']) + await self.bot.process_new_updates([get_update('/tsya')]) + self.assertEqual(RESULTS[0][0]['text'], "-тся и -ться в глаголах") + self.assertTrue("https://tsya.ru/" in RESULTS[0][0]['reply_markup']) async def test_nometa(self): text = """Не задавай мета-вопросов вроде:\n «Можно задать вопрос?»\n «Кто-нибудь пользовался .. ?»\n «Привет, мне нужна помощь по .. !»\n\nПросто спроси сразу! И чем лучше объяснишь проблему, тем вероятнее получишь помощь.""" - # Нужно пропустить проверку на спам - with patch("src.filters.is_antispam_enabled") as mocked: - mocked.return_value = False - await self.bot.process_new_updates([get_update('/nometa', reply_to="кто тут питонист?")]) - self.assertEqual(RESULTS[0][0]['text'], text) + await self.bot.process_new_updates([get_update('/nometa', reply_to="кто тут питонист?")]) + self.assertEqual(RESULTS[0][0]['text'], text) async def test_neprivet(self): - # Нужно пропустить проверку на спам - with patch("src.filters.is_antispam_enabled") as mocked: - mocked.return_value = False - await self.bot.process_new_updates([get_update('/neprivet', reply_to="привет всем")]) - self.assertEqual(RESULTS[0][0]['text'], "Пожалуйста, не пишите просто «Привет» в чате.") - self.assertTrue("https://neprivet.com/" in RESULTS[0][0]['reply_markup']) + await self.bot.process_new_updates([get_update('/neprivet', reply_to="привет всем")]) + self.assertEqual(RESULTS[0][0]['text'], "Пожалуйста, не пишите просто «Привет» в чате.") + self.assertTrue("https://neprivet.com/" in RESULTS[0][0]['reply_markup']) async def test_lutz(self): - # Нужно пропустить проверку на спам - with patch("src.filters.is_antispam_enabled") as mocked: - mocked.return_value = False - await self.bot.process_new_updates([get_update('/lutz', reply_to="привет всем")]) - self.assertEqual(RESULTS[0][0]['caption'], "вот, не позорься") - self.assertEqual(RESULTS[0][0]['document'], LUTZ_ID) - self.assertEqual(RESULTS[0][1], "sendDocument") - self.assertEqual(RESULTS[1][1], "deleteMessage") + await self.bot.process_new_updates([get_update('/lutz', reply_to="привет всем")]) + self.assertEqual(RESULTS[0][0]['caption'], "вот, не позорься") + self.assertEqual(RESULTS[0][0]['document'], LUTZ_ID) + self.assertEqual(RESULTS[0][1], "sendDocument") + self.assertEqual(RESULTS[1][1], "deleteMessage") async def test_bdmtss(self): - # Нужно пропустить проверку на спам - with patch("src.filters.is_antispam_enabled") as mocked: - mocked.return_value = False - await self.bot.process_new_updates([get_update('/bdmtss', reply_to="привет всем")]) - self.assertEqual(RESULTS[0][0]['voice'], BDMTSS_ID) - self.assertEqual(RESULTS[0][1], "sendVoice") - self.assertEqual(RESULTS[1][1], "deleteMessage") + await self.bot.process_new_updates([get_update('/bdmtss', reply_to="привет всем")]) + self.assertEqual(RESULTS[0][0]['voice'], BDMTSS_ID) + self.assertEqual(RESULTS[0][1], "sendVoice") + self.assertEqual(RESULTS[1][1], "deleteMessage") async def test_google(self): - # Нужно пропустить проверку на спам - with patch("src.filters.is_antispam_enabled") as mocked: - mocked.return_value = False - await self.bot.process_new_updates([get_update('/g', reply_to="python")]) - self.assertEqual(RESULTS[0][0]['text'], "Ищем «python»...") - self.assertTrue("https://www.google.com/search?q=python" in RESULTS[0][0]['reply_markup']) + await self.bot.process_new_updates([get_update('/g', reply_to="python")]) + self.assertEqual(RESULTS[0][0]['text'], "Ищем «python»...") + self.assertTrue("https://www.google.com/search?q=python" in RESULTS[0][0]['reply_markup']) async def test_nojob(self): text = """Мы здесь не для того, чтобы за тебя решать задачи.\n\nЗдесь помогают по конкретным вопросам в ТВОЁМ коде, поэтому тебе нужно показать код, который ты написал сам и объяснить где и почему застрял... всё просто. 🤷🏼️""" - # Нужно пропустить проверку на спам - with patch("src.filters.is_antispam_enabled") as mocked: - mocked.return_value = False - await self.bot.process_new_updates([get_update('/nojob', reply_to="есть работа")]) - self.assertEqual(RESULTS[0][0]['text'], text) + await self.bot.process_new_updates([get_update('/nojob', reply_to="есть работа")]) + self.assertEqual(RESULTS[0][0]['text'], text) async def test_nobot(self): text = """Внимание:\nТелеграм бот не должен быть твоим первым проектом на Python. Пожалуйста, изучи основы Python, работу с модулями, основы веб-технологий, асинхронное программирование и отладку до начала работы с Телеграм ботами. Существует много ресурсов для этого в интернете.""" - # Нужно пропустить проверку на спам - with patch("src.filters.is_antispam_enabled") as mocked: - mocked.return_value = False - await self.bot.process_new_updates([get_update('/nobot', reply_to="как бота написать?")]) - self.assertEqual(RESULTS[0][0]['text'], text) + await self.bot.process_new_updates([get_update('/nobot', reply_to="как бота написать?")]) + self.assertEqual(RESULTS[0][0]['text'], text) async def test_nogui(self): text = """Внимание:\nGUI приложение не должно быть твоим первым проектом на Python. Пожалуйста, изучи основы Python, работу с модулями, циклы событий и отладку до начала работы с какими-либо GUI-фреймворками. Существует много ресурсов для этого в интернете.""" - # Нужно пропустить проверку на спам - with patch("src.filters.is_antispam_enabled") as mocked: - mocked.return_value = False - await self.bot.process_new_updates([get_update('/nogui', reply_to="как gui написать?")]) - self.assertEqual(RESULTS[0][0]['text'], text) + await self.bot.process_new_updates([get_update('/nogui', reply_to="как gui написать?")]) + self.assertEqual(RESULTS[0][0]['text'], text) async def test_spam_text(self): # Нужно подставить антиспам вкл. @@ -314,44 +270,38 @@ class TestBot(IsolatedAsyncioTestCase): with patch("src.commands.update_stats") as update_stats: mocked.return_value = True update_stats.side_effect = lambda x: None - await self.bot.process_new_updates( - [get_update('/unban_id 1', reply_to="shit text", user_id=ADMIN_ID)]) + await self.bot.process_new_updates([get_update('/unban_id 1', reply_to="shit text", + user_id=ADMIN_ID)]) self.assertEqual(RESULTS[0][1], 'unbanChatMember') async def test_white_id_can_add_quote(self): - # Нужно подставить список цитат - with patch("src.stan.all_chat_quotes") as mocked: - # Нужно подменить инкремент, нельзя реально в шелве работать в тестах - with patch("src.commands.increment") as _mocked_shelve: - # Подменяем обновление статистики, нам важно что оно просто было вызвано - with patch("src.stan.add_quote") as aq: - mocked.return_value = [] - aq.side_effect = lambda *a: None - await self.bot.process_new_updates([get_update('/add', reply_to="good text", user_id=100)]) - self.assertEqual(RESULTS[0][1], 'sendMessage') - self.assertEqual(RESULTS[0][0]['text'], '➕\n └ good text') - self.assertEqual(RESULTS[1][1], 'deleteMessage') + # Нужно подменить инкремент, нельзя реально в шелве работать в тестах + with patch("src.commands.increment") as _mocked_shelve: + # Подменяем обновление статистики, нам важно что оно просто было вызвано + with patch("src.stan.add_quote") as aq: + aq.side_effect = lambda *a: None + await self.bot.process_new_updates([get_update('/add', reply_to="good text", user_id=100)]) + self.assertEqual(RESULTS[0][1], 'sendMessage') + self.assertEqual(RESULTS[0][0]['text'], '➕\n └ good text') + self.assertEqual(RESULTS[1][1], 'deleteMessage') async def test_white_id_cant_add_existing_quote(self): - # Нужно подставить список цитат - with patch("src.stan.all_chat_quotes") as mocked: + # Нужно подставить наличие цитаты + with patch("src.stan.is_quote_in_chat") as mocked: # Нужно подменить инкремент, нельзя реально в шелве работать в тестах with patch("src.commands.increment") as _mocked_shelve: # Подменяем реальное добавление цитаты, нам важно что оно просто было вызвано with patch("src.stan.add_quote") as add_qoute: - mocked.return_value = [["good text"]] + mocked.return_value = True add_qoute.side_effect = lambda x, y: None await self.bot.process_new_updates([get_update('/add', reply_to="good text", user_id=100)]) self.assertEqual(RESULTS[0][1], 'sendMessage') self.assertEqual(RESULTS[0][0]['text'], '⛔️ Не добавил, есть токое\n └ good text') async def test_white_id_cant_remove_non_existing_quote(self): - # Нужно вернуть запрос о наличии цитаты - with patch("src.stan.is_quote_in_chat") as mocked: - mocked.return_value = False - await self.bot.process_new_updates([get_update('/remove', reply_to="good text", user_id=100)]) - self.assertEqual(RESULTS[0][1], 'sendMessage') - self.assertEqual(RESULTS[0][0]['text'], '⛔️ Нет такого\n └ good text') + await self.bot.process_new_updates([get_update('/remove', reply_to="good text", user_id=100)]) + self.assertEqual(RESULTS[0][1], 'sendMessage') + self.assertEqual(RESULTS[0][0]['text'], '⛔️ Нет такого\n └ good text') async def test_white_id_can_remove_existing_quote(self): # Нужно вернуть запрос о наличии цитаты @@ -375,6 +325,148 @@ class TestBot(IsolatedAsyncioTestCase): await self.bot.process_new_updates([member(leave=True)]) self.assertEqual(RESULTS[0][1], 'deleteMessage') + async def test_me(self): + with patch("src.admin_commands.me") as mocked: + mocked.side_effect = lambda m: m.text + await self.bot.process_new_updates([get_update('/me')]) + self.assertEqual(RESULTS[0][1], 'sendMessage') + self.assertEqual(RESULTS[0][0]['text'], '/me') + + async def test_admin_can_get_ip(self): + with patch("src.admin_commands.my_ip") as mocked: + mocked.return_value = '127.0.0.1' + await self.bot.process_new_updates([get_update('/ip', user_id=ADMIN_ID)]) + self.assertEqual(RESULTS[0][1], 'sendMessage') + self.assertEqual(RESULTS[0][0]['text'], '127.0.0.1') + + async def test_non_admin_cant_do(self): + comands = ('ip', 'jobs', 'stats', 'reset_stats', 'enable_stan', 'disable_stan', 'set_antispam_report_reminder', + 'set_rules', 'get_quotes', 'get_group_info') + for command in comands: + with self.subTest(f"user cant use {command}"): + # Нужно подменить инкремент, нельзя реально в шелве работать в тестах + with patch("src.commands.increment") as _mocked_shelve: + # Подменяем обновление статистики, нам важно что оно просто было вызвано + with patch("src.commands.update_stats") as update_stats: + update_stats.side_effect = lambda x: None + await self.bot.process_new_updates([get_update(f'/{command}')]) + self.assertEqual(RESULTS, []) + + async def test_remind_manually(self): + params = ( + ("Формат даты: MM-DD-YYYY\n\nПримеры:\n/remind 09-12-2024\n/remind 09-13-2022", ''), + ("Не удалось разобрать дату!\ntime data 'shit' does not match format '%m-%d-%Y'", ' shit'), + ("Не удалось разобрать дату!\ntime data '13-13-2020' does not match format '%m-%d-%Y'", ' 13-13-2020'), + ("2023-10-10 00:00:00", ' 10-10-2023'), + ) + for expected, date in params: + with self.subTest(f"remind_manually {date}"): + with patch("src.admin_commands.remind") as mocked: + mocked.side_effect = lambda x, y: RESULTS.append([{'text': str(y)}]) + await self.bot.process_new_updates([get_update(f'/remind{date}')]) + self.assertEqual(expected, RESULTS[0][0]['text']) + RESULTS.clear() + + async def test_admin_can_get_get_jobs(self): + with patch("src.admin_commands.print_get_jobs") as mocked: + mocked.return_value = 'jobs' + await self.bot.process_new_updates([get_update('/jobs', user_id=ADMIN_ID)]) + self.assertEqual(RESULTS[0][1], 'sendMessage') + self.assertEqual(RESULTS[0][0]['text'], 'jobs') + self.assertEqual(RESULTS[0][0]['chat_id'], str(ADMIN_ID)) + + async def test_send_stats(self): + params = ( + ("11", ''), + ("99", ' 99'), + ([], ' 0'), + ) + for expected, stats in params: + with self.subTest(f"stats {stats}"): + with patch("src.admin_commands.create_report_text") as mocked: + with patch("src.admin_commands.current_data") as mocked2: + mocked.side_effect = lambda x: str(x) if x != '0' else None + mocked2.return_value = 0, 0 + await self.bot.process_new_updates([get_update(f'/stats{stats}', user_id=ADMIN_ID)]) + if stats != ' 0': + self.assertEqual(expected, RESULTS[0][0]['text']) + else: + self.assertEqual(expected, RESULTS) + RESULTS.clear() + + async def test_enable_stan(self): + params = ( + ('Группа "None" добавлена в БД.\n/get_group_info - узнать текущие настройки', False), + ('Отказ. Группа уже включена', True), + ) + for expected, exists in params: + with self.subTest(f"enable_stan chat_exists={exists}"): + with patch("src.admin_commands.is_chat_exists") as mocked: + with patch("src.admin_commands.add_chat") as mocked2: + mocked.return_value = exists + mocked2.side_effect = lambda a, b: None + await self.bot.process_new_updates([get_update('/enable_stan', user_id=100)]) + self.assertEqual(expected, RESULTS[0][0]['text']) + RESULTS.clear() + + async def test_disable_stan(self): + params = ( + ('Отказ. Этой группы нет в БД.', False), + ('Группа DELETED и все связанные с ней цитаты удалены!', True), + ) + for expected, exists in params: + with self.subTest(f"enable_stan chat_exists={exists}"): + with patch("src.admin_commands.is_chat_exists") as mocked: + with patch("src.admin_commands.delete_chat") as mocked2: + mocked.return_value = exists + mocked2.side_effect = lambda a: 'DELETED' + await self.bot.process_new_updates([get_update('/disable_stan', user_id=ADMIN_ID)]) + self.assertEqual(expected, RESULTS[0][0]['text']) + RESULTS.clear() + + async def test_set_antispam_report_reminder(self): + params = ( + ((11, 1, 1, 1), 'set_rules 1 1 1'), + ((11, 1, 1, 1), 'set_antispam_report_reminder 1 1 1'), + ) + for expected, command in params: + with self.subTest(f"set_antispam_report_reminder {command}"): + with patch("src.admin_commands.update_chat") as mocked: + mocked.side_effect = lambda *a: RESULTS.append(a) + await self.bot.process_new_updates([get_update(f'/{command}', user_id=100)]) + self.assertEqual(expected, RESULTS[0]) + self.assertEqual("Настройки обновлены. Проверить: /get_group_info", RESULTS[1][0]['text']) + RESULTS.clear() + + async def test_get_quotes(self): + params = ( + ("Цитаты отсутствуют. Подробнее: /get_group_info", []), + ("Количество цитат: 1\n\\Последние добавленные\n\n· one", ['one']), + ("Количество цитат: 2\n\\Последние добавленные\n\n· one\n· two", ['one', 'two']), + ) + for expected, quotes in params: + with self.subTest(f"get_quotes {quotes}"): + with patch("src.admin_commands.all_chat_quotes") as mocked: + mocked.side_effect = lambda *a: quotes + await self.bot.process_new_updates([get_update('/get_quotes', user_id=100)]) + self.assertEqual(expected, RESULTS[0][0]['text']) + RESULTS.clear() + + async def test_get_group_info(self): + exp = f"Группа: Title\nID группы: 11 \n\nКоличество цитат: 0\nПоследние добавленные: /get_quotes\n\n" \ + f"Текущие настройки:\n Антиспам: 1\n Ежедневные отчёты: 1\n Праздники: 1" + params = ( + ("Группа не включена. Включить: /enable_stan", None), + (exp, MagicMock(title="Title", chat_id=11, quotes=[], antispam=1, report=1, reminder=1)), + ) + for expected, chat in params: + with self.subTest(f"get_group_info {chat}"): + with patch("src.admin_commands.chat_by_id") as mocked: + mocked.side_effect = lambda *a: chat + await self.bot.process_new_updates([get_update('/get_group_info', user_id=100)]) + self.assertEqual(expected, RESULTS[0][0]['text']) + RESULTS.clear() + if __name__ == '__main__': main() diff --git a/tests/test_helpers.py b/tests/test_helpers.py index 02c0fa0..a1540aa 100644 --- a/tests/test_helpers.py +++ b/tests/test_helpers.py @@ -5,7 +5,8 @@ from src.helpers import is_spam, is_mixed, is_ban_words_in_caption, is_in_not_al CASES = ( "Оформим резидeнтствo ОАЭ без предoплаты за 5000 дирхам. Пoдробнoсти в лc", - "Всем привет Я в поисках людей в kpиптo кoмaнду У нас вы пройдете 6ecплaтноe обученue 1-3 дня Можно совмещать с основной paбoтoй" + "Всем привет Я в поисках людей в kpиптo кoмaнду У нас вы пройдете 6ecплaтноe обученue 1-3 дня Можно совмещать с основной paбoтoй", + "Здравствуйте! Нужны 2 человека от 200 $/день. Можно без опыта Пишите в личные сообщения +", )