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