61 lines
2.6 KiB
Python
61 lines
2.6 KiB
Python
from contextlib import closing
|
|
from pathlib import Path
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import dbutil
|
|
from room import Room
|
|
|
|
|
|
class Event(SimpleNamespace):
|
|
pass
|
|
|
|
|
|
class Response(SimpleNamespace):
|
|
pass
|
|
|
|
|
|
class GetVotesTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_both_commands_report_current_emoji_votes_without_consuming_titles(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
room = Room.__new__(Room)
|
|
room.db_path = str(Path(directory) / "suggestions.db")
|
|
room.init_db()
|
|
with closing(dbutil.connect(room.db_path)) as conn, conn:
|
|
conn.execute("INSERT INTO suggestions (user, display_name, message, timestamp, age) "
|
|
"VALUES ('alice', 'Alice', 'Pending title', 100, 0)")
|
|
room.config = SimpleNamespace(room_id="test", command_sync_timeout=1)
|
|
room.user_id = "bot"
|
|
room.count_votes = AsyncMock(side_effect=[
|
|
"Suggestion 1: 1 vote", "Suggestion 1: 2 votes",
|
|
])
|
|
room.alert_post = AsyncMock()
|
|
events = [
|
|
Event(body=body, sender="alice", event_id=event_id, server_timestamp=timestamp)
|
|
for body, event_id, timestamp in [
|
|
("!countvotes", "$count", 100000),
|
|
("!countvotes", "$count", 100000), # replay ignored
|
|
("!getvotes", "$get", 101000),
|
|
("!gettitles", "$titles", 102000),
|
|
]
|
|
]
|
|
room.client = SimpleNamespace(sync=AsyncMock(return_value=Response(
|
|
rooms=SimpleNamespace(join={"test": SimpleNamespace(
|
|
timeline=SimpleNamespace(events=events)
|
|
)})
|
|
)))
|
|
with patch("room.RoomMessageText", Event), patch("room.SyncResponse", Response):
|
|
self.assertEqual(await room.wait_for_commands(since_ms=0), "gettitles")
|
|
self.assertEqual(room.count_votes.await_count, 2)
|
|
messages = [call.args[0] for call in room.alert_post.await_args_list]
|
|
self.assertEqual(messages, [
|
|
"Counting votes... This may take a moment...", "Suggestion 1: 1 vote",
|
|
"Counting votes... This may take a moment...", "Suggestion 1: 2 votes",
|
|
])
|
|
room.read_db(only_unposted=True)
|
|
self.assertEqual([post.content for post in room.posts], ["Pending title"])
|
|
self.assertEqual(sorted(path.name for path in Path(directory).iterdir()),
|
|
["suggestions.db"])
|