refactor: organize repository around legal platform
This commit is contained in:
238
tools/telegram-bot/test_bot.py
Normal file
238
tools/telegram-bot/test_bot.py
Normal file
@@ -0,0 +1,238 @@
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
from bot import (
|
||||
APP_VERSION,
|
||||
Config,
|
||||
archive_update,
|
||||
connect,
|
||||
export_markdown,
|
||||
handle_command,
|
||||
next_update_id,
|
||||
)
|
||||
|
||||
|
||||
class SecretaryTest(unittest.TestCase):
|
||||
def test_archives_allowed_topic_and_exports_reply(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
config = Config(
|
||||
token="test",
|
||||
chat_id=-1004242041275,
|
||||
owner_id=7,
|
||||
report_thread_id=37,
|
||||
thread_ids=frozenset({2}),
|
||||
database=Path(directory) / "bot.sqlite3",
|
||||
timezone=ZoneInfo("Asia/Bishkek"),
|
||||
)
|
||||
connection = connect(config.database)
|
||||
message = {
|
||||
"message_id": 12,
|
||||
"message_thread_id": 2,
|
||||
"date": 1_700_000_000,
|
||||
"chat": {"id": config.chat_id},
|
||||
"from": {"id": 7, "first_name": "Айжан"},
|
||||
"text": "Зафиксируем это решение.",
|
||||
"reply_to_message": {"message_id": 11},
|
||||
}
|
||||
|
||||
archived = archive_update(
|
||||
connection, {"update_id": 40, "message": message}, config
|
||||
)
|
||||
markdown = export_markdown(
|
||||
connection, config, thread_id=2, days=None, now=1_700_000_001
|
||||
)
|
||||
|
||||
self.assertEqual(archived, message)
|
||||
self.assertEqual(next_update_id(connection), 41)
|
||||
self.assertIn("Айжан", markdown)
|
||||
self.assertIn("Зафиксируем это решение.", markdown)
|
||||
self.assertIn("Ответ на сообщение: #11", markdown)
|
||||
self.assertIn(f"Акылдаш v{APP_VERSION}", markdown)
|
||||
|
||||
def test_ignores_other_chat_but_advances_offset(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
config = Config(
|
||||
token="test",
|
||||
chat_id=-1,
|
||||
owner_id=7,
|
||||
report_thread_id=37,
|
||||
thread_ids=frozenset({0}),
|
||||
database=Path(directory) / "bot.sqlite3",
|
||||
timezone=ZoneInfo("UTC"),
|
||||
)
|
||||
connection = connect(config.database)
|
||||
update = {
|
||||
"update_id": 5,
|
||||
"message": {
|
||||
"message_id": 1,
|
||||
"date": 1,
|
||||
"chat": {"id": -2},
|
||||
"text": "Не наша группа",
|
||||
},
|
||||
}
|
||||
|
||||
self.assertIsNone(archive_update(connection, update, config))
|
||||
self.assertEqual(next_update_id(connection), 6)
|
||||
self.assertEqual(connection.execute("SELECT COUNT(*) FROM messages").fetchone()[0], 0)
|
||||
|
||||
def test_omits_forum_topic_root_reply_from_export(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
config = Config(
|
||||
token="test",
|
||||
chat_id=-1,
|
||||
owner_id=7,
|
||||
report_thread_id=37,
|
||||
thread_ids=frozenset({2}),
|
||||
database=Path(directory) / "bot.sqlite3",
|
||||
timezone=ZoneInfo("UTC"),
|
||||
)
|
||||
connection = connect(config.database)
|
||||
message = {
|
||||
"message_id": 12,
|
||||
"message_thread_id": 2,
|
||||
"date": 1,
|
||||
"chat": {"id": config.chat_id},
|
||||
"from": {"id": config.owner_id},
|
||||
"text": "Сообщение темы",
|
||||
"reply_to_message": {"message_id": 2},
|
||||
}
|
||||
|
||||
archive_update(connection, {"update_id": 1, "message": message}, config)
|
||||
markdown = export_markdown(connection, config, 2, None)
|
||||
|
||||
self.assertNotIn("Ответ на сообщение: #2", markdown)
|
||||
|
||||
def test_rejects_export_from_non_owner(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
config = Config(
|
||||
token="test",
|
||||
chat_id=-1,
|
||||
owner_id=7,
|
||||
report_thread_id=37,
|
||||
thread_ids=frozenset({0}),
|
||||
database=Path(directory) / "bot.sqlite3",
|
||||
timezone=ZoneInfo("UTC"),
|
||||
)
|
||||
connection = connect(config.database)
|
||||
message = {
|
||||
"message_id": 1,
|
||||
"date": 1,
|
||||
"chat": {"id": config.chat_id},
|
||||
"from": {"id": 8},
|
||||
"text": "/export все",
|
||||
}
|
||||
|
||||
with patch("bot.send_text") as send_text, patch(
|
||||
"bot.send_document"
|
||||
) as send_document:
|
||||
handle_command(connection, config, message)
|
||||
|
||||
send_text.assert_called_once_with(
|
||||
config, message, "Экспорт доступен только владельцу бота."
|
||||
)
|
||||
send_document.assert_not_called()
|
||||
|
||||
def test_sends_report_to_reports_topic_and_marks_discussion(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
config = Config(
|
||||
token="test",
|
||||
chat_id=-100123,
|
||||
owner_id=7,
|
||||
report_thread_id=37,
|
||||
thread_ids=frozenset({2}),
|
||||
database=Path(directory) / "bot.sqlite3",
|
||||
timezone=ZoneInfo("UTC"),
|
||||
)
|
||||
connection = connect(config.database)
|
||||
message = {
|
||||
"message_id": 12,
|
||||
"message_thread_id": 2,
|
||||
"date": 1,
|
||||
"chat": {"id": config.chat_id},
|
||||
"from": {"id": config.owner_id},
|
||||
"text": "/export",
|
||||
}
|
||||
|
||||
with patch("bot.send_text") as send_text, patch(
|
||||
"bot.send_document", return_value={"message_id": 99}
|
||||
) as send_document:
|
||||
handle_command(connection, config, message)
|
||||
|
||||
document = send_document.call_args.args
|
||||
self.assertEqual(document[1], config.report_thread_id)
|
||||
self.assertIn("#report_12", document[4])
|
||||
self.assertIn("https://t.me/c/123/12", document[4])
|
||||
marker = send_text.call_args.args[2]
|
||||
self.assertIn("ОБСУЖДЕНИЕ ЗАВЕРШЕНО", marker)
|
||||
self.assertIn("#report_12", marker)
|
||||
self.assertIn("https://t.me/c/123/99", marker)
|
||||
|
||||
def test_next_export_starts_after_previous_cutoff(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
config = Config(
|
||||
token="test",
|
||||
chat_id=-100123,
|
||||
owner_id=7,
|
||||
report_thread_id=37,
|
||||
thread_ids=frozenset({2}),
|
||||
database=Path(directory) / "bot.sqlite3",
|
||||
timezone=ZoneInfo("UTC"),
|
||||
)
|
||||
connection = connect(config.database)
|
||||
|
||||
def archive(message_id, text):
|
||||
archive_update(
|
||||
connection,
|
||||
{
|
||||
"update_id": message_id,
|
||||
"message": {
|
||||
"message_id": message_id,
|
||||
"message_thread_id": 2,
|
||||
"date": message_id,
|
||||
"chat": {"id": config.chat_id},
|
||||
"from": {"id": config.owner_id},
|
||||
"text": text,
|
||||
},
|
||||
},
|
||||
config,
|
||||
)
|
||||
|
||||
archive(10, "Первое обсуждение")
|
||||
with patch("bot.send_text"), patch(
|
||||
"bot.send_document",
|
||||
side_effect=[{"message_id": 90}, {"message_id": 91}],
|
||||
) as send_document:
|
||||
handle_command(
|
||||
connection,
|
||||
config,
|
||||
{
|
||||
"message_id": 12,
|
||||
"message_thread_id": 2,
|
||||
"from": {"id": config.owner_id},
|
||||
"text": "/export",
|
||||
},
|
||||
)
|
||||
archive(13, "Второе обсуждение")
|
||||
handle_command(
|
||||
connection,
|
||||
config,
|
||||
{
|
||||
"message_id": 14,
|
||||
"message_thread_id": 2,
|
||||
"from": {"id": config.owner_id},
|
||||
"text": "/export",
|
||||
},
|
||||
)
|
||||
|
||||
first_export = send_document.call_args_list[0].args[3].decode()
|
||||
second_export = send_document.call_args_list[1].args[3].decode()
|
||||
self.assertIn("Первое обсуждение", first_export)
|
||||
self.assertNotIn("Первое обсуждение", second_export)
|
||||
self.assertIn("Второе обсуждение", second_export)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user