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):