""" Slack API simulator. Provides realistic mock responses for Slack API actions. """ from typing import Optional from .base import BaseSimulator from datetime import datetime import time class SlackSimulator(BaseSimulator): """Simulator for Slack API.""" def __init__(self): super().__init__('slack') def load_mock_responses(self): """Load Slack mock response templates.""" self.mock_responses = { 'post_message': self._post_message_template, 'create_channel': self._create_channel_template, 'upload_file': self._upload_file_template, 'read_messages': self._read_messages_template, 'update_message': self._update_message_template, 'add_reaction': self._add_reaction_template, } def get_required_permissions(self, action: str) -> set[str]: """Get required Slack scopes for an action.""" permissions_map = { 'post_message': {'chat:write'}, 'create_channel': {'channels:manage'}, 'upload_file': {'files:write'}, 'read_messages': {'channels:history'}, 'update_message': {'chat:write'}, 'add_reaction': {'reactions:write'}, } return permissions_map.get(action, set()) def validate_params(self, action: str, params: dict) -> tuple[bool, Optional[str]]: """Validate parameters for Slack actions.""" if action == 'post_message': required = {'channel', 'text'} missing = required - set(params.keys()) if missing: return False, f"Missing required parameters: {missing}" elif action == 'create_channel': required = {'name'} missing = required - set(params.keys()) if missing: return False, f"Missing required parameters: {missing}" # Validate channel name format name = params['name'] if not name.islower() or ' ' in name: return False, "Channel name must be lowercase without spaces" elif action == 'upload_file': required = {'channels', 'content'} missing = required - set(params.keys()) if missing: return False, f"Missing required parameters: {missing}" elif action == 'read_messages': required = {'channel'} missing = required - set(params.keys()) if missing: return False, f"Missing required parameters: {missing}" elif action == 'update_message': required = {'channel', 'ts', 'text'} missing = required - set(params.keys()) if missing: return False, f"Missing required parameters: {missing}" elif action == 'add_reaction': required = {'channel', 'timestamp', 'name'} missing = required - set(params.keys()) if missing: return False, f"Missing required parameters: {missing}" else: return False, f"Unknown action: {action}" return True, None def generate_mock_response(self, action: str, params: dict) -> dict: """Generate realistic Slack API response.""" if action not in self.mock_responses: raise ValueError(f"Unknown action: {action}") template_func = self.mock_responses[action] return template_func(params) def _post_message_template(self, params: dict) -> dict: """Mock response for posting a message.""" ts = f"{int(time.time())}.{int(time.time() * 1000000) % 1000000}" return { "ok": True, "channel": params['channel'], "ts": ts, "message": { "type": "message", "subtype": None, "text": params['text'], "ts": ts, "username": "My AI Agent", "bot_id": "B07XXXXXX", "app_id": "A07XXXXXX", "team": "T07XXXXXX", "blocks": params.get('blocks', []), "thread_ts": params.get('thread_ts') } } def _create_channel_template(self, params: dict) -> dict: """Mock response for creating a channel.""" channel_id = f"C{int(time.time())}" return { "ok": True, "channel": { "id": channel_id, "name": params['name'], "is_channel": True, "is_group": False, "is_im": False, "is_mpim": False, "is_private": params.get('is_private', False), "created": int(time.time()), "is_archived": False, "is_general": False, "unlinked": 0, "name_normalized": params['name'], "is_shared": False, "is_org_shared": False, "is_pending_ext_shared": False, "pending_shared": [], "context_team_id": "T07XXXXXX", "updated": int(time.time()) * 1000, "creator": "U07XXXXXX", "is_member": True, "topic": { "value": "", "creator": "", "last_set": 0 }, "purpose": { "value": params.get('purpose', ''), "creator": "U07XXXXXX", "last_set": int(time.time()) } } } def _upload_file_template(self, params: dict) -> dict: """Mock response for uploading a file.""" file_id = f"F{int(time.time())}" return { "ok": True, "file": { "id": file_id, "created": int(time.time()), "timestamp": int(time.time()), "name": params.get('filename', 'file.txt'), "title": params.get('title', 'Uploaded File'), "mimetype": params.get('filetype', 'text/plain'), "filetype": params.get('filetype', 'txt'), "pretty_type": "Plain Text", "user": "U07XXXXXX", "mode": "hosted", "editable": False, "is_external": False, "external_type": "", "is_public": False, "public_url_shared": False, "display_as_bot": False, "username": "", "size": len(params['content']), "url_private": f"https://files.slack.com/files-pri/T07XXXXXX-{file_id}/file.txt", "url_private_download": f"https://files.slack.com/files-pri/T07XXXXXX-{file_id}/download/file.txt", "permalink": f"https://myworkspace.slack.com/files/U07XXXXXX/{file_id}/file.txt", "channels": params['channels'] if isinstance(params['channels'], list) else [params['channels']], "groups": [], "ims": [], "has_rich_preview": False } } def _read_messages_template(self, params: dict) -> dict: """Mock response for reading channel messages.""" # Generate some realistic message history base_ts = int(time.time()) - 3600 # 1 hour ago messages = [] for i in range(5): ts = f"{base_ts + (i * 300)}.{i * 111111}" messages.append({ "type": "message", "user": f"U07XXXX{i:02d}", "text": f"This is message {i+1} in the channel", "ts": ts, "team": "T07XXXXXX", "blocks": [ { "type": "rich_text", "block_id": f"block{i}", "elements": [ { "type": "rich_text_section", "elements": [ { "type": "text", "text": f"This is message {i+1} in the channel" } ] } ] } ] }) return { "ok": True, "messages": messages, "has_more": False, "pin_count": 0, "channel_actions_ts": None, "channel_actions_count": 0 } def _update_message_template(self, params: dict) -> dict: """Mock response for updating a message.""" return { "ok": True, "channel": params['channel'], "ts": params['ts'], "text": params['text'], "message": { "type": "message", "user": "U07XXXXXX", "text": params['text'], "ts": params['ts'], "team": "T07XXXXXX", "edited": { "user": "U07XXXXXX", "ts": f"{int(time.time())}.000000" } } } def _add_reaction_template(self, params: dict) -> dict: """Mock response for adding a reaction to a message.""" return { "ok": True }