4885be0752
Phase 1: Ruff lint fixes - Remove unused imports across all files - Remove unused variables (now_utc, tz, ctx) - Fix f-string without placeholders - Fix E402 import order with noqa comments Phase 2: Remove confusing hard delete from storage - Removed delete_bounty() from RoomStorage Protocol (never used by app) - Removed delete_bounty() from JsonFileRoomStorage (was hard delete) - Removed corresponding tests (hard delete was never used) Phase 3: Sync SPEC.md with actual code behavior - Updated overview: admins can add/edit/delete (not 'anyone' + 'creator') - Updated command table: /add, /edit, /delete are admin only - Updated error handling messages Test results: 96 passed (2 hard delete tests removed)
565 lines
18 KiB
Python
565 lines
18 KiB
Python
"""Tests for commands.py — parsing, formatting, and command handlers."""
|
|
|
|
import time
|
|
from unittest.mock import MagicMock, patch, AsyncMock
|
|
|
|
import pytest
|
|
|
|
from telegram import Update, Message, User, Chat
|
|
from telegram.ext import ContextTypes
|
|
|
|
from commands import (
|
|
extract_args,
|
|
parse_args,
|
|
format_bounty,
|
|
cmd_bounty,
|
|
cmd_my,
|
|
cmd_add,
|
|
cmd_update,
|
|
cmd_delete,
|
|
cmd_track,
|
|
cmd_untrack,
|
|
cmd_start,
|
|
cmd_help,
|
|
is_group,
|
|
get_group_id,
|
|
get_user_id,
|
|
get_room_id,
|
|
BOUNTY_SERVICE,
|
|
TRACKING_SERVICE,
|
|
)
|
|
|
|
|
|
class TestExtractArgs:
|
|
def test_basic_split(self):
|
|
assert extract_args("/add hello world") == ["hello", "world"]
|
|
assert extract_args("/track 42") == ["42"]
|
|
|
|
def test_no_args(self):
|
|
assert extract_args("/bounty") == []
|
|
assert extract_args("/add") == []
|
|
|
|
def test_preserves_urls(self):
|
|
result = extract_args("/add check https://example.com stuff")
|
|
assert result == ["check", "https://example.com", "stuff"]
|
|
|
|
def test_none_input(self):
|
|
assert extract_args(None) == []
|
|
|
|
def test_strips_whitespace(self):
|
|
assert extract_args(" /add hello world ") == ["hello", "world"]
|
|
|
|
|
|
class TestParseArgs:
|
|
def test_text_only(self):
|
|
text, link, due = parse_args(["hello", "world"])
|
|
assert text == "hello world"
|
|
assert link is None
|
|
assert due is None
|
|
|
|
def test_link_extracted(self):
|
|
text, link, due = parse_args(["hello", "https://example.com"])
|
|
# "hello" is non-link non-date → becomes text; only the URL becomes link
|
|
assert text == "hello"
|
|
assert link == "https://example.com"
|
|
assert due is None
|
|
|
|
def test_text_and_link(self):
|
|
text, link, due = parse_args(["hello", "world", "https://example.com"])
|
|
assert text == "hello world"
|
|
assert link == "https://example.com"
|
|
|
|
def test_due_date_parsed(self):
|
|
text, link, due = parse_args(["hello", "tomorrow"])
|
|
assert text == "hello"
|
|
assert due is not None
|
|
# Should be some time in the future
|
|
assert due > int(time.time())
|
|
|
|
def test_all_three(self):
|
|
text, link, due = parse_args(["hello", "https://example.com", "tomorrow"])
|
|
assert text == "hello"
|
|
assert link == "https://example.com"
|
|
assert due is not None
|
|
|
|
def test_http_and_https_both_detected(self):
|
|
_, link1, _ = parse_args(["http://example.com"])
|
|
_, link2, _ = parse_args(["https://example.com"])
|
|
assert link1 == "http://example.com"
|
|
assert link2 == "https://example.com"
|
|
|
|
def test_non_url_non_date_becomes_text(self):
|
|
text, link, due = parse_args(["fix", "the", "bug"])
|
|
assert text == "fix the bug"
|
|
assert link is None
|
|
assert due is None
|
|
|
|
def test_multiple_links_first_only(self):
|
|
_, link, _ = parse_args(["text", "https://first.com", "https://second.com"])
|
|
assert link == "https://first.com"
|
|
|
|
def test_due_date_after_link(self):
|
|
text, link, due = parse_args(["task", "https://example.com", "in 5 days"])
|
|
assert text == "task"
|
|
assert link == "https://example.com"
|
|
assert due is not None
|
|
|
|
def test_empty_args(self):
|
|
text, link, due = parse_args([])
|
|
assert text is None
|
|
assert link is None
|
|
assert due is None
|
|
|
|
def test_date_parser_failure_returns_none(self):
|
|
# "asdfjkl" is not parseable → goes to text
|
|
text, link, due = parse_args(["hello", "asdfjkl"])
|
|
assert text == "hello asdfjkl"
|
|
assert due is None
|
|
|
|
def test_link_takes_first_match(self):
|
|
# Even if it's not a valid URL, starts with https://
|
|
_, link, _ = parse_args(["skip", "https://not-real.but-still-a-link"])
|
|
assert link == "https://not-real.but-still-a-link"
|
|
|
|
|
|
class TestFormatBounty:
|
|
def _row(
|
|
self,
|
|
id=1,
|
|
text="Test bounty",
|
|
link="https://example.com",
|
|
due_date_ts=None,
|
|
created_by_user_id=123456,
|
|
):
|
|
row = MagicMock()
|
|
row.id = id
|
|
row.text = text
|
|
row.link = link
|
|
row.due_date_ts = due_date_ts
|
|
row.created_by_user_id = created_by_user_id
|
|
return row
|
|
|
|
def test_shows_id(self):
|
|
b = self._row(id=42)
|
|
out = format_bounty(b, show_id=True)
|
|
assert "[#42]" in out
|
|
|
|
def test_hides_id_when_requested(self):
|
|
b = self._row(id=42)
|
|
out = format_bounty(b, show_id=False)
|
|
assert "[#42]" not in out
|
|
|
|
def test_text_included(self):
|
|
b = self._row(text="Fix the login bug")
|
|
out = format_bounty(b)
|
|
assert "Fix the login bug" in out
|
|
|
|
def test_link_shown(self):
|
|
b = self._row(link="https://github.com/bob/repo")
|
|
out = format_bounty(b)
|
|
assert "https://github.com/bob/repo" in out
|
|
assert "🔗" in out
|
|
|
|
def test_no_link(self):
|
|
b = self._row(link=None)
|
|
out = format_bounty(b)
|
|
assert "🔗" not in out
|
|
|
|
def test_due_date_future(self):
|
|
future = int(time.time()) + 3 * 86400
|
|
b = self._row(due_date_ts=future)
|
|
out = format_bounty(b)
|
|
assert "⏰" in out
|
|
assert "3d" in out
|
|
|
|
def test_due_date_today(self):
|
|
today = int(time.time()) + 3600 # within today
|
|
b = self._row(due_date_ts=today)
|
|
out = format_bounty(b)
|
|
assert "TODAY" in out
|
|
|
|
def test_due_date_overdue(self):
|
|
past = int(time.time()) - 86400 # yesterday
|
|
b = self._row(due_date_ts=past)
|
|
out = format_bounty(b)
|
|
assert "OVERDUE" in out
|
|
|
|
def test_created_by_shown(self):
|
|
b = self._row(created_by_user_id=999)
|
|
out = format_bounty(b)
|
|
assert "999" in out
|
|
|
|
|
|
def create_mock_update(
|
|
user_id=123,
|
|
chat_id=-456,
|
|
chat_type="group",
|
|
message_text="/bounty",
|
|
):
|
|
"""Create a mock Telegram Update with common values."""
|
|
user = MagicMock(spec=User)
|
|
user.id = user_id
|
|
|
|
chat = MagicMock(spec=Chat)
|
|
chat.id = chat_id
|
|
chat.type = chat_type
|
|
|
|
message = MagicMock(spec=Message)
|
|
message.text = message_text
|
|
message.reply_text = AsyncMock()
|
|
message.user = user
|
|
|
|
update = MagicMock(spec=Update)
|
|
update.effective_user = user
|
|
update.effective_chat = chat
|
|
update.message = message
|
|
|
|
return update
|
|
|
|
|
|
class TestHelperFunctions:
|
|
"""Test helper functions."""
|
|
|
|
def test_is_group_true(self):
|
|
update = create_mock_update(chat_type="group")
|
|
assert is_group(update) is True
|
|
|
|
def test_is_group_false_for_private(self):
|
|
update = create_mock_update(chat_type="private")
|
|
assert is_group(update) is False
|
|
|
|
def test_get_group_id(self):
|
|
update = create_mock_update(chat_id=-789)
|
|
assert get_group_id(update) == -789
|
|
|
|
def test_get_user_id(self):
|
|
update = create_mock_update(user_id=999)
|
|
assert get_user_id(update) == 999
|
|
|
|
def test_get_room_id_group(self):
|
|
update = create_mock_update(chat_id=-456, chat_type="group", user_id=123)
|
|
assert get_room_id(update) == -456
|
|
|
|
def test_get_room_id_private(self):
|
|
update = create_mock_update(chat_id=123, chat_type="private", user_id=123)
|
|
assert get_room_id(update) == 123
|
|
|
|
|
|
class TestCmdBounty:
|
|
"""Test cmd_bounty command."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lists_bounties(self):
|
|
update = create_mock_update()
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
mock_bounty = MagicMock()
|
|
mock_bounty.id = 1
|
|
mock_bounty.text = "Test"
|
|
mock_bounty.link = None
|
|
mock_bounty.due_date_ts = None
|
|
mock_bounty.created_by_user_id = 123
|
|
|
|
with patch.object(
|
|
BOUNTY_SERVICE, "list_bounties", return_value=[mock_bounty]
|
|
) as mock_list:
|
|
await cmd_bounty(update, ctx)
|
|
mock_list.assert_called_once_with(-456)
|
|
update.message.reply_text.assert_called_once()
|
|
call_args = update.message.reply_text.call_args[0][0]
|
|
assert "[#1]" in call_args
|
|
assert "Test" in call_args
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_bounties(self):
|
|
update = create_mock_update()
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
with patch.object(BOUNTY_SERVICE, "list_bounties", return_value=[]):
|
|
await cmd_bounty(update, ctx)
|
|
update.message.reply_text.assert_called_once_with("No bounties yet.")
|
|
|
|
|
|
class TestCmdMy:
|
|
"""Test cmd_my command."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_in_group_shows_tracked(self):
|
|
update = create_mock_update(chat_type="group")
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
mock_bounty = MagicMock()
|
|
mock_bounty.id = 1
|
|
mock_bounty.text = "Tracked"
|
|
mock_bounty.link = None
|
|
mock_bounty.due_date_ts = None
|
|
mock_bounty.created_by_user_id = 123
|
|
|
|
with patch.object(
|
|
TRACKING_SERVICE, "get_tracked_bounties", return_value=[mock_bounty]
|
|
) as mock_track:
|
|
await cmd_my(update, ctx)
|
|
mock_track.assert_called_once_with(-456, 123)
|
|
update.message.reply_text.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_in_private_shows_personal(self):
|
|
update = create_mock_update(chat_type="private", chat_id=123)
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
mock_bounty = MagicMock()
|
|
mock_bounty.id = 2
|
|
mock_bounty.text = "Personal"
|
|
mock_bounty.link = None
|
|
mock_bounty.due_date_ts = None
|
|
mock_bounty.created_by_user_id = 123
|
|
|
|
with patch.object(
|
|
BOUNTY_SERVICE, "list_bounties", return_value=[mock_bounty]
|
|
) as mock_list:
|
|
await cmd_my(update, ctx)
|
|
mock_list.assert_called_once_with(123)
|
|
update.message.reply_text.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_bounties_tracked(self):
|
|
update = create_mock_update(chat_type="group")
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
with patch.object(TRACKING_SERVICE, "get_tracked_bounties", return_value=[]):
|
|
await cmd_my(update, ctx)
|
|
update.message.reply_text.assert_called_once_with(
|
|
"You are not tracking any bounties."
|
|
)
|
|
|
|
|
|
class TestCmdAdd:
|
|
"""Test cmd_add command."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_bounty_success(self):
|
|
update = create_mock_update(message_text="/add Fix the bug")
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
mock_bounty = MagicMock()
|
|
mock_bounty.id = 42
|
|
|
|
with patch.object(
|
|
BOUNTY_SERVICE, "add_bounty", return_value=mock_bounty
|
|
) as mock_add:
|
|
await cmd_add(update, ctx)
|
|
mock_add.assert_called_once()
|
|
call_kwargs = mock_add.call_args[1]
|
|
assert call_kwargs["room_id"] == -456
|
|
assert call_kwargs["user_id"] == 123
|
|
assert call_kwargs["text"] == "Fix the bug"
|
|
update.message.reply_text.assert_called_once()
|
|
assert "✅" in update.message.reply_text.call_args[0][0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_without_args(self):
|
|
update = create_mock_update(message_text="/add")
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
await cmd_add(update, ctx)
|
|
update.message.reply_text.assert_called_once()
|
|
assert "Usage" in update.message.reply_text.call_args[0][0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_needs_text_or_link(self):
|
|
update = create_mock_update(message_text="/add")
|
|
|
|
_, link, _ = parse_args([])
|
|
if not "test" and not link:
|
|
await update.message.reply_text("A bounty needs at least text or a link.")
|
|
update.message.reply_text.assert_called_once()
|
|
|
|
|
|
class TestCmdUpdate:
|
|
"""Test cmd_update command."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_bounty_success(self):
|
|
update = create_mock_update(message_text="/update 1 New text")
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
with patch.object(
|
|
BOUNTY_SERVICE, "update_bounty", return_value=True
|
|
) as mock_update:
|
|
await cmd_update(update, ctx)
|
|
mock_update.assert_called_once()
|
|
call_kwargs = mock_update.call_args[1]
|
|
assert call_kwargs["room_id"] == -456
|
|
assert call_kwargs["bounty_id"] == 1
|
|
assert call_kwargs["text"] == "New text"
|
|
update.message.reply_text.assert_called_once()
|
|
assert "✅" in update.message.reply_text.call_args[0][0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_without_args(self):
|
|
update = create_mock_update(message_text="/update")
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
await cmd_update(update, ctx)
|
|
update.message.reply_text.assert_called_once()
|
|
assert "Usage" in update.message.reply_text.call_args[0][0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_invalid_id(self):
|
|
update = create_mock_update(message_text="/update abc")
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
await cmd_update(update, ctx)
|
|
update.message.reply_text.assert_called_once_with("Invalid bounty ID.")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_permission_denied(self):
|
|
update = create_mock_update(message_text="/update 1 new text")
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
with patch.object(
|
|
BOUNTY_SERVICE,
|
|
"update_bounty",
|
|
side_effect=PermissionError("Not your bounty"),
|
|
):
|
|
await cmd_update(update, ctx)
|
|
update.message.reply_text.assert_called_once()
|
|
assert "⛔" in update.message.reply_text.call_args[0][0]
|
|
|
|
|
|
class TestCmdDelete:
|
|
"""Test cmd_delete command."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_bounty_success(self):
|
|
update = create_mock_update(message_text="/delete 1")
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
with patch.object(
|
|
BOUNTY_SERVICE, "delete_bounty", return_value=True
|
|
) as mock_delete:
|
|
await cmd_delete(update, ctx)
|
|
mock_delete.assert_called_once_with(room_id=-456, bounty_id=1, user_id=123)
|
|
update.message.reply_text.assert_called_once()
|
|
assert "✅" in update.message.reply_text.call_args[0][0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_without_args(self):
|
|
update = create_mock_update(message_text="/delete")
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
await cmd_delete(update, ctx)
|
|
update.message.reply_text.assert_called_once()
|
|
assert "Usage" in update.message.reply_text.call_args[0][0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_invalid_id(self):
|
|
update = create_mock_update(message_text="/delete abc")
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
await cmd_delete(update, ctx)
|
|
update.message.reply_text.assert_called_once_with("Invalid bounty ID.")
|
|
|
|
|
|
class TestCmdTrack:
|
|
"""Test cmd_track command."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_track_in_group(self):
|
|
update = create_mock_update(chat_type="group", message_text="/track 1")
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
with patch.object(
|
|
TRACKING_SERVICE, "track_bounty", return_value=True
|
|
) as mock_track:
|
|
await cmd_track(update, ctx)
|
|
mock_track.assert_called_once_with(-456, 123, 1)
|
|
update.message.reply_text.assert_called_once()
|
|
assert "✅" in update.message.reply_text.call_args[0][0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_track_not_in_group(self):
|
|
update = create_mock_update(chat_type="private", message_text="/track 1")
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
await cmd_track(update, ctx)
|
|
update.message.reply_text.assert_called_once()
|
|
assert "⛔" in update.message.reply_text.call_args[0][0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_track_without_args(self):
|
|
update = create_mock_update(chat_type="group", message_text="/track")
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
await cmd_track(update, ctx)
|
|
update.message.reply_text.assert_called_once()
|
|
assert "Usage" in update.message.reply_text.call_args[0][0]
|
|
|
|
|
|
class TestCmdUntrack:
|
|
"""Test cmd_untrack command."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_untrack_in_group(self):
|
|
update = create_mock_update(chat_type="group", message_text="/untrack 1")
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
with patch.object(
|
|
TRACKING_SERVICE, "untrack_bounty", return_value=True
|
|
) as mock_untrack:
|
|
await cmd_untrack(update, ctx)
|
|
mock_untrack.assert_called_once_with(-456, 123, 1)
|
|
update.message.reply_text.assert_called_once()
|
|
assert "✅" in update.message.reply_text.call_args[0][0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_untrack_not_in_group(self):
|
|
update = create_mock_update(chat_type="private", message_text="/untrack 1")
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
await cmd_untrack(update, ctx)
|
|
update.message.reply_text.assert_called_once()
|
|
assert "⛔" in update.message.reply_text.call_args[0][0]
|
|
|
|
|
|
class TestCmdStart:
|
|
"""Test cmd_start command."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_in_group(self):
|
|
update = create_mock_update(chat_type="group")
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
await cmd_start(update, ctx)
|
|
update.message.reply_text.assert_called_once()
|
|
text = update.message.reply_text.call_args[0][0]
|
|
assert "👻" in text
|
|
assert "/bounty" in text
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_in_private(self):
|
|
update = create_mock_update(chat_type="private")
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
await cmd_start(update, ctx)
|
|
update.message.reply_text.assert_called_once()
|
|
text = update.message.reply_text.call_args[0][0]
|
|
assert "👻" in text
|
|
assert "/my" in text
|
|
|
|
|
|
class TestCmdHelp:
|
|
"""Test cmd_help command."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_help_shows_commands(self):
|
|
update = create_mock_update()
|
|
ctx = MagicMock(spec=ContextTypes.DEFAULT_TYPE)
|
|
|
|
await cmd_help(update, ctx)
|
|
update.message.reply_text.assert_called_once()
|
|
text = update.message.reply_text.call_args[0][0]
|
|
assert "/bounty" in text
|
|
assert "/add" in text
|
|
assert "/help" in text
|