"""Tests for the remind plugin.""" import asyncio import importlib.util import sys from pathlib import Path from derp.irc import Message # plugins/ is not a Python package -- load the module from file path _spec = importlib.util.spec_from_file_location( "plugins.remind", Path(__file__).resolve().parent.parent / "plugins" / "remind.py", ) _mod = importlib.util.module_from_spec(_spec) sys.modules[_spec.name] = _mod _spec.loader.exec_module(_mod) from plugins.remind import ( # noqa: E402 _by_user, _cleanup, _format_duration, _make_id, _parse_duration, _remind_once, _remind_repeat, _reminders, cmd_remind, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- class _FakeBot: """Minimal bot stand-in that captures sent/replied messages.""" def __init__(self): self.sent: list[tuple[str, str]] = [] self.replied: list[str] = [] async def send(self, target: str, text: str) -> None: self.sent.append((target, text)) async def reply(self, message, text: str) -> None: self.replied.append(text) def _msg(text: str, nick: str = "alice", target: str = "#test") -> Message: """Create a channel PRIVMSG.""" return Message( raw="", prefix=f"{nick}!~{nick}@host", nick=nick, command="PRIVMSG", params=[target, text], tags={}, ) def _pm(text: str, nick: str = "alice") -> Message: """Create a private PRIVMSG (target = bot nick).""" return Message( raw="", prefix=f"{nick}!~{nick}@host", nick=nick, command="PRIVMSG", params=["botname", text], tags={}, ) def _clear() -> None: """Reset global module state between tests.""" for entry in _reminders.values(): task = entry[0] if task is not None and not task.done(): task.cancel() _reminders.clear() _by_user.clear() async def _run_cmd(bot, msg): """Run cmd_remind and cancel any spawned tasks afterward.""" await cmd_remind(bot, msg) # Yield so tasks can start await asyncio.sleep(0) async def _run_cmd_and_cleanup(bot, msg): """Run cmd_remind, yield, then cancel all spawned tasks.""" await cmd_remind(bot, msg) await asyncio.sleep(0) for entry in list(_reminders.values()): if entry[0] is not None and not entry[0].done(): entry[0].cancel() await asyncio.sleep(0) # --------------------------------------------------------------------------- # Pure helpers: _parse_duration # --------------------------------------------------------------------------- class TestParseDurationRawSeconds: def test_positive_integer(self): assert _parse_duration("60") == 60 def test_zero_returns_none(self): assert _parse_duration("0") is None def test_negative_returns_none(self): assert _parse_duration("-5") is None def test_large_value(self): assert _parse_duration("86400") == 86400 class TestParseDurationSpecs: def test_minutes(self): assert _parse_duration("5m") == 300 def test_hours_and_minutes(self): assert _parse_duration("1h30m") == 5400 def test_days(self): assert _parse_duration("2d") == 172800 def test_seconds_suffix(self): assert _parse_duration("90s") == 90 def test_full_combo(self): assert _parse_duration("1d12h30m15s") == 131415 class TestParseDurationInvalid: def test_empty_string(self): assert _parse_duration("") is None def test_letters_only(self): assert _parse_duration("abc") is None def test_all_zeros(self): assert _parse_duration("0m0s") is None class TestParseDurationEdgeCases: def test_uppercase_works(self): assert _parse_duration("5M") == 300 def test_hours_with_zero_minutes(self): assert _parse_duration("1h0m") == 3600 def test_all_zeros_except_one_second(self): assert _parse_duration("0d0h0m1s") == 1 # --------------------------------------------------------------------------- # Pure helpers: _format_duration # --------------------------------------------------------------------------- class TestFormatDuration: def test_minutes_and_seconds(self): assert _format_duration(90) == "1m30s" def test_exact_hour(self): assert _format_duration(3600) == "1h" def test_exact_day(self): assert _format_duration(86400) == "1d" def test_zero(self): assert _format_duration(0) == "0s" def test_full_combo(self): assert _format_duration(90061) == "1d1h1m1s" def test_seconds_only(self): assert _format_duration(45) == "45s" # --------------------------------------------------------------------------- # Pure helpers: _make_id # --------------------------------------------------------------------------- class TestMakeId: def test_returns_six_char_hex(self): rid = _make_id("user", "check oven") assert len(rid) == 6 assert all(c in "0123456789abcdef" for c in rid) def test_different_inputs_differ(self): rid1 = _make_id("alice", "task one") rid2 = _make_id("bob", "task two") assert rid1 != rid2 # --------------------------------------------------------------------------- # _cleanup # --------------------------------------------------------------------------- class TestCleanup: def test_removes_from_both_structures(self): _clear() _reminders["abc123"] = (None, "#ch", "alice", "label", "12:00", False) _by_user[("#ch", "alice")] = ["abc123"] _cleanup("abc123", "#ch", "alice") assert "abc123" not in _reminders assert ("#ch", "alice") not in _by_user def test_removes_single_entry_from_multi(self): _clear() _reminders["aaa"] = (None, "#ch", "alice", "", "12:00", False) _reminders["bbb"] = (None, "#ch", "alice", "", "12:00", False) _by_user[("#ch", "alice")] = ["aaa", "bbb"] _cleanup("aaa", "#ch", "alice") assert "aaa" not in _reminders assert _by_user[("#ch", "alice")] == ["bbb"] def test_missing_rid_no_error(self): _clear() _cleanup("nonexistent", "#ch", "alice") def test_missing_user_key_no_error(self): _clear() _reminders["abc"] = (None, "#ch", "alice", "", "12:00", False) _cleanup("abc", "#ch", "bob") # different nick, user key absent assert "abc" not in _reminders # --------------------------------------------------------------------------- # _remind_once # --------------------------------------------------------------------------- class TestRemindOnce: def test_fires_metadata_and_label(self): _clear() bot = _FakeBot() async def inner(): rid = "once01" task = asyncio.create_task( _remind_once(bot, rid, "#ch", "alice", "check oven", 0, "12:00:00 UTC"), ) _reminders[rid] = (task, "#ch", "alice", "check oven", "12:00:00 UTC", False) _by_user[("#ch", "alice")] = [rid] await task asyncio.run(inner()) assert len(bot.sent) == 2 assert "alice: reminder #once01" in bot.sent[0][1] assert "12:00:00 UTC" in bot.sent[0][1] assert bot.sent[1] == ("#ch", "check oven") assert "once01" not in _reminders def test_empty_label_sends_one_line(self): _clear() bot = _FakeBot() async def inner(): rid = "once02" task = asyncio.create_task( _remind_once(bot, rid, "#ch", "bob", "", 0, "12:00:00 UTC"), ) _reminders[rid] = (task, "#ch", "bob", "", "12:00:00 UTC", False) _by_user[("#ch", "bob")] = [rid] await task asyncio.run(inner()) assert len(bot.sent) == 1 def test_cancellation_cleans_up(self): _clear() bot = _FakeBot() async def inner(): rid = "once03" task = asyncio.create_task( _remind_once(bot, rid, "#ch", "alice", "text", 9999, "12:00:00 UTC"), ) _reminders[rid] = (task, "#ch", "alice", "text", "12:00:00 UTC", False) _by_user[("#ch", "alice")] = [rid] await asyncio.sleep(0) task.cancel() await asyncio.gather(task, return_exceptions=True) asyncio.run(inner()) assert len(bot.sent) == 0 assert "once03" not in _reminders # --------------------------------------------------------------------------- # _remind_repeat # --------------------------------------------------------------------------- class TestRemindRepeat: def test_fires_at_least_once(self): _clear() bot = _FakeBot() async def inner(): rid = "rpt01" task = asyncio.create_task( _remind_repeat(bot, rid, "#ch", "alice", "hydrate", 0, "12:00:00 UTC"), ) _reminders[rid] = (task, "#ch", "alice", "hydrate", "12:00:00 UTC", True) _by_user[("#ch", "alice")] = [rid] for _ in range(5): await asyncio.sleep(0) task.cancel() await asyncio.gather(task, return_exceptions=True) asyncio.run(inner()) assert len(bot.sent) >= 2 # at least one fire (metadata + label) assert any("rpt01" in t for _, t in bot.sent) assert "rpt01" not in _reminders def test_cancellation_cleans_up(self): _clear() bot = _FakeBot() async def inner(): rid = "rpt02" task = asyncio.create_task( _remind_repeat(bot, rid, "#ch", "bob", "stretch", 9999, "12:00:00 UTC"), ) _reminders[rid] = (task, "#ch", "bob", "stretch", "12:00:00 UTC", True) _by_user[("#ch", "bob")] = [rid] await asyncio.sleep(0) task.cancel() await asyncio.gather(task, return_exceptions=True) asyncio.run(inner()) assert len(bot.sent) == 0 assert "rpt02" not in _reminders # --------------------------------------------------------------------------- # cmd_remind: usage / errors # --------------------------------------------------------------------------- class TestCmdRemindUsage: def test_no_args_shows_usage(self): _clear() bot = _FakeBot() asyncio.run(cmd_remind(bot, _msg("!remind"))) assert "Usage:" in bot.replied[0] def test_invalid_duration(self): _clear() bot = _FakeBot() asyncio.run(cmd_remind(bot, _msg("!remind xyz some text"))) assert "Invalid duration" in bot.replied[0] def test_every_no_args(self): _clear() bot = _FakeBot() asyncio.run(cmd_remind(bot, _msg("!remind every"))) assert "Invalid duration" in bot.replied[0] def test_every_invalid_duration(self): _clear() bot = _FakeBot() asyncio.run(cmd_remind(bot, _msg("!remind every abc"))) assert "Invalid duration" in bot.replied[0] # --------------------------------------------------------------------------- # cmd_remind: one-shot creation # --------------------------------------------------------------------------- class TestCmdRemindOneshot: def test_creates_with_duration(self): _clear() bot = _FakeBot() async def inner(): await _run_cmd_and_cleanup(bot, _msg("!remind 5m check the oven")) asyncio.run(inner()) assert len(bot.replied) == 1 assert "set (5m)" in bot.replied[0] assert "#" in bot.replied[0] def test_no_label(self): _clear() bot = _FakeBot() async def inner(): await _run_cmd_and_cleanup(bot, _msg("!remind 5m")) asyncio.run(inner()) assert "set (5m)" in bot.replied[0] def test_stores_in_tracking(self): _clear() bot = _FakeBot() async def inner(): await _run_cmd(bot, _msg("!remind 9999s task")) assert len(_reminders) == 1 entry = next(iter(_reminders.values())) assert entry[1] == "#test" # target assert entry[2] == "alice" # nick assert entry[3] == "task" # label assert entry[5] is False # not repeating assert ("#test", "alice") in _by_user # cleanup for e in _reminders.values(): e[0].cancel() await asyncio.sleep(0) asyncio.run(inner()) def test_days_duration(self): _clear() bot = _FakeBot() async def inner(): await _run_cmd_and_cleanup(bot, _msg("!remind 2d12h renew cert")) asyncio.run(inner()) assert "set (2d12h)" in bot.replied[0] # --------------------------------------------------------------------------- # cmd_remind: repeating creation # --------------------------------------------------------------------------- class TestCmdRemindRepeat: def test_creates_repeating(self): _clear() bot = _FakeBot() async def inner(): await _run_cmd_and_cleanup(bot, _msg("!remind every 1h drink water")) asyncio.run(inner()) assert len(bot.replied) == 1 assert "every 1h" in bot.replied[0] def test_repeating_stores_flag(self): _clear() bot = _FakeBot() async def inner(): await _run_cmd(bot, _msg("!remind every 30m stretch")) entry = next(iter(_reminders.values())) assert entry[5] is True # repeating flag for e in _reminders.values(): e[0].cancel() await asyncio.sleep(0) asyncio.run(inner()) def test_repeating_no_label(self): _clear() bot = _FakeBot() async def inner(): await _run_cmd_and_cleanup(bot, _msg("!remind every 1h")) asyncio.run(inner()) assert "every 1h" in bot.replied[0] # --------------------------------------------------------------------------- # cmd_remind: list # --------------------------------------------------------------------------- class TestCmdRemindList: def test_empty_list(self): _clear() bot = _FakeBot() asyncio.run(cmd_remind(bot, _msg("!remind list"))) assert "No active reminders" in bot.replied[0] def test_shows_active(self): _clear() bot = _FakeBot() async def inner(): await _run_cmd(bot, _msg("!remind 9999s task")) bot.replied.clear() await cmd_remind(bot, _msg("!remind list")) for e in _reminders.values(): e[0].cancel() await asyncio.sleep(0) asyncio.run(inner()) assert "Reminders:" in bot.replied[0] assert "#" in bot.replied[0] def test_shows_repeat_tag(self): _clear() bot = _FakeBot() async def inner(): await _run_cmd(bot, _msg("!remind every 9999s task")) bot.replied.clear() await cmd_remind(bot, _msg("!remind list")) for e in _reminders.values(): e[0].cancel() await asyncio.sleep(0) asyncio.run(inner()) assert "(repeat)" in bot.replied[0] # --------------------------------------------------------------------------- # cmd_remind: cancel # --------------------------------------------------------------------------- class TestCmdRemindCancel: def _extract_rid(self, reply: str) -> str: """Extract reminder ID from 'Reminder #abc123 set (...)' reply.""" return reply.split("#")[1].split(" ")[0] def test_cancel_valid(self): _clear() bot = _FakeBot() async def inner(): await _run_cmd(bot, _msg("!remind 9999s task")) rid = self._extract_rid(bot.replied[0]) bot.replied.clear() await cmd_remind(bot, _msg(f"!remind cancel {rid}")) await asyncio.sleep(0) asyncio.run(inner()) assert "Cancelled" in bot.replied[0] def test_cancel_with_hash_prefix(self): _clear() bot = _FakeBot() async def inner(): await _run_cmd(bot, _msg("!remind 9999s task")) rid = self._extract_rid(bot.replied[0]) bot.replied.clear() await cmd_remind(bot, _msg(f"!remind cancel #{rid}")) await asyncio.sleep(0) asyncio.run(inner()) assert "Cancelled" in bot.replied[0] def test_cancel_wrong_user(self): _clear() bot = _FakeBot() async def inner(): await _run_cmd(bot, _msg("!remind 9999s secret")) rid = self._extract_rid(bot.replied[0]) bot.replied.clear() await cmd_remind(bot, _msg(f"!remind cancel {rid}", nick="eve")) for e in _reminders.values(): e[0].cancel() await asyncio.sleep(0) asyncio.run(inner()) assert "No active reminder" in bot.replied[0] def test_cancel_nonexistent(self): _clear() bot = _FakeBot() asyncio.run(cmd_remind(bot, _msg("!remind cancel ffffff"))) assert "No active reminder" in bot.replied[0] def test_cancel_no_id(self): _clear() bot = _FakeBot() asyncio.run(cmd_remind(bot, _msg("!remind cancel"))) assert "Usage:" in bot.replied[0] # --------------------------------------------------------------------------- # cmd_remind: target routing # --------------------------------------------------------------------------- class TestCmdRemindTarget: def test_channel_target(self): _clear() bot = _FakeBot() async def inner(): await _run_cmd(bot, _msg("!remind 9999s task", target="#ops")) entry = next(iter(_reminders.values())) assert entry[1] == "#ops" for e in _reminders.values(): e[0].cancel() await asyncio.sleep(0) asyncio.run(inner()) def test_pm_uses_nick(self): _clear() bot = _FakeBot() async def inner(): await _run_cmd(bot, _pm("!remind 9999s task")) entry = next(iter(_reminders.values())) assert entry[1] == "alice" # nick, not "botname" for e in _reminders.values(): e[0].cancel() await asyncio.sleep(0) asyncio.run(inner())