From fff38f11a7be0cb9e5ef0cb1fc8502706d4e94fe Mon Sep 17 00:00:00 2001 From: Kenneth Zhao Date: Mon, 22 Jun 2026 19:47:00 +0000 Subject: [PATCH] feat: add Mattermost channel support --- AGENTS.md | 2 +- README.md | 4 +- docs/README.md | 2 +- docs/concepts.md | 2 +- docs/configuration.md | 4 +- docs/quick-start.md | 2 +- nanobot/channels/mattermost.py | 661 +++++++++++++++++++ tests/channels/test_mattermost_channel.py | 757 ++++++++++++++++++++++ 8 files changed, 1426 insertions(+), 8 deletions(-) create mode 100644 nanobot/channels/mattermost.py create mode 100644 tests/channels/test_mattermost_channel.py diff --git a/AGENTS.md b/AGENTS.md index bba7f4fd..1e6ab6c6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -36,7 +36,7 @@ Messages flow through an async `MessageBus` (`nanobot/bus/queue.py`) that decoup - **Agent Loop** (`nanobot/agent/loop.py`, `runner.py`): The core processing engine. `AgentLoop` manages session keys, hooks, and context building. `AgentRunner` executes the multi-turn LLM conversation with tool execution. - **LLM Providers** (`nanobot/providers/`): Provider implementations (Anthropic, OpenAI-compatible, OpenAI Responses API, Azure, Bedrock, GitHub Copilot, OpenAI Codex, etc.) built on a common base (`base.py`). Includes image generation (`image_generation.py`) and audio transcription (`transcription.py`). `factory.py` and `registry.py` handle instantiation and model discovery. -- **Channels** (`nanobot/channels/`): Platform integrations (Telegram, Discord, Slack, Feishu, Matrix, WhatsApp, QQ, WeChat, WeCom, DingTalk, Email, MoChat, MS Teams, WebSocket). `manager.py` discovers and coordinates them. Channels are auto-discovered via `pkgutil` scan + entry-point plugins. +- **Channels** (`nanobot/channels/`): Platform integrations (Telegram, Discord, Slack, Mattermost, Feishu, Matrix, WhatsApp, QQ, WeChat, WeCom, DingTalk, Email, MoChat, MS Teams, WebSocket). `manager.py` discovers and coordinates them. Channels are auto-discovered via `pkgutil` scan + entry-point plugins. - **Tools** (`nanobot/agent/tools/`): Agent capabilities exposed to the LLM: filesystem (read/write/edit/list), shell execution (with sandbox backends), web search/fetch, MCP servers, cron, notebook editing, subagent spawning, long-running tasks / sustained goals (`long_task.py`), image generation, and self-modification. Tools are auto-discovered via `pkgutil` scan + entry-point plugins. - **Memory** (`nanobot/agent/memory.py`): Session history persistence with Dream two-phase memory consolidation. Uses atomic writes with fsync for durability. - **Session Management** (`nanobot/session/`): Per-session history, context compaction, TTL-based auto-compaction (`manager.py`), and sustained goal state tracking (`goal_state.py`). diff --git a/README.md b/README.md index f57ae212..2951fcfb 100644 --- a/README.md +++ b/README.md @@ -43,7 +43,7 @@ | Install nanobot with no terminal/config background | [Start Without Technical Background](./docs/start-without-technical-background.md) | | Install quickly and get one CLI reply | [Install](#-install) and [Quick Start](#-quick-start) | | Open the bundled browser UI | [WebUI](#-webui) | -| Connect Telegram, Discord, WeChat, Slack, Email, or another chat app | [Chat Apps](./docs/chat-apps.md) | +| Connect Telegram, Discord, WeChat, Slack, Mattermost, Email, or another chat app | [Chat Apps](./docs/chat-apps.md) | | Configure providers, fallback models, Langfuse, MCP, web tools, or security | [Docs](./docs/README.md) and [Configuration](./docs/configuration.md) | | Understand or extend the internals | [Architecture](./docs/architecture.md) and [Development](./docs/development.md) | @@ -205,7 +205,7 @@ ## 💡 Why nanobot - **Persistent workflows**: goals, memory, tools, and chat context survive long-running work. -- **Chat-native reach**: WebUI, API, Telegram, Feishu, Slack, Discord, Teams, and email. +- **Chat-native reach**: WebUI, API, Telegram, Feishu, Slack, Mattermost, Discord, Teams, and email. - **Model freedom**: OpenAI-compatible APIs, local LLMs, image generation, search, and fallbacks. - **Small core**: readable internals with MCP, memory, deployment, and automation built in. - **Own your stack**: inspect, customize, self-host, and extend without a giant platform. diff --git a/docs/README.md b/docs/README.md index d2cda4c3..0c2e6846 100644 --- a/docs/README.md +++ b/docs/README.md @@ -50,7 +50,7 @@ If a local `nanobot agent` session can already answer normally, you can also ask | Goal | Read | Outcome | |---|---|---| | Open the bundled browser UI | [`webui.md`](./webui.md) | `nanobot webui`, chat workspace, Apps, Skills, Automations, and settings | -| Connect Telegram, Discord, WeChat, Slack, and other apps | [`chat-apps.md`](./chat-apps.md) | A gateway-backed chat channel with access control | +| Connect Telegram, Discord, WeChat, Slack, Mattermost, and other apps | [`chat-apps.md`](./chat-apps.md) | A gateway-backed chat channel with access control | | Use slash commands and automations | [`chat-commands.md`](./chat-commands.md) | Pairing, model presets, local triggers, heartbeat tasks, and chat-side controls | | Generate images | [`image-generation.md`](./image-generation.md) | Image provider config, WebUI image mode, and artifact behavior | | Run several isolated bots | [`multiple-instances.md`](./multiple-instances.md) | Separate configs, workspaces, ports, and sessions | diff --git a/docs/concepts.md b/docs/concepts.md index 926d6e36..20ae96de 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -12,7 +12,7 @@ nanobot has one small core loop and several ways to enter it: |---|---| | Agent loop | Builds context, selects the session, calls the provider, runs tools, and publishes replies | | Providers | LLM backends such as OpenRouter, Anthropic, OpenAI, Bedrock, Ollama, vLLM, and other OpenAI-compatible APIs | -| Channels | User-facing transports such as CLI, WebUI/WebSocket, Telegram, Discord, Slack, Feishu, WeChat, Email, and others | +| Channels | User-facing transports such as CLI, WebUI/WebSocket, Telegram, Discord, Slack, Mattermost, Feishu, WeChat, Email, and others | | Tools | Capabilities the model may call, including files, shell, web search/fetch, MCP, cron, image generation, and subagents | | Memory | Workspace files and session history that keep useful context across turns | | Gateway | Long-running process that connects enabled channels and serves the health endpoint | diff --git a/docs/configuration.md b/docs/configuration.md index 4f6b558e..29a5865b 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -1525,7 +1525,7 @@ Global settings that apply to all channels. Configure under the `channels` secti |---------|---------|-------------| | `sendProgress` | `true` | Stream agent's text progress to the channel | | `sendToolHints` | `false` | Stream tool-call hints (e.g. `read_file("…")`) | -| `showReasoning` | `true` | Allow channels to surface model reasoning/thinking content (DeepSeek-R1 `reasoning_content`, Anthropic `thinking_blocks`, inline `` tags). Reasoning flows as a dedicated stream with `_reasoning_delta` / `_reasoning_end` markers — channels override `send_reasoning_delta` / `send_reasoning_end` to render in-place updates. Even with `true`, channels without those overrides stay no-op silently. Currently surfaced on CLI and WebSocket/WebUI (italic shimmer header, auto-collapses after the stream ends); Telegram / Slack / Discord / Feishu / WeChat / Matrix keep the base no-op until their bubble UI is adapted. Independent of `sendProgress`. | +| `showReasoning` | `true` | Allow channels to surface model reasoning/thinking content (DeepSeek-R1 `reasoning_content`, Anthropic `thinking_blocks`, inline `` tags). Reasoning flows as a dedicated stream with `_reasoning_delta` / `_reasoning_end` markers — channels override `send_reasoning_delta` / `send_reasoning_end` to render in-place updates. Even with `true`, channels without those overrides stay no-op silently. Currently surfaced on CLI and WebSocket/WebUI (italic shimmer header, auto-collapses after the stream ends); Telegram / Slack / Mattermost / Discord / Feishu / WeChat / Matrix keep the base no-op until their bubble UI is adapted. Independent of `sendProgress`. | | `extractDocumentText` | `true` | Extract supported document/text attachments into the model prompt. Install parser dependencies with `nanobot plugins enable documents`. If you used document parsing before those parsers became optional, run that command after upgrading. Set to `false` to keep document content out of the prompt and include attachment path references instead. | | `sendMaxRetries` | `3` | Max delivery attempts per outbound message, including the initial send (0-10 configured, minimum 1 actual attempt) | @@ -1950,7 +1950,7 @@ Pairing lets users get access to the bot through a simple code exchange — no c ### How it works -1. A user sends a DM to the bot on any channel (Telegram, Discord, Slack, etc.) where they aren't yet approved. +1. A user sends a DM to the bot on any channel (Telegram, Discord, Slack, Mattermost, etc.) where they aren't yet approved. 2. The bot replies with a pairing code (like `ABCD-EFGH`) and tells them to forward it to you. 3. You approve the code: diff --git a/docs/quick-start.md b/docs/quick-start.md index 0d33e03e..86ee9086 100644 --- a/docs/quick-start.md +++ b/docs/quick-start.md @@ -288,7 +288,7 @@ Exit interactive mode with `exit`, `quit`, `/exit`, `/quit`, `:q`, or `Ctrl+D`. | Copy another provider or local model setup | [`provider-cookbook.md`](./provider-cookbook.md) | | Understand provider/model matching | [`providers.md`](./providers.md) | | Open the bundled browser UI | [`webui.md`](./webui.md) | -| Connect Telegram, Discord, WeChat, Slack, Email, or another chat app | [`chat-apps.md`](./chat-apps.md) | +| Connect Telegram, Discord, WeChat, Slack, Mattermost, Email, or another chat app | [`chat-apps.md`](./chat-apps.md) | | Configure web search, MCP, security, memory, gateway, or runtime settings | [`configuration.md`](./configuration.md) | | Run with Docker, systemd, or LaunchAgent | [`deployment.md`](./deployment.md) | | Debug a failure | [`troubleshooting.md`](./troubleshooting.md) | diff --git a/nanobot/channels/mattermost.py b/nanobot/channels/mattermost.py new file mode 100644 index 00000000..a9a5a02d --- /dev/null +++ b/nanobot/channels/mattermost.py @@ -0,0 +1,661 @@ +"""Mattermost channel implementation using WebSocket + REST API.""" + +from __future__ import annotations + +import asyncio +import json +import re +from pathlib import Path +from typing import Any + +import httpx +from pydantic import Field + +from nanobot.bus.events import OutboundMessage +from nanobot.bus.queue import MessageBus +from nanobot.channels.base import BaseChannel +from nanobot.config.paths import get_media_dir +from nanobot.config_base import Base +from nanobot.utils.helpers import split_message + +MATTERMOST_MAX_MESSAGE_LEN = 16383 +MATTERMOST_WS_RECONNECT_BASE_DELAY = 1 +MATTERMOST_WS_RECONNECT_MAX_DELAY = 30 + +_CHANNEL_TYPES = { + "O": "public", + "P": "private", + "D": "dm", + "G": "group", +} + + +class MattermostDMConfig(Base): + """Mattermost DM policy configuration.""" + enabled: bool = True + policy: str = "open" + allow_from: list[str] = Field(default_factory=list) + + +class MattermostConfig(Base): + """Mattermost channel configuration.""" + enabled: bool = False + server_url: str = "" + token: str = "" + team_id: str = "" + allow_from_match_mode: str = "id" + allow_from: list[str] = Field(default_factory=list) + group_policy: str = "mention" + group_allow_from: list[str] = Field(default_factory=list) + reply_in_thread: bool = True + include_thread_context: bool = True + thread_context_limit: int = 20 + streaming: bool = True + streaming_max_chars: int = 16000 + react_emoji: str = "eyes" + done_emoji: str = "white_check_mark" + send_progress: bool = True + send_tool_hints: bool = False + dm: MattermostDMConfig = Field(default_factory=MattermostDMConfig) + + +def _server_url_to_ws_url(server_url: str) -> str: + if server_url.startswith("https://"): + return server_url.replace("https://", "wss://", 1) + "/api/v4/websocket" + if server_url.startswith("http://"): + return server_url.replace("http://", "ws://", 1) + "/api/v4/websocket" + return server_url + "/api/v4/websocket" + + +class MattermostChannel(BaseChannel): + """Mattermost channel using WebSocket + REST API.""" + + name = "mattermost" + display_name = "Mattermost" + + _BOT_MENTION_RE = re.compile(r"@\S+") + + @classmethod + def default_config(cls) -> dict[str, Any]: + return MattermostConfig().model_dump(by_alias=True) + + def __init__(self, config: Any, bus: MessageBus): + if isinstance(config, dict): + config = MattermostConfig.model_validate(config) + super().__init__(config, bus) + self.config: MattermostConfig = config + self._server_url = config.server_url.rstrip("/") + self._ws_url = _server_url_to_ws_url(self._server_url) + self._http_client: httpx.AsyncClient | None = None + self._ws_task: asyncio.Task | None = None + self._self_id: str | None = None + self._self_username: str | None = None + self._self_email: str | None = None + self._usernames: dict[str, str] = {} + self._user_emails: dict[str, str] = {} + self._channel_types: dict[str, str] = {} + self._stream_posts: dict[str, str] = {} + self._stream_buffers: dict[str, str] = {} + self._stream_last_content: dict[str, str] = {} + self._stream_committed: dict[str, str] = {} + self._stream_root_ids: dict[str, str] = {} + self._thread_context_attempted: set[str] = set() + + # Lifecycle ---------------------------------------------------------------- + + async def start(self) -> None: + if not self.config.server_url or not self.config.token: + self.logger.error("serverUrl and token must be configured") + return + + if self._http_client is None: + self._http_client = httpx.AsyncClient( + base_url=self._server_url, + headers={"Authorization": f"Bearer {self.config.token}"}, + timeout=30.0, + ) + + try: + resp = await self._http_client.get("/api/v4/users/me") + resp.raise_for_status() + me = resp.json() + self._self_id = me.get("id") + self._self_username = me.get("username") + self._self_email = me.get("email", "") + self.logger.info("bot @{} connected", self._self_username) + except Exception as e: + self.logger.error("Failed to identify bot user: {}", e) + await self._cleanup_http() + return + + self._running = True + self._ws_task = asyncio.create_task(self._ws_listen_loop()) + + async def stop(self) -> None: + self._running = False + if self._ws_task: + self._ws_task.cancel() + try: + await self._ws_task + except asyncio.CancelledError: + pass + self._ws_task = None + await self._cleanup_http() + + async def _cleanup_http(self) -> None: + if self._http_client: + await self._http_client.aclose() + self._http_client = None + + # WebSocket ---------------------------------------------------------------- + + async def _ws_listen_loop(self) -> None: + import websockets + + delay = MATTERMOST_WS_RECONNECT_BASE_DELAY + while self._running: + try: + async with websockets.connect( + self._ws_url, + additional_headers={"Authorization": f"Bearer {self.config.token}"}, + ping_interval=20, + ping_timeout=10, + ) as ws: + self.logger.debug("websocket connected") + delay = MATTERMOST_WS_RECONNECT_BASE_DELAY + async for raw in ws: + await self._handle_ws_message(json.loads(raw)) + except asyncio.CancelledError: + break + except Exception as e: + if not self._running: + break + self.logger.warning("websocket error: {} (reconnect in {}s)", e, delay) + await asyncio.sleep(delay) + delay = min(delay * 2, MATTERMOST_WS_RECONNECT_MAX_DELAY) + + async def _handle_ws_message(self, msg: dict[str, Any]) -> None: + event = msg.get("event", "") + if event == "posted": + await self._handle_posted_event(msg) + elif event == "action": + await self._handle_action_event(msg) + elif event == "post_deleted": + await self._handle_post_deleted_event(msg) + + # Event: posted ------------------------------------------------------------ + + async def _handle_posted_event(self, msg: dict[str, Any]) -> None: + data = msg.get("data", {}) + broadcast = msg.get("broadcast", {}) + + raw_post = data.get("post", "{}") + try: + post = json.loads(raw_post) if isinstance(raw_post, str) else raw_post + except json.JSONDecodeError: + self.logger.warning("failed to parse post json") + return + + sender_id = post.get("user_id", "") + channel_id = post.get("channel_id", "") + message_text = post.get("message", "") + root_id = post.get("root_id", "") or "" + post_id = post.get("id", "") + file_ids: list[str] = post.get("file_ids", []) + + if self._self_id and sender_id == self._self_id: + return + if not sender_id or not channel_id: + return + + channel_type_code = data.get("channel_type", "") + channel_type = _CHANNEL_TYPES.get(channel_type_code, "public") + is_dm = channel_type == "dm" + + team_id = broadcast.get("team_id", "") + if self.config.team_id and team_id and team_id != self.config.team_id: + if not is_dm: + return + + if not await self._is_allowed(sender_id, channel_id, channel_type): + if is_dm and self.config.dm.enabled: + await self._handle_message( + sender_id=sender_id, + chat_id=channel_id, + content="", + is_dm=True, + ) + return + + if not is_dm and not self._should_respond_in_channel(message_text, channel_id): + return + + message_text = self._strip_bot_mention(message_text) + + thread_ts = root_id if root_id else None + if self.config.reply_in_thread and not thread_ts and not is_dm: + thread_ts = post_id + session_key = ( + f"mattermost:{channel_id}:{thread_ts}" if thread_ts and root_id else None + ) + + try: + await self._add_reaction(channel_id, post_id, self.config.react_emoji) + except Exception: + self.logger.debug("add reaction failed") + + media_paths: list[str] = [] + for fid in file_ids: + path = await self._download_file(fid) + if path: + media_paths.append(path) + + content = message_text + if root_id and self.config.include_thread_context: + content = await self._with_thread_context( + content, channel_id=channel_id, root_id=root_id, + ) + + mm_meta: dict[str, Any] = { + "post_id": post_id, + "root_id": root_id, + "channel_type": channel_type, + } + if thread_ts: + mm_meta["thread_ts"] = thread_ts + + await self._handle_message( + sender_id=sender_id, + chat_id=channel_id, + content=content, + media=media_paths, + metadata={ + "mattermost": mm_meta, + "message_id": post_id, + }, + session_key=session_key, + is_dm=is_dm, + ) + + # Event: action ------------------------------------------------------------ + + async def _handle_action_event(self, msg: dict[str, Any]) -> None: + data = msg.get("data", {}) + sender_id = data.get("user_id", "") + channel_id = data.get("channel_id", "") + context = data.get("context", {}) or {} + value = context.get("selected_option", "") + + if not sender_id or not channel_id or not value: + return + + channel_type = self._channel_types.get(channel_id, "public") + if not await self._is_allowed(sender_id, channel_id, channel_type): + return + + await self._handle_message( + sender_id=sender_id, + chat_id=channel_id, + content=value, + metadata={"mattermost": {"channel_type": channel_type, "is_action": True}}, + ) + + # Event: post_deleted ------------------------------------------------------ + + async def _handle_post_deleted_event(self, msg: dict[str, Any]) -> None: + data = msg.get("data", {}) + raw_post = data.get("post", "{}") + try: + post = json.loads(raw_post) if isinstance(raw_post, str) else raw_post + except json.JSONDecodeError: + return + post_id = post.get("id", "") + if not post_id: + return + to_remove = [sid for sid, pid in self._stream_posts.items() if pid == post_id] + for sid in to_remove: + self._stream_posts.pop(sid, None) + self._stream_buffers.pop(sid, None) + self._stream_last_content.pop(sid, None) + self._stream_committed.pop(sid, None) + + # Permission / policy ------------------------------------------------------ + + def is_allowed(self, sender_id: str) -> bool: + return True + + async def _is_allowed(self, sender_id: str, chat_id: str, channel_type: str) -> bool: + if channel_type == "dm": + if not self.config.dm.enabled: + return False + if self.config.dm.policy == "allowlist": + return await self._match_sender(sender_id, self.config.dm.allow_from) + return True + + if self.config.group_policy == "allowlist": + return chat_id in self.config.group_allow_from + return True + + def _should_respond_in_channel(self, text: str, chat_id: str) -> bool: + if self.config.group_policy == "open": + return True + if self.config.group_policy == "mention": + return self._is_mentioned(text) + if self.config.group_policy == "allowlist": + return chat_id in self.config.group_allow_from + return False + + def _is_mentioned(self, text: str) -> bool: + if not self._self_username: + return False + return f"@{self._self_username}" in text + + def _strip_bot_mention(self, text: str) -> str: + if not text or not self._self_username: + return text + return re.sub(rf"@{re.escape(self._self_username)}\s*", "", text).strip() + + async def _match_sender(self, sender_id: str, allow_list: list[str]) -> bool: + if not allow_list: + return False + if "*" in allow_list: + return True + mode = self.config.allow_from_match_mode + if mode == "id": + return sender_id in allow_list + if mode == "username": + username = await self._resolve_username(sender_id) + return username in allow_list if username else False + if mode == "email": + email = await self._resolve_email(sender_id) + return email in allow_list if email else False + return False + + async def _resolve_username(self, user_id: str) -> str | None: + if user_id in self._usernames: + return self._usernames[user_id] + try: + user = await self._api_get(f"/api/v4/users/{user_id}") + self._usernames[user_id] = user.get("username", "") + return self._usernames[user_id] + except Exception as e: + self.logger.warning("failed to resolve username for {}: {}", user_id, e) + return None + + async def _resolve_email(self, user_id: str) -> str | None: + if user_id in self._user_emails: + return self._user_emails[user_id] + try: + user = await self._api_get(f"/api/v4/users/{user_id}") + self._user_emails[user_id] = user.get("email", "").lower() + return self._user_emails[user_id] + except Exception as e: + self.logger.warning("failed to resolve email for {}: {}", user_id, e) + return None + + # Thread context ----------------------------------------------------------- + + async def _with_thread_context(self, text: str, *, channel_id: str, root_id: str) -> str: + key = f"{channel_id}:{root_id}" + if key in self._thread_context_attempted: + return text + self._thread_context_attempted.add(key) + + try: + data = await self._api_get( + f"/api/v4/posts/{root_id}/thread?perPage={max(1, self.config.thread_context_limit)}", + ) + except Exception as e: + self.logger.warning("thread context unavailable for {}: {}", key, e) + return text + + posts = data.get("posts", {}) + order = data.get("order", []) + if not order: + return text + + lines: list[str] = [] + for pid in order: + post = posts.get(pid, {}) + if post.get("id") == root_id: + continue + if post.get("user_id") == self._self_id: + label = "bot" + else: + label = f"<{post.get('user_id', 'unknown')}>" + msg_text = (post.get("message", "") or "").strip() + if not msg_text: + continue + if len(msg_text) > 500: + msg_text = msg_text[:500] + "\u2026" + lines.append(f"- {label}: {msg_text}") + + if not lines: + return text + return "Mattermost thread context before this mention:\n" + "\n".join(lines) + f"\n\nCurrent message:\n{text}" + + # Send --------------------------------------------------------------------- + + async def send(self, msg: OutboundMessage) -> None: + if not self._http_client: + self.logger.warning("client not initialized") + return + + try: + chat_id = msg.chat_id + meta = msg.metadata or {} + mm_meta = meta.get("mattermost", {}) or {} + root_id = mm_meta.get("root_id") or mm_meta.get("thread_ts") or meta.get("root_id") + + file_ids: list[str] = [] + for media_path in msg.media or []: + try: + fid = await self._upload_file(chat_id, media_path) + if fid: + file_ids.append(fid) + except Exception: + self.logger.exception("Failed to upload file {}", media_path) + + if msg.content or file_ids: + text = msg.content or " " + chunks = split_message(text, MATTERMOST_MAX_MESSAGE_LEN) + for chunk in chunks: + await self._create_post( + chat_id, chunk, + root_id=root_id if self.config.reply_in_thread else None, + file_ids=file_ids or None, + ) + + if not meta.get("_progress") and meta.get("message_id"): + try: + await self._remove_reaction(meta["message_id"], self.config.react_emoji) + except Exception: + self.logger.debug("remove reaction failed") + if self.config.done_emoji: + try: + await self._add_reaction(chat_id, meta["message_id"], self.config.done_emoji) + except Exception: + self.logger.debug("done reaction failed") + + except Exception: + self.logger.exception("Error sending message") + raise + + # Streaming ----------------------------------------------------------------- + + async def send_delta(self, chat_id: str, delta: str, metadata: dict[str, Any] | None = None) -> None: + if not self._http_client: + return + + meta = metadata or {} + stream_id = meta.get("_stream_id", chat_id) + + if meta.get("_stream_end"): + self._stream_buffers.pop(stream_id, None) + self._stream_last_content.pop(stream_id, None) + post_id = self._stream_posts.pop(stream_id, None) + final = self._stream_committed.pop(stream_id, None) + if post_id and final and self.config.done_emoji: + try: + await self._add_reaction(chat_id, post_id, self.config.done_emoji) + except Exception: + self.logger.debug("done reaction failed") + if not meta.get("_progress") and meta.get("message_id"): + try: + await self._remove_reaction(meta["message_id"], self.config.react_emoji) + except Exception: + self.logger.debug("remove reaction failed") + return + + if not delta.strip(): + return + + committed = self._stream_committed.get(stream_id, "") + buf = committed + delta + self._stream_buffers[stream_id] = buf + + if stream_id not in self._stream_posts: + try: + mm_meta = (meta.get("mattermost", {}) or {}) if isinstance(meta.get("mattermost"), dict) else {} + root_id = mm_meta.get("root_id") or mm_meta.get("thread_ts") or meta.get("root_id") + post = await self._create_post( + chat_id, buf, + root_id=root_id if self.config.reply_in_thread else None, + ) + self._stream_posts[stream_id] = post["id"] + self._stream_committed[stream_id] = buf + if root_id and self.config.reply_in_thread: + self._stream_root_ids[stream_id] = root_id + except Exception as e: + self.logger.warning("stream initial post failed: {}", e) + raise + else: + post_id = self._stream_posts[stream_id] + if buf == self._stream_last_content.get(stream_id): + return + self._stream_last_content[stream_id] = buf + if len(buf) > self.config.streaming_max_chars: + try: + stream_root = self._stream_root_ids.get(stream_id) + post = await self._create_post( + chat_id, buf, + root_id=stream_root if self.config.reply_in_thread else None, + ) + self._stream_posts[stream_id] = post["id"] + self._stream_committed[stream_id] = buf + except Exception as e: + self.logger.warning("stream overflow post failed: {}", e) + raise + else: + try: + await self._edit_post(post_id, buf) + self._stream_committed[stream_id] = buf + except Exception as e: + self.logger.warning("stream edit failed: {}", e) + raise + + # API helpers --------------------------------------------------------------- + + async def _api_get(self, path: str) -> dict[str, Any]: + resp = await self._http_client.get(path) + resp.raise_for_status() + return resp.json() + + async def _api_post(self, path: str, json_data: dict[str, Any]) -> dict[str, Any]: + resp = await self._http_client.post(path, json=json_data) + resp.raise_for_status() + return resp.json() + + async def _api_put(self, path: str, json_data: dict[str, Any]) -> dict[str, Any]: + resp = await self._http_client.put(path, json=json_data) + resp.raise_for_status() + return resp.json() + + async def _create_post( + self, + channel_id: str, + message: str, + *, + root_id: str | None = None, + file_ids: list[str] | None = None, + ) -> dict[str, Any]: + body: dict[str, Any] = { + "channel_id": channel_id, + "message": message, + } + if root_id: + body["root_id"] = root_id + if file_ids: + body["file_ids"] = file_ids + return await self._api_post("/api/v4/posts", body) + + async def _edit_post(self, post_id: str, message: str) -> dict[str, Any]: + return await self._api_put(f"/api/v4/posts/{post_id}", {"id": post_id, "message": message}) + + async def _upload_file(self, channel_id: str, file_path: str) -> str | None: + path = Path(file_path) + if not path.exists(): + self.logger.warning("file not found: {}", file_path) + return None + + try: + files = {"files": (path.name, path.read_bytes())} + resp = await self._http_client.post( + "/api/v4/files", + data={"channel_id": channel_id}, + files=files, + ) + resp.raise_for_status() + data = resp.json() + infos = data.get("file_infos", []) + if infos: + return infos[0].get("id") + except Exception as e: + self.logger.warning("file upload failed for {}: {}", file_path, e) + return None + + async def _download_file(self, file_id: str) -> str | None: + try: + resp = await self._http_client.get(f"/api/v4/files/{file_id}") + resp.raise_for_status() + info = resp.json() + name = info.get("name", file_id) + out = Path(get_media_dir("mattermost")) / f"{file_id}_{name}" + out.parent.mkdir(parents=True, exist_ok=True) + + dl = await self._http_client.get(f"/api/v4/files/{file_id}/download") + dl.raise_for_status() + out.write_bytes(dl.content) + return str(out) + except Exception as e: + self.logger.warning("file download failed for {}: {}", file_id, e) + return None + + async def _add_reaction(self, channel_id: str, post_id: str, emoji: str) -> None: + if not self._self_id or not emoji: + return + await self._api_post("/api/v4/reactions", { + "user_id": self._self_id, + "post_id": post_id, + "emoji_name": emoji, + }) + + async def _remove_reaction(self, post_id: str, emoji: str) -> None: + if not self._self_id or not emoji: + return + resp = await self._http_client.delete( + f"/api/v4/users/{self._self_id}/posts/{post_id}/reactions/{emoji}", + ) + if resp.status_code >= 400: + self.logger.debug("remove reaction failed: {} {}", resp.status_code, resp.text) + + async def resolve_channel_type(self, channel_id: str) -> str: + if channel_id in self._channel_types: + return self._channel_types[channel_id] + try: + data = await self._api_get(f"/api/v4/channels/{channel_id}") + ctype = _CHANNEL_TYPES.get(data.get("type", ""), "public") + self._channel_types[channel_id] = ctype + return ctype + except Exception: + return "public" diff --git a/tests/channels/test_mattermost_channel.py b/tests/channels/test_mattermost_channel.py new file mode 100644 index 00000000..4c2f5988 --- /dev/null +++ b/tests/channels/test_mattermost_channel.py @@ -0,0 +1,757 @@ +"""Tests for the Mattermost channel implementation.""" + +from __future__ import annotations + +import json +from typing import Any +from unittest.mock import AsyncMock, patch + +import httpx +import pytest + +from nanobot.bus.events import OutboundMessage +from nanobot.bus.queue import MessageBus +from nanobot.channels.mattermost import ( + MATTERMOST_MAX_MESSAGE_LEN, + MattermostChannel, + MattermostConfig, +) + + +class _FakeHTTPClient: + """Mock httpx.AsyncClient that records calls and returns canned responses.""" + + def __init__(self) -> None: + self.get_calls: list[dict[str, Any]] = [] + self.post_calls: list[dict[str, Any]] = [] + self.put_calls: list[dict[str, Any]] = [] + self.delete_calls: list[dict[str, Any]] = [] + self._get_responses: dict[str, Any] = {} + self._post_responses: dict[str, Any] = {} + self._put_responses: dict[str, Any] = {} + self._delete_status: int | None = None + + def _req(self, method: str, path: str) -> httpx.Request: + return httpx.Request(method, f"https://chat.example.com{path}") + + def _resp(self, status: int, json_data: Any, method: str = "GET", path: str = "/") -> httpx.Response: + return httpx.Response(status, json=json_data, request=self._req(method, path)) + + def set_get_response(self, path: str, data: Any) -> None: + self._get_responses[path] = data + + def set_post_response(self, path: str, data: Any) -> None: + self._post_responses[path] = data + + def set_put_response(self, path: str, data: Any) -> None: + self._put_responses[path] = data + + def set_delete_status(self, status: int) -> None: + self._delete_status = status + + async def get(self, path: str, **kwargs) -> httpx.Response: + self.get_calls.append({"path": path, **kwargs}) + data = self._get_responses.get(path, {"id": "resp_" + path.split("/")[-1]}) + return self._resp(200, data, "GET", path) + + async def post(self, path: str, *, json: dict[str, Any] | None = None, data: Any = None, files: Any = None, **kwargs) -> httpx.Response: + call: dict[str, Any] = {"path": path} + if json is not None: + call["json"] = json + if data is not None: + call["data"] = data + if files is not None: + call["files"] = files + self.post_calls.append(call) + data = self._post_responses.get(path, {"id": "new_id"}) + return self._resp(201, data, "POST", path) + + async def put(self, path: str, *, json: dict[str, Any] | None = None, **kwargs) -> httpx.Response: + self.put_calls.append({"path": path, "json": json}) + data = self._put_responses.get(path, {"id": path.split("/")[-1]}) + return self._resp(200, data, "PUT", path) + + async def delete(self, path: str, **kwargs) -> httpx.Response: + self.delete_calls.append({"path": path}) + status = self._delete_status if self._delete_status is not None else 200 + return self._resp(status, {}, "DELETE", path) + + async def aclose(self) -> None: + pass + + +def _make_channel( + overrides: dict[str, Any] | None = None, + bus: MessageBus | None = None, +) -> tuple[MattermostChannel, _FakeHTTPClient]: + config_dict: dict[str, Any] = { + "enabled": True, + "serverUrl": "https://chat.example.com", + "token": "test_token", + **(overrides or {}), + } + config = MattermostConfig.model_validate(config_dict) + if bus is None: + bus = MessageBus() + channel = MattermostChannel(config, bus) + fake = _FakeHTTPClient() + fake.set_get_response("/api/v4/users/me", { + "id": "botuserid123", + "username": "nanobot", + "email": "bot@example.com", + }) + fake.set_post_response("/api/v4/posts", {"id": "post_new_id"}) + channel._http_client = fake + return channel, fake + + +# --------------------------------------------------------------------------- +# Config parsing +# --------------------------------------------------------------------------- + + +def test_config_defaults(): + config = MattermostConfig() + assert config.enabled is False + assert config.server_url == "" + assert config.token == "" + assert config.streaming is True + assert config.streaming_max_chars == 16000 + assert config.dm.enabled is True + assert config.dm.policy == "open" + assert config.reply_in_thread is True + + +def test_config_camelcase_aliases(): + raw = { + "serverUrl": "https://mm.example.com", + "token": "abc123", + "allowFromMatchMode": "username", + "streamingMaxChars": 8000, + "replyInThread": False, + } + config = MattermostConfig.model_validate(raw) + assert config.server_url == "https://mm.example.com" + assert config.token == "abc123" + assert config.allow_from_match_mode == "username" + assert config.streaming_max_chars == 8000 + assert config.reply_in_thread is False + + +def test_config_default_config_classmethod(): + d = MattermostChannel.default_config() + assert d["enabled"] is False + assert d["serverUrl"] == "" + assert d["token"] == "" + + +# --------------------------------------------------------------------------- +# Self-identification on start +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_start_identifies_bot(): + channel, fake = _make_channel({"serverUrl": "https://chat.example.com", "token": "tok"}) + calls_before = len(fake.get_calls) + with patch("websockets.connect", AsyncMock(side_effect=Exception("no-op"))): + await channel.start() + assert channel._self_id == "botuserid123" + assert channel._self_username == "nanobot" + assert channel._self_email == "bot@example.com" + user_me_calls = [c for c in fake.get_calls[calls_before:] if "/api/v4/users/me" in c["path"]] + assert len(user_me_calls) == 1 + await channel.stop() + + +@pytest.mark.asyncio +async def test_start_missing_config(): + channel, fake = _make_channel({"serverUrl": "", "token": ""}) + await channel.start() + assert channel._self_id is None + + +# --------------------------------------------------------------------------- +# Server URL normalization +# --------------------------------------------------------------------------- + + +def test_server_url_normalization(): + config = MattermostConfig.model_validate({ + "serverUrl": "https://chat.example.com/", + "token": "tok", + }) + channel = MattermostChannel(config, MessageBus()) + assert channel._server_url == "https://chat.example.com" + assert "/api/v4/websocket" in channel._ws_url + assert channel._ws_url.startswith("wss://") + + +def test_server_url_no_trailing_slash(): + config = MattermostConfig.model_validate({ + "serverUrl": "https://chat.example.com", + "token": "tok", + }) + channel = MattermostChannel(config, MessageBus()) + assert channel._server_url == "https://chat.example.com" + + +# --------------------------------------------------------------------------- +# Inbound routing: posted event +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_posted_event_routes_to_handle_message(): + channel, fake = _make_channel() + channel._self_id = "botuserid123" + channel._self_username = "nanobot" + with patch.object(channel, "_handle_message", AsyncMock()) as mock_handle: + ws_msg = { + "event": "posted", + "data": { + "channel_type": "D", + "post": json.dumps({ + "id": "post_abc", + "user_id": "user_42", + "channel_id": "chan_1", + "message": "hello", + "root_id": "", + }), + }, + "broadcast": {"channel_id": "chan_1", "team_id": ""}, + } + await channel._handle_ws_message(ws_msg) + mock_handle.assert_awaited_once() + args, kwargs = mock_handle.call_args + assert kwargs["sender_id"] == "user_42" + assert kwargs["chat_id"] == "chan_1" + assert kwargs["content"] == "hello" + assert kwargs["is_dm"] is True + + +@pytest.mark.asyncio +async def test_posted_event_self_message_ignored(): + channel, fake = _make_channel() + channel._self_id = "bot_id" + with patch.object(channel, "_handle_message", AsyncMock()) as mock_handle: + ws_msg = { + "event": "posted", + "data": { + "channel_type": "D", + "post": json.dumps({ + "id": "p1", "user_id": "bot_id", + "channel_id": "c1", "message": "ignore me", "root_id": "", + }), + }, + "broadcast": {}, + } + await channel._handle_ws_message(ws_msg) + mock_handle.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_posted_event_channel_type_detection(): + channel, fake = _make_channel() + channel._self_id = "bot_id" + + for code, expected_dm in [("D", True), ("O", False), ("P", False), ("G", False)]: + with patch.object(channel, "_handle_message", AsyncMock()) as mock_handle: + with patch.object(channel, "_should_respond_in_channel", return_value=True): + with patch.object(channel, "_is_allowed", AsyncMock(return_value=True)): + ws_msg = { + "event": "posted", + "data": { + "channel_type": code, + "post": json.dumps({ + "id": "p1", "user_id": "u1", + "channel_id": "c1", "message": "hi", "root_id": "", + }), + }, + "broadcast": {}, + } + await channel._handle_ws_message(ws_msg) + mock_handle.assert_called_once() + assert mock_handle.call_args[1]["is_dm"] == expected_dm + + +# --------------------------------------------------------------------------- +# Bot @mention stripping +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_strip_bot_mention_from_incoming(): + channel, fake = _make_channel() + channel._self_id = "bot_id" + channel._self_username = "nanobot" + with patch.object(channel, "_handle_message", AsyncMock()) as mock_handle: + with patch.object(channel, "_is_allowed", AsyncMock(return_value=True)): + with patch.object(channel, "_should_respond_in_channel", return_value=True): + ws_msg = { + "event": "posted", + "data": { + "channel_type": "O", + "post": json.dumps({ + "id": "p1", "user_id": "u1", + "channel_id": "c1", "message": "@nanobot hello there", "root_id": "", + }), + }, + "broadcast": {}, + } + await channel._handle_ws_message(ws_msg) + assert mock_handle.call_args[1]["content"] == "hello there" + + +# --------------------------------------------------------------------------- +# DM policy: open / allowlist +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_dm_policy_open(): + channel, fake = _make_channel({"dm": {"policy": "open"}}) + result = await channel._is_allowed("any_user", "dm_chan", "dm") + assert result is True + + +@pytest.mark.asyncio +async def test_dm_policy_allowlist_match(): + channel, fake = _make_channel({"dm": {"policy": "allowlist", "allowFrom": ["user_1", "user_2"]}}) + assert await channel._is_allowed("user_1", "dm_chan", "dm") is True + assert await channel._is_allowed("user_3", "dm_chan", "dm") is False + + +@pytest.mark.asyncio +async def test_dm_disabled(): + channel, fake = _make_channel({"dm": {"enabled": False}}) + assert await channel._is_allowed("u1", "dm_chan", "dm") is False + + +# --------------------------------------------------------------------------- +# Group policy: mention / open / allowlist +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_group_policy_mention(): + channel, fake = _make_channel({"groupPolicy": "mention"}) + channel._self_username = "nanobot" + assert channel._should_respond_in_channel("hello", "c1") is False + assert channel._should_respond_in_channel("@nanobot hello", "c1") is True + + +@pytest.mark.asyncio +async def test_group_policy_open(): + channel, fake = _make_channel({"groupPolicy": "open"}) + assert channel._should_respond_in_channel("anything", "c1") is True + + +@pytest.mark.asyncio +async def test_group_policy_allowlist(): + channel, fake = _make_channel({"groupPolicy": "allowlist", "groupAllowFrom": ["c1"]}) + assert channel._should_respond_in_channel("msg", "c1") is True + assert channel._should_respond_in_channel("msg", "c2") is False + + +# --------------------------------------------------------------------------- +# Match mode: id / username / email +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_match_mode_id(): + channel, fake = _make_channel({"allowFromMatchMode": "id", "allowFrom": ["u1", "u2"]}) + assert await channel._match_sender("u1", ["u1", "u2"]) is True + assert await channel._match_sender("u3", ["u1", "u2"]) is False + + +@pytest.mark.asyncio +async def test_match_mode_username(): + channel, fake = _make_channel({"allowFromMatchMode": "username", "allowFrom": ["alice"]}) + fake.set_get_response("/api/v4/users/u1", {"id": "u1", "username": "alice", "email": "alice@x.com"}) + assert await channel._match_sender("u1", ["alice"]) is True + assert await channel._match_sender("u2", ["alice"]) is False + + +@pytest.mark.asyncio +async def test_match_mode_email(): + channel, fake = _make_channel({"allowFromMatchMode": "email", "allowFrom": ["alice@x.com"]}) + fake.set_get_response("/api/v4/users/u1", {"id": "u1", "username": "alice", "email": "alice@x.com"}) + assert await channel._match_sender("u1", ["alice@x.com"]) is True + assert await channel._match_sender("u2", ["alice@x.com"]) is False + + +# --------------------------------------------------------------------------- +# Identity cache +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_identity_cache_username(): + channel, fake = _make_channel({"allowFromMatchMode": "username", "allowFrom": ["alice"]}) + fake.set_get_response("/api/v4/users/u1", {"id": "u1", "username": "alice", "email": ""}) + + calls_before = len(fake.get_calls) + await channel._match_sender("u1", ["alice"]) + assert len(fake.get_calls) == calls_before + 1 + + await channel._match_sender("u1", ["alice"]) + assert len(fake.get_calls) == calls_before + 1 + + +# --------------------------------------------------------------------------- +# Send +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_send_creates_post(): + channel, fake = _make_channel() + channel._self_id = "bot_id" + msg = OutboundMessage( + channel="mattermost", + chat_id="chan_1", + content="hello world", + ) + await channel.send(msg) + posts = [c for c in fake.post_calls if c["path"] == "/api/v4/posts"] + assert len(posts) == 1 + assert posts[0]["json"]["channel_id"] == "chan_1" + assert posts[0]["json"]["message"] == "hello world" + + +@pytest.mark.asyncio +async def test_send_with_file_upload(): + channel, fake = _make_channel() + channel._self_id = "bot_id" + fake.set_post_response("/api/v4/files", { + "file_infos": [{"id": "file_abc", "name": "test.txt"}], + }) + + with patch("nanobot.channels.mattermost.Path.exists", return_value=True): + with patch("nanobot.channels.mattermost.Path.read_bytes", return_value=b"data"): + msg = OutboundMessage( + channel="mattermost", + chat_id="chan_1", + content="with file", + media=["/tmp/test.txt"], + ) + await channel.send(msg) + + file_uploads = [c for c in fake.post_calls if c["path"] == "/api/v4/files"] + assert len(file_uploads) == 1 + + posts = [c for c in fake.post_calls if c["path"] == "/api/v4/posts"] + assert len(posts) == 1 + assert posts[0]["json"]["file_ids"] == ["file_abc"] + + +@pytest.mark.asyncio +async def test_send_with_thread_root_id(): + channel, fake = _make_channel({"replyInThread": True}) + channel._self_id = "bot_id" + msg = OutboundMessage( + channel="mattermost", + chat_id="chan_1", + content="reply in thread", + metadata={"mattermost": {"root_id": "root_42"}}, + ) + await channel.send(msg) + posts = [c for c in fake.post_calls if c["path"] == "/api/v4/posts"] + assert len(posts) == 1 + assert posts[0]["json"]["root_id"] == "root_42" + + +@pytest.mark.asyncio +async def test_send_reaction_on_completion(): + channel, fake = _make_channel() + channel._self_id = "bot_id" + msg = OutboundMessage( + channel="mattermost", + chat_id="chan_1", + content="done", + metadata={"message_id": "orig_post_1"}, + ) + await channel.send(msg) + reactions = [c for c in fake.post_calls if c["path"] == "/api/v4/reactions"] + assert len(reactions) == 1 + assert reactions[0]["json"]["emoji_name"] == "white_check_mark" + + +# --------------------------------------------------------------------------- +# Streaming +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_stream_first_delta_creates_post(): + channel, fake = _make_channel() + channel._self_id = "bot_id" + fake.set_post_response("/api/v4/posts", {"id": "stream_post_1"}) + + await channel.send_delta("chan_1", "Hello", {"_stream_id": "s1"}) + posts = [c for c in fake.post_calls if c["path"] == "/api/v4/posts"] + assert len(posts) == 1 + assert "Hello" in posts[0]["json"]["message"] + assert channel._stream_posts["s1"] == "stream_post_1" + + +@pytest.mark.asyncio +async def test_stream_subsequent_delta_edits_post(): + channel, fake = _make_channel() + channel._self_id = "bot_id" + fake.set_post_response("/api/v4/posts", {"id": "stream_post_1"}) + + await channel.send_delta("chan_1", "Hello", {"_stream_id": "s1"}) + assert channel._stream_buffers["s1"] == "Hello" + + await channel.send_delta("chan_1", " world", {"_stream_id": "s1"}) + edits = [c for c in fake.put_calls if c["path"] == "/api/v4/posts/stream_post_1"] + assert len(edits) == 1 + assert edits[0]["json"]["message"] == "Hello world" + + +@pytest.mark.asyncio +async def test_stream_end_adds_done_emoji(): + channel, fake = _make_channel() + channel._self_id = "bot_id" + fake.set_post_response("/api/v4/posts", {"id": "stream_post_1"}) + + await channel.send_delta("chan_1", "Hello", {"_stream_id": "s1"}) + await channel.send_delta("chan_1", "", {"_stream_id": "s1", "_stream_end": True}) + reactions = [c for c in fake.post_calls if c["path"] == "/api/v4/reactions" and c["json"]["emoji_name"] == "white_check_mark"] + assert len(reactions) >= 1 + assert channel._stream_posts.get("s1") is None + + +@pytest.mark.asyncio +async def test_stream_chunk_boundary_finalizes_and_creates_new(): + channel, fake = _make_channel({"streamingMaxChars": 10}) + channel._self_id = "bot_id" + fake.set_post_response("/api/v4/posts", {"id": "post_1"}) + + await channel.send_delta("chan_1", "Hello ", {"_stream_id": "s1"}) + await channel.send_delta("chan_1", "world", {"_stream_id": "s1"}) + + posts = [c for c in fake.post_calls if c["path"] == "/api/v4/posts"] + assert len(posts) == 2 + + +# --------------------------------------------------------------------------- +# Reactions +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_reaction_add_on_receipt(): + channel, fake = _make_channel() + channel._self_id = "bot_id" + await channel._add_reaction("chan_1", "post_1", "eyes") + reactions = [c for c in fake.post_calls if c["path"] == "/api/v4/reactions"] + assert len(reactions) >= 1 + assert reactions[-1]["json"]["emoji_name"] == "eyes" + + +@pytest.mark.asyncio +async def test_reaction_remove(): + channel, fake = _make_channel() + channel._self_id = "bot_id" + await channel._remove_reaction("post_1", "eyes") + assert len(fake.delete_calls) >= 1 + assert "post_1" in fake.delete_calls[-1]["path"] + assert "eyes" in fake.delete_calls[-1]["path"] + + +# --------------------------------------------------------------------------- +# Team filtering +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_team_filtering_rejects_wrong_team(): + channel, fake = _make_channel({"teamId": "team_a"}) + channel._self_id = "bot_id" + with patch.object(channel, "_handle_message", AsyncMock()) as mock_handle: + ws_msg = { + "event": "posted", + "data": { + "channel_type": "O", + "post": json.dumps({ + "id": "p1", "user_id": "u1", + "channel_id": "c1", "message": "hi", "root_id": "", + }), + }, + "broadcast": {"channel_id": "c1", "team_id": "team_b"}, + } + await channel._handle_ws_message(ws_msg) + mock_handle.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_team_filtering_allows_correct_team(): + channel, fake = _make_channel({"teamId": "team_a"}) + channel._self_id = "bot_id" + with patch.object(channel, "_handle_message", AsyncMock()) as mock_handle: + with patch.object(channel, "_is_allowed", AsyncMock(return_value=True)): + with patch.object(channel, "_should_respond_in_channel", return_value=True): + ws_msg = { + "event": "posted", + "data": { + "channel_type": "O", + "post": json.dumps({ + "id": "p1", "user_id": "u1", + "channel_id": "c1", "message": "hi", "root_id": "", + }), + }, + "broadcast": {"channel_id": "c1", "team_id": "team_a"}, + } + await channel._handle_ws_message(ws_msg) + mock_handle.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_team_filtering_dm_bypass(): + channel, fake = _make_channel({"teamId": "team_a"}) + channel._self_id = "bot_id" + with patch.object(channel, "_handle_message", AsyncMock()) as mock_handle: + ws_msg = { + "event": "posted", + "data": { + "channel_type": "D", + "post": json.dumps({ + "id": "p1", "user_id": "u1", + "channel_id": "dm_chan", "message": "hi", "root_id": "", + }), + }, + "broadcast": {"channel_id": "dm_chan", "team_id": ""}, + } + await channel._handle_ws_message(ws_msg) + mock_handle.assert_awaited_once() + + +# --------------------------------------------------------------------------- +# Thread session key +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_thread_session_key(): + channel, fake = _make_channel() + channel._self_id = "bot_id" + with patch.object(channel, "_handle_message", AsyncMock()) as mock_handle: + with patch.object(channel, "_is_allowed", AsyncMock(return_value=True)): + with patch.object(channel, "_should_respond_in_channel", return_value=True): + ws_msg = { + "event": "posted", + "data": { + "channel_type": "O", + "post": json.dumps({ + "id": "post_1", "user_id": "u1", + "channel_id": "c1", "message": "in thread", + "root_id": "root_99", + }), + }, + "broadcast": {}, + } + await channel._handle_ws_message(ws_msg) + kwargs = mock_handle.call_args[1] + assert kwargs["session_key"] == "mattermost:c1:root_99" + + +# --------------------------------------------------------------------------- +# Action event (interactive buttons) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_action_event(): + channel, fake = _make_channel() + channel._self_id = "bot_id" + with patch.object(channel, "_handle_message", AsyncMock()) as mock_handle: + ws_msg = { + "event": "action", + "data": { + "user_id": "u1", + "channel_id": "c1", + "context": {"selected_option": "Approve"}, + }, + "broadcast": {}, + } + await channel._handle_ws_message(ws_msg) + mock_handle.assert_awaited_once_with( + sender_id="u1", + chat_id="c1", + content="Approve", + metadata={"mattermost": {"channel_type": "public", "is_action": True}}, + ) + + +# --------------------------------------------------------------------------- +# Post deleted event +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_post_deleted_cleans_stream_state(): + channel, fake = _make_channel() + channel._stream_posts["s1"] = "del_post_1" + channel._stream_posts["s2"] = "keep_post_2" + + ws_msg = { + "event": "post_deleted", + "data": { + "channel_id": "c1", + "post": json.dumps({"id": "del_post_1", "delete_at": 123}), + }, + "broadcast": {}, + } + await channel._handle_ws_message(ws_msg) + assert "s1" not in channel._stream_posts + assert channel._stream_posts["s2"] == "keep_post_2" + + +# --------------------------------------------------------------------------- +# Auth failure +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_auth_failure_prevents_start(): + channel, fake = _make_channel() + fake.set_get_response("/api/v4/users/me", {"id": "", "username": ""}) + with patch.object(fake, "get", side_effect=Exception("401 Unauthorized")): + await channel.start() + assert channel._self_id is None + + +# --------------------------------------------------------------------------- +# DM allowlist with match mode +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_dm_allowlist_with_username_match(): + channel, fake = _make_channel({ + "allowFromMatchMode": "username", + "dm": {"policy": "allowlist", "allowFrom": ["alice"]}, + }) + fake.set_get_response("/api/v4/users/u1", {"id": "u1", "username": "alice", "email": ""}) + assert await channel._is_allowed("u1", "dm_chan", "dm") is True + assert await channel._is_allowed("u2", "dm_chan", "dm") is False + + +# --------------------------------------------------------------------------- +# split_message helper +# --------------------------------------------------------------------------- + + +def test_message_splitting(): + from nanobot.utils.helpers import split_message + short = "short message" + assert split_message(short, MATTERMOST_MAX_MESSAGE_LEN) == [short] + + long_text = "A" * (MATTERMOST_MAX_MESSAGE_LEN + 100) + chunks = split_message(long_text, MATTERMOST_MAX_MESSAGE_LEN) + assert all(len(c) <= MATTERMOST_MAX_MESSAGE_LEN for c in chunks) + assert "".join(chunks) == long_text