Refactoring: step3

- вся работа с shelve унесена в report
- вся работа с БД унесена в моделс (крме команд админа)
- ид книги лутца и звука вынесены в константы
- исправлен баг в проверке на смешанный алфавит при определении спама, добавлены константы для более точного выявления
- написаны новые тесты в том числе на цитаты и действия админов
This commit is contained in:
kotolex
2023-12-23 16:32:22 +05:00
parent af606017c4
commit ec1b11088b
10 changed files with 288 additions and 78 deletions
+10 -20
View File
@@ -1,16 +1,16 @@
import logging import logging
import logging.handlers import logging.handlers
import shelve
from telebot import types 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 .filters import in_spam_list, in_caption_spam_list, in_delete_list
from .helpers import represent_as_get, detect_args, is_admin 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 .rules import fetch_rule
from .stan import act, speak from .stan import act, speak
from .report import update_stats
async def send_or_reply(message: types.Message, answer, **kwargs): 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.""" """Ban user and delete their message."""
await bot.delete_message(message.chat.id, message.id) await bot.delete_message(message.chat.id, message.id)
await bot.ban_chat_member(message.chat.id, message.from_user.id) await bot.ban_chat_member(message.chat.id, message.from_user.id)
with shelve.open(f"{DATA}{message.chat.id}") as shelve_db: increment(message.chat.id, banned=True)
shelve_db["Banned"] += 1
@bot.message_handler(func=in_caption_spam_list, content_types=["video"], chat_types=["supergroup", "group"]) @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""" """Catch offensive videos"""
await bot.delete_message(message.chat.id, message.id) await bot.delete_message(message.chat.id, message.id)
await bot.ban_chat_member(message.chat.id, message.from_user.id) await bot.ban_chat_member(message.chat.id, message.from_user.id)
with shelve.open(f"{DATA}{message.chat.id}") as shelve_db: increment(message.chat.id, banned=True)
shelve_db["Banned"] += 1
@bot.edited_message_handler(func=in_delete_list, chat_types=["supergroup", "group"]) @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): async def delete_message(message: types.Message):
"""Delete unwanted message.""" """Delete unwanted message."""
await bot.delete_message(message.chat.id, message.id) await bot.delete_message(message.chat.id, message.id)
with shelve.open(f"{DATA}{message.chat.id}") as shelve_db: increment(message.chat.id, banned=False)
shelve_db["Deleted"] += 1
""" [ COMMANDS ] """ """ [ COMMANDS ] """
@@ -122,11 +119,7 @@ async def send_lutz(message: types.Message):
message.text, message.text,
) )
) )
await bot.send_document( await bot.send_document(message.chat.id, document=LUTZ_ID, caption="вот, не позорься")
message.chat.id,
document="BQACAgQAAxkBAAPBYsWJG9Ml0fPrnbU9UyzTQiQSuHkAAjkDAAIstCxSkuRbXAlcqeQpBA",
caption="вот, не позорься",
)
await bot.delete_message(message.chat.id, message.id) await bot.delete_message(message.chat.id, message.id)
@@ -141,10 +134,7 @@ async def send_bdmtss_audio(message: types.Message):
message.text, message.text,
) )
) )
await bot.send_voice( await bot.send_voice(message.chat.id, BDMTSS_ID)
message.chat.id,
"AwACAgIAAxkBAAIJrWOg2WUvLwrf7ahyJxQHB8_nqllwAAL5JQAC2_IJSbhfQIO5YnVmLAQ",
)
await bot.delete_message(message.chat.id, message.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 = 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( await send_or_reply(
message, message,
"""Не задавай мета-вопросов вроде: """Не задавай мета-вопросов вроде:
+3 -1
View File
@@ -1,9 +1,11 @@
import logging import logging
import os import os
from .helpers import is_url_reachable
from dotenv import load_dotenv from dotenv import load_dotenv
from telebot import logger from telebot import logger
from .helpers import is_url_reachable
load_dotenv() load_dotenv()
TOKEN = os.environ.get("LUTZPYBOT") TOKEN = os.environ.get("LUTZPYBOT")
+5 -2
View File
@@ -1,8 +1,8 @@
import re import re
import string
from telebot import types from telebot import types
LOG_TEXT = "[%s] %s: %s" LOG_TEXT = "[%s] %s: %s"
LOG_COMM = "[%s] [%s] %s: %s" LOG_COMM = "[%s] [%s] %s: %s"
LOGGING_LEVEL_DEBUG = 10 LOGGING_LEVEL_DEBUG = 10
@@ -37,6 +37,8 @@ ALLOWED_WORDS = [
] ]
RUS = """ёйцукенгшщзхъфывапролджэячсмитьбю.Ё!"№;%:?ЙЦУКЕНГШЩЗХЪ/ФЫВАПРОЛДЖЭЯЧСМИТЬБЮ,""" RUS = """ёйцукенгшщзхъфывапролджэячсмитьбю.Ё!"№;%:?ЙЦУКЕНГШЩЗХЪ/ФЫВАПРОЛДЖЭЯЧСМИТЬБЮ,"""
ONLY_RUS_LETTERS = "ёйцукенгшщзхъфывапролджэячсмитьбю"
RUS_LETTERS_AND_PUNCTUATION = ONLY_RUS_LETTERS + string.punctuation
ENG = """`qwertyuiop[]asdfghjkl;'zxcvbnm,./~!@#$%^&QWERTYUIOP{}|ASDFGHJKL:"ZXCVBNM<>?""" ENG = """`qwertyuiop[]asdfghjkl;'zxcvbnm,./~!@#$%^&QWERTYUIOP{}|ASDFGHJKL:"ZXCVBNM<>?"""
RUS_ENG_TABLE = str.maketrans(RUS, ENG) RUS_ENG_TABLE = str.maketrans(RUS, ENG)
ENG_RUS_TABLE = str.maketrans(ENG, RUS) 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.", "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!", "Namespaces are one honking great idea — let's do more of those!",
] ]
LUTZ_ID = "BQACAgQAAxkBAAPBYsWJG9Ml0fPrnbU9UyzTQiQSuHkAAjkDAAIstCxSkuRbXAlcqeQpBA"
BDMTSS_ID = "AwACAgIAAxkBAAIJrWOg2WUvLwrf7ahyJxQHB8_nqllwAAL5JQAC2_IJSbhfQIO5YnVmLAQ"
+3 -4
View File
@@ -5,12 +5,11 @@ from telebot import types
from .config import WHITEIDS from .config import WHITEIDS
from .constants import LOG_COMM, ALLOWED_WORDS, URL_RX from .constants import LOG_COMM, ALLOWED_WORDS, URL_RX
from .helpers import is_spam, is_in_not_allowed, is_ban_words_in_caption 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: def in_spam_list(message: types.Message) -> bool:
"""Check for mentioning unwanted persons in text.""" antispam_is_enabled = is_antispam_enabled(message.chat.id)
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 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 from_user_id not in WHITEIDS and antispam_is_enabled:
if is_spam(message.text): 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: def in_delete_list(message: types.Message) -> bool:
"""Check for URLs in message and delete.""" """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 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 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): if URL_RX.search(message.text) and is_in_not_allowed(ALLOWED_WORDS, message.text):
+10 -2
View File
@@ -6,7 +6,7 @@ from urllib.request import urlopen
from telebot import types 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: 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: def is_spam(message_text: str) -> bool:
"""
Проверяет текст на спам - сначала проверка смешанного алфавита, потом на константный набор спам-слов
"""
if is_mixed(message_text): if is_mixed(message_text):
return True return True
return any(text.casefold() in message_text.casefold() for text in SPAM) return any(text.casefold() in message_text.casefold() for text in SPAM)
def is_mixed(text: str) -> bool: def is_mixed(text: str) -> bool:
"""
Разбивает текст на слова и проверяет на смешанный алфавит, спамеры используют его. Слово допускается только
полностью на русском или русский + знаки пунктуации
"""
for word in text.split(): 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 True
return False return False
+27
View File
@@ -2,6 +2,7 @@ from pathlib import Path
from sqlalchemy import Column, String, Integer, Boolean, ForeignKey from sqlalchemy import Column, String, Integer, Boolean, ForeignKey
from sqlalchemy import create_engine from sqlalchemy import create_engine
from sqlalchemy import delete
from sqlalchemy.orm import relationship, declarative_base, sessionmaker from sqlalchemy.orm import relationship, declarative_base, sessionmaker
Base = declarative_base() Base = declarative_base()
@@ -49,3 +50,29 @@ class Chat(Base):
def __repr__(self): def __repr__(self):
return f'Chat({self.chat_id=}, {self.title=})' 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()
+10 -3
View File
@@ -5,10 +5,17 @@ import random
import shelve import shelve
from pathlib import Path from pathlib import Path
from telebot import types
from .config import ROLLBACK 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): def create_report_text(chat_id):
+8 -11
View File
@@ -3,21 +3,21 @@ import asyncio
import html import html
import random import random
from sqlalchemy import delete
from telebot import types from telebot import types
from .filters import is_white_id from .filters import is_white_id
from .helpers import is_nongrata 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 from .robot import bot
TYPING_TIMEOUT = 0.13 / 4 # Reading time is quarter of the same text writing time 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) number = random.randint(0, chance_of)
if number == 0: 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): 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): async def add_stan_quote(message: types.Message):
if message.reply_to_message and message.reply_to_message.text: if message.reply_to_message and message.reply_to_message.text:
quote = 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()]: if quote not in {i[0] for i in all_chat_quotes(message.chat.id)}:
session.add(Quote(chat_id=message.chat.id, text=quote.replace("\n", " "))) add_quote(message.chat.id, 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) await bot.delete_message(message.chat.id, message.id)
else: else:
@@ -52,10 +51,8 @@ async def add_stan_quote(message: types.Message):
async def remove_stan_quote(message: types.Message): async def remove_stan_quote(message: types.Message):
if message.reply_to_message and message.reply_to_message.text: if message.reply_to_message and message.reply_to_message.text:
quote = 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 is_quote_in_chat(quote, message.chat.id):
if already_exist: delete_quote_in_chat(quote, message.chat.id)
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.send_message(message.chat.id, f"➖ \n └ {quote}", parse_mode='Markdown')
await bot.delete_message(message.chat.id, message.id) await bot.delete_message(message.chat.id, message.id)
else: else:
+207 -35
View File
@@ -1,10 +1,11 @@
import logging import logging
from unittest import IsolatedAsyncioTestCase, main from unittest import IsolatedAsyncioTestCase, main, TestCase
import os import os
from unittest.mock import patch, MagicMock
from telebot import asyncio_helper, types, logger from telebot import asyncio_helper, types, logger
# Надо сделать до импорта бота, иначе упадет, так ка нет енва # Надо сделать до импорта бота, иначе упадет, так как нет енва в тестах
os.environ["LUTZPYBOT"] = "00000:AAAAAAAAAAAA" os.environ["LUTZPYBOT"] = "00000:AAAAAAAAAAAA"
os.environ["whiteids"] = "100,200,300" os.environ["whiteids"] = "100,200,300"
os.environ["rollback"] = "1,2,3" os.environ["rollback"] = "1,2,3"
@@ -15,20 +16,26 @@ from src import helpers
# нужно не дать делать реальные запросы в тестах, конкретно тут -за Правилами чата # нужно не дать делать реальные запросы в тестах, конкретно тут -за Правилами чата
helpers.is_url_reachable = lambda a: True helpers.is_url_reachable = lambda a: True
from src.commands import bot 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) 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} params = {'text': text}
chat = types.Chat(id=11, type='private') chat = types.Chat(id=11, type='group')
user = types.User(id=10, is_bot=False, first_name='Some User') user = types.User(id=user_id, is_bot=False, first_name='Some User')
if reply_to: 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="") 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, 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 edited_message = None
channel_post = None channel_post = None
edited_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) my_chat_member, chat_member, chat_join_request)
# сюда будут складываться результаты, перед началом теста очищается
RESULTS = []
async def custom_sender(token, url, method='get', params=None, files=None, **kwargs): 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 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 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): class TestBot(IsolatedAsyncioTestCase):
def setUp(self) -> None: def setUp(self) -> None:
self.bot = bot self.bot = bot
RESULTS.clear() RESULTS.clear()
async def test_start(self): async def test_start(self):
await self.bot.process_new_updates([get_update('/start')]) 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): async def test_non_grata(self):
await self.bot.process_new_updates([get_update('как вам дударь?')]) 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): async def test_non_grata2(self):
await self.bot.process_new_updates([get_update('смотрю хауди')]) await self.bot.process_new_updates([get_update('смотрю хауди')])
self.assertEqual(RESULTS[0]['text'], "у нас тут таких не любят") self.assertEqual(RESULTS[0][0]['text'], "у нас тут таких не любят")
async def test_rules(self): async def test_rules(self):
await self.bot.process_new_updates([get_update('/rules')]) await self.bot.process_new_updates([get_update('/rules')])
self.assertEqual(RESULTS[0]['text'], "Читай...") self.assertEqual(RESULTS[0][0]['text'], "Читай...")
self.assertTrue("https://telegra.ph/pythonchatru-07-07" in RESULTS[0]['reply_markup']) self.assertTrue("https://telegra.ph/pythonchatru-07-07" in RESULTS[0][0]['reply_markup'])
async def test_faq(self): async def test_faq(self):
await self.bot.process_new_updates([get_update('/faq')]) await self.bot.process_new_updates([get_update('/faq')])
self.assertEqual(RESULTS[0]['text'], "Читай...") self.assertEqual(RESULTS[0][0]['text'], "Читай...")
self.assertTrue("https://telegra.ph/faq-10-07-4" in RESULTS[0]['reply_markup']) self.assertTrue("https://telegra.ph/faq-10-07-4" in RESULTS[0][0]['reply_markup'])
async def test_lib(self): async def test_lib(self):
await self.bot.process_new_updates([get_update('/lib')]) await self.bot.process_new_updates([get_update('/lib')])
self.assertEqual(RESULTS[0]['text'], "Читай...") self.assertEqual(RESULTS[0][0]['text'], "Читай...")
self.assertTrue("https://telegra.ph/what-to-read-10-06" in RESULTS[0]['reply_markup']) self.assertTrue("https://telegra.ph/what-to-read-10-06" in RESULTS[0][0]['reply_markup'])
async def test_tr(self): async def test_tr(self):
await self.bot.process_new_updates([get_update('/tr', reply_to="ghbdtn")]) 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): async def test_tr2(self):
await self.bot.process_new_updates([get_update('/tr', reply_to="руддщ")]) 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): async def test_tsya(self):
await self.bot.process_new_updates([get_update('/tsya')]) await self.bot.process_new_updates([get_update('/tsya')])
self.assertEqual(RESULTS[0]['text'], "<i>-тся</i> и <i>-ться</i> в глаголах") self.assertEqual(RESULTS[0][0]['text'], "<i>-тся</i> и <i>-ться</i> в глаголах")
self.assertTrue("https://tsya.ru/" in RESULTS[0]['reply_markup']) self.assertTrue("https://tsya.ru/" in RESULTS[0][0]['reply_markup'])
async def test_nometa(self): async def test_nometa(self):
text = """Не задавай мета-вопросов вроде:\n<i> «Можно задать вопрос?»\n «Кто-нибудь пользовался .. ?»\n «Привет, мне нужна помощь по .. !»</i>\n\nПросто спроси сразу! И чем лучше объяснишь проблему, тем вероятнее получишь помощь.""" text = """Не задавай мета-вопросов вроде:\n<i> «Можно задать вопрос?»\n «Кто-нибудь пользовался .. ?»\n «Привет, мне нужна помощь по .. !»</i>\n\nПросто спроси сразу! И чем лучше объяснишь проблему, тем вероятнее получишь помощь."""
await self.bot.process_new_updates([get_update('/nometa', reply_to="кто тут питонист?")]) 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): async def test_neprivet(self):
await self.bot.process_new_updates([get_update('/neprivet', reply_to="привет всем")]) await self.bot.process_new_updates([get_update('/neprivet', reply_to="привет всем")])
self.assertEqual(RESULTS[0]['text'], "Пожалуйста, не пишите просто «Привет» в чате.") self.assertEqual(RESULTS[0][0]['text'], "Пожалуйста, не пишите просто «Привет» в чате.")
self.assertTrue("https://neprivet.com/" in RESULTS[0]['reply_markup']) 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'], "<i>Ищем «<i>python</i>»...</i>")
self.assertTrue("https://www.google.com/search?q=python" in RESULTS[0][0]['reply_markup'])
async def test_nojob(self): async def test_nojob(self):
text = """Мы здесь не для того, чтобы за тебя решать задачи.\n\nЗдесь помогают по конкретным вопросам в <u>ТВОЁМ</u> коде, поэтому тебе нужно показать код, который ты написал сам и объяснить где и почему застрял... всё просто. 🤷🏼️""" text = """Мы здесь не для того, чтобы за тебя решать задачи.\n\nЗдесь помогают по конкретным вопросам в <u>ТВОЁМ</u> коде, поэтому тебе нужно показать код, который ты написал сам и объяснить где и почему застрял... всё просто. 🤷🏼️"""
await self.bot.process_new_updates([get_update('/nojob', reply_to="есть работа")]) 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): async def test_nobot(self):
text = """<b>Внимание</b>:\nТелеграм бот <i>не должен</i> быть твоим первым проектом на Python. Пожалуйста, изучи <code>основы Python</code>, <code>работу с модулями</code>, <code>основы веб-технологий</code>, <code>асинхронное программирование</code> и <code>отладку</code> до начала работы с Телеграм ботами. Существует много ресурсов для этого в интернете.""" text = """<b>Внимание</b>:\nТелеграм бот <i>не должен</i> быть твоим первым проектом на Python. Пожалуйста, изучи <code>основы Python</code>, <code>работу с модулями</code>, <code>основы веб-технологий</code>, <code>асинхронное программирование</code> и <code>отладку</code> до начала работы с Телеграм ботами. Существует много ресурсов для этого в интернете."""
await self.bot.process_new_updates([get_update('/nobot', reply_to="как бота написать?")]) 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): async def test_nogui(self):
text = """<b>Внимание</b>:\nGUI приложение <i>не должно</i> быть твоим первым проектом на Python. Пожалуйста, изучи <code>основы Python</code>, <code>работу с модулями</code>, <code>циклы событий</code> и <code>отладку</code> до начала работы с какими-либо GUI-фреймворками. Существует много ресурсов для этого в интернете.""" text = """<b>Внимание</b>:\nGUI приложение <i>не должно</i> быть твоим первым проектом на Python. Пожалуйста, изучи <code>основы Python</code>, <code>работу с модулями</code>, <code>циклы событий</code> и <code>отладку</code> до начала работы с какими-либо GUI-фреймворками. Существует много ресурсов для этого в интернете."""
await self.bot.process_new_updates([get_update('/nogui', reply_to="как 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__': if __name__ == '__main__':
+5
View File
@@ -26,6 +26,11 @@ class TestSpam(TestCase):
def test_is_mixed(self): def test_is_mixed(self):
self.assertTrue(is_mixed("привeт")) # e английская self.assertTrue(is_mixed("привeт")) # e английская
self.assertFalse(is_mixed("привет ребята!")) 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): class TestOthers(TestCase):