!community doctor <room_id> for detailed analysis of specific rooms!community doctor <room_id> for detailed analysis of specific rooms{title}
{content}
Error: {error}
" + else: + return f"Error: {error}" + + @staticmethod + def build_success_response(message: str, allow_html: bool = True) -> str: + """Build a success response. + + Args: + message: Success message + allow_html: Whether to allow HTML formatting + + Returns: + str: Formatted success response + """ + if allow_html: + return f"Success: {message}
" + else: + return f"Success: {message}" + + @staticmethod + def build_list_response(title: str, items: List[str], allow_html: bool = True) -> str: + """Build a list response. + + Args: + title: List title + items: List items + allow_html: Whether to allow HTML formatting + + Returns: + str: Formatted list response + """ + if not items: + return ResponseBuilder.build_html_response(title, "No items found.", allow_html) + + if allow_html: + items_html = "{title}
{items_html}
Users inactive for between {warn_threshold} and {kick_threshold} days:
"
+ f"{warn_list}
Users inactive for at least {kick_threshold} days:
"
+ f"{kick_list}
Ignored users:
{ignored_list}
Users banned:{ban_list_html}
Errors:{error_list_html}
No users were banned.
") + + return "".join(response_parts) + + @staticmethod + def build_sync_results_response(results: Dict[str, List[str]]) -> str: + """Build a sync results response. + + Args: + results: Sync results data + + Returns: + str: Formatted sync results + """ + added = results.get("added", []) + dropped = results.get("dropped", []) + + response_parts = [] + + if added: + added_html = "Added:
{added_html}
Dropped:
{dropped_html}
No changes made.
") + + return "".join(response_parts) + + @staticmethod + def build_doctor_report_response(report: Dict[str, Any]) -> str: + """Build a doctor report response. + + Args: + report: Doctor report data + + Returns: + str: Formatted doctor report + """ + response_parts = [] + + # Space information + if report.get("space"): + space = report["space"] + space_info = f"Space: {space.get('room_id', 'Unknown')}{space_info}
") + + # Room information + if report.get("rooms"): + rooms_info = "Rooms:{rooms_info}
") + + # Issues + if report.get("issues"): + issues_html = "Issues:
{issues_html}
Warnings:
{warnings_html}
No issues found.
") + + return "".join(response_parts) diff --git a/community/helpers/room_creation_utils.py b/community/helpers/room_creation_utils.py new file mode 100644 index 0000000..b6b7cbe --- /dev/null +++ b/community/helpers/room_creation_utils.py @@ -0,0 +1,269 @@ +"""Room creation utility functions for the community bot.""" + +import re +import asyncio +from typing import Optional, Tuple, List, Dict, Any +from mautrix.types import MessageEvent, PowerLevelStateEventContent, EventType +from mautrix.client import Client + + +async def validate_room_creation_params( + roomname: str, + config: dict, + evt: Optional[MessageEvent] = None +) -> Tuple[str, bool, bool, str]: + """Validate and process room creation parameters. + + Args: + roomname: Original room name + config: Bot configuration + evt: Optional MessageEvent for error responses + + Returns: + Tuple of (sanitized_name, force_encryption, force_unencryption, error_msg) + """ + # Check for encryption flags + encrypted_flag_regex = re.compile(r"(\s+|^)-+encrypt(ed)?\s?") + unencrypted_flag_regex = re.compile(r"(\s+|^)-+unencrypt(ed)?\s?") + force_encryption = bool(encrypted_flag_regex.search(roomname)) + force_unencryption = bool(unencrypted_flag_regex.search(roomname)) + + # Clean up room name + if force_encryption: + roomname = encrypted_flag_regex.sub("", roomname) + if force_unencryption: + roomname = unencrypted_flag_regex.sub("", roomname) + + sanitized_name = re.sub(r"[^a-zA-Z0-9]", "", roomname).lower() + + # Check if community slug is configured + if not config.get("community_slug"): + error_msg = "No community slug configured. Please run initialize command first." + return sanitized_name, force_encryption, force_unencryption, error_msg + + return sanitized_name, force_encryption, force_unencryption, "" + + +async def prepare_room_creation_data( + sanitized_name: str, + config: dict, + client: Client, + invitees: Optional[List[str]] = None +) -> Tuple[str, str, List[str], str]: + """Prepare data needed for room creation. + + Args: + sanitized_name: Sanitized room name + config: Bot configuration + client: Matrix client + invitees: Optional list of users to invite + + Returns: + Tuple of (alias_localpart, server, room_invitees, parent_room) + """ + # Create alias with community slug + alias_localpart = f"{sanitized_name}-{config['community_slug']}" + + # Get server and invitees + server = client.parse_user_id(client.mxid)[1] + room_invitees = invitees if invitees is not None else config["invitees"] + parent_room = config["parent_room"] + + return alias_localpart, server, room_invitees, parent_room + + +async def prepare_power_levels( + client: Client, + config: dict, + parent_room: str, + power_level_override: Optional[PowerLevelStateEventContent] = None +) -> PowerLevelStateEventContent: + """Prepare power levels for room creation. + + Args: + client: Matrix client + config: Bot configuration + parent_room: Parent room ID + power_level_override: Optional existing power level override + + Returns: + PowerLevelStateEventContent for room creation + """ + if power_level_override: + return power_level_override + + if parent_room: + # Get parent room power levels to extract user power levels + parent_power_levels = await client.get_state_event( + parent_room, EventType.ROOM_POWER_LEVELS + ) + + # Create new power levels with server defaults, not copying all permissions from space + power_levels = PowerLevelStateEventContent() + + # Copy only user power levels from parent space, not the entire permission set + if parent_power_levels.users: + user_power_levels = parent_power_levels.users.copy() + # Ensure bot has highest power + user_power_levels[client.mxid] = 1000 + power_levels.users = user_power_levels + else: + power_levels.users = { + client.mxid: 1000, # Bot gets highest power + } + + # Set explicit config values + power_levels.invite = config["invite_power_level"] + + return power_levels + else: + # If no parent room, create default power levels + power_levels = PowerLevelStateEventContent() + power_levels.users = { + client.mxid: 1000, # Bot gets highest power + } + power_levels.invite = config["invite_power_level"] + return power_levels + + +def prepare_initial_state( + config: dict, + parent_room: str, + server: str, + force_encryption: bool, + force_unencryption: bool, + creation_content: Optional[Dict[str, Any]] = None +) -> List[Dict[str, Any]]: + """Prepare initial state events for room creation. + + Args: + config: Bot configuration + parent_room: Parent room ID + server: Server name + force_encryption: Whether to force encryption + force_unencryption: Whether to force no encryption + creation_content: Optional creation content + + Returns: + List of initial state events + """ + initial_state = [] + + # Only add space parent state if we have a parent room + if parent_room: + initial_state.extend([ + { + "type": str(EventType.SPACE_PARENT), + "state_key": parent_room, + "content": { + "via": [server], + "canonical": True + } + }, + { + "type": str(EventType.ROOM_JOIN_RULES), + "content": { + "join_rule": "restricted", + "allow": [{ + "type": "m.room_membership", + "room_id": parent_room + }] + } + } + ]) + + # Add encryption if needed + if (config.get("encrypt", False) and not force_unencryption) or force_encryption: + initial_state.append({ + "type": str(EventType.ROOM_ENCRYPTION), + "content": { + "algorithm": "m.megolm.v1.aes-sha2" + } + }) + + # Add history visibility if specified in creation_content + if creation_content and "m.room.history_visibility" in creation_content: + initial_state.append({ + "type": str(EventType.ROOM_HISTORY_VISIBILITY), + "content": { + "history_visibility": creation_content["m.room.history_visibility"] + } + }) + + return initial_state + + +def adjust_power_levels_for_modern_rooms( + power_levels: PowerLevelStateEventContent, + room_version: str +) -> PowerLevelStateEventContent: + """Adjust power levels for modern room versions. + + Args: + power_levels: Power level state content + room_version: Room version string + + Returns: + Adjusted power level state content + """ + # For modern room versions (12+), remove the bot from power levels + # as creators have unlimited power by default and cannot appear in power levels + if room_version and int(room_version) >= 12 and power_levels: + if power_levels.users: + # Remove bot from users list but keep other important settings + power_levels.users.pop("bot_mxid", None) # Will be replaced with actual bot mxid + + return power_levels + + +async def add_room_to_space( + client: Client, + parent_room: str, + room_id: str, + server: str, + sleep_duration: float +) -> None: + """Add created room to parent space. + + Args: + client: Matrix client + parent_room: Parent room ID + room_id: Created room ID + server: Server name + sleep_duration: Sleep duration between operations + """ + if parent_room: + await client.send_state_event( + parent_room, + EventType.SPACE_CHILD, + { + "via": [server], + "suggested": False + }, + state_key=room_id + ) + await asyncio.sleep(sleep_duration) + + +async def verify_room_creation( + client: Client, + room_id: str, + expected_version: str, + logger +) -> None: + """Verify that room was created with correct settings. + + Args: + client: Matrix client + room_id: Created room ID + expected_version: Expected room version + logger: Logger instance + """ + try: + from .room_utils import get_room_version_and_creators + actual_version, actual_creators = await get_room_version_and_creators(client, room_id, logger) + logger.info(f"Room {room_id} created with version {actual_version} (requested: {expected_version})") + if actual_version != expected_version: + logger.warning(f"Room version mismatch: requested {expected_version}, got {actual_version}") + except Exception as e: + logger.warning(f"Could not verify room version for {room_id}: {e}") diff --git a/community/helpers/room_utils.py b/community/helpers/room_utils.py new file mode 100644 index 0000000..9975776 --- /dev/null +++ b/community/helpers/room_utils.py @@ -0,0 +1,170 @@ +"""Room and space utility functions.""" + +import re +from typing import Optional, Tuple, List +from mautrix.types import EventType, PowerLevelStateEventContent +from mautrix.errors import MNotFound + + +async def validate_room_alias(client, alias_localpart: str, server: str) -> bool: + """Check if a room alias already exists. + + Args: + client: Matrix client instance + alias_localpart: The localpart of the alias (without # and :server) + server: The server domain + + Returns: + bool: True if alias is available, False if it already exists + """ + try: + full_alias = f"#{alias_localpart}:{server}" + await client.resolve_room_alias(full_alias) + # If we get here, the alias exists + return False + except MNotFound: + # Alias doesn't exist, so it's available + return True + except Exception as e: + # For other errors, assume alias is available to be safe + return True + + +async def validate_room_aliases(client, room_names: list[str], community_slug: str, server: str) -> Tuple[bool, List[str]]: + """Validate that all room aliases are available. + + Args: + client: Matrix client instance + room_names: List of room names to validate + community_slug: The community slug to append + server: The server domain + + Returns: + tuple: (is_valid, list_of_conflicting_aliases) + """ + if not community_slug: + return False, [] + + conflicting_aliases = [] + + for room_name in room_names: + # Clean the room name and create alias + from .message_utils import sanitize_room_name + sanitized_name = sanitize_room_name(room_name) + alias_localpart = f"{sanitized_name}-{community_slug}" + + # Check if alias is available + is_available = await validate_room_alias(client, alias_localpart, server) + if not is_available: + conflicting_aliases.append(f"#{alias_localpart}:{server}") + + return len(conflicting_aliases) == 0, conflicting_aliases + + +async def get_room_version_and_creators(client, room_id: str) -> Tuple[str, List[str]]: + """Get the room version and creators for a room. + + Args: + client: Matrix client instance + room_id: The room ID to check + + Returns: + tuple: (room_version, list_of_creators) + """ + try: + # Get all state events to find the creation event + state_events = await client.get_state(room_id) + + # Find the m.room.create event + creation_event = None + for event in state_events: + if event.type == EventType.ROOM_CREATE: + creation_event = event + break + + if not creation_event: + # Default to version 1 if no creation event found + return "1", [] + + room_version = creation_event.content.get("room_version", "1") + creators = [] + + # Add the sender of the creation event as a creator + if creation_event.sender: + creators.append(creation_event.sender) + + # Add any additional creators from the content + additional_creators = creation_event.content.get("additional_creators", []) + if isinstance(additional_creators, list): + creators.extend(additional_creators) + + return room_version, creators + + except Exception: + # Default to version 1 if there's an error + return "1", [] + + +def is_modern_room_version(room_version: str) -> bool: + """Check if a room version is 12 or newer (modern room versions). + + Args: + room_version: The room version string to check + + Returns: + bool: True if room version is 12 or newer + """ + try: + version_num = int(room_version) + return version_num >= 12 + except (ValueError, TypeError): + # If we can't parse the version, assume it's not modern + return False + + +async def user_has_unlimited_power(client, user_id: str, room_id: str) -> bool: + """Check if a user has unlimited power in a room (creator in modern room versions). + + Args: + client: Matrix client instance + user_id: The user ID to check + room_id: The room ID to check in + + Returns: + bool: True if user has unlimited power + """ + try: + room_version, creators = await get_room_version_and_creators(client, room_id) + + # In modern room versions (12+), creators have unlimited power + if is_modern_room_version(room_version): + return user_id in creators + + # In older room versions, creators don't have special unlimited power + return False + + except Exception: + return False + + +async def get_moderators_and_above(client, parent_room: str) -> List[str]: + """Get list of users with moderator or higher permissions from the parent space. + + Args: + client: Matrix client instance + parent_room: The parent room ID + + Returns: + list: List of user IDs with power level >= 50 (moderator or above) + """ + try: + power_levels = await client.get_state_event( + parent_room, EventType.ROOM_POWER_LEVELS + ) + moderators = [] + for user, level in power_levels.users.items(): + if level >= 50: # Moderator level or above + moderators.append(user) + return moderators + except Exception: + return [] diff --git a/community/helpers/user_utils.py b/community/helpers/user_utils.py new file mode 100644 index 0000000..d473660 --- /dev/null +++ b/community/helpers/user_utils.py @@ -0,0 +1,199 @@ +"""User management utility functions.""" + +import fnmatch +import re +import time +from typing import List, Dict, Tuple +from mautrix.types import EventType, UserID +from mautrix.errors import MNotFound + + +async def check_if_banned(client, userid: str, banlists: List[str], logger) -> bool: + """Check if a user is banned according to banlists. + + Args: + client: Matrix client instance + userid: The user ID to check + banlists: List of banlist room IDs or aliases + logger: Logger instance for error reporting + + Returns: + bool: True if user is banned + """ + is_banned = False + myrooms = await client.get_joined_rooms() + banlist_roomids = await get_banlist_roomids(client, banlists, logger) + + for list_id in banlist_roomids: + if list_id not in myrooms: + logger.error( + f"Bot must be in {list_id} before attempting to use it as a banlist." + ) + continue + + try: + list_state = await client.get_state(list_id) + user_policies = list( + filter(lambda p: p.type.t == "m.policy.rule.user", list_state) + ) + except Exception as e: + logger.error(e) + continue + + for rule in user_policies: + try: + if bool( + fnmatch.fnmatch(userid, rule["content"]["entity"]) + ) and bool(re.search("ban$", rule["content"]["recommendation"])): + return True + except Exception: + # Skip invalid rules + pass + + return is_banned + + +async def get_banlist_roomids(client, banlists: List[str], logger) -> List[str]: + """Get room IDs for all configured banlists. + + Args: + client: Matrix client instance + banlists: List of banlist room IDs or aliases + logger: Logger instance for error reporting + + Returns: + list: List of room IDs for banlists + """ + banlist_roomids = [] + for banlist in banlists: + if banlist.startswith("#"): + try: + room_info = await client.resolve_room_alias(banlist) + list_id = room_info["room_id"] + banlist_roomids.append(list_id) + except Exception as e: + logger.error(f"Banlist fetching failed for {banlist}: {e}") + continue + else: + list_id = banlist + banlist_roomids.append(list_id) + + return banlist_roomids + + +async def ban_user_from_rooms(client, user: str, roomlist: List[str], reason: str = "banned", + all_rooms: bool = False, redact_on_ban: bool = False, + get_messages_to_redact_func=None, database=None, + sleep_time: float = 0.1, logger=None) -> Dict: + """Ban a user from a list of rooms. + + Args: + client: Matrix client instance + user: User ID to ban + roomlist: List of room IDs to ban from + reason: Reason for the ban + all_rooms: Whether to ban even if user is not in room + redact_on_ban: Whether to queue messages for redaction + get_messages_to_redact_func: Function to get messages to redact + database: Database instance for redaction tasks + sleep_time: Sleep time between operations + logger: Logger instance + + Returns: + dict: Ban results with success/error lists + """ + ban_event_map = {"ban_list": {}, "error_list": {}} + ban_event_map["ban_list"][user] = [] + + for room in roomlist: + try: + roomname = None + try: + roomnamestate = await client.get_state_event(room, "m.room.name") + roomname = roomnamestate["name"] + except: + pass + + # ban user even if they're not in the room! + if not all_rooms: + await client.get_state_event(room, EventType.ROOM_MEMBER, user) + + await client.ban_user(room, user, reason=reason) + if roomname: + ban_event_map["ban_list"][user].append(roomname) + else: + ban_event_map["ban_list"][user].append(room) + time.sleep(sleep_time) + except MNotFound: + pass + except Exception as e: + if logger: + logger.warning(e) + ban_event_map["error_list"][user] = [] + ban_event_map["error_list"][user].append(roomname or room) + + if redact_on_ban and get_messages_to_redact_func and database: + messages = await get_messages_to_redact_func(room, user) + # Queue messages for redaction + for msg in messages: + await database.execute( + "INSERT INTO redaction_tasks (event_id, room_id) VALUES ($1, $2)", + msg.event_id, + room, + ) + if logger: + logger.info( + f"Queued {len(messages)} messages for redaction in {roomname or room}" + ) + + return ban_event_map + + +async def user_permitted(client, user_id: UserID, parent_room: str, min_level: int = 50, + room_id: str = None, logger=None) -> bool: + """Check if a user has sufficient power level in a room. + + Args: + client: Matrix client instance + user_id: The Matrix ID of the user to check + parent_room: The parent room ID + min_level: Minimum required power level (default 50 for moderator) + room_id: The room ID to check permissions in. If None, uses parent room. + logger: Logger instance for error reporting + + Returns: + bool: True if user has sufficient power level + """ + try: + target_room = room_id or parent_room + + # First check if user has unlimited power (creator in modern room versions) + from .room_utils import user_has_unlimited_power + if await user_has_unlimited_power(client, user_id, target_room): + return True + + # Then check power level + power_levels = await client.get_state_event( + target_room, EventType.ROOM_POWER_LEVELS + ) + user_level = power_levels.get_user_level(user_id) + return user_level >= min_level + except Exception as e: + if logger: + logger.error(f"Failed to check user power level: {e}") + return False + + +async def user_has_unlimited_power(client, user_id: str, room_id: str) -> bool: + """Check if a user has unlimited power in a room (creator in modern room versions). + + Args: + client: Matrix client instance + user_id: The user ID to check + room_id: The room ID to check in + + Returns: + bool: True if user has unlimited power + """ + from .room_utils import user_has_unlimited_power as room_user_has_unlimited_power + return await room_user_has_unlimited_power(client, user_id, room_id) diff --git a/example-standalone-config.yaml b/example-standalone-config.yaml index c1d6195..8a4222e 100644 --- a/example-standalone-config.yaml +++ b/example-standalone-config.yaml @@ -93,24 +93,18 @@ plugin_config: # subject to removal by the purge command. kick_threshold_days: 60 - # track users? if false, will disable all tracking and avoid writing anything to the database. - track_users: True + # track users? empty list disables all tracking, otherwise specify what to track: + # - "messages": track user activity from messages + # - "reactions": track user activity from reactions + # examples: + # track_users: [] # disable all tracking + # track_users: ["messages"] # track only messages + # track_users: ["reactions"] # track only reactions + # track_users: ["messages", "reactions"] # track both + track_users: ["messages", "reactions"] - # track messages? if false, will not track user activity timestamps. enable if you'd like to track - # inactive users in your community - track_messages: True - - # track reactions? if false, will only track activity based on normal message events, but if true - # will update the user's last-active date when they add a reaction to a message - track_reactions: True - - # list of users who can use administrative commands. these users will also be made room admins (PL100) - # DEPRECATED: set user powerlevels in the parent room instead. - admins: [] - - # list of users who should be considered community moderators. these users will be made room mods (PL50) - # DEPRECATED: set userpowerlevels in the parent room instead. - moderators: [] + # Note: Admin and moderator permissions are now managed through power levels + # in the parent room. Set user power levels directly in the parent space. # list of users who should be invited to new rooms immediately (other bots, moderators, perhaps) # use full matrix IDs here diff --git a/maubot.yaml b/maubot.yaml index 96f93e2..2dff14f 100644 --- a/maubot.yaml +++ b/maubot.yaml @@ -1,6 +1,6 @@ maubot: 0.1.0 id: org.jobmachine.communitybot -version: 0.2.12 +version: 0.3.0 license: MIT modules: - community diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..7c9f712 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,6 @@ +[tool:pytest] +testpaths = tests +python_files = test_*.py +python_classes = Test* +python_functions = test_* +addopts = -v --tb=short diff --git a/run_tests.py b/run_tests.py new file mode 100755 index 0000000..4b70202 --- /dev/null +++ b/run_tests.py @@ -0,0 +1,35 @@ +#!/usr/bin/env python3 +"""Test runner for the community bot project.""" + +import sys +import subprocess +import os + +def run_tests(): + """Run all tests for the community bot project.""" + print("Running community bot tests...") + + # Change to the project directory + project_dir = os.path.dirname(os.path.abspath(__file__)) + os.chdir(project_dir) + + # Run pytest + try: + result = subprocess.run([ + sys.executable, "-m", "pytest", + "tests/", + "-v", + "--tb=short", + "--color=yes" + ], check=True) + print("\n✅ All tests passed!") + return 0 + except subprocess.CalledProcessError as e: + print(f"\n❌ Tests failed with exit code {e.returncode}") + return e.returncode + except FileNotFoundError: + print("❌ pytest not found. Please install it with: pip install pytest") + return 1 + +if __name__ == "__main__": + sys.exit(run_tests()) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..3fd35c0 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +# Test package for community bot diff --git a/tests/test_bot_commands.py b/tests/test_bot_commands.py new file mode 100644 index 0000000..7f0619d --- /dev/null +++ b/tests/test_bot_commands.py @@ -0,0 +1,430 @@ +"""Tests for bot command handlers.""" + +import pytest +from unittest.mock import Mock, AsyncMock, patch, MagicMock +from mautrix.types import EventType, UserID, MessageEvent, StateEvent +from mautrix.errors import MNotFound + +from community.bot import CommunityBot + + +class TestBotCommands: + """Test cases for bot command handlers.""" + + @pytest.fixture + def bot(self): + """Create a mock bot instance for testing.""" + bot = Mock(spec=CommunityBot) + bot.client = Mock() + bot.database = Mock() + bot.log = Mock() + bot.config = { + "parent_room": "!parent:example.com", + "community_slug": "test", + "track_users": True, + "warn_threshold_days": 7, + "kick_threshold_days": 14, + "sleep": 0.1, + "censor_wordlist": [r"badword"], + "censor_files": False, + "censor": True, + "banlists": ["!banlist:example.com"], + "redact_on_ban": False, + "admins": [], + "moderators": [] + } + return bot + + @pytest.fixture + def mock_evt(self): + """Create a mock MessageEvent for testing.""" + evt = Mock(spec=MessageEvent) + evt.sender = "@user:example.com" + evt.room_id = "!room:example.com" + evt.reply = AsyncMock() + evt.respond = AsyncMock() + return evt + + @pytest.mark.asyncio + async def test_check_parent_room_configured(self, bot, mock_evt): + """Test check_parent_room when parent room is configured.""" + # Use the mock bot instance + bot.config = {"parent_room": "!parent:example.com"} + + result = await bot.check_parent_room(mock_evt) + + assert result == True + mock_evt.reply.assert_not_called() + + @pytest.mark.asyncio + async def test_check_parent_room_not_configured(self, bot, mock_evt): + """Test check_parent_room when parent room is not configured.""" + bot.config = {"parent_room": None} + + result = await bot.check_parent_room(mock_evt) + + assert result == False + mock_evt.reply.assert_called_once() + + @pytest.mark.asyncio + async def test_check_banlists_command(self, bot, mock_evt): + """Test the check_banlists command.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.log = bot.log + + # Mock the check_if_banned method + with patch.object(real_bot, 'check_if_banned', return_value=True): + await real_bot.check_banlists(mock_evt, "@test:example.com") + + mock_evt.reply.assert_called_once_with("user on banlist: True") + + @pytest.mark.asyncio + async def test_sync_space_members_command(self, bot, mock_evt): + """Test the sync_space_members command.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.database = bot.database + real_bot.log = bot.log + + # Mock required methods + with patch.object(real_bot, 'user_permitted', return_value=True), \ + patch.object(real_bot, 'do_sync', return_value={"added": [], "dropped": []}): + + await real_bot.sync_space_members(mock_evt) + + mock_evt.respond.assert_called() + + @pytest.mark.asyncio + async def test_sync_space_members_no_permission(self, bot, mock_evt): + """Test sync_space_members command without permission.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.log = bot.log + + with patch.object(real_bot, 'user_permitted', return_value=False): + await real_bot.sync_space_members(mock_evt) + + mock_evt.reply.assert_called_once_with("You don't have permission to use this command") + + @pytest.mark.asyncio + async def test_sync_space_members_tracking_disabled(self, bot, mock_evt): + """Test sync_space_members command when tracking is disabled.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = {**bot.config, "track_users": False} + real_bot.client = bot.client + real_bot.log = bot.log + + with patch.object(real_bot, 'user_permitted', return_value=True): + await real_bot.sync_space_members(mock_evt) + + mock_evt.respond.assert_called_once_with("user tracking is disabled") + + @pytest.mark.asyncio + async def test_ignore_command(self, bot, mock_evt): + """Test the ignore command.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.database = bot.database + real_bot.log = bot.log + + # Mock database operations + real_bot.database.execute = AsyncMock() + + with patch.object(real_bot, 'user_permitted', return_value=True): + await real_bot.ignore_user(mock_evt, "@test:example.com") + + real_bot.database.execute.assert_called() + mock_evt.reply.assert_called() + + @pytest.mark.asyncio + async def test_unignore_command(self, bot, mock_evt): + """Test the unignore command.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.database = bot.database + real_bot.log = bot.log + + # Mock database operations + real_bot.database.execute = AsyncMock() + + with patch.object(real_bot, 'user_permitted', return_value=True): + await real_bot.unignore_user(mock_evt, "@test:example.com") + + real_bot.database.execute.assert_called() + mock_evt.reply.assert_called() + + @pytest.mark.asyncio + async def test_kick_command(self, bot, mock_evt): + """Test the kick command.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.log = bot.log + + # Mock required methods + with patch.object(real_bot, 'user_permitted', return_value=True), \ + patch.object(real_bot, 'get_space_roomlist', return_value=["!room1:example.com"]), \ + patch.object(real_bot, 'ban_this_user', return_value={"ban_list": {}, "error_list": {}}): + + await real_bot.kick_user(mock_evt, "@test:example.com") + + mock_evt.reply.assert_called() + + @pytest.mark.asyncio + async def test_ban_command(self, bot, mock_evt): + """Test the ban command.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.log = bot.log + + # Mock required methods + with patch.object(real_bot, 'user_permitted', return_value=True), \ + patch.object(real_bot, 'get_space_roomlist', return_value=["!room1:example.com"]), \ + patch.object(real_bot, 'ban_this_user', return_value={"ban_list": {}, "error_list": {}}): + + await real_bot.ban_user(mock_evt, "@test:example.com") + + mock_evt.reply.assert_called() + + @pytest.mark.asyncio + async def test_doctor_command(self, bot, mock_evt): + """Test the doctor command.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.database = bot.database + real_bot.log = bot.log + + # Mock required methods + with patch.object(real_bot, 'user_permitted', return_value=True), \ + patch.object(real_bot, 'get_space_roomlist', return_value=["!room1:example.com"]): + + await real_bot.doctor(mock_evt) + + mock_evt.respond.assert_called() + + @pytest.mark.asyncio + async def test_doctor_room_detail_command(self, bot, mock_evt): + """Test the doctor room detail command.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.database = bot.database + real_bot.log = bot.log + + # Mock required methods + with patch.object(real_bot, 'user_permitted', return_value=True), \ + patch.object(real_bot, '_doctor_room_detail', return_value=None): + + await real_bot.doctor_room_detail(mock_evt, "!room:example.com") + + mock_evt.respond.assert_called() + + @pytest.mark.asyncio + async def test_initialize_command(self, bot, mock_evt): + """Test the initialize command.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.database = bot.database + real_bot.log = bot.log + + # Mock required methods + with patch.object(real_bot, 'user_permitted', return_value=True), \ + patch.object(real_bot, 'create_space', return_value=("!space:example.com", "#space:example.com")): + + await real_bot.initialize(mock_evt, "Test Community") + + mock_evt.respond.assert_called() + + @pytest.mark.asyncio + async def test_create_room_command(self, bot, mock_evt): + """Test the create_room command.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.log = bot.log + + # Mock required methods + with patch.object(real_bot, 'user_permitted', return_value=True), \ + patch.object(real_bot, 'validate_room_aliases', return_value=(True, [])), \ + patch.object(real_bot, 'create_room', return_value=("!room:example.com", "#room:example.com")): + + await real_bot.create_room(mock_evt, "Test Room") + + mock_evt.respond.assert_called() + + @pytest.mark.asyncio + async def test_create_room_command_alias_conflict(self, bot, mock_evt): + """Test create_room command with alias conflict.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.log = bot.log + + # Mock required methods + with patch.object(real_bot, 'user_permitted', return_value=True), \ + patch.object(real_bot, 'validate_room_aliases', return_value=(False, ["#conflict:example.com"])): + + await real_bot.create_room(mock_evt, "Test Room") + + mock_evt.respond.assert_called() + # Should mention the conflict + + @pytest.mark.asyncio + async def test_archive_room_command(self, bot, mock_evt): + """Test the archive_room command.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.log = bot.log + + # Mock required methods + with patch.object(real_bot, 'user_permitted', return_value=True), \ + patch.object(real_bot, 'do_archive_room', return_value=None): + + await real_bot.archive_room(mock_evt, "!room:example.com") + + mock_evt.respond.assert_called() + + @pytest.mark.asyncio + async def test_remove_room_command(self, bot, mock_evt): + """Test the remove_room command.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.log = bot.log + + # Mock required methods + with patch.object(real_bot, 'user_permitted', return_value=True), \ + patch.object(real_bot, 'remove_room_aliases', return_value=[]): + + await real_bot.remove_room(mock_evt, "!room:example.com") + + mock_evt.respond.assert_called() + + @pytest.mark.asyncio + async def test_join_room_command(self, bot, mock_evt): + """Test the join_room command.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.log = bot.log + + # Mock required methods + with patch.object(real_bot, 'user_permitted', return_value=True), \ + patch.object(real_bot, 'join_room', return_value="!room:example.com"): + + await real_bot.join_room(mock_evt, "!room:example.com") + + mock_evt.respond.assert_called() + + @pytest.mark.asyncio + async def test_leave_room_command(self, bot, mock_evt): + """Test the leave_room command.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.log = bot.log + + # Mock required methods + with patch.object(real_bot, 'user_permitted', return_value=True), \ + patch.object(real_bot, 'leave_room', return_value=None): + + await real_bot.leave_room(mock_evt, "!room:example.com") + + mock_evt.respond.assert_called() + + @pytest.mark.asyncio + async def test_verify_command(self, bot, mock_evt): + """Test the verify command.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.database = bot.database + real_bot.log = bot.log + + # Mock required methods + with patch.object(real_bot, 'user_permitted', return_value=True), \ + patch.object(real_bot, 'create_verification_dm', return_value="!dm:example.com"): + + await real_bot.verify_user(mock_evt, "@test:example.com", "!room:example.com") + + mock_evt.respond.assert_called() + + @pytest.mark.asyncio + async def test_commands_require_permission(self, bot, mock_evt): + """Test that commands require proper permissions.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.log = bot.log + + # Test various commands that require permission + commands_to_test = [ + ('sync_space_members', []), + ('ignore_user', ['@test:example.com']), + ('unignore_user', ['@test:example.com']), + ('kick_user', ['@test:example.com']), + ('ban_user', ['@test:example.com']), + ('doctor', []), + ('doctor_room_detail', ['!room:example.com']), + ('initialize', ['Test Community']), + ('create_room', ['Test Room']), + ('archive_room', ['!room:example.com']), + ('remove_room', ['!room:example.com']), + ('join_room', ['!room:example.com']), + ('leave_room', ['!room:example.com']), + ('verify_user', ['@test:example.com', '!room:example.com']) + ] + + for command_name, args in commands_to_test: + with patch.object(real_bot, 'user_permitted', return_value=False): + command_func = getattr(real_bot, command_name) + await command_func(mock_evt, *args) + + # Should respond with permission denied message + mock_evt.reply.assert_called() + mock_evt.reply.reset_mock() diff --git a/tests/test_bot_events.py b/tests/test_bot_events.py new file mode 100644 index 0000000..9ccb5b3 --- /dev/null +++ b/tests/test_bot_events.py @@ -0,0 +1,452 @@ +"""Tests for bot event handlers.""" + +import pytest +from unittest.mock import Mock, AsyncMock, patch, MagicMock +from mautrix.types import EventType, UserID, MessageEvent, StateEvent, ReactionEvent +from mautrix.errors import MNotFound + +from community.bot import CommunityBot + + +class TestBotEvents: + """Test cases for bot event handlers.""" + + @pytest.fixture + def bot(self): + """Create a mock bot instance for testing.""" + bot = Mock(spec=CommunityBot) + bot.client = Mock() + bot.database = Mock() + bot.log = Mock() + bot.config = { + "parent_room": "!parent:example.com", + "community_slug": "test", + "track_users": True, + "track_messages": True, + "track_reactions": True, + "warn_threshold_days": 7, + "kick_threshold_days": 14, + "sleep": 0.1, + "censor_wordlist": [r"badword"], + "censor_files": False, + "censor": True, + "banlists": ["!banlist:example.com"], + "redact_on_ban": False, + "proactive_banning": True, + "check_if_human": True, + "verification_phrases": ["test phrase"], + "verification_attempts": 3, + "verification_message": "Please verify", + "invite_power_level": 50, + "uncensor_pl": 50 + } + return bot + + @pytest.fixture + def mock_message_evt(self): + """Create a mock MessageEvent for testing.""" + evt = Mock(spec=MessageEvent) + evt.sender = "@user:example.com" + evt.room_id = "!room:example.com" + evt.timestamp = 1234567890 + evt.content = Mock() + evt.content.body = "test message" + evt.content.msgtype = "m.text" + evt.reply = AsyncMock() + evt.respond = AsyncMock() + evt.react = AsyncMock() + return evt + + @pytest.fixture + def mock_state_evt(self): + """Create a mock StateEvent for testing.""" + evt = Mock(spec=StateEvent) + evt.sender = "@user:example.com" + evt.room_id = "!room:example.com" + evt.state_key = "@user:example.com" + evt.content = { + "entity": "@banned:example.com", + "recommendation": "ban" + } + evt.prev_content = {} + return evt + + @pytest.fixture + def mock_reaction_evt(self): + """Create a mock ReactionEvent for testing.""" + evt = Mock(spec=ReactionEvent) + evt.sender = "@user:example.com" + evt.room_id = "!room:example.com" + evt.content = Mock() + evt.content.relates_to = Mock() + evt.content.relates_to.event_id = "!msg:example.com" + evt.content.relates_to.key = "👍" + return evt + + @pytest.mark.asyncio + async def test_check_ban_event_proactive_banning_enabled(self, bot, mock_state_evt): + """Test ban event handler with proactive banning enabled.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.log = bot.log + + # Mock required methods + with patch.object(real_bot, 'get_banlist_roomids', return_value=["!banlist:example.com"]), \ + patch.object(real_bot, 'ban_this_user', return_value={"ban_list": {}, "error_list": {}}): + + await real_bot.check_ban_event(mock_state_evt) + + # Should call ban_this_user + real_bot.ban_this_user.assert_called_once_with("@banned:example.com") + + @pytest.mark.asyncio + async def test_check_ban_event_proactive_banning_disabled(self, bot, mock_state_evt): + """Test ban event handler with proactive banning disabled.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = {**bot.config, "proactive_banning": False} + real_bot.client = bot.client + real_bot.log = bot.log + + with patch.object(real_bot, 'get_banlist_roomids', return_value=["!banlist:example.com"]): + await real_bot.check_ban_event(mock_state_evt) + + # Should not call ban_this_user + real_bot.ban_this_user.assert_not_called() + + @pytest.mark.asyncio + async def test_check_ban_event_wrong_room(self, bot, mock_state_evt): + """Test ban event handler with wrong room.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.log = bot.log + + with patch.object(real_bot, 'get_banlist_roomids', return_value=["!other:example.com"]): + await real_bot.check_ban_event(mock_state_evt) + + # Should not call ban_this_user + real_bot.ban_this_user.assert_not_called() + + @pytest.mark.asyncio + async def test_sync_power_levels(self, bot, mock_state_evt): + """Test power levels sync event handler.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.log = bot.log + + # Mock power level changes + mock_state_evt.prev_content = {"users": {"@user:example.com": 25}} + mock_state_evt.content = {"users": {"@user:example.com": 50}} + + with patch.object(real_bot, 'get_space_roomlist', return_value=["!room1:example.com", "!room2:example.com"]), \ + patch.object(real_bot, 'sync_power_levels_to_room', return_value=None): + + await real_bot.sync_power_levels(mock_state_evt) + + # Should sync to all rooms + assert real_bot.sync_power_levels_to_room.call_count == 2 + + @pytest.mark.asyncio + async def test_sync_power_levels_wrong_room(self, bot, mock_state_evt): + """Test power levels sync with wrong room.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.log = bot.log + + mock_state_evt.room_id = "!other:example.com" + + with patch.object(real_bot, 'get_space_roomlist', return_value=["!room1:example.com"]): + await real_bot.sync_power_levels(mock_state_evt) + + # Should not sync to any rooms + real_bot.sync_power_levels_to_room.assert_not_called() + + @pytest.mark.asyncio + async def test_handle_leave_events(self, bot, mock_state_evt): + """Test leave events handler.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.database = bot.database + real_bot.log = bot.log + + # Mock database operations + real_bot.database.execute = AsyncMock() + + with patch.object(real_bot, 'get_space_roomlist', return_value=["!room1:example.com", "!room2:example.com"]): + await real_bot.handle_leave_events(mock_state_evt) + + # Should delete user from database + real_bot.database.execute.assert_called() + + @pytest.mark.asyncio + async def test_handle_leave(self, bot, mock_state_evt): + """Test leave event handler.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.database = bot.database + real_bot.log = bot.log + + with patch.object(real_bot, 'handle_leave_events', return_value=None): + await real_bot.handle_leave(mock_state_evt) + + real_bot.handle_leave_events.assert_called_once_with(mock_state_evt) + + @pytest.mark.asyncio + async def test_handle_kick(self, bot, mock_state_evt): + """Test kick event handler.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.database = bot.database + real_bot.log = bot.log + + with patch.object(real_bot, 'handle_leave_events', return_value=None): + await real_bot.handle_kick(mock_state_evt) + + real_bot.handle_leave_events.assert_called_once_with(mock_state_evt) + + @pytest.mark.asyncio + async def test_handle_ban(self, bot, mock_state_evt): + """Test ban event handler.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.database = bot.database + real_bot.log = bot.log + + with patch.object(real_bot, 'handle_leave_events', return_value=None): + await real_bot.handle_ban(mock_state_evt) + + real_bot.handle_leave_events.assert_called_once_with(mock_state_evt) + + @pytest.mark.asyncio + async def test_newjoin_event(self, bot, mock_state_evt): + """Test new join event handler.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.database = bot.database + real_bot.log = bot.log + + # Mock database operations + real_bot.database.execute = AsyncMock() + + with patch.object(real_bot, 'get_space_roomlist', return_value=["!room1:example.com", "!room2:example.com"]), \ + patch.object(real_bot, 'upsert_user_timestamp', return_value=None): + + await real_bot.newjoin(mock_state_evt) + + # Should update user timestamp + real_bot.upsert_user_timestamp.assert_called() + + @pytest.mark.asyncio + async def test_update_message_timestamp_tracking_enabled(self, bot, mock_message_evt): + """Test message timestamp update with tracking enabled.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.database = bot.database + real_bot.log = bot.log + + # Mock power levels + power_levels = Mock() + power_levels.get_user_level.return_value = 25 + + real_bot.client.get_state_event = AsyncMock(return_value=power_levels) + real_bot.database.execute = AsyncMock() + + with patch.object(real_bot, 'get_space_roomlist', return_value=["!room1:example.com"]), \ + patch.object(real_bot, 'upsert_user_timestamp', return_value=None): + + await real_bot.update_message_timestamp(mock_message_evt) + + # Should update user timestamp + real_bot.upsert_user_timestamp.assert_called() + + @pytest.mark.asyncio + async def test_update_message_timestamp_tracking_disabled(self, bot, mock_message_evt): + """Test message timestamp update with tracking disabled.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = {**bot.config, "track_messages": False} + real_bot.client = bot.client + real_bot.database = bot.database + real_bot.log = bot.log + + with patch.object(real_bot, 'get_space_roomlist', return_value=["!room1:example.com"]): + await real_bot.update_message_timestamp(mock_message_evt) + + # Should not update user timestamp + real_bot.upsert_user_timestamp.assert_not_called() + + @pytest.mark.asyncio + async def test_handle_verification(self, bot, mock_message_evt): + """Test verification message handler.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.database = bot.database + real_bot.log = bot.log + + # Mock verification state + verification_state = { + "user_id": "@user:example.com", + "target_room_id": "!room:example.com", + "verification_phrase": "test phrase", + "attempts_remaining": 3, + "required_power_level": 50 + } + + real_bot.database.fetchrow = AsyncMock(return_value=verification_state) + real_bot.database.execute = AsyncMock() + + # Mock message content + mock_message_evt.content.body = "test phrase" + + with patch.object(real_bot, 'user_permitted', return_value=True), \ + patch.object(real_bot, 'join_room', return_value="!room:example.com"): + + await real_bot.handle_verification(mock_message_evt) + + # Should process verification + real_bot.database.execute.assert_called() + + @pytest.mark.asyncio + async def test_handle_verification_wrong_phrase(self, bot, mock_message_evt): + """Test verification with wrong phrase.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.database = bot.database + real_bot.log = bot.log + + # Mock verification state + verification_state = { + "user_id": "@user:example.com", + "target_room_id": "!room:example.com", + "verification_phrase": "correct phrase", + "attempts_remaining": 3, + "required_power_level": 50 + } + + real_bot.database.fetchrow = AsyncMock(return_value=verification_state) + real_bot.database.execute = AsyncMock() + + # Mock message content with wrong phrase + mock_message_evt.content.body = "wrong phrase" + + await real_bot.handle_verification(mock_message_evt) + + # Should decrement attempts + real_bot.database.execute.assert_called() + + @pytest.mark.asyncio + async def test_handle_verification_no_state(self, bot, mock_message_evt): + """Test verification with no verification state.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.database = bot.database + real_bot.log = bot.log + + # Mock no verification state + real_bot.database.fetchrow = AsyncMock(return_value=None) + + await real_bot.handle_verification(mock_message_evt) + + # Should not process verification + real_bot.database.execute.assert_not_called() + + @pytest.mark.asyncio + async def test_handle_reaction_tracking_enabled(self, bot, mock_reaction_evt): + """Test reaction event handler with tracking enabled.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.database = bot.database + real_bot.log = bot.log + + # Mock power levels + power_levels = Mock() + power_levels.get_user_level.return_value = 25 + + real_bot.client.get_state_event = AsyncMock(return_value=power_levels) + real_bot.database.execute = AsyncMock() + + with patch.object(real_bot, 'get_space_roomlist', return_value=["!room1:example.com"]), \ + patch.object(real_bot, 'upsert_user_timestamp', return_value=None): + + await real_bot.handle_reaction(mock_reaction_evt) + + # Should update user timestamp + real_bot.upsert_user_timestamp.assert_called() + + @pytest.mark.asyncio + async def test_handle_reaction_tracking_disabled(self, bot, mock_reaction_evt): + """Test reaction event handler with tracking disabled.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = {**bot.config, "track_reactions": False} + real_bot.client = bot.client + real_bot.database = bot.database + real_bot.log = bot.log + + with patch.object(real_bot, 'get_space_roomlist', return_value=["!room1:example.com"]): + await real_bot.handle_reaction(mock_reaction_evt) + + # Should not update user timestamp + real_bot.upsert_user_timestamp.assert_not_called() + + @pytest.mark.asyncio + async def test_handle_reaction_wrong_room(self, bot, mock_reaction_evt): + """Test reaction event handler with wrong room.""" + from community.bot import CommunityBot + + real_bot = CommunityBot() + real_bot.config = bot.config + real_bot.client = bot.client + real_bot.database = bot.database + real_bot.log = bot.log + + with patch.object(real_bot, 'get_space_roomlist', return_value=["!other:example.com"]): + await real_bot.handle_reaction(mock_reaction_evt) + + # Should not update user timestamp + real_bot.upsert_user_timestamp.assert_not_called() diff --git a/tests/test_database_utils.py b/tests/test_database_utils.py new file mode 100644 index 0000000..65bb44c --- /dev/null +++ b/tests/test_database_utils.py @@ -0,0 +1,253 @@ +"""Tests for database utility functions.""" + +import pytest +from unittest.mock import Mock, AsyncMock, patch +import asyncio + +from community.helpers.database_utils import ( + get_messages_to_redact, redact_messages, upsert_user_timestamp, + get_inactive_users, cleanup_stale_verification_states, + get_verification_state, create_verification_state, + update_verification_attempts, delete_verification_state +) + + +class TestDatabaseUtils: + """Test cases for database utility functions.""" + + @pytest.mark.asyncio + async def test_get_messages_to_redact_success(self): + """Test getting messages to redact successfully.""" + client = Mock() + + # Mock message events + msg1 = Mock() + msg1.content = Mock() + msg1.content.serialize.return_value = {"body": "test"} + + msg2 = Mock() + msg2.content = None + + msg3 = Mock() + msg3.content = Mock() + msg3.content.serialize.return_value = {"body": "test2"} + + messages = Mock() + messages.events = [msg1, msg2, msg3] + + client.get_messages = AsyncMock(return_value=messages) + + logger = Mock() + result = await get_messages_to_redact(client, "!room:example.com", "@user:example.com", logger) + + assert len(result) == 2 # Only msg1 and msg3 have content + assert msg1 in result + assert msg3 in result + assert msg2 not in result + + @pytest.mark.asyncio + async def test_get_messages_to_redact_error(self): + """Test getting messages to redact with error.""" + client = Mock() + client.get_messages = AsyncMock(side_effect=Exception("Network error")) + + logger = Mock() + result = await get_messages_to_redact(client, "!room:example.com", "@user:example.com", logger) + + assert result == [] + logger.error.assert_called() + + @pytest.mark.asyncio + async def test_redact_messages_success(self): + """Test redacting messages successfully.""" + client = Mock() + client.redact = AsyncMock() + + database = Mock() + database.fetch = AsyncMock(return_value=[ + {"event_id": "!msg1:example.com"}, + {"event_id": "!msg2:example.com"} + ]) + database.execute = AsyncMock() + + logger = Mock() + + with patch('asyncio.sleep', new_callable=AsyncMock): + result = await redact_messages(client, database, "!room:example.com", 0.1, logger) + + assert result["success"] == 2 + assert result["failure"] == 0 + assert client.redact.call_count == 2 + assert database.execute.call_count == 2 + + @pytest.mark.asyncio + async def test_redact_messages_rate_limited(self): + """Test redacting messages with rate limiting.""" + client = Mock() + client.redact = AsyncMock(side_effect=Exception("Too Many Requests")) + + database = Mock() + database.fetch = AsyncMock(return_value=[ + {"event_id": "!msg1:example.com"} + ]) + + logger = Mock() + + with patch('asyncio.sleep', new_callable=AsyncMock): + result = await redact_messages(client, database, "!room:example.com", 0.1, logger) + + assert result["success"] == 0 + assert result["failure"] == 0 # Rate limited, so no failure count + logger.warning.assert_called() + + @pytest.mark.asyncio + async def test_upsert_user_timestamp_success(self): + """Test upserting user timestamp successfully.""" + database = Mock() + database.execute = AsyncMock() + + logger = Mock() + + await upsert_user_timestamp(database, "@user:example.com", 1234567890, logger) + + database.execute.assert_called_once() + + @pytest.mark.asyncio + async def test_get_inactive_users_success(self): + """Test getting inactive users successfully.""" + database = Mock() + database.fetch = AsyncMock(side_effect=[ + [{"mxid": "@user1:example.com"}, {"mxid": "@user2:example.com"}], # warn results + [{"mxid": "@user3:example.com"}] # kick results + ]) + + logger = Mock() + + with patch('time.time', return_value=1234567890): + result = await get_inactive_users(database, 7, 14, logger) + + assert len(result["warn"]) == 2 + assert len(result["kick"]) == 1 + assert "@user1:example.com" in result["warn"] + assert "@user3:example.com" in result["kick"] + + @pytest.mark.asyncio + async def test_get_inactive_users_error(self): + """Test getting inactive users with error.""" + database = Mock() + database.fetch = AsyncMock(side_effect=Exception("Database error")) + + logger = Mock() + + result = await get_inactive_users(database, 7, 14, logger) + + assert result == {"warn": [], "kick": []} + logger.error.assert_called() + + @pytest.mark.asyncio + async def test_cleanup_stale_verification_states_success(self): + """Test cleaning up stale verification states successfully.""" + database = Mock() + database.execute = AsyncMock() + + logger = Mock() + + await cleanup_stale_verification_states(database, logger) + + database.execute.assert_called_once() + + @pytest.mark.asyncio + async def test_cleanup_stale_verification_states_error(self): + """Test cleaning up stale verification states with error.""" + database = Mock() + database.execute = AsyncMock(side_effect=Exception("Database error")) + + logger = Mock() + + await cleanup_stale_verification_states(database, logger) + + logger.error.assert_called() + + @pytest.mark.asyncio + async def test_get_verification_state_success(self): + """Test getting verification state successfully.""" + database = Mock() + database.fetchrow = AsyncMock(return_value={ + "dm_room_id": "!dm:example.com", + "user_id": "@user:example.com", + "target_room_id": "!room:example.com", + "verification_phrase": "test phrase", + "attempts_remaining": 3, + "required_power_level": 50 + }) + + result = await get_verification_state(database, "!dm:example.com") + + assert result is not None + assert result["dm_room_id"] == "!dm:example.com" + assert result["user_id"] == "@user:example.com" + + @pytest.mark.asyncio + async def test_get_verification_state_not_found(self): + """Test getting verification state when not found.""" + database = Mock() + database.fetchrow = AsyncMock(return_value=None) + + result = await get_verification_state(database, "!dm:example.com") + + assert result is None + + @pytest.mark.asyncio + async def test_get_verification_state_error(self): + """Test getting verification state with error.""" + database = Mock() + database.fetchrow = AsyncMock(side_effect=Exception("Database error")) + + result = await get_verification_state(database, "!dm:example.com") + + assert result is None + + @pytest.mark.asyncio + async def test_create_verification_state_success(self): + """Test creating verification state successfully.""" + database = Mock() + database.execute = AsyncMock() + + await create_verification_state( + database, "!dm:example.com", "@user:example.com", + "!room:example.com", "test phrase", 3, 50 + ) + + database.execute.assert_called_once() + + @pytest.mark.asyncio + async def test_create_verification_state_error(self): + """Test creating verification state with error (should not raise).""" + database = Mock() + database.execute = AsyncMock(side_effect=Exception("Database error")) + + # Should not raise exception + await create_verification_state( + database, "!dm:example.com", "@user:example.com", + "!room:example.com", "test phrase", 3, 50 + ) + + @pytest.mark.asyncio + async def test_update_verification_attempts_success(self): + """Test updating verification attempts successfully.""" + database = Mock() + database.execute = AsyncMock() + + await update_verification_attempts(database, "!dm:example.com", 2) + + database.execute.assert_called_once() + + @pytest.mark.asyncio + async def test_delete_verification_state_success(self): + """Test deleting verification state successfully.""" + database = Mock() + database.execute = AsyncMock() + + await delete_verification_state(database, "!dm:example.com") + + database.execute.assert_called_once() diff --git a/tests/test_message_utils.py b/tests/test_message_utils.py new file mode 100644 index 0000000..2f1f44e --- /dev/null +++ b/tests/test_message_utils.py @@ -0,0 +1,106 @@ +"""Tests for message utility functions.""" + +import pytest +from unittest.mock import Mock +from mautrix.types import MessageType, MediaMessageEventContent + +from community.helpers.message_utils import ( + flag_message, flag_instaban, censor_room, + sanitize_room_name, generate_community_slug +) + + +class TestMessageUtils: + """Test cases for message utility functions.""" + + def test_flag_message_file_types(self): + """Test that file messages are flagged when censor_files is True.""" + msg = Mock() + msg.content.msgtype = MessageType.FILE + msg.content.body = "test file" + + assert flag_message(msg, [], True) == True + assert flag_message(msg, [], False) == False + + def test_flag_message_wordlist(self): + """Test that messages are flagged based on wordlist patterns.""" + msg = Mock() + msg.content.msgtype = MessageType.TEXT + msg.content.body = "This is a test message with badword" + + wordlist = [r"badword", r"another.*pattern"] + + assert flag_message(msg, wordlist, False) == True + + msg.content.body = "This is a clean message" + assert flag_message(msg, wordlist, False) == False + + def test_flag_message_invalid_regex(self): + """Test that invalid regex patterns are handled gracefully.""" + msg = Mock() + msg.content.msgtype = MessageType.TEXT + msg.content.body = "test message" + + wordlist = [r"valid.*pattern", r"[invalid", r"another.*pattern"] + + # Should not raise exception and should work with valid patterns + result = flag_message(msg, wordlist, False) + assert isinstance(result, bool) + + def test_flag_instaban(self): + """Test instant ban flagging.""" + msg = Mock() + msg.content.msgtype = MessageType.TEXT + msg.content.body = "This contains instaban_word" + + instaban_list = [r"instaban_word", r"another.*instaban"] + + assert flag_instaban(msg, instaban_list) == True + + msg.content.body = "This is clean" + assert flag_instaban(msg, instaban_list) == False + + def test_censor_room_boolean_config(self): + """Test room censoring with boolean configuration.""" + msg = Mock() + msg.room_id = "!room123:example.com" + + assert censor_room(msg, True) == True + assert censor_room(msg, False) == False + + def test_censor_room_list_config(self): + """Test room censoring with list configuration.""" + msg = Mock() + msg.room_id = "!room123:example.com" + + censor_list = ["!room123:example.com", "!room456:example.com"] + + assert censor_room(msg, censor_list) == True + + msg.room_id = "!room789:example.com" + assert censor_room(msg, censor_list) == False + + def test_censor_room_invalid_config(self): + """Test room censoring with invalid configuration.""" + msg = Mock() + msg.room_id = "!room123:example.com" + + assert censor_room(msg, "invalid") == False + assert censor_room(msg, None) == False + + def test_sanitize_room_name(self): + """Test room name sanitization.""" + assert sanitize_room_name("Test Room 123") == "testroom123" + assert sanitize_room_name("Special@#$%Characters") == "specialcharacters" + assert sanitize_room_name("UPPERCASE") == "uppercase" + assert sanitize_room_name("123 Numbers") == "123numbers" + assert sanitize_room_name("") == "" + + def test_generate_community_slug(self): + """Test community slug generation.""" + assert generate_community_slug("Test Community") == "tc" + assert generate_community_slug("My Awesome Community") == "mac" + assert generate_community_slug("Single") == "s" + assert generate_community_slug("Multiple Spaces") == "ms" + assert generate_community_slug("") == "" + assert generate_community_slug(" ") == "" diff --git a/tests/test_report_utils.py b/tests/test_report_utils.py new file mode 100644 index 0000000..51cae50 --- /dev/null +++ b/tests/test_report_utils.py @@ -0,0 +1,163 @@ +"""Tests for report utility functions.""" + +import pytest + +from community.helpers.report_utils import ( + generate_activity_report, split_doctor_report, format_ban_results, + format_sync_results +) + + +class TestReportUtils: + """Test cases for report utility functions.""" + + def test_generate_activity_report_success(self): + """Test generating activity report successfully.""" + database_results = { + "active": [{"mxid": "@user1:example.com"}, {"mxid": "@user2:example.com"}], + "inactive": [{"mxid": "@user3:example.com"}], + "ignored": [{"mxid": "@user4:example.com"}] + } + + result = generate_activity_report(database_results) + + assert result["active"] == ["@user1:example.com", "@user2:example.com"] + assert result["inactive"] == ["@user3:example.com"] + assert result["ignored"] == ["@user4:example.com"] + + def test_generate_activity_report_empty(self): + """Test generating activity report with empty results.""" + database_results = { + "active": [], + "inactive": [], + "ignored": [] + } + + result = generate_activity_report(database_results) + + assert result["active"] == ["none"] + assert result["inactive"] == ["none"] + assert result["ignored"] == ["none"] + + def test_generate_activity_report_missing_keys(self): + """Test generating activity report with missing keys.""" + database_results = {} + + result = generate_activity_report(database_results) + + assert result["active"] == ["none"] + assert result["inactive"] == ["none"] + assert result["ignored"] == ["none"] + + def test_split_doctor_report_small(self): + """Test splitting small report that doesn't need splitting.""" + report_text = "This is a small report that fits in one chunk." + + result = split_doctor_report(report_text, 1000) + + assert len(result) == 1 + assert result[0] == report_text + + def test_split_doctor_report_large(self): + """Test splitting large report into chunks.""" + # Create a large report + lines = [f"Line {i}: This is a test line for splitting" for i in range(100)] + report_text = "\n".join(lines) + + result = split_doctor_report(report_text, 100) # Small chunk size + + assert len(result) > 1 + assert all(len(chunk) <= 100 for chunk in result) + + # Verify all content is preserved (account for newlines) + combined = "\n".join(result) + assert combined == report_text + + def test_split_doctor_report_with_sections(self): + """Test splitting report with section headers.""" + report_text = """