From 9103669cac50e6b03918cb4aa0a63897b6e373fb Mon Sep 17 00:00:00 2001 From: kotolex Date: Fri, 22 Dec 2023 19:16:57 +0500 Subject: [PATCH 1/9] Refactoring: step 1 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - пакет app переименован в src - файл БД теперь должен лежать в data - удалена папка коммандс, все команды вынесены в модуль коммандс или админ_коммандс - нет импортов в ините - убраны все импорты * - теперь в .env можно указать use_reminder="FALSE", тогда не будет стартовать ремайндер - можно указать урл для Правил - в конфиге теперь только конфиг -то, что из енва получаем - константы теперь лежат отдельно - добавлены явные return там, где не было - бот и вебхук разнесены отдельно - при старте проверяем, что Правила доступны - добавлено еще логирование --- app/__init__.py | 1 - app/commands/get.py | 27 ------- app/helpers.py | 39 --------- requirements.txt | Bin 1732 -> 2020 bytes src/__init__.py | 1 + .../service.py => src/admin_commands.py | 71 +++++++---------- app/bot.py => src/commands.py | 75 ++++++++---------- src/config.py | 26 ++++++ app/config.py => src/constants.py | 43 ++-------- {app => src}/filters.py | 40 ++++------ src/helpers.py | 62 +++++++++++++++ {app => src}/models.py | 22 ++--- {app => src}/reminder.py | 31 +++++--- {app => src}/report.py | 22 ++--- src/robot.py | 14 ++++ {app => src}/rules.py | 7 +- {app => src}/stan.py | 18 ++--- src/webhook.py | 20 +++++ wsgi.py | 7 +- 19 files changed, 268 insertions(+), 258 deletions(-) delete mode 100644 app/__init__.py delete mode 100644 app/commands/get.py delete mode 100644 app/helpers.py create mode 100644 src/__init__.py rename app/commands/service.py => src/admin_commands.py (74%) rename app/bot.py => src/commands.py (88%) create mode 100644 src/config.py rename app/config.py => src/constants.py (76%) rename {app => src}/filters.py (62%) create mode 100644 src/helpers.py rename {app => src}/models.py (66%) rename {app => src}/reminder.py (75%) rename {app => src}/report.py (85%) create mode 100644 src/robot.py rename {app => src}/rules.py (75%) rename {app => src}/stan.py (86%) create mode 100644 src/webhook.py diff --git a/app/__init__.py b/app/__init__.py deleted file mode 100644 index 2b3c21e..0000000 --- a/app/__init__.py +++ /dev/null @@ -1 +0,0 @@ -from .commands import service diff --git a/app/commands/get.py b/app/commands/get.py deleted file mode 100644 index 415fa1c..0000000 --- a/app/commands/get.py +++ /dev/null @@ -1,27 +0,0 @@ -from urllib.request import urlopen - - -def my_ip(): - ip_service = "https://icanhazip.com" - ip = urlopen( - ip_service, - ) - return ip.read().decode("utf-8") - - -def me(message): - msg = f"Telegram ID: {message.from_user.id}\n\ - ├ Is bot: {message.from_user.is_bot}\n\ - ├ First name: {message.from_user.first_name}\n" - if message.from_user.last_name: - msg += f" ├ Last name: {message.from_user.last_name}\n" - if message.from_user.username: - msg += f" ├ Username: {message.from_user.username}\n" - msg += f" ├ Language code: {message.from_user.language_code}\n" - if message.from_user.is_premium is True: - msg += f" ├ Is Premium: ⭐️\n" - else: - msg += f" ├ Is Premium: No\n" - msg += f" ├ Chat ID: {message.chat.id}" - - return msg diff --git a/app/helpers.py b/app/helpers.py deleted file mode 100644 index 2204944..0000000 --- a/app/helpers.py +++ /dev/null @@ -1,39 +0,0 @@ -""" Send User's info. """ -import shelve -import logging -from . import report -from .config import DATA -from urllib import parse -from telebot import types - - -def represent_as_get(message: types.Message): - return parse.quote_plus(detect_args(message)) - - -def detect_args(message: types.Message): - if message.reply_to_message and message.reply_to_message.text: - if len(message.text.split()) == 1: - return message.reply_to_message.text - else: - return " ".join(message.text.split()[1:]) - else: - if len(message.text.split()) > 1: - return " ".join(message.text.split()[1:]) - - -def update_stats(message: types.Message): - with shelve.open(f"{DATA}{message.chat.id}", writeback=True) as s: - if "Messages" not in s: - report.reset_report_stats(message.chat.id) - - if message.from_user.id not in s["Messages"]: - s["Messages"][message.from_user.id] = { - "User": message.from_user, - "Count": 1, - } - logging.info( - f"[{message.chat.title[:10]}] [{message.from_user.id}] {message.from_user.first_name}" - ) - else: - s["Messages"][message.from_user.id]["Count"] += 1 diff --git a/requirements.txt b/requirements.txt index 951a96faac5d486b7724517951788db38a3e2efa..de8b4d45c18db4b96fa692b3fb88c9e2732ce424 100644 GIT binary patch delta 310 zcmYk2I|{-;6h-eOoe=D8#L_}BF$f}7qTn(_V+`gaM#0Z2Br6fOU?(oY&fXoU_sv+y zFfhYCbKZM%b)MzX=RHA$3=3?yR_JrJ(L|Tqg@Y1Io|6iZ@)m2xWiFgyPS$FWW6ykw z20alH1hj5>7NbwHrC%7M)LJ9gnfLRMI4=WL53r*pU@zjy=nTn<%XBA|J@FMcRj 1 and args[-1].isdigit() and 0 < int(args[-1]): await send_or_reply( message, - f"Правило {args[-1]}\n{rules.fetch_rule(args[-1])}", + f"Правило {args[-1]}\n{fetch_rule(args[-1])}", reply_markup=markup, ) else: @@ -164,7 +176,7 @@ async def stan_speak(message: types.Message): message.text, ) ) - await bot.send_message(message.chat.id, stan.speak(0, message.chat.id), parse_mode='Markdown') + await bot.send_message(message.chat.id, speak(0, message.chat.id), parse_mode='Markdown') await bot.delete_message(message.chat.id, message.id) @@ -180,10 +192,7 @@ async def send_tsya(message: types.Message): ) ) markup = types.InlineKeyboardMarkup() - markup.add( - types.InlineKeyboardButton("🧑🏼‍🎓 Читать правило", url="https://tsya.ru/"), - row_width=1, - ) + markup.add(types.InlineKeyboardButton("🧑🏼‍🎓 Читать правило", url="https://tsya.ru/"), row_width=1) await send_or_reply(message, "-тся и -ться в глаголах", reply_markup=markup) await bot.delete_message(message.chat.id, message.id) @@ -200,10 +209,7 @@ async def send_nometa(message: types.Message): ) ) markup = types.InlineKeyboardMarkup() - markup.add( - types.InlineKeyboardButton("❓ nometa.xyz", url="https://nometa.xyz/ru.html"), - row_width=1, - ) + markup.add(types.InlineKeyboardButton("❓ nometa.xyz", url="https://nometa.xyz/ru.html"), row_width=1) await send_or_reply( message, """Не задавай мета-вопросов вроде: @@ -229,13 +235,8 @@ async def send_neprivet(message: types.Message): ) ) markup = types.InlineKeyboardMarkup() - markup.add( - types.InlineKeyboardButton("👋 Непривет", url="https://neprivet.com/"), - row_width=1, - ) - await send_or_reply( - message, "Пожалуйста, не пишите просто «Привет» в чате.", reply_markup=markup - ) + markup.add(types.InlineKeyboardButton("👋 Непривет", url="https://neprivet.com/"), row_width=1) + await send_or_reply(message, "Пожалуйста, не пишите просто «Привет» в чате.", reply_markup=markup) await bot.delete_message(message.chat.id, message.id) @@ -282,8 +283,7 @@ async def google_it(message: types.Message): ) ) query = f"{detect_args(message)}" - get_query = "https://www.google.com/search?q=" + represent_as_get(message) - + get_query = f"https://www.google.com/search?q={represent_as_get(message)}" markup = types.InlineKeyboardMarkup() markup.add(types.InlineKeyboardButton("🔍 Google Поиск", url=get_query), row_width=1) await send_or_reply(message, f"Ищем «{query}»...", reply_markup=markup) @@ -360,22 +360,9 @@ async def default_query(inline_query): """ [ COUNTER ] """ -@bot.message_handler(content_types=["text", "sticker", "photo", "animation", "video", "audio", "document", ], +@bot.message_handler(content_types=["text", "sticker", "photo", "animation", "video", "audio", "document"], chat_types=["supergroup", "group"]) async def handle_msg(message: types.Message): """Count messages, Stan.""" update_stats(message) - await stan.act(message) - - -""" [ WEBHOOK ] """ - - -@app.post(f"/bot{TOKEN}/") -async def webhook(update: dict): - """Parse POST requests from Telegram.""" - if update: - update = telebot.types.Update.de_json(update) - await bot.process_new_updates([update]) - else: - return + await act(message) diff --git a/src/config.py b/src/config.py new file mode 100644 index 0000000..0e34a9e --- /dev/null +++ b/src/config.py @@ -0,0 +1,26 @@ +import logging +import os +from urllib.error import URLError +from urllib.request import urlopen + +from dotenv import load_dotenv +from telebot import logger + +load_dotenv() + +TOKEN = os.environ.get("LUTZPYBOT") +if not TOKEN: + raise LookupError("LUTZPYBOT (token) has not been found in .env") + +WHITEIDS = {int(i) for i in os.environ.get("whiteids").split(",")} +ROLLBACK = {int(i) for i in os.environ.get("rollback").split(",")} +USE_REMINDER = os.environ.get("use_reminder", "TRUE") == "TRUE" + +RULES_URL = os.environ.get("rules_url", "https://telegra.ph/pythonchatru-07-07") +try: + assert urlopen(RULES_URL).status == 200 +except (URLError, AssertionError): + raise LookupError(f"LUTZPYBOT rules url({RULES_URL}) is not reachable") + +logging.basicConfig(level=logging.DEBUG, format="%(asctime)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S") +logger.setLevel(logging.DEBUG) diff --git a/app/config.py b/src/constants.py similarity index 76% rename from app/config.py rename to src/constants.py index 75976c4..271b4b3 100644 --- a/app/config.py +++ b/src/constants.py @@ -1,40 +1,15 @@ -import os import re -import logging -import telebot.types -from fastapi import FastAPI -from telebot.async_telebot import AsyncTeleBot -from telebot import logger, types -from dotenv import load_dotenv +from telebot import types -logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S") -logger.setLevel(logging.ERROR) -logging.warning("[START] Stan") - -load_dotenv() - -TOKEN = os.environ.get("LUTZPYBOT") -if not TOKEN: - raise LookupError("LUTZPYBOT (token) has not been found in .env") - -bot = AsyncTeleBot( - TOKEN, - "HTML", - disable_web_page_preview=True, - allow_sending_without_reply=True, - colorful_logs=True, -) LOG_TEXT = "[%s] %s: %s" LOG_COMM = "[%s] [%s] %s: %s" -app = FastAPI(docs=None, redoc_url=None) - +LOGGING_LEVEL_DEBUG = 10 +LOGGING_LEVEL_INFO = 20 DATA = "data/chat" - ADMIN_ID = 280887861 # Rykov7 PYTHONCHATRU = -1001338616632 # pythonchatru - URL_RX = re.compile(r"\w+\.\w+/(\+)?\w+") ALLOWED_WORDS = [ "paste", @@ -61,20 +36,13 @@ ALLOWED_WORDS = [ "jetbrains", ] -WHITEIDS = {int(i) for i in os.environ.get("whiteids").split(",")} -ROLLBACK = {int(i) for i in os.environ.get("rollback").split(",")} - RUS = """ёйцукенгшщзхъфывапролджэячсмитьбю.Ё!"№;%:?ЙЦУКЕНГШЩЗХЪ/ФЫВАПРОЛДЖЭЯЧСМИТЬБЮ,""" ENG = """`qwertyuiop[]asdfghjkl;'zxcvbnm,./~!@#$%^&QWERTYUIOP{}|ASDFGHJKL:"ZXCVBNM<>?""" RUS_ENG_TABLE = str.maketrans(RUS, ENG) ENG_RUS_TABLE = str.maketrans(ENG, RUS) -RULES = types.InlineKeyboardButton( - "🟡 Правила чата", url="https://telegra.ph/pythonchatru-07-07" -) -FAQ = types.InlineKeyboardButton( - "🔵 Частые вопросы", url="https://telegra.ph/faq-10-07-4" -) +RULES = types.InlineKeyboardButton("🟡 Правила чата", url="https://telegra.ph/pythonchatru-07-07") +FAQ = types.InlineKeyboardButton("🔵 Частые вопросы", url="https://telegra.ph/faq-10-07-4") LIB = types.InlineKeyboardButton("📚 Книги", url="https://telegra.ph/what-to-read-10-06") SPAM = {"me.sv/", "tg.sv/", "goo.by/", "go.sv/", "intim.video/", "uclck.ru/", 'раб0т', @@ -88,6 +56,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'} + NON_GRATA = {"дудар", "хауди", "dudar"} BAN_WORDS = {"GREEN"} diff --git a/app/filters.py b/src/filters.py similarity index 62% rename from app/filters.py rename to src/filters.py index 432d0e1..6a803a9 100644 --- a/app/filters.py +++ b/src/filters.py @@ -1,20 +1,23 @@ import logging.handlers -from .models import session +from telebot import types + +from .config import WHITEIDS +from .constants import BAN_WORDS, LOG_COMM, SPAM, ALLOWED_WORDS, URL_RX, NON_GRATA, ADMIN_ID from .models import Chat -from .config import * +from .models import session def in_spam_list(message: types.Message) -> bool: """Check for mentioning unwanted persons in text.""" antispam_is_enabled = session.query(Chat.antispam).filter_by(chat_id=message.chat.id).first()[0] - if message.from_user.id not in WHITEIDS and antispam_is_enabled: + from_user_id, title, from_user_name = message.from_user.id, message.chat.title, message.from_user.first_name + if from_user_id not in WHITEIDS and antispam_is_enabled: for text in SPAM: if text.casefold() in message.text.casefold(): - logging.info( - "!BAN!" + LOG_COMM % ( - message.chat.title, message.from_user.id, message.from_user.first_name, message.text)) + logging.info("!BAN!" + LOG_COMM % (title, from_user_id, from_user_name, message.text)) return True + return False def in_caption_spam_list(message: types.Message) -> bool: @@ -24,6 +27,7 @@ def in_caption_spam_list(message: types.Message) -> bool: logging.info("!BAN!" + LOG_COMM % ( message.chat.title, message.from_user.id, message.from_user.first_name, message.video.file_name)) return True + return False def in_not_allowed(word_list, msg): @@ -36,21 +40,17 @@ def in_not_allowed(word_list, msg): def in_delete_list(message: types.Message) -> bool: """Check for URLs in message and delete.""" antispam_is_enabled = session.query(Chat.antispam).filter_by(chat_id=message.chat.id).first()[0] - if message.from_user.id not in WHITEIDS and antispam_is_enabled: - if URL_RX.search(message.text) and in_not_allowed( - ALLOWED_WORDS, message.text - ): - logging.info( - f"[DELETE] [{message.chat.title}] [{message.from_user.id}] {message.from_user.first_name}: {message.text}" - ) + from_user_id, title, from_user_name = message.from_user.id, message.chat.title, message.from_user.first_name + if from_user_id not in WHITEIDS and antispam_is_enabled: + if URL_RX.search(message.text) and in_not_allowed(ALLOWED_WORDS, message.text): + logging.info(f"[DELETE] [{title}] [{from_user_id}] {from_user_name}: {message.text}") return True if message.entities: for entity in message.entities: if entity.url and in_not_allowed(ALLOWED_WORDS, entity.url): - logging.info( - f"[DELETE] [{message.chat.title}] [{message.from_user.id}] {message.from_user.first_name}: [entity] {message.text}" - ) + logging.info(f"[DELETE] [{title}] [{from_user_id}] {from_user_name}: [entity] {message.text}") return True + return False def is_nongrata(type_message: types.Message) -> bool: @@ -58,6 +58,7 @@ def is_nongrata(type_message: types.Message) -> bool: for phrase in NON_GRATA: if phrase in type_message.text.casefold(): return True + return False def is_admin(message: types.Message) -> bool: @@ -66,10 +67,3 @@ def is_admin(message: types.Message) -> bool: def is_white(message: types.Message) -> bool: return message.from_user.id in WHITEIDS - - -async def send_or_reply(m: types.Message, answer, **kwargs): - if m.reply_to_message: - await bot.reply_to(m.reply_to_message, answer, **kwargs) - else: - await bot.send_message(m.chat.id, answer, **kwargs) diff --git a/src/helpers.py b/src/helpers.py new file mode 100644 index 0000000..7a5127b --- /dev/null +++ b/src/helpers.py @@ -0,0 +1,62 @@ +""" Send User's info. """ +import logging +import shelve +from pathlib import Path +from urllib import parse +from urllib.request import urlopen + +from telebot import types + +from .constants import DATA +from .report import reset_report_stats + + +def my_ip(): + ip_service = "https://icanhazip.com" + ip = urlopen(ip_service) + return ip.read().decode("utf-8") + + +def me(message: types.Message): + msg = f"Telegram ID: {message.from_user.id}\n\ + ├ Is bot: {message.from_user.is_bot}\n\ + ├ First name: {message.from_user.first_name}\n" + if message.from_user.last_name: + msg += f" ├ Last name: {message.from_user.last_name}\n" + if message.from_user.username: + msg += f" ├ Username: {message.from_user.username}\n" + msg += f" ├ Language code: {message.from_user.language_code}\n" + if message.from_user.is_premium is True: + msg += f" ├ Is Premium: ⭐️\n" + else: + msg += f" ├ Is Premium: No\n" + msg += f" ├ Chat ID: {message.chat.id}" + return msg + + +def represent_as_get(message: types.Message): + return parse.quote_plus(detect_args(message)) + + +def detect_args(message: types.Message): + if message.reply_to_message and message.reply_to_message.text: + if len(message.text.split()) == 1: + return message.reply_to_message.text + else: + return " ".join(message.text.split()[1:]) + else: + if len(message.text.split()) > 1: + return " ".join(message.text.split()[1:]) + + +def update_stats(message: types.Message): + if not Path(f"{DATA}{message.chat.id}").exists(): + reset_report_stats(message.chat.id) + with shelve.open(f"{DATA}{message.chat.id}", writeback=True) as shelve_db: + if "Messages" not in shelve_db: + reset_report_stats(message.chat.id) + if message.from_user.id not in shelve_db["Messages"]: + shelve_db["Messages"][message.from_user.id] = {"User": message.from_user, "Count": 1} + logging.info(f"[{message.chat.title[:10]}] [{message.from_user.id}] {message.from_user.first_name}") + else: + shelve_db["Messages"][message.from_user.id]["Count"] += 1 diff --git a/app/models.py b/src/models.py similarity index 66% rename from app/models.py rename to src/models.py index 51c6c96..a2a1543 100644 --- a/app/models.py +++ b/src/models.py @@ -1,25 +1,27 @@ -import os +from pathlib import Path + from sqlalchemy import Column, String, Integer, Boolean, ForeignKey +from sqlalchemy import create_engine from sqlalchemy.orm import relationship, declarative_base, sessionmaker -from sqlalchemy import create_engine - Base = declarative_base() -basedir = os.path.abspath(os.path.dirname(__file__)) -engine = create_engine('sqlite:///' + os.path.join(basedir, 'db/db.sqlite')) +root = Path(__file__).parent.parent +data_folder = root / "data" +db_path = data_folder / "db.sqlite" +engine = create_engine(f'sqlite:///{db_path}') + Base.metadata.create_all(engine) - Session = sessionmaker(bind=engine) session = Session() class Permission: - USE_COMMANDS = 1 # Пользоваться обычными командами. - SEND_LINKS = 2 # Может отправлять ссылки без удаления. - MODIFY_QUOTES = 4 # Может добавлять и удалять цитаты. + USE_COMMANDS = 1 # Пользоваться обычными командами. + SEND_LINKS = 2 # Может отправлять ссылки без удаления. + MODIFY_QUOTES = 4 # Может добавлять и удалять цитаты. MODERATE = 8 - ADMIN = 16 # Может банить юзеров и удалять сообщения. + ADMIN = 16 # Может банить юзеров и удалять сообщения. class Quote(Base): diff --git a/app/reminder.py b/src/reminder.py similarity index 75% rename from app/reminder.py rename to src/reminder.py index e09b987..3591abe 100644 --- a/app/reminder.py +++ b/src/reminder.py @@ -1,15 +1,23 @@ """ Schedule holidays. """ -from schedule import repeat, every, get_jobs, run_pending -import time -from threading import Thread -from datetime import datetime as dt import csv -from .config import PYTHONCHATRU, TOKEN -from . import report +import logging +import time +from datetime import datetime as dt +from threading import Thread + +from schedule import repeat, every, get_jobs, run_pending from telebot import TeleBot -# We need synchronous instance of bot here as we use synchronous library 'schedule' in a separate thread. -bot = TeleBot(TOKEN, "HTML", disable_web_page_preview=True, allow_sending_without_reply=True) +from .config import TOKEN, USE_REMINDER +from .constants import PYTHONCHATRU +from .report import create_report_text, reset_report_stats + +if USE_REMINDER: + logging.debug("Starts reminder!!!") + # We need synchronous instance of bot here as we use synchronous library 'schedule' in a separate thread. + bot = TeleBot(TOKEN, "HTML", disable_web_page_preview=True, allow_sending_without_reply=True) +else: + bot = None def scheduler(): @@ -42,10 +50,10 @@ def remind(chat_to_repeat, today): @repeat(every().day.at("06:00"), PYTHONCHATRU) def stat_report(chat_to_repeat): """Everyday group statistic info.""" - rep = report.create_report_text(PYTHONCHATRU) + rep = create_report_text(PYTHONCHATRU) if rep: bot.send_message(chat_to_repeat, rep) - report.reset_report_stats(PYTHONCHATRU) + reset_report_stats(PYTHONCHATRU) def print_get_jobs(): @@ -58,4 +66,5 @@ def print_get_jobs(): return text -Thread(target=scheduler).start() +if USE_REMINDER: + Thread(target=scheduler).start() diff --git a/app/report.py b/src/report.py similarity index 85% rename from app/report.py rename to src/report.py index 9953e10..df17683 100644 --- a/app/report.py +++ b/src/report.py @@ -1,8 +1,10 @@ import html +import os import random import shelve -import os -from .config import DATA, ROLLBACK + +from .config import ROLLBACK +from .constants import DATA def create_report_text(chat_id): @@ -44,7 +46,7 @@ def create_report_text(chat_id): if s["Banned"]: report += f""" Криптоидов поймано: {s['Banned']} - └ [ {s['Banned']*current_spammer_face} ] + └ [ {s['Banned'] * current_spammer_face} ] """ return report else: @@ -52,12 +54,12 @@ def create_report_text(chat_id): def reset_report_stats(chat_id): - with shelve.open(f"{DATA}{chat_id}") as s: - s["Messages"] = {} - s["Banned"] = 0 - s["Deleted"] = 0 + with shelve.open(f"{DATA}{chat_id}") as shelve_db: + shelve_db["Messages"] = {} + shelve_db["Banned"] = 0 + shelve_db["Deleted"] = 0 return f"""Chat ID: {chat_id} -{s['Messages']=} -{s['Banned']=} -{s['Deleted']=}""" +{shelve_db['Messages']=} +{shelve_db['Banned']=} +{shelve_db['Deleted']=}""" diff --git a/src/robot.py b/src/robot.py new file mode 100644 index 0000000..4ba4d95 --- /dev/null +++ b/src/robot.py @@ -0,0 +1,14 @@ +import logging + +from telebot.async_telebot import AsyncTeleBot + +from .config import TOKEN + +logging.warning("[START] Stan") +bot = AsyncTeleBot( + TOKEN, + "HTML", + disable_web_page_preview=True, + allow_sending_without_reply=True, + colorful_logs=True, +) diff --git a/app/rules.py b/src/rules.py similarity index 75% rename from app/rules.py rename to src/rules.py index 6a34b94..8af710b 100644 --- a/app/rules.py +++ b/src/rules.py @@ -1,7 +1,10 @@ -from bs4 import BeautifulSoup from urllib.request import urlopen -html = BeautifulSoup(urlopen("https://telegra.ph/pythonchatru-07-07"), "lxml") +from bs4 import BeautifulSoup + +from .config import RULES_URL + +html = BeautifulSoup(urlopen(RULES_URL), "lxml") def fetch_rule(li): diff --git a/app/stan.py b/src/stan.py similarity index 86% rename from app/stan.py rename to src/stan.py index 87160a4..a184cac 100644 --- a/app/stan.py +++ b/src/stan.py @@ -4,11 +4,14 @@ import html import random from sqlalchemy import delete +from telebot import types -from .config import bot, types from .filters import is_white, is_nongrata -from .models import session from .models import Quote +from .models import session +from .robot import bot + +TYPING_TIMEOUT = 0.13 / 4 # Reading time is quarter of the same text writing time def speak(chance_of, group_id): @@ -20,7 +23,7 @@ def speak(chance_of, group_id): async def send_quote(after_sec, message, quote): """Pretend Reading, pretend Typing, send.""" if message.text: - await asyncio.sleep(len(message.text) * 0.13 / 4) # Reading time is quarter of the same text writing time + await asyncio.sleep(len(message.text) * TYPING_TIMEOUT) await bot.send_chat_action(message.chat.id, action="typing") await asyncio.sleep(after_sec) # Typing time await bot.send_message(message.chat.id, html.escape(quote)) @@ -29,7 +32,7 @@ async def send_quote(after_sec, message, quote): async def act(message: types.Message): quote = speak(50, message.chat.id) if quote: - await send_quote(len(quote) * 0.13, message, quote ) + await send_quote(len(quote) * 0.13, message, quote) @bot.message_handler(func=is_white, commands=["add"]) @@ -39,8 +42,7 @@ async def add_stan_quote(message: types.Message): if quote not in [i[0] for i in session.query(Quote.text).filter(Quote.chat_id == message.chat.id).all()]: session.add(Quote(chat_id=message.chat.id, text=quote.replace("\n", " "))) session.commit() - await bot.send_message(message.chat.id, "➕\n └ " + quote.replace("\n", " "), - parse_mode='Markdown') + await bot.send_message(message.chat.id, "➕\n └ " + quote.replace("\n", " "), parse_mode='Markdown') await bot.delete_message(message.chat.id, message.id) else: await bot.send_message(message.chat.id, f"⛔️ Не добавил, есть токое\n └ {quote}", parse_mode='Markdown') @@ -49,13 +51,11 @@ async def add_stan_quote(message: types.Message): @bot.message_handler(func=is_white, commands=["remove"]) async def remove_stan_quote(message: types.Message): if message.reply_to_message and message.reply_to_message.text: - quote = message.reply_to_message.text already_exist = session.query(Quote.text).filter_by(text=quote, chat_id=message.chat.id).first() if already_exist: session.execute(delete(Quote).filter_by(text=quote, chat_id=message.chat.id)) session.commit() - await bot.send_message(message.chat.id, f"➖ \n └ {quote}", parse_mode='Markdown') await bot.delete_message(message.chat.id, message.id) else: @@ -65,4 +65,4 @@ async def remove_stan_quote(message: types.Message): @bot.message_handler(func=is_nongrata) async def tease_nongrata(message: types.Message): """Reply to non grata mentions.""" - await bot.reply_to(message, f"у нас тут таких не любят") + await bot.reply_to(message, "у нас тут таких не любят") diff --git a/src/webhook.py b/src/webhook.py new file mode 100644 index 0000000..1a7929e --- /dev/null +++ b/src/webhook.py @@ -0,0 +1,20 @@ +import logging + +import telebot +from fastapi import FastAPI + +from .commands import bot +from .config import TOKEN + +logging.warning("[START] Webhook server") +app = FastAPI(docs=None, redoc_url=None) + + +@app.post(f"/bot{TOKEN}/") +async def webhook(update: dict): + """Parse POST requests from Telegram.""" + if update: + update = telebot.types.Update.de_json(update) + await bot.process_new_updates([update]) + else: + return None diff --git a/wsgi.py b/wsgi.py index b334fe5..c8a4b7f 100644 --- a/wsgi.py +++ b/wsgi.py @@ -1 +1,6 @@ -from app.bot import app \ No newline at end of file +import asyncio + +from src.webhook import app, bot + +if __name__ == '__main__': + asyncio.run(bot.infinity_polling()) From 52347c19f780713efb784d6aae7f9e61d8522164 Mon Sep 17 00:00:00 2001 From: kotolex Date: Fri, 22 Dec 2023 22:17:07 +0500 Subject: [PATCH 2/9] Refactoring: step 2 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - независимые функции вынесены в хелперы - добавлены тесты на хелперы и простые команды бота - проверка урла Правил вынесена в хелпер - уровень логирования повышен для бота - добавлено пара кейсов и слов в спам - при проверке на спам идет сработка если кроме русского и препинания есть еще какие-то символы - --- src/admin_commands.py | 14 ++--- src/commands.py | 5 +- src/config.py | 12 ++-- src/constants.py | 3 +- src/filters.py | 47 +++++----------- src/helpers.py | 63 +++++++++++++++------ src/report.py | 17 ++++++ src/stan.py | 10 ++-- tests/test_bot.py | 127 ++++++++++++++++++++++++++++++++++++++++++ tests/test_helpers.py | 73 ++++++++++++++++++++++++ wsgi.py | 7 +-- 11 files changed, 297 insertions(+), 81 deletions(-) create mode 100644 tests/test_bot.py create mode 100644 tests/test_helpers.py diff --git a/src/admin_commands.py b/src/admin_commands.py index 1d86d7b..13daa6c 100644 --- a/src/admin_commands.py +++ b/src/admin_commands.py @@ -4,8 +4,8 @@ from datetime import datetime as dt from telebot import types, logger from .constants import ADMIN_ID, LOGGING_LEVEL_DEBUG, LOGGING_LEVEL_INFO -from .filters import is_white, is_admin -from .helpers import me, my_ip +from .filters import is_white_id +from .helpers import me, my_ip, is_admin from .models import Chat, Quote, session from .reminder import remind, print_get_jobs, create_report_text from .report import reset_report_stats @@ -67,7 +67,7 @@ async def send_stats(message: types.Message): await bot.send_message(message.chat.id, reset_report_stats(message.chat.id)) -@bot.message_handler(func=is_white, commands=["enable_stan"], chat_types=["supergroup", "group"]) +@bot.message_handler(func=is_white_id, commands=["enable_stan"], chat_types=["supergroup", "group"]) 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}") @@ -96,7 +96,7 @@ async def disable_stan(message: types.Message): @bot.message_handler( - func=is_white, + func=is_white_id, commands=["set_antispam_report_reminder", "set_rules"], chat_types=["supergroup", "group"], ) @@ -117,7 +117,7 @@ async def set_antispam_report_reminder(message: types.Message): await bot.send_message(message.chat.id, f"""Настройки обновлены. Проверить: /get_group_info""") -@bot.message_handler(func=is_white, commands=["get_quotes"], chat_types=["supergroup", "group"]) +@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(Quote.chat_id == message.chat.id).all() @@ -130,7 +130,7 @@ async def get_quotes(message: types.Message): await bot.send_message(message.chat.id, f"Цитаты отсутствуют. Подробнее: /get_group_info") -@bot.message_handler(func=is_white, commands=["get_group_info"], chat_types=["supergroup", "group"]) +@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(Chat.chat_id == message.chat.id).first() if group: @@ -151,7 +151,7 @@ ID группы: {group.chat_id} await bot.send_message(message.chat.id, f"Группа не включена. Включить: /enable_stan") -@bot.message_handler(func=is_white, commands=["set_logging_level"], chat_types=["supergroup", "group"]) +@bot.message_handler(func=is_white_id, commands=["set_logging_level"], chat_types=["supergroup", "group"]) async def set_logging_level(message: types.Message): args = message.text.split() if args[-1] == '10': diff --git a/src/commands.py b/src/commands.py index b192469..83ce911 100644 --- a/src/commands.py +++ b/src/commands.py @@ -5,11 +5,12 @@ import shelve from telebot import types from .constants import DATA, LOG_COMM, FAQ, LIB, RULES, RUS, RUS_ENG_TABLE, ENG_RUS_TABLE, PYTHONCHATRU, ZEN -from .filters import in_spam_list, in_caption_spam_list, in_delete_list, is_admin -from .helpers import represent_as_get, detect_args, update_stats +from .filters import in_spam_list, in_caption_spam_list, in_delete_list +from .helpers import represent_as_get, detect_args, is_admin from .admin_commands import bot from .rules import fetch_rule from .stan import act, speak +from .report import update_stats async def send_or_reply(message: types.Message, answer, **kwargs): diff --git a/src/config.py b/src/config.py index 0e34a9e..3fe0bea 100644 --- a/src/config.py +++ b/src/config.py @@ -1,8 +1,6 @@ import logging import os -from urllib.error import URLError -from urllib.request import urlopen - +from .helpers import is_url_reachable from dotenv import load_dotenv from telebot import logger @@ -17,10 +15,8 @@ ROLLBACK = {int(i) for i in os.environ.get("rollback").split(",")} USE_REMINDER = os.environ.get("use_reminder", "TRUE") == "TRUE" RULES_URL = os.environ.get("rules_url", "https://telegra.ph/pythonchatru-07-07") -try: - assert urlopen(RULES_URL).status == 200 -except (URLError, AssertionError): +if not is_url_reachable(RULES_URL): raise LookupError(f"LUTZPYBOT rules url({RULES_URL}) is not reachable") -logging.basicConfig(level=logging.DEBUG, format="%(asctime)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S") -logger.setLevel(logging.DEBUG) +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S") +logger.setLevel(logging.INFO) diff --git a/src/constants.py b/src/constants.py index 271b4b3..7d1c9e5 100644 --- a/src/constants.py +++ b/src/constants.py @@ -55,7 +55,8 @@ SPAM = {"me.sv/", "tg.sv/", "goo.by/", "go.sv/", "intim.video/", "uclck.ru/", ' 'заработка в крипте', 'Веду набор', 'Идёт набор', 'мне в ЛС', 'Доход от', 'Trading', 'P2P', 'Идет набор', 'в лс', 'нaбop', 'комaндy', 'kpuпт', 'бupж', '💰', 'пишитe', '💸', '$ в неделю', 'обучаю с нуля', 'доходы для всех', 'по трейдингу', 'фальшивые рубли', '‼️', - 'зароботку', 'Набuраю', 'цuфр', 'Обучу вас', 'прuб', 'остались курсы', 'Crypt', 'kр', 'рu'} + 'зароботку', 'Набuраю', 'цuфр', 'Обучу вас', 'прuб', 'остались курсы', 'Crypt', 'kр', 'рu', + 'в лc'} NON_GRATA = {"дудар", "хауди", "dudar"} BAN_WORDS = {"GREEN"} diff --git a/src/filters.py b/src/filters.py index 6a803a9..c140498 100644 --- a/src/filters.py +++ b/src/filters.py @@ -3,9 +3,9 @@ import logging.handlers from telebot import types from .config import WHITEIDS -from .constants import BAN_WORDS, LOG_COMM, SPAM, ALLOWED_WORDS, URL_RX, NON_GRATA, ADMIN_ID -from .models import Chat -from .models import session +from .constants import LOG_COMM, ALLOWED_WORDS, URL_RX +from .helpers import is_spam, is_in_not_allowed, is_ban_words_in_caption +from .models import Chat, session def in_spam_list(message: types.Message) -> bool: @@ -13,57 +13,36 @@ def in_spam_list(message: types.Message) -> bool: antispam_is_enabled = session.query(Chat.antispam).filter_by(chat_id=message.chat.id).first()[0] from_user_id, title, from_user_name = message.from_user.id, message.chat.title, message.from_user.first_name if from_user_id not in WHITEIDS and antispam_is_enabled: - for text in SPAM: - if text.casefold() in message.text.casefold(): - logging.info("!BAN!" + LOG_COMM % (title, from_user_id, from_user_name, message.text)) - return True + if is_spam(message.text): + logging.info("!BAN!" + LOG_COMM % (title, from_user_id, from_user_name, message.text)) + return True return False def in_caption_spam_list(message: types.Message) -> bool: """Check for mentioning unwanted words in caption.""" - for phrase in BAN_WORDS: - if message.caption and phrase in message.caption: - logging.info("!BAN!" + LOG_COMM % ( - message.chat.title, message.from_user.id, message.from_user.first_name, message.video.file_name)) - return True + if message.caption and is_ban_words_in_caption(message.caption): + from_user_id, title, from_user_name = message.from_user.id, message.chat.title, message.from_user.first_name + logging.info("!BAN!" + LOG_COMM % (title, from_user_id, from_user_name, message.video.file_name)) + return True return False -def in_not_allowed(word_list, msg): - for word in word_list: - if word in msg.casefold(): - return False - return True - - def in_delete_list(message: types.Message) -> bool: """Check for URLs in message and delete.""" antispam_is_enabled = session.query(Chat.antispam).filter_by(chat_id=message.chat.id).first()[0] from_user_id, title, from_user_name = message.from_user.id, message.chat.title, message.from_user.first_name if from_user_id not in WHITEIDS and antispam_is_enabled: - if URL_RX.search(message.text) and in_not_allowed(ALLOWED_WORDS, message.text): + if URL_RX.search(message.text) and is_in_not_allowed(ALLOWED_WORDS, message.text): logging.info(f"[DELETE] [{title}] [{from_user_id}] {from_user_name}: {message.text}") return True if message.entities: for entity in message.entities: - if entity.url and in_not_allowed(ALLOWED_WORDS, entity.url): + if entity.url and is_in_not_allowed(ALLOWED_WORDS, entity.url): logging.info(f"[DELETE] [{title}] [{from_user_id}] {from_user_name}: [entity] {message.text}") return True return False -def is_nongrata(type_message: types.Message) -> bool: - """Check for non grata persons.""" - for phrase in NON_GRATA: - if phrase in type_message.text.casefold(): - return True - return False - - -def is_admin(message: types.Message) -> bool: - return message.from_user.id == ADMIN_ID - - -def is_white(message: types.Message) -> bool: +def is_white_id(message: types.Message) -> bool: return message.from_user.id in WHITEIDS diff --git a/src/helpers.py b/src/helpers.py index 7a5127b..5e2739a 100644 --- a/src/helpers.py +++ b/src/helpers.py @@ -1,14 +1,54 @@ """ Send User's info. """ -import logging -import shelve -from pathlib import Path + from urllib import parse +from urllib.error import URLError from urllib.request import urlopen from telebot import types -from .constants import DATA -from .report import reset_report_stats +from .constants import SPAM, RUS, BAN_WORDS, NON_GRATA, ADMIN_ID + + +def is_url_reachable(url: str) -> bool: + try: + return urlopen(url).status == 200 + except URLError: + return False + + +def is_spam(message_text: str) -> bool: + if is_mixed(message_text): + return True + return any(text.casefold() in message_text.casefold() for text in SPAM) + + +def is_mixed(text: str) -> bool: + for word in text.split(): + if not all(e in RUS for e in word.strip().lower()): + return True + return False + + +def is_in_not_allowed(word_list: list, msg: str) -> bool: + for word in word_list: + if word in msg.casefold(): + return False + return True + + +def is_ban_words_in_caption(caption: str) -> bool: + return any(ban_word in caption for ban_word in BAN_WORDS) + + +def is_nongrata(type_message: types.Message) -> bool: + for phrase in NON_GRATA: + if phrase in type_message.text.casefold(): + return True + return False + + +def is_admin(message: types.Message) -> bool: + return message.from_user.id == ADMIN_ID def my_ip(): @@ -47,16 +87,3 @@ def detect_args(message: types.Message): else: if len(message.text.split()) > 1: return " ".join(message.text.split()[1:]) - - -def update_stats(message: types.Message): - if not Path(f"{DATA}{message.chat.id}").exists(): - reset_report_stats(message.chat.id) - with shelve.open(f"{DATA}{message.chat.id}", writeback=True) as shelve_db: - if "Messages" not in shelve_db: - reset_report_stats(message.chat.id) - if message.from_user.id not in shelve_db["Messages"]: - shelve_db["Messages"][message.from_user.id] = {"User": message.from_user, "Count": 1} - logging.info(f"[{message.chat.title[:10]}] [{message.from_user.id}] {message.from_user.first_name}") - else: - shelve_db["Messages"][message.from_user.id]["Count"] += 1 diff --git a/src/report.py b/src/report.py index df17683..66a2c59 100644 --- a/src/report.py +++ b/src/report.py @@ -1,7 +1,11 @@ import html +import logging import os import random import shelve +from pathlib import Path + +from telebot import types from .config import ROLLBACK from .constants import DATA @@ -63,3 +67,16 @@ def reset_report_stats(chat_id): {shelve_db['Messages']=} {shelve_db['Banned']=} {shelve_db['Deleted']=}""" + + +def update_stats(message: types.Message): + if not Path(f"{DATA}{message.chat.id}").exists(): + reset_report_stats(message.chat.id) + with shelve.open(f"{DATA}{message.chat.id}", writeback=True) as shelve_db: + if "Messages" not in shelve_db: + reset_report_stats(message.chat.id) + if message.from_user.id not in shelve_db["Messages"]: + shelve_db["Messages"][message.from_user.id] = {"User": message.from_user, "Count": 1} + logging.info(f"[{message.chat.title[:10]}] [{message.from_user.id}] {message.from_user.first_name}") + else: + shelve_db["Messages"][message.from_user.id]["Count"] += 1 diff --git a/src/stan.py b/src/stan.py index a184cac..1414337 100644 --- a/src/stan.py +++ b/src/stan.py @@ -6,9 +6,9 @@ import random from sqlalchemy import delete from telebot import types -from .filters import is_white, is_nongrata -from .models import Quote -from .models import session +from .filters import is_white_id +from .helpers import is_nongrata +from .models import Quote, session from .robot import bot TYPING_TIMEOUT = 0.13 / 4 # Reading time is quarter of the same text writing time @@ -35,7 +35,7 @@ async def act(message: types.Message): await send_quote(len(quote) * 0.13, message, quote) -@bot.message_handler(func=is_white, commands=["add"]) +@bot.message_handler(func=is_white_id, commands=["add"]) 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 @@ -48,7 +48,7 @@ async def add_stan_quote(message: types.Message): await bot.send_message(message.chat.id, f"⛔️ Не добавил, есть токое\n └ {quote}", parse_mode='Markdown') -@bot.message_handler(func=is_white, commands=["remove"]) +@bot.message_handler(func=is_white_id, commands=["remove"]) async def remove_stan_quote(message: types.Message): if message.reply_to_message and message.reply_to_message.text: quote = message.reply_to_message.text diff --git a/tests/test_bot.py b/tests/test_bot.py new file mode 100644 index 0000000..8d8183f --- /dev/null +++ b/tests/test_bot.py @@ -0,0 +1,127 @@ +import logging +from unittest import IsolatedAsyncioTestCase, main + +from telebot import asyncio_helper, types, logger + +from src import helpers + +# нужно не дать делать реальные запросы в тестах, конкретно тут -за Правилами чата +helpers.is_url_reachable = lambda a: True +from src.commands import bot + +# отключаем логи бота, в тестах они не нужны +logger.setLevel(logging.ERROR) + + +def get_update(text, reply_to=None): + params = {'text': text} + chat = types.Chat(id=11, type='private') + user = types.User(id=10, is_bot=False, first_name='Some User') + if reply_to: + params["reply_to_message"] = types.Message(message_id=1, from_user=user, date=None, chat=chat, + content_type='text', options={'text': reply_to}, json_string="") + mess = types.Message(message_id=1, from_user=user, date=None, chat=chat, content_type='text', options=params, + json_string="") + edited_message = None + channel_post = None + edited_channel_post = None + inline_query = None + chosen_inline_result = None + callback_query = None + shipping_query = None + pre_checkout_query = None + poll = None + poll_answer = None + my_chat_member = None + chat_member = None + chat_join_request = None + return types.Update(1001234038283, mess, edited_message, channel_post, edited_channel_post, inline_query, + chosen_inline_result, callback_query, shipping_query, pre_checkout_query, poll, poll_answer, + my_chat_member, chat_member, chat_join_request) + +# сюда будут складываться результаты, перед началом теста очищается +RESULTS = [] + + +async def custom_sender(token, url, method='get', params=None, files=None, **kwargs): + RESULTS.append(params) + result = {"message_id": 1, "date": 1, "chat": {"id": 1, "type": "private"}} + return result + +# Чтобы перехватить запросы к телеге +asyncio_helper._process_request = custom_sender + + +class TestBot(IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.bot = bot + RESULTS.clear() + + async def test_start(self): + await self.bot.process_new_updates([get_update('/start')]) + self.assertEqual(RESULTS[0]['text'], "Начни с прочтения") + + async def test_non_grata(self): + await self.bot.process_new_updates([get_update('как вам дударь?')]) + self.assertEqual(RESULTS[0]['text'], "у нас тут таких не любят") + + async def test_non_grata2(self): + await self.bot.process_new_updates([get_update('смотрю хауди')]) + self.assertEqual(RESULTS[0]['text'], "у нас тут таких не любят") + + async def test_rules(self): + await self.bot.process_new_updates([get_update('/rules')]) + self.assertEqual(RESULTS[0]['text'], "Читай...") + self.assertTrue("https://telegra.ph/pythonchatru-07-07" in RESULTS[0]['reply_markup']) + + async def test_faq(self): + await self.bot.process_new_updates([get_update('/faq')]) + self.assertEqual(RESULTS[0]['text'], "Читай...") + self.assertTrue("https://telegra.ph/faq-10-07-4" in RESULTS[0]['reply_markup']) + + async def test_lib(self): + await self.bot.process_new_updates([get_update('/lib')]) + self.assertEqual(RESULTS[0]['text'], "Читай...") + self.assertTrue("https://telegra.ph/what-to-read-10-06" in RESULTS[0]['reply_markup']) + + async def test_tr(self): + await self.bot.process_new_updates([get_update('/tr', reply_to="ghbdtn")]) + self.assertEqual(RESULTS[0]['text'], "привет") + + async def test_tr2(self): + await self.bot.process_new_updates([get_update('/tr', reply_to="руддщ")]) + self.assertEqual(RESULTS[0]['text'], "hello") + + async def test_tsya(self): + await self.bot.process_new_updates([get_update('/tsya')]) + self.assertEqual(RESULTS[0]['text'], "-тся и -ться в глаголах") + self.assertTrue("https://tsya.ru/" in RESULTS[0]['reply_markup']) + + async def test_nometa(self): + text = """Не задавай мета-вопросов вроде:\n «Можно задать вопрос?»\n «Кто-нибудь пользовался .. ?»\n «Привет, мне нужна помощь по .. !»\n\nПросто спроси сразу! И чем лучше объяснишь проблему, тем вероятнее получишь помощь.""" + await self.bot.process_new_updates([get_update('/nometa', reply_to="кто тут питонист?")]) + self.assertEqual(RESULTS[0]['text'], text) + + async def test_neprivet(self): + await self.bot.process_new_updates([get_update('/neprivet', reply_to="привет всем")]) + self.assertEqual(RESULTS[0]['text'], "Пожалуйста, не пишите просто «Привет» в чате.") + self.assertTrue("https://neprivet.com/" in RESULTS[0]['reply_markup']) + + async def test_nojob(self): + text = """Мы здесь не для того, чтобы за тебя решать задачи.\n\nЗдесь помогают по конкретным вопросам в ТВОЁМ коде, поэтому тебе нужно показать код, который ты написал сам и объяснить где и почему застрял... всё просто. 🤷🏼️""" + await self.bot.process_new_updates([get_update('/nojob', reply_to="есть работа")]) + self.assertEqual(RESULTS[0]['text'], text) + + async def test_nobot(self): + text = """Внимание:\nТелеграм бот не должен быть твоим первым проектом на Python. Пожалуйста, изучи основы Python, работу с модулями, основы веб-технологий, асинхронное программирование и отладку до начала работы с Телеграм ботами. Существует много ресурсов для этого в интернете.""" + await self.bot.process_new_updates([get_update('/nobot', reply_to="как бота написать?")]) + self.assertEqual(RESULTS[0]['text'], text) + + async def test_nogui(self): + text = """Внимание:\nGUI приложение не должно быть твоим первым проектом на Python. Пожалуйста, изучи основы Python, работу с модулями, циклы событий и отладку до начала работы с какими-либо GUI-фреймворками. Существует много ресурсов для этого в интернете.""" + await self.bot.process_new_updates([get_update('/nogui', reply_to="как gui написать?")]) + self.assertEqual(RESULTS[0]['text'], text) + + +if __name__ == '__main__': + main() diff --git a/tests/test_helpers.py b/tests/test_helpers.py new file mode 100644 index 0000000..60da9e9 --- /dev/null +++ b/tests/test_helpers.py @@ -0,0 +1,73 @@ +from unittest import TestCase, main +from unittest.mock import MagicMock +from src.constants import SPAM, BAN_WORDS, ADMIN_ID +from src.helpers import is_spam, is_mixed, is_ban_words_in_caption, is_in_not_allowed, is_nongrata, is_admin + +CASES = ( + "Оформим резидeнтствo ОАЭ без предoплаты за 5000 дирхам. Пoдробнoсти в лc", + "Всем привет Я в поисках людей в kpиптo кoмaнду У нас вы пройдете 6ecплaтноe обученue 1-3 дня Можно совмещать с основной paбoтoй" +) + + +class TestSpam(TestCase): + def test_simple(self): + for spam_word in SPAM: + with self.subTest(f"check spam word({spam_word})"): + self.assertTrue(is_spam(spam_word)) + + def test_case_first(self): + for text in CASES: + with self.subTest(f"check case({text})"): + self.assertTrue(is_spam(text)) + + def test_one(self): + self.assertTrue(is_spam("привeт")) # e английская + + def test_is_mixed(self): + self.assertTrue(is_mixed("привeт")) # e английская + self.assertFalse(is_mixed("привет ребята!")) + + +class TestOthers(TestCase): + def test_ban(self): + for word in BAN_WORDS: + with self.subTest(f"check ban word({word})"): + self.assertTrue(is_ban_words_in_caption(word)) + + def test_allowed(self): + params = ( + (True, [], 'any'), + (True, ['one'], 'two'), + (False, ['one', 'two'], 'two'), + (False, ['one', 'two'], 'Two'), + (False, ['one', 'two'], 'TWO'), + ) + for expected, words, mess in params: + with self.subTest(f"check allowed({words, mess})"): + self.assertEqual(is_in_not_allowed(words, mess), expected) + + def test_non_grata(self): + params = ( + (False, 'any'), + (True, 'дудар'), + (True, 'хауди'), + (True, 'dudar'), + ) + for expected, text in params: + with self.subTest(f"check non-grata({text})"): + m = MagicMock(text=text) + self.assertEqual(is_nongrata(m), expected) + + def test_is_admin(self): + params = ( + (False, 'any'), + (True, ADMIN_ID), + ) + for expected, text in params: + with self.subTest(f"check non-grata({text})"): + m = MagicMock(from_user=MagicMock(id=text)) + self.assertEqual(is_admin(m), expected) + + +if __name__ == '__main__': + main() diff --git a/wsgi.py b/wsgi.py index c8a4b7f..8fc75a2 100644 --- a/wsgi.py +++ b/wsgi.py @@ -1,6 +1 @@ -import asyncio - -from src.webhook import app, bot - -if __name__ == '__main__': - asyncio.run(bot.infinity_polling()) +from src.webhook import app From d54dc6cf04171bf6060ecc393aab97038c2ed275 Mon Sep 17 00:00:00 2001 From: kotolex Date: Fri, 22 Dec 2023 22:21:16 +0500 Subject: [PATCH 3/9] Create python-package.yml --- .github/workflows/python-package.yml | 33 ++++++++++++++++++++++++++++ 1 file changed, 33 insertions(+) create mode 100644 .github/workflows/python-package.yml diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml new file mode 100644 index 0000000..9d641ae --- /dev/null +++ b/.github/workflows/python-package.yml @@ -0,0 +1,33 @@ +# This workflow will install Python dependencies, run tests and lint with a variety of Python versions +# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python + +name: Python package + +on: + push: + branches: [ "master" ] + pull_request: + branches: [ "master" ] + +jobs: + build: + + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ["3.10"] + + steps: + - uses: actions/checkout@v3 + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v3 + with: + python-version: ${{ matrix.python-version }} + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + - name: Test with unittest + run: | + python -m unittest discover -s tests From 86dea6a6e35100801e808e9338103814b38334e1 Mon Sep 17 00:00:00 2001 From: kotolex Date: Fri, 22 Dec 2023 22:25:49 +0500 Subject: [PATCH 4/9] minor: add fake token to tests --- tests/test_bot.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/test_bot.py b/tests/test_bot.py index 8d8183f..34a8ace 100644 --- a/tests/test_bot.py +++ b/tests/test_bot.py @@ -1,8 +1,11 @@ import logging from unittest import IsolatedAsyncioTestCase, main +import os from telebot import asyncio_helper, types, logger +# Надо сделать до импорта бота, иначе упадет по токену +os.environ["LUTZPYBOT"] = "00000:AAAAAAAAAAAA" from src import helpers # нужно не дать делать реальные запросы в тестах, конкретно тут -за Правилами чата From 03d135df9c0558b0485c7fc909c9867b854c298f Mon Sep 17 00:00:00 2001 From: kotolex Date: Fri, 22 Dec 2023 22:28:58 +0500 Subject: [PATCH 5/9] minor: add fake env params to tests --- tests/test_bot.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/test_bot.py b/tests/test_bot.py index 34a8ace..6019f36 100644 --- a/tests/test_bot.py +++ b/tests/test_bot.py @@ -4,8 +4,11 @@ import os from telebot import asyncio_helper, types, logger -# Надо сделать до импорта бота, иначе упадет по токену +# Надо сделать до импорта бота, иначе упадет, так ка нет енва os.environ["LUTZPYBOT"] = "00000:AAAAAAAAAAAA" +os.environ["whiteids"] = "100,200,300" +os.environ["rollback"] = "1,2,3" + from src import helpers # нужно не дать делать реальные запросы в тестах, конкретно тут -за Правилами чата From f874f4754d202d10e827f4dd12a6472b06da723a Mon Sep 17 00:00:00 2001 From: kotolex Date: Fri, 22 Dec 2023 22:32:50 +0500 Subject: [PATCH 6/9] minor: change github workflow --- .github/workflows/python-package.yml | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml index 9d641ae..4e36e6d 100644 --- a/.github/workflows/python-package.yml +++ b/.github/workflows/python-package.yml @@ -1,7 +1,7 @@ -# This workflow will install Python dependencies, run tests and lint with a variety of Python versions +# This workflow will install Python dependencies, run tests and lint with a single version of Python # For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python -name: Python package +name: Python application on: push: @@ -9,21 +9,20 @@ on: pull_request: branches: [ "master" ] +permissions: + contents: read + jobs: build: runs-on: ubuntu-latest - strategy: - fail-fast: false - matrix: - python-version: ["3.10"] steps: - uses: actions/checkout@v3 - - name: Set up Python ${{ matrix.python-version }} + - name: Set up Python 3.10 uses: actions/setup-python@v3 with: - python-version: ${{ matrix.python-version }} + python-version: "3.10" - name: Install dependencies run: | python -m pip install --upgrade pip From af606017c41cae96194c870de5b10517f214a27a Mon Sep 17 00:00:00 2001 From: kotolex Date: Fri, 22 Dec 2023 22:46:20 +0500 Subject: [PATCH 7/9] minor: add fake reminder to test --- tests/test_bot.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/test_bot.py b/tests/test_bot.py index 6019f36..de5f243 100644 --- a/tests/test_bot.py +++ b/tests/test_bot.py @@ -8,6 +8,7 @@ from telebot import asyncio_helper, types, logger os.environ["LUTZPYBOT"] = "00000:AAAAAAAAAAAA" os.environ["whiteids"] = "100,200,300" os.environ["rollback"] = "1,2,3" +os.environ["use_reminder"] = "FALSE" from src import helpers @@ -45,6 +46,7 @@ def get_update(text, reply_to=None): chosen_inline_result, callback_query, shipping_query, pre_checkout_query, poll, poll_answer, my_chat_member, chat_member, chat_join_request) + # сюда будут складываться результаты, перед началом теста очищается RESULTS = [] @@ -54,15 +56,20 @@ async def custom_sender(token, url, method='get', params=None, files=None, **kwa result = {"message_id": 1, "date": 1, "chat": {"id": 1, "type": "private"}} return result + # Чтобы перехватить запросы к телеге asyncio_helper._process_request = custom_sender class TestBot(IsolatedAsyncioTestCase): + def setUp(self) -> None: self.bot = bot RESULTS.clear() + + + async def test_start(self): await self.bot.process_new_updates([get_update('/start')]) self.assertEqual(RESULTS[0]['text'], "Начни с прочтения") From ec1b11088b3781a5031550fe26a997985514d2bb Mon Sep 17 00:00:00 2001 From: kotolex Date: Sat, 23 Dec 2023 16:32:22 +0500 Subject: [PATCH 8/9] Refactoring: step3 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - вся работа с shelve унесена в report - вся работа с БД унесена в моделс (крме команд админа) - ид книги лутца и звука вынесены в константы - исправлен баг в проверке на смешанный алфавит при определении спама, добавлены константы для более точного выявления - написаны новые тесты в том числе на цитаты и действия админов --- src/commands.py | 30 ++---- src/config.py | 4 +- src/constants.py | 7 +- src/filters.py | 7 +- src/helpers.py | 12 ++- src/models.py | 27 +++++ src/report.py | 13 ++- src/stan.py | 19 ++-- tests/test_bot.py | 242 ++++++++++++++++++++++++++++++++++++------ tests/test_helpers.py | 5 + 10 files changed, 288 insertions(+), 78 deletions(-) diff --git a/src/commands.py b/src/commands.py index 83ce911..bdcc949 100644 --- a/src/commands.py +++ b/src/commands.py @@ -1,16 +1,16 @@ import logging import logging.handlers -import shelve from telebot import types -from .constants import DATA, LOG_COMM, FAQ, LIB, RULES, RUS, RUS_ENG_TABLE, ENG_RUS_TABLE, PYTHONCHATRU, ZEN +from .admin_commands import bot +from .constants import (LOG_COMM, FAQ, LIB, RULES, RUS, RUS_ENG_TABLE, ENG_RUS_TABLE, PYTHONCHATRU, ZEN, LUTZ_ID, + BDMTSS_ID) from .filters import in_spam_list, in_caption_spam_list, in_delete_list from .helpers import represent_as_get, detect_args, is_admin -from .admin_commands import bot +from .report import update_stats, increment from .rules import fetch_rule from .stan import act, speak -from .report import update_stats async def send_or_reply(message: types.Message, answer, **kwargs): @@ -29,8 +29,7 @@ async def moderate_messages(message: types.Message): """Ban user and delete their message.""" await bot.delete_message(message.chat.id, message.id) await bot.ban_chat_member(message.chat.id, message.from_user.id) - with shelve.open(f"{DATA}{message.chat.id}") as shelve_db: - shelve_db["Banned"] += 1 + increment(message.chat.id, banned=True) @bot.message_handler(func=in_caption_spam_list, content_types=["video"], chat_types=["supergroup", "group"]) @@ -38,8 +37,7 @@ async def catch_videos(message: types.Message): """Catch offensive videos""" await bot.delete_message(message.chat.id, message.id) await bot.ban_chat_member(message.chat.id, message.from_user.id) - with shelve.open(f"{DATA}{message.chat.id}") as shelve_db: - shelve_db["Banned"] += 1 + increment(message.chat.id, banned=True) @bot.edited_message_handler(func=in_delete_list, chat_types=["supergroup", "group"]) @@ -47,8 +45,7 @@ async def catch_videos(message: types.Message): async def delete_message(message: types.Message): """Delete unwanted message.""" await bot.delete_message(message.chat.id, message.id) - with shelve.open(f"{DATA}{message.chat.id}") as shelve_db: - shelve_db["Deleted"] += 1 + increment(message.chat.id, banned=False) """ [ COMMANDS ] """ @@ -122,11 +119,7 @@ async def send_lutz(message: types.Message): message.text, ) ) - await bot.send_document( - message.chat.id, - document="BQACAgQAAxkBAAPBYsWJG9Ml0fPrnbU9UyzTQiQSuHkAAjkDAAIstCxSkuRbXAlcqeQpBA", - caption="вот, не позорься", - ) + await bot.send_document(message.chat.id, document=LUTZ_ID, caption="вот, не позорься") await bot.delete_message(message.chat.id, message.id) @@ -141,10 +134,7 @@ async def send_bdmtss_audio(message: types.Message): message.text, ) ) - await bot.send_voice( - message.chat.id, - "AwACAgIAAxkBAAIJrWOg2WUvLwrf7ahyJxQHB8_nqllwAAL5JQAC2_IJSbhfQIO5YnVmLAQ", - ) + await bot.send_voice(message.chat.id, BDMTSS_ID) await bot.delete_message(message.chat.id, message.id) @@ -210,7 +200,7 @@ async def send_nometa(message: types.Message): ) ) markup = types.InlineKeyboardMarkup() - markup.add(types.InlineKeyboardButton("❓ nometa.xyz", url="https://nometa.xyz/ru.html"), row_width=1) + markup.add(types.InlineKeyboardButton("❓ nometa.xyz", url="https://nometa.xyz/ru.html"), row_width=1) await send_or_reply( message, """Не задавай мета-вопросов вроде: diff --git a/src/config.py b/src/config.py index 3fe0bea..1ca5646 100644 --- a/src/config.py +++ b/src/config.py @@ -1,9 +1,11 @@ import logging import os -from .helpers import is_url_reachable + from dotenv import load_dotenv from telebot import logger +from .helpers import is_url_reachable + load_dotenv() TOKEN = os.environ.get("LUTZPYBOT") diff --git a/src/constants.py b/src/constants.py index 7d1c9e5..15c30dc 100644 --- a/src/constants.py +++ b/src/constants.py @@ -1,8 +1,8 @@ import re +import string from telebot import types - LOG_TEXT = "[%s] %s: %s" LOG_COMM = "[%s] [%s] %s: %s" LOGGING_LEVEL_DEBUG = 10 @@ -37,6 +37,8 @@ ALLOWED_WORDS = [ ] RUS = """ёйцукенгшщзхъфывапролджэячсмитьбю.Ё!"№;%:?ЙЦУКЕНГШЩЗХЪ/ФЫВАПРОЛДЖЭЯЧСМИТЬБЮ,""" +ONLY_RUS_LETTERS = "ёйцукенгшщзхъфывапролджэячсмитьбю" +RUS_LETTERS_AND_PUNCTUATION = ONLY_RUS_LETTERS + string.punctuation ENG = """`qwertyuiop[]asdfghjkl;'zxcvbnm,./~!@#$%^&QWERTYUIOP{}|ASDFGHJKL:"ZXCVBNM<>?""" RUS_ENG_TABLE = str.maketrans(RUS, ENG) ENG_RUS_TABLE = str.maketrans(ENG, RUS) @@ -78,4 +80,5 @@ ZEN = [ "If the implementation is easy to explain, it may be a good idea.", "Namespaces are one honking great idea — let's do more of those!", ] - +LUTZ_ID = "BQACAgQAAxkBAAPBYsWJG9Ml0fPrnbU9UyzTQiQSuHkAAjkDAAIstCxSkuRbXAlcqeQpBA" +BDMTSS_ID = "AwACAgIAAxkBAAIJrWOg2WUvLwrf7ahyJxQHB8_nqllwAAL5JQAC2_IJSbhfQIO5YnVmLAQ" diff --git a/src/filters.py b/src/filters.py index c140498..87f68bb 100644 --- a/src/filters.py +++ b/src/filters.py @@ -5,12 +5,11 @@ from telebot import types from .config import WHITEIDS from .constants import LOG_COMM, ALLOWED_WORDS, URL_RX from .helpers import is_spam, is_in_not_allowed, is_ban_words_in_caption -from .models import Chat, session +from .models import is_antispam_enabled def in_spam_list(message: types.Message) -> bool: - """Check for mentioning unwanted persons in text.""" - antispam_is_enabled = session.query(Chat.antispam).filter_by(chat_id=message.chat.id).first()[0] + antispam_is_enabled = is_antispam_enabled(message.chat.id) from_user_id, title, from_user_name = message.from_user.id, message.chat.title, message.from_user.first_name if from_user_id not in WHITEIDS and antispam_is_enabled: if is_spam(message.text): @@ -30,7 +29,7 @@ def in_caption_spam_list(message: types.Message) -> bool: def in_delete_list(message: types.Message) -> bool: """Check for URLs in message and delete.""" - antispam_is_enabled = session.query(Chat.antispam).filter_by(chat_id=message.chat.id).first()[0] + antispam_is_enabled = is_antispam_enabled(message.chat.id) from_user_id, title, from_user_name = message.from_user.id, message.chat.title, message.from_user.first_name if from_user_id not in WHITEIDS and antispam_is_enabled: if URL_RX.search(message.text) and is_in_not_allowed(ALLOWED_WORDS, message.text): diff --git a/src/helpers.py b/src/helpers.py index 5e2739a..b8f57e2 100644 --- a/src/helpers.py +++ b/src/helpers.py @@ -6,7 +6,7 @@ from urllib.request import urlopen from telebot import types -from .constants import SPAM, RUS, BAN_WORDS, NON_GRATA, ADMIN_ID +from .constants import SPAM, BAN_WORDS, NON_GRATA, ADMIN_ID, ONLY_RUS_LETTERS, RUS_LETTERS_AND_PUNCTUATION def is_url_reachable(url: str) -> bool: @@ -17,14 +17,22 @@ def is_url_reachable(url: str) -> bool: def is_spam(message_text: str) -> bool: + """ + Проверяет текст на спам - сначала проверка смешанного алфавита, потом на константный набор спам-слов + """ if is_mixed(message_text): return True return any(text.casefold() in message_text.casefold() for text in SPAM) def is_mixed(text: str) -> bool: + """ + Разбивает текст на слова и проверяет на смешанный алфавит, спамеры используют его. Слово допускается только + полностью на русском или русский + знаки пунктуации + """ for word in text.split(): - if not all(e in RUS for e in word.strip().lower()): + word = word.strip().lower() + if any(e in ONLY_RUS_LETTERS for e in word) and not all(e in RUS_LETTERS_AND_PUNCTUATION for e in word): return True return False diff --git a/src/models.py b/src/models.py index a2a1543..50acc92 100644 --- a/src/models.py +++ b/src/models.py @@ -2,6 +2,7 @@ from pathlib import Path from sqlalchemy import Column, String, Integer, Boolean, ForeignKey from sqlalchemy import create_engine +from sqlalchemy import delete from sqlalchemy.orm import relationship, declarative_base, sessionmaker Base = declarative_base() @@ -49,3 +50,29 @@ class Chat(Base): def __repr__(self): return f'Chat({self.chat_id=}, {self.title=})' + + +def is_antispam_enabled(chat_id: int) -> bool: + try: + return session.query(Chat.antispam).filter_by(chat_id=chat_id).first()[0] + except TypeError: + return False + + +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) + + +def delete_quote_in_chat(text: str, chat_id: int): + session.execute(delete(Quote).filter_by(text=text, chat_id=chat_id)) + session.commit() + + +def all_chat_quotes(chat_id: int) -> list: + return session.query(Quote.text).filter(Quote.chat_id == chat_id).all() + + +def add_quote(chat_id: int, text: str): + session.add(Quote(chat_id=chat_id, text=text)) + session.commit() diff --git a/src/report.py b/src/report.py index 66a2c59..e09c571 100644 --- a/src/report.py +++ b/src/report.py @@ -5,10 +5,17 @@ import random import shelve from pathlib import Path -from telebot import types - from .config import ROLLBACK -from .constants import DATA +from .constants import DATA, types + + +def increment(chat_id: int, banned: bool = True): + """ + Увеличивает счетчик банов или удаленных + """ + category = "Banned" if banned else "Deleted" + with shelve.open(f"{DATA}{chat_id}") as shelve_db: + shelve_db[category] += 1 def create_report_text(chat_id): diff --git a/src/stan.py b/src/stan.py index 1414337..42b7e56 100644 --- a/src/stan.py +++ b/src/stan.py @@ -3,21 +3,21 @@ import asyncio import html import random -from sqlalchemy import delete from telebot import types from .filters import is_white_id from .helpers import is_nongrata -from .models import Quote, session +from .models import all_chat_quotes, add_quote, is_quote_in_chat, delete_quote_in_chat from .robot import bot TYPING_TIMEOUT = 0.13 / 4 # Reading time is quarter of the same text writing time -def speak(chance_of, group_id): +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 session.query(Quote.text).filter(Quote.chat_id == group_id).all()]) + return random.choice([i[0] for i in all_chat_quotes(chat_id)]) + return None async def send_quote(after_sec, message, quote): @@ -39,9 +39,8 @@ 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 session.query(Quote.text).filter(Quote.chat_id == message.chat.id).all()]: - session.add(Quote(chat_id=message.chat.id, text=quote.replace("\n", " "))) - session.commit() + if quote not in {i[0] for i in all_chat_quotes(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) else: @@ -52,10 +51,8 @@ async def add_stan_quote(message: types.Message): async def remove_stan_quote(message: types.Message): if message.reply_to_message and message.reply_to_message.text: quote = message.reply_to_message.text - already_exist = session.query(Quote.text).filter_by(text=quote, chat_id=message.chat.id).first() - if already_exist: - session.execute(delete(Quote).filter_by(text=quote, chat_id=message.chat.id)) - session.commit() + if is_quote_in_chat(quote, message.chat.id): + delete_quote_in_chat(quote, message.chat.id) await bot.send_message(message.chat.id, f"➖ \n └ {quote}", parse_mode='Markdown') await bot.delete_message(message.chat.id, message.id) else: diff --git a/tests/test_bot.py b/tests/test_bot.py index de5f243..eb7fd6d 100644 --- a/tests/test_bot.py +++ b/tests/test_bot.py @@ -1,10 +1,11 @@ import logging -from unittest import IsolatedAsyncioTestCase, main +from unittest import IsolatedAsyncioTestCase, main, TestCase import os +from unittest.mock import patch, MagicMock from telebot import asyncio_helper, types, logger -# Надо сделать до импорта бота, иначе упадет, так ка нет енва +# Надо сделать до импорта бота, иначе упадет, так как нет енва в тестах os.environ["LUTZPYBOT"] = "00000:AAAAAAAAAAAA" os.environ["whiteids"] = "100,200,300" os.environ["rollback"] = "1,2,3" @@ -15,20 +16,26 @@ from src import helpers # нужно не дать делать реальные запросы в тестах, конкретно тут -за Правилами чата helpers.is_url_reachable = lambda a: True from src.commands import bot +from src.constants import BDMTSS_ID, LUTZ_ID, ADMIN_ID +from src.filters import is_white_id # отключаем логи бота, в тестах они не нужны logger.setLevel(logging.ERROR) +# сюда будут складываться результаты, перед началом теста очищается +RESULTS = [] -def get_update(text, reply_to=None): + +def get_update(text, reply_to=None, user_id=10): + """Хелпер, генерирующий событие обновления для бота""" params = {'text': text} - chat = types.Chat(id=11, type='private') - user = types.User(id=10, is_bot=False, first_name='Some User') + chat = types.Chat(id=11, type='group') + user = types.User(id=user_id, is_bot=False, first_name='Some User') if reply_to: - params["reply_to_message"] = types.Message(message_id=1, from_user=user, date=None, chat=chat, + params["reply_to_message"] = types.Message(message_id=2, from_user=user, date=None, chat=chat, content_type='text', options={'text': reply_to}, json_string="") mess = types.Message(message_id=1, from_user=user, date=None, chat=chat, content_type='text', options=params, - json_string="") + json_string='') edited_message = None channel_post = None edited_channel_post = None @@ -47,13 +54,10 @@ def get_update(text, reply_to=None): my_chat_member, chat_member, chat_join_request) -# сюда будут складываться результаты, перед началом теста очищается -RESULTS = [] - - async def custom_sender(token, url, method='get', params=None, files=None, **kwargs): - RESULTS.append(params) - result = {"message_id": 1, "date": 1, "chat": {"id": 1, "type": "private"}} + """Замена для тестов, чтобы не слать запросы в телегу""" + RESULTS.append((params, url)) + result = {"message_id": 1000, "date": 1, "chat": {"id": 1000, "type": "group"}} return result @@ -61,79 +65,247 @@ async def custom_sender(token, url, method='get', params=None, files=None, **kwa asyncio_helper._process_request = custom_sender +class TestOther(TestCase): + def test_is_white(self): + params = ( + (False, 1), + (False, 10), + (True, 100), + (True, 200), + (True, 300), + ) + for expected, text in params: + with self.subTest(f"check non-is_white_id({text})"): + m = MagicMock(from_user=MagicMock(id=text)) + self.assertEqual(is_white_id(m), expected) + + class TestBot(IsolatedAsyncioTestCase): def setUp(self) -> None: self.bot = bot RESULTS.clear() - - - async def test_start(self): await self.bot.process_new_updates([get_update('/start')]) - self.assertEqual(RESULTS[0]['text'], "Начни с прочтения") + self.assertEqual(RESULTS[0][0]['text'], "Начни с прочтения") async def test_non_grata(self): await self.bot.process_new_updates([get_update('как вам дударь?')]) - self.assertEqual(RESULTS[0]['text'], "у нас тут таких не любят") + self.assertEqual(RESULTS[0][0]['text'], "у нас тут таких не любят") async def test_non_grata2(self): await self.bot.process_new_updates([get_update('смотрю хауди')]) - self.assertEqual(RESULTS[0]['text'], "у нас тут таких не любят") + self.assertEqual(RESULTS[0][0]['text'], "у нас тут таких не любят") async def test_rules(self): await self.bot.process_new_updates([get_update('/rules')]) - self.assertEqual(RESULTS[0]['text'], "Читай...") - self.assertTrue("https://telegra.ph/pythonchatru-07-07" in RESULTS[0]['reply_markup']) + 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): await self.bot.process_new_updates([get_update('/faq')]) - self.assertEqual(RESULTS[0]['text'], "Читай...") - self.assertTrue("https://telegra.ph/faq-10-07-4" in RESULTS[0]['reply_markup']) + 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): await self.bot.process_new_updates([get_update('/lib')]) - self.assertEqual(RESULTS[0]['text'], "Читай...") - self.assertTrue("https://telegra.ph/what-to-read-10-06" in RESULTS[0]['reply_markup']) + 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): await self.bot.process_new_updates([get_update('/tr', reply_to="ghbdtn")]) - self.assertEqual(RESULTS[0]['text'], "привет") + self.assertEqual(RESULTS[0][0]['text'], "привет") async def test_tr2(self): await self.bot.process_new_updates([get_update('/tr', reply_to="руддщ")]) - self.assertEqual(RESULTS[0]['text'], "hello") + self.assertEqual(RESULTS[0][0]['text'], "hello") async def test_tsya(self): await self.bot.process_new_updates([get_update('/tsya')]) - self.assertEqual(RESULTS[0]['text'], "-тся и -ться в глаголах") - self.assertTrue("https://tsya.ru/" in RESULTS[0]['reply_markup']) + 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Просто спроси сразу! И чем лучше объяснишь проблему, тем вероятнее получишь помощь.""" await self.bot.process_new_updates([get_update('/nometa', reply_to="кто тут питонист?")]) - self.assertEqual(RESULTS[0]['text'], text) + self.assertEqual(RESULTS[0][0]['text'], text) async def test_neprivet(self): await self.bot.process_new_updates([get_update('/neprivet', reply_to="привет всем")]) - self.assertEqual(RESULTS[0]['text'], "Пожалуйста, не пишите просто «Привет» в чате.") - self.assertTrue("https://neprivet.com/" in RESULTS[0]['reply_markup']) + self.assertEqual(RESULTS[0][0]['text'], "Пожалуйста, не пишите просто «Привет» в чате.") + self.assertTrue("https://neprivet.com/" in RESULTS[0][0]['reply_markup']) + + async def test_lutz(self): + 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): + 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): + 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Здесь помогают по конкретным вопросам в ТВОЁМ коде, поэтому тебе нужно показать код, который ты написал сам и объяснить где и почему застрял... всё просто. 🤷🏼️""" await self.bot.process_new_updates([get_update('/nojob', reply_to="есть работа")]) - self.assertEqual(RESULTS[0]['text'], text) + self.assertEqual(RESULTS[0][0]['text'], text) async def test_nobot(self): text = """Внимание:\nТелеграм бот не должен быть твоим первым проектом на Python. Пожалуйста, изучи основы Python, работу с модулями, основы веб-технологий, асинхронное программирование и отладку до начала работы с Телеграм ботами. Существует много ресурсов для этого в интернете.""" await self.bot.process_new_updates([get_update('/nobot', reply_to="как бота написать?")]) - self.assertEqual(RESULTS[0]['text'], text) + self.assertEqual(RESULTS[0][0]['text'], text) async def test_nogui(self): text = """Внимание:\nGUI приложение не должно быть твоим первым проектом на Python. Пожалуйста, изучи основы Python, работу с модулями, циклы событий и отладку до начала работы с какими-либо GUI-фреймворками. Существует много ресурсов для этого в интернете.""" await self.bot.process_new_updates([get_update('/nogui', reply_to="как gui написать?")]) - self.assertEqual(RESULTS[0]['text'], text) + self.assertEqual(RESULTS[0][0]['text'], text) + + async def test_spam_text(self): + # Нужно подставить антиспам вкл. + with patch("src.filters.is_antispam_enabled") as mocked: + # Нужно подменить инкремент, нельзя реально в шелве работать в тестах + with patch("src.commands.increment") as _mocked_shelve: + mocked.return_value = True + await self.bot.process_new_updates([get_update('пишите мне в ЛС, Писать в ЛС')]) + self.assertEqual(RESULTS[0][1], "deleteMessage") + self.assertEqual(RESULTS[1][1], "banChatMember") + self.assertEqual(RESULTS[1][0], {'chat_id': 11, 'user_id': 10, 'until_date': None}) + + async def test_wrong_url(self): + # Нужно подставить антиспам вкл. + with patch("src.filters.is_antispam_enabled") as mocked: + # Нужно подменить инкремент, нельзя реально в шелве работать в тестах + with patch("src.commands.increment") as _mocked_shelve: + mocked.return_value = True + await self.bot.process_new_updates([get_update('тут подробнее https://zvuk.com/artist/61375')]) + self.assertEqual(RESULTS[0][1], "deleteMessage") + self.assertEqual(RESULTS[0][0], {'chat_id': 11, 'message_id': 1}) + + async def test_update_stats(self): + # Нужно подставить антиспам вкл. + with patch("src.filters.is_antispam_enabled") as mocked: + # Нужно подменить инкремент, нельзя реально в шелве работать в тестах + with patch("src.commands.increment") as _mocked_shelve: + # Подменяем обновление статистики, нам важно что оно просто было вызвано + with patch("src.commands.update_stats") as update_stats: + mocked.return_value = True + update_stats.side_effect = lambda x: RESULTS.append(x) + await self.bot.process_new_updates([get_update('просто текст')]) + self.assertEqual(RESULTS[0].text, "просто текст") + + async def test_admin_can_delete(self): + # Нужно подставить антиспам вкл. + with patch("src.filters.is_antispam_enabled") as mocked: + # Нужно подменить инкремент, нельзя реально в шелве работать в тестах + with patch("src.commands.increment") as _mocked_shelve: + # Подменяем обновление статистики, нам важно что оно просто было вызвано + 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('/ddel', reply_to="shit text", user_id=ADMIN_ID)]) + self.assertEqual(RESULTS[0], ({'chat_id': 11, 'message_id': 1}, 'deleteMessage')) + self.assertEqual(RESULTS[1], ({'chat_id': 11, 'message_id': 2}, 'deleteMessage')) + + async def test_user_cant(self): + for command in ('/ddel', '/bban', '/unban_id', '/unban_id 1', '/add', '/remove'): + with self.subTest(f"user cant use {command}"): + # Нужно подставить антиспам вкл. + with patch("src.filters.is_antispam_enabled") as mocked: + # Нужно подменить инкремент, нельзя реально в шелве работать в тестах + with patch("src.commands.increment") as _mocked_shelve: + # Подменяем обновление статистики, нам важно что оно просто было вызвано + with patch("src.commands.update_stats") as update_stats: + # Подменяем speak, чтобы Стен не начал отвечать + with patch("src.stan.speak") as speak: + speak.return_value = None + mocked.return_value = True + update_stats.side_effect = lambda x: None + await self.bot.process_new_updates([get_update(command, reply_to="shit text")]) + self.assertEqual(RESULTS, []) + + async def test_admin_can_ban(self): + # Нужно подставить антиспам вкл. + with patch("src.filters.is_antispam_enabled") as mocked: + # Нужно подменить инкремент, нельзя реально в шелве работать в тестах + with patch("src.commands.increment") as _mocked_shelve: + # Подменяем обновление статистики, нам важно что оно просто было вызвано + 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('/bban', reply_to="shit text", user_id=ADMIN_ID)]) + self.assertEqual(RESULTS[0], ({'chat_id': 11, 'message_id': 1}, 'deleteMessage')) + self.assertEqual(RESULTS[1], ({'chat_id': 11, 'message_id': 2}, 'deleteMessage')) + self.assertEqual(RESULTS[2][1], 'banChatMember') + + async def test_admin_can_unban(self): + # Нужно подставить антиспам вкл. + with patch("src.filters.is_antispam_enabled") as mocked: + # Нужно подменить инкремент, нельзя реально в шелве работать в тестах + with patch("src.commands.increment") as _mocked_shelve: + # Подменяем обновление статистики, нам важно что оно просто было вызвано + 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)]) + 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.commands.update_stats") as update_stats: + mocked.return_value = [] + update_stats.side_effect = lambda x: 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.commands.increment") as _mocked_shelve: + # Подменяем реальное добавление цитаты, нам важно что оно просто было вызвано + with patch("src.stan.add_quote") as add_qoute: + mocked.return_value = [["good text"]] + 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') + + async def test_white_id_can_remove_existing_quote(self): + # Нужно вернуть запрос о наличии цитаты + with patch("src.stan.is_quote_in_chat") as mocked: + # Подменяем удаление цитаты, нам важно что оно просто было вызвано + with patch("src.stan.delete_quote_in_chat") as delete_qoute: + mocked.return_value = [["good text"]] + delete_qoute.side_effect = lambda x, y: None + mocked.return_value = True + 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[1][1], 'deleteMessage') + self.assertEqual(RESULTS[0][0]['text'], '➖ \n └ good text') if __name__ == '__main__': diff --git a/tests/test_helpers.py b/tests/test_helpers.py index 60da9e9..9ad709b 100644 --- a/tests/test_helpers.py +++ b/tests/test_helpers.py @@ -26,6 +26,11 @@ class TestSpam(TestCase): def test_is_mixed(self): self.assertTrue(is_mixed("привeт")) # e английская self.assertFalse(is_mixed("привет ребята!")) + self.assertFalse(is_mixed("hi guys")) + self.assertFalse(is_mixed("hi guys! #%^&%^$ @Вася привет g! п@")) + self.assertFalse(is_mixed("тут подробнее https://hack.comp.com")) + self.assertTrue(is_mixed("hi guys! #%^&%^$ ghb привет g! пQ")) + class TestOthers(TestCase): From a263b28a70ee9c7b0d621a51ec1f3cc3f56c8361 Mon Sep 17 00:00:00 2001 From: kotolex Date: Sat, 23 Dec 2023 16:51:28 +0500 Subject: [PATCH 9/9] =?UTF-8?q?fix=20test:=20=D0=B4=D0=BE=D0=B1=D0=B0?= =?UTF-8?q?=D0=B2=D0=B8=D0=BB=20=D0=BC=D0=BE=D0=BA=D0=B8=20=D0=B0=D0=BD?= =?UTF-8?q?=D1=82=D0=B8=D1=81=D0=BF=D0=B0=D0=BC=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_bot.py | 166 ++++++++++++++++++++++++++++------------------ 1 file changed, 101 insertions(+), 65 deletions(-) diff --git a/tests/test_bot.py b/tests/test_bot.py index eb7fd6d..c6da398 100644 --- a/tests/test_bot.py +++ b/tests/test_bot.py @@ -87,8 +87,11 @@ class TestBot(IsolatedAsyncioTestCase): RESULTS.clear() async def test_start(self): - await self.bot.process_new_updates([get_update('/start')]) - self.assertEqual(RESULTS[0][0]['text'], "Начни с прочтения") + # Нужно пропустить проверку на спам + 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'], "Начни с прочтения") async def test_non_grata(self): await self.bot.process_new_updates([get_update('как вам дударь?')]) @@ -99,96 +102,134 @@ class TestBot(IsolatedAsyncioTestCase): self.assertEqual(RESULTS[0][0]['text'], "у нас тут таких не любят") async def test_rules(self): - 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']) + # Нужно пропустить проверку на спам + 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']) async def test_faq(self): - 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']) + # Нужно пропустить проверку на спам + 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']) async def test_lib(self): - 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']) + # Нужно пропустить проверку на спам + 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']) async def test_tr(self): - await self.bot.process_new_updates([get_update('/tr', reply_to="ghbdtn")]) - self.assertEqual(RESULTS[0][0]['text'], "привет") + # Нужно пропустить проверку на спам + 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'], "привет") async def test_tr2(self): - await self.bot.process_new_updates([get_update('/tr', reply_to="руддщ")]) - self.assertEqual(RESULTS[0][0]['text'], "hello") + # Нужно пропустить проверку на спам + 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") async def test_tsya(self): - 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']) + # Нужно пропустить проверку на спам + 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']) async def test_nometa(self): text = """Не задавай мета-вопросов вроде:\n «Можно задать вопрос?»\n «Кто-нибудь пользовался .. ?»\n «Привет, мне нужна помощь по .. !»\n\nПросто спроси сразу! И чем лучше объяснишь проблему, тем вероятнее получишь помощь.""" - await self.bot.process_new_updates([get_update('/nometa', reply_to="кто тут питонист?")]) - self.assertEqual(RESULTS[0][0]['text'], text) + # Нужно пропустить проверку на спам + 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) async def test_neprivet(self): - 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']) + # Нужно пропустить проверку на спам + 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']) async def test_lutz(self): - 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") + # Нужно пропустить проверку на спам + 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") async def test_bdmtss(self): - 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") + # Нужно пропустить проверку на спам + 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") async def test_google(self): - 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']) + # Нужно пропустить проверку на спам + 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']) async def test_nojob(self): text = """Мы здесь не для того, чтобы за тебя решать задачи.\n\nЗдесь помогают по конкретным вопросам в ТВОЁМ коде, поэтому тебе нужно показать код, который ты написал сам и объяснить где и почему застрял... всё просто. 🤷🏼️""" - await self.bot.process_new_updates([get_update('/nojob', reply_to="есть работа")]) - self.assertEqual(RESULTS[0][0]['text'], text) + # Нужно пропустить проверку на спам + 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) async def test_nobot(self): text = """Внимание:\nТелеграм бот не должен быть твоим первым проектом на Python. Пожалуйста, изучи основы Python, работу с модулями, основы веб-технологий, асинхронное программирование и отладку до начала работы с Телеграм ботами. Существует много ресурсов для этого в интернете.""" - await self.bot.process_new_updates([get_update('/nobot', reply_to="как бота написать?")]) - self.assertEqual(RESULTS[0][0]['text'], text) + # Нужно пропустить проверку на спам + 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) async def test_nogui(self): text = """Внимание:\nGUI приложение не должно быть твоим первым проектом на Python. Пожалуйста, изучи основы Python, работу с модулями, циклы событий и отладку до начала работы с какими-либо GUI-фреймворками. Существует много ресурсов для этого в интернете.""" - await self.bot.process_new_updates([get_update('/nogui', reply_to="как gui написать?")]) - self.assertEqual(RESULTS[0][0]['text'], text) + # Нужно пропустить проверку на спам + 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) async def test_spam_text(self): # Нужно подставить антиспам вкл. with patch("src.filters.is_antispam_enabled") as mocked: - # Нужно подменить инкремент, нельзя реально в шелве работать в тестах - with patch("src.commands.increment") as _mocked_shelve: - mocked.return_value = True - await self.bot.process_new_updates([get_update('пишите мне в ЛС, Писать в ЛС')]) - self.assertEqual(RESULTS[0][1], "deleteMessage") - self.assertEqual(RESULTS[1][1], "banChatMember") - self.assertEqual(RESULTS[1][0], {'chat_id': 11, 'user_id': 10, 'until_date': None}) + mocked.return_value = True + await self.bot.process_new_updates([get_update('пишите мне в ЛС, Писать в ЛС')]) + self.assertEqual(RESULTS[0][1], "deleteMessage") + self.assertEqual(RESULTS[1][1], "banChatMember") + self.assertEqual(RESULTS[1][0], {'chat_id': 11, 'user_id': 10, 'until_date': None}) async def test_wrong_url(self): # Нужно подставить антиспам вкл. with patch("src.filters.is_antispam_enabled") as mocked: - # Нужно подменить инкремент, нельзя реально в шелве работать в тестах - with patch("src.commands.increment") as _mocked_shelve: - mocked.return_value = True - await self.bot.process_new_updates([get_update('тут подробнее https://zvuk.com/artist/61375')]) - self.assertEqual(RESULTS[0][1], "deleteMessage") - self.assertEqual(RESULTS[0][0], {'chat_id': 11, 'message_id': 1}) + mocked.return_value = True + await self.bot.process_new_updates([get_update('тут подробнее https://zvuk.com/artist/61375')]) + self.assertEqual(RESULTS[0][1], "deleteMessage") + self.assertEqual(RESULTS[0][0], {'chat_id': 11, 'message_id': 1}) async def test_update_stats(self): # Нужно подставить антиспам вкл. @@ -205,15 +246,10 @@ class TestBot(IsolatedAsyncioTestCase): async def test_admin_can_delete(self): # Нужно подставить антиспам вкл. with patch("src.filters.is_antispam_enabled") as mocked: - # Нужно подменить инкремент, нельзя реально в шелве работать в тестах - with patch("src.commands.increment") as _mocked_shelve: - # Подменяем обновление статистики, нам важно что оно просто было вызвано - 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('/ddel', reply_to="shit text", user_id=ADMIN_ID)]) - self.assertEqual(RESULTS[0], ({'chat_id': 11, 'message_id': 1}, 'deleteMessage')) - self.assertEqual(RESULTS[1], ({'chat_id': 11, 'message_id': 2}, 'deleteMessage')) + mocked.return_value = True + await self.bot.process_new_updates([get_update('/ddel', reply_to="shit text", user_id=ADMIN_ID)]) + self.assertEqual(RESULTS[0], ({'chat_id': 11, 'message_id': 1}, 'deleteMessage')) + self.assertEqual(RESULTS[1], ({'chat_id': 11, 'message_id': 2}, 'deleteMessage')) async def test_user_cant(self): for command in ('/ddel', '/bban', '/unban_id', '/unban_id 1', '/add', '/remove'): @@ -265,9 +301,9 @@ class TestBot(IsolatedAsyncioTestCase): # Нужно подменить инкремент, нельзя реально в шелве работать в тестах with patch("src.commands.increment") as _mocked_shelve: # Подменяем обновление статистики, нам важно что оно просто было вызвано - with patch("src.commands.update_stats") as update_stats: + with patch("src.stan.add_quote") as aq: mocked.return_value = [] - update_stats.side_effect = lambda x: None + 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')