85 lines
3.0 KiB
Python
85 lines
3.0 KiB
Python
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
from typing import Any
|
||
from unittest.mock import AsyncMock
|
||
|
||
from telegram.constants import ChatType, ParseMode
|
||
from telegram.ext import ApplicationHandlerStop
|
||
|
||
from assistant_bot.auth import authentication_guard
|
||
from assistant_bot.storage import AssistantStorage
|
||
from assistant_bot.telegram_utils import render_markdown_chunks
|
||
|
||
|
||
class AuthenticationGuardTests(unittest.IsolatedAsyncioTestCase):
|
||
def setUp(self) -> None:
|
||
self.temporary_directory = tempfile.TemporaryDirectory()
|
||
self.storage = AssistantStorage(
|
||
Path(self.temporary_directory.name) / "assistant.sqlite3"
|
||
)
|
||
|
||
def tearDown(self) -> None:
|
||
self.temporary_directory.cleanup()
|
||
|
||
def make_update_and_context(self, text: str) -> tuple[Any, Any]:
|
||
message = SimpleNamespace(text=text, reply_text=AsyncMock())
|
||
update = SimpleNamespace(
|
||
effective_user=SimpleNamespace(id=42),
|
||
effective_chat=SimpleNamespace(type=ChatType.PRIVATE),
|
||
effective_message=message,
|
||
inline_query=None,
|
||
)
|
||
context = SimpleNamespace(
|
||
application=SimpleNamespace(
|
||
bot_data={
|
||
"storage": self.storage,
|
||
"assistant_password": "секрет",
|
||
}
|
||
)
|
||
)
|
||
return update, context
|
||
|
||
async def test_correct_password_authorizes_user(self) -> None:
|
||
update, context = self.make_update_and_context("секрет")
|
||
|
||
with self.assertRaises(ApplicationHandlerStop):
|
||
await authentication_guard(update, context)
|
||
|
||
self.assertTrue(self.storage.is_user_authorized(42))
|
||
rendered = render_markdown_chunks(
|
||
"✅ **Пароль принят.** Доступ открыт — повторно вводить его не нужно."
|
||
)[0][0]
|
||
update.effective_message.reply_text.assert_awaited_once_with(
|
||
rendered,
|
||
parse_mode=ParseMode.MARKDOWN_V2,
|
||
)
|
||
|
||
async def test_wrong_password_does_not_authorize_user(self) -> None:
|
||
update, context = self.make_update_and_context("неверно")
|
||
|
||
with self.assertRaises(ApplicationHandlerStop):
|
||
await authentication_guard(update, context)
|
||
|
||
self.assertFalse(self.storage.is_user_authorized(42))
|
||
rendered = render_markdown_chunks(
|
||
"🔐 **Неверный пароль.** Попробуй ещё раз."
|
||
)[0][0]
|
||
update.effective_message.reply_text.assert_awaited_once_with(
|
||
rendered,
|
||
parse_mode=ParseMode.MARKDOWN_V2,
|
||
)
|
||
|
||
async def test_authorized_user_passes_guard(self) -> None:
|
||
self.storage.authorize_user(42)
|
||
update, context = self.make_update_and_context("обычное сообщение")
|
||
|
||
await authentication_guard(update, context)
|
||
|
||
update.effective_message.reply_text.assert_not_awaited()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|