70 lines
2.9 KiB
Python
70 lines
2.9 KiB
Python
from contextlib import closing
|
|
from pathlib import Path
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import AsyncMock
|
|
|
|
from nio import RoomSendResponse
|
|
|
|
import dbutil
|
|
from post import Post
|
|
from room import Room
|
|
|
|
|
|
class PostSuggestionsTests(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.room = Room.__new__(Room)
|
|
self.room.db_path = str(Path(self.temp.name) / "suggestions.db")
|
|
self.room.init_db()
|
|
self.room.config = SimpleNamespace(room_id="test", post_interval=0)
|
|
self.room.client = SimpleNamespace(room_send=AsyncMock())
|
|
self.room.alert_post = AsyncMock()
|
|
self.room.is_banned = lambda user: False
|
|
|
|
def suggestions(self, *titles):
|
|
self.room.posts = [Post("alice", "Alice", title, i, age=0)
|
|
for i, title in enumerate(titles, start=1)]
|
|
self.room.write_to_db()
|
|
|
|
def sent(self, event_id):
|
|
return RoomSendResponse.from_dict({"event_id": event_id}, "test")
|
|
|
|
async def test_posting_consumes_queue_and_repeat_gettitles_does_not_restart_vote(self):
|
|
self.suggestions("Title")
|
|
self.room.client.room_send.side_effect = [self.sent("$title"), self.sent("$reaction")]
|
|
await self.room.post_suggestions_individually()
|
|
banner = self.room.alert_post.await_args.args[0]
|
|
self.assertIn("!countvotes", banner)
|
|
self.assertIn("!getvotes", banner)
|
|
self.assertNotIn("!voteurl", banner)
|
|
self.assertNotIn("HTTP VOTING", banner)
|
|
self.room.read_db(only_unposted=True)
|
|
self.assertEqual(self.room.posts, [])
|
|
await self.room.post_suggestions_individually()
|
|
self.room.alert_post.assert_awaited_with("No new title suggestions posted.")
|
|
self.assertEqual(self.room.client.room_send.await_count, 2)
|
|
|
|
async def test_failed_title_send_stays_pending(self):
|
|
self.suggestions("Posted", "Retry me")
|
|
self.room.client.room_send.side_effect = [self.sent("$title"), self.sent("$reaction"), object()]
|
|
await self.room.post_suggestions_individually()
|
|
self.room.read_db(only_unposted=True)
|
|
self.assertEqual([post.content for post in self.room.posts], ["Retry me"])
|
|
with closing(dbutil.connect(self.room.db_path)) as conn:
|
|
self.assertEqual(conn.execute("SELECT COUNT(*) FROM suggestions").fetchone()[0], 2)
|
|
|
|
async def test_reaction_failure_does_not_repost_successful_title(self):
|
|
self.suggestions("Title")
|
|
self.room.client.room_send.side_effect = [self.sent("$title"), RuntimeError("Reaction failed")]
|
|
with self.assertRaises(RuntimeError):
|
|
await self.room.post_suggestions_individually()
|
|
self.room.read_db(only_unposted=True)
|
|
self.assertEqual(self.room.posts, [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|