109 lines
4.7 KiB
Python
109 lines
4.7 KiB
Python
import asyncio
|
|
from contextlib import closing
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import dbutil
|
|
import post_results
|
|
from room import Room
|
|
|
|
|
|
class ClearSuggestionsTests(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()
|
|
with closing(dbutil.connect(self.room.db_path)) as conn, conn:
|
|
conn.executemany(
|
|
"INSERT INTO suggestions (message, is_posted) VALUES (?, ?)",
|
|
[("pending one", 0), ("pending two", 0), ("already posted", 1)],
|
|
)
|
|
|
|
def test_clear_preserves_rows_and_excludes_pending_from_voting(self):
|
|
self.assertEqual(self.room.clear_suggestions("alice", 100, "$clear"), 2)
|
|
self.room.read_db(only_unposted=True)
|
|
self.assertEqual(self.room.posts, [])
|
|
with closing(dbutil.connect(self.room.db_path)) as conn, conn:
|
|
self.assertEqual(conn.execute("SELECT COUNT(*) FROM suggestions").fetchone()[0], 3)
|
|
conn.execute("INSERT INTO suggestions (message) VALUES ('new title')")
|
|
self.assertIsNone(self.room.clear_suggestions("alice", 100, "$clear"))
|
|
with closing(dbutil.connect(self.room.db_path)) as conn, conn:
|
|
self.assertEqual(conn.execute("SELECT COUNT(*) FROM suggestions WHERE is_posted = 0").fetchone()[0], 1)
|
|
self.assertEqual(self.room.clear_suggestions("alice", 101, "$next"), 1)
|
|
self.assertEqual(self.room.clear_suggestions("alice", 102, "$empty"), 0)
|
|
|
|
async def test_command_confirms_clear_and_keeps_waiting_for_gettitles(self):
|
|
class Event(SimpleNamespace):
|
|
pass
|
|
|
|
class Response(SimpleNamespace):
|
|
pass
|
|
|
|
def response(body, event_id, timestamp):
|
|
event = Event(body=body, sender="alice", event_id=event_id, server_timestamp=timestamp)
|
|
return Response(rooms=SimpleNamespace(join={
|
|
"test": SimpleNamespace(timeline=SimpleNamespace(events=[event]))
|
|
}))
|
|
|
|
self.room.config = SimpleNamespace(room_id="test", command_sync_timeout=1)
|
|
self.room.user_id = "bot"
|
|
self.room.client = SimpleNamespace(sync=AsyncMock(side_effect=[
|
|
response("!clear", "$clear", 100000),
|
|
response("!clear", "$clear", 100000),
|
|
response("!gettitles", "$titles", 101000),
|
|
]))
|
|
self.room.alert_post = AsyncMock()
|
|
with patch("room.RoomMessageText", Event), patch("room.SyncResponse", Response):
|
|
self.assertEqual(await self.room.wait_for_commands(since_ms=0), "gettitles")
|
|
self.room.alert_post.assert_awaited_once_with(
|
|
"Cleared 2 pending title suggestions. They will not be posted for voting."
|
|
)
|
|
|
|
async def test_clear_then_gettitles_in_same_batch_reports_no_titles(self):
|
|
class Event(SimpleNamespace):
|
|
pass
|
|
|
|
class Response(SimpleNamespace):
|
|
pass
|
|
|
|
events = [
|
|
Event(body="!clear", sender="alice", event_id="$clear", server_timestamp=100000),
|
|
Event(body="!gettitles", sender="alice", event_id="$titles", server_timestamp=101000),
|
|
]
|
|
response = Response(rooms=SimpleNamespace(join={
|
|
"test": SimpleNamespace(timeline=SimpleNamespace(events=events))
|
|
}))
|
|
self.room.config = SimpleNamespace(
|
|
room_id="test", command_sync_timeout=1, initial_sync_timeout=1, post_loop_interval=1
|
|
)
|
|
self.room.user_id = "bot"
|
|
self.room.client = SimpleNamespace(sync=AsyncMock(side_effect=[None, response]))
|
|
self.room.login_to_matrix = AsyncMock(return_value=True)
|
|
self.room.close = AsyncMock()
|
|
self.room.alert_post = AsyncMock()
|
|
self.room.post_suggestions_individually = AsyncMock()
|
|
with (
|
|
patch("room.RoomMessageText", Event),
|
|
patch("room.SyncResponse", Response),
|
|
patch("post_results.Room", return_value=self.room),
|
|
patch("post_results.time.time", return_value=99),
|
|
patch("post_results.asyncio.sleep", side_effect=asyncio.CancelledError),
|
|
):
|
|
with self.assertRaises(asyncio.CancelledError):
|
|
await post_results.main()
|
|
self.assertEqual(
|
|
self.room.alert_post.await_args_list[-1].args,
|
|
("No new title suggestions posted.",),
|
|
)
|
|
self.room.post_suggestions_individually.assert_not_awaited()
|
|
self.room.close.assert_awaited_once()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|