-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: add telegram bots healthcheck (#42)
* feat: add telegram bots healthcheck * chore: add tests
- Loading branch information
1 parent
2349c38
commit 3a2c242
Showing
8 changed files
with
195 additions
and
10 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,2 +1,8 @@ | ||
bot_token: "" | ||
chat_ids: [] | ||
token: "" | ||
chat_ids: [] | ||
api_id: "" | ||
api_hash: "" | ||
bots_to_check: | ||
- username: "" | ||
command: "" | ||
expected_response: "" |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,3 +1,4 @@ | ||
httpx == 0.23.0 | ||
httpx[http2] == 0.23.0 | ||
pyyaml == 6.0.1 | ||
pyrogram == 2.0.106 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,65 @@ | ||
from pyrogram import Client, filters | ||
from pyrogram.errors import RPCError | ||
from pyrogram.types import Message | ||
from os.path import exists | ||
from asyncio import sleep | ||
|
||
config_map = None | ||
app = Client('healthcheck_user_client') | ||
received_answer = [] | ||
|
||
@app.on_message(filters.text & filters.bot) | ||
async def check_welcome(client: Client, message: Message) -> None: | ||
global received_answer | ||
|
||
for bots_to_check in config_map['bots_to_check']: | ||
if message.from_user.username == bots_to_check['username']: | ||
if message.text == bots_to_check['expected_response']: | ||
received_answer.insert(0, message.from_user.username) | ||
return | ||
|
||
|
||
def bot_checker(config: dict) -> None: | ||
global config_map | ||
config_map = config | ||
|
||
# No bots to check | ||
if 'bots_to_check' not in config_map or not config_map['bots_to_check']: | ||
return | ||
|
||
# Checking if already exists a session | ||
if not exists('healthcheck_user_client.session'): | ||
# We need to check that all the needed parameters are available | ||
for key in ['api_id', 'api_hash']: | ||
if key not in config_map: | ||
print(f'Missing required parameter: {key}') | ||
return | ||
|
||
app.api_id=config_map['api_id'] | ||
app.api_hash=config_map['api_hash'] | ||
|
||
print('Starting login, after the client is connected you can exit with CTRL+C') | ||
|
||
app.run(main()) | ||
|
||
|
||
async def main() -> None: | ||
for bot in config_map['bots_to_check']: | ||
if not all(key in bot for key in ['username', 'command', 'expected_response']): | ||
print(f'Skipping {bot["username"] if "id" in bot else "a bot without a specified username c:"}') | ||
continue | ||
|
||
try: | ||
await app.send_message(chat_id=bot['username'], text=bot['command']) | ||
except RPCError as e: | ||
print(f'There was a problem communicating with {bot["username"]}:', e) | ||
|
||
await sleep(10) # Just to be sure the bot is not busy | ||
# To prevent circular imports | ||
# TODO: in a future refactor we could split the main code in another file only for webpages, | ||
# would improve modularity and reuse of code with an helper file | ||
from src.main import make_request_to_telegram | ||
for bot in config_map['bots_to_check']: | ||
if bot['username'] not in received_answer: | ||
for user_to_notify in config_map['chat_ids']: | ||
await make_request_to_telegram(f'@{bot["username"]}', 'Telegram', user_to_notify) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,82 @@ | ||
import pytest | ||
from pytest_mock import MockerFixture | ||
from unittest.mock import AsyncMock | ||
import src.telegram_checker as tg | ||
import src.main as main | ||
from pyrogram.client import Client | ||
import asyncio | ||
|
||
tests = [ | ||
{ | ||
'func': tg.bot_checker, | ||
'expected_res': None, | ||
'arg': ({'bots_to_check': [{'username':'WebpageBot', 'command':'/start', 'expected_response':'Hello, I\'m the Webpage Bot!\nPlease send me up to 10 links and I will update previews for them.'}]},), | ||
'is_async': False | ||
}, | ||
{ | ||
'func': tg.bot_checker, | ||
'expected_res': None, | ||
'arg': ({},), | ||
'is_async': False | ||
}, | ||
{ | ||
'func': tg.bot_checker, | ||
'expected_res': None, | ||
'arg': ({'api_id':'12345', 'api_hash':'0123456789abcdef0123456789abcdef','bots_to_check': [{'username':'WebpageBot', 'command':'/start', 'expected_response':'Hello, I\'m the Webpage Bot!\nPlease send me up to 10 links and I will update previews for them.'}]},), | ||
'mock_obj': [Client], | ||
'mock_func': ['run'], | ||
'mock_ret': [None], | ||
'is_async': False | ||
}, | ||
{ | ||
'func': tg.main, | ||
'expected_res': None, | ||
'arg': tuple(), | ||
'mock_obj': [main, Client], | ||
'mock_func': ['make_request_to_telegram', 'send_message'], | ||
'mock_ret': [None, None], | ||
'mock_dict': {'bots_to_check': [{'username':'WebpageBot', 'command':'/start', 'expected_response':'Hello, I\'m the Webpage Bot!\nPlease send me up to 10 links and I will update previews for them.'}]}, | ||
'is_async': True | ||
}, | ||
{ | ||
'func': tg.main, | ||
'expected_res': None, | ||
'arg': tuple(), | ||
'mock_obj': [main, Client], | ||
'mock_func': ['make_request_to_telegram', 'send_message'], | ||
'mock_ret': [None, None], | ||
'mock_dict': {'bots_to_check': [{'command':'/start'}]}, | ||
'is_async': True | ||
}, | ||
{ | ||
'func': tg.check_welcome, | ||
'expected_res': None, | ||
'arg': (None, {'from_user': {'username': 'username'}, 'text': 'A response'},), | ||
'mock_dict': {'bots_to_check': [{'username':'WebpageBot', 'command':'/start', 'expected_response':'Hello, I\'m the Webpage Bot!\nPlease send me up to 10 links and I will update previews for them.'}]}, | ||
'is_async': True | ||
} | ||
] | ||
|
||
@pytest.mark.asyncio | ||
@pytest.mark.parametrize('test', tests) | ||
async def test_generic(mocker: MockerFixture, test: dict) -> None: | ||
spyed_objs = [] | ||
spyed_dicts = [] | ||
|
||
if test.get('mock_obj'): | ||
for index, obj in enumerate(test['mock_obj']): | ||
mocker.patch.object(obj, test['mock_func'][index], return_value=test['mock_ret'][index]) | ||
spyed_objs.append(mocker.spy(obj, test['mock_func'][index])) | ||
|
||
if test.get('mock_dict'): | ||
mocker.patch.dict(tg.config_map, test['mock_dict']) | ||
spyed_dicts.append(mocker.spy(tg, 'config_map')) | ||
|
||
if test.get('is_async'): | ||
res = await test['func'](*test['arg']) | ||
else: | ||
res = test['func'](*test['arg']) | ||
|
||
assert res == test['expected_res'] | ||
for index, spy in enumerate(spyed_objs): | ||
assert spy.spy_return == test['mock_ret'][index] |