test(mutmut): stabilize runs and extend telegram coverage (#157)

pyproject.toml

@@ -54,6 +54,7 @@ build-backend = "uv_build"
 
 [dependency-groups]
 dev = [
+    "mutmut>=3.4.0",
     "pytest>=9.0.2",
     "pytest-anyio>=0.0.0",
     "pytest-cov>=7.0.0",
@@ -66,9 +67,15 @@ docs = [
 ]
 
 [tool.pytest.ini_options]
-addopts = ["--cov=takopi", "--cov-branch", "--cov-report=term-missing", "--cov-fail-under=80"]
+addopts = ["--cov=takopi", "--cov-branch", "--cov-report=term-missing", "--cov-fail-under=81"]
 testpaths = ["tests"]
 
+[tool.mutmut]
+paths_to_mutate = ["src/takopi"]
+tests_dir = ["tests"]
+pytest_add_cli_args = ["-q", "--no-cov"]
+do_not_mutate = ["src/takopi/cli/*"]
+
 [tool.ruff.lint]
 extend-select = ["B", "BLE001", "C4", "PERF", "RUF043", "S110", "SIM", "UP"]
 

src/takopi/config_watch.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 from dataclasses import dataclass
 from pathlib import Path
 from collections.abc import Awaitable, Callable, Iterable
@@ -75,6 +76,9 @@ async def watch_config(
     reserved: Iterable[str] = RESERVED_CHAT_COMMANDS,
     on_reload: Callable[[ConfigReload], Awaitable[None]] | None = None,
 ) -> None:
+    # Mutmut sets MUTANT_UNDER_TEST; disable watchers to avoid native crashes.
+    if os.environ.get("MUTANT_UNDER_TEST"):
+        return
     reserved_tuple = tuple(reserved)
     config_path = config_path.expanduser().resolve()
     watch_root = config_path.parent

src/takopi/logging.py

@@ -206,7 +206,7 @@ class SafeWriter(io.TextIOBase):
 
 
 def setup_logging(
-    *, debug: bool = False, cache_logger_on_first_use: bool = True
+    *, debug: bool = False, cache_logger_on_first_use: bool = False
 ) -> None:
     global _MIN_LEVEL, _PIPELINE_LEVEL_NAME
     global _log_file_handle

tests/test_onboarding_helpers.py

@@ -4,6 +4,7 @@ from pathlib import Path
 from typing import Any, cast
 
 import pytest
+from rich.console import Console
 from rich.table import Table
 from rich.text import Text
 
@@ -258,6 +259,42 @@ def test_render_engine_table_prints() -> None:
     assert ui.printed
 
 
+def test_debug_onboarding_paths_prints_table() -> None:
+    console = Console(record=True, width=120)
+    onboarding.debug_onboarding_paths(console=console)
+    output = console.export_text()
+    assert "onboarding paths (15)" in output
+    assert "workspace" in output
+    assert "assistant" in output
+    assert "handoff" in output
+
+
+@pytest.mark.anyio
+async def test_confirm_prompt_returns_question_result(monkeypatch) -> None:
+    seen: dict[str, object] = {}
+
+    class _PromptSession:
+        def __init__(self, *args, **kwargs) -> None:
+            seen["args"] = args
+            seen["kwargs"] = kwargs
+            self.app = object()
+
+    class _Question:
+        def __init__(self, app) -> None:
+            seen["app"] = app
+
+        async def ask_async(self) -> bool | None:
+            return True
+
+    monkeypatch.setattr(onboarding, "PromptSession", _PromptSession)
+    monkeypatch.setattr(onboarding, "Question", _Question)
+
+    result = await onboarding.confirm_prompt("continue?", default=False)
+
+    assert result is True
+    assert "kwargs" in seen
+
+
 @pytest.mark.anyio
 async def test_get_bot_info_retries(monkeypatch) -> None:
     class _Bot:

tests/test_telegram_queue.py

@@ -3,7 +3,15 @@ from typing import Any
 import anyio
 import pytest
 
-from takopi.telegram.api_models import Chat, File, Message, Update, User
+from takopi.telegram.api_models import (
+    Chat,
+    ChatMember,
+    File,
+    ForumTopic,
+    Message,
+    Update,
+    User,
+)
 from takopi.telegram.client import BotClient, TelegramClient, TelegramRetryAfter
 
 
@@ -13,6 +21,16 @@ class FakeBot(BotClient):
         self.edit_calls: list[str] = []
         self.delete_calls: list[tuple[int, int]] = []
         self.topic_calls: list[tuple[int, int, str]] = []
+        self.document_calls: list[
+            tuple[int, str, bytes, int | None, int | None, bool | None, str | None]
+        ] = []
+        self.command_calls: list[
+            tuple[list[dict[str, Any]], dict[str, Any] | None, str | None]
+        ] = []
+        self.callback_calls: list[tuple[str, str | None, bool | None]] = []
+        self.chat_calls: list[int] = []
+        self.chat_member_calls: list[tuple[int, int]] = []
+        self.create_topic_calls: list[tuple[int, str]] = []
         self._edit_attempts = 0
         self._updates_attempts = 0
         self.retry_after: float | None = None
@@ -51,7 +69,9 @@ class FakeBot(BotClient):
         disable_notification: bool | None = False,
         caption: str | None = None,
     ) -> Message | None:
-        _ = (
+        self.calls.append("send_document")
+        self.document_calls.append(
+            (
                 chat_id,
                 filename,
                 content,
@@ -60,7 +80,7 @@ class FakeBot(BotClient):
                 disable_notification,
                 caption,
             )
-        self.calls.append("send_document")
+        )
         return Message(message_id=1, chat=Chat(id=chat_id, type="private"))
 
     async def edit_message_text(
@@ -104,9 +124,8 @@ class FakeBot(BotClient):
         scope: dict[str, Any] | None = None,
         language_code: str | None = None,
     ) -> bool:
-        _ = commands
-        _ = scope
-        _ = language_code
+        self.calls.append("set_my_commands")
+        self.command_calls.append((commands, scope, language_code))
         return True
 
     async def get_updates(
@@ -144,7 +163,8 @@ class FakeBot(BotClient):
         text: str | None = None,
         show_alert: bool | None = None,
     ) -> bool:
-        _ = callback_query_id, text, show_alert
+        self.calls.append("answer_callback_query")
+        self.callback_calls.append((callback_query_id, text, show_alert))
         return True
 
     async def edit_forum_topic(
@@ -154,6 +174,21 @@ class FakeBot(BotClient):
         self.topic_calls.append((chat_id, message_thread_id, name))
         return True
 
+    async def get_chat(self, chat_id: int) -> Chat | None:
+        self.calls.append("get_chat")
+        self.chat_calls.append(chat_id)
+        return Chat(id=chat_id, type="private")
+
+    async def get_chat_member(self, chat_id: int, user_id: int) -> ChatMember | None:
+        self.calls.append("get_chat_member")
+        self.chat_member_calls.append((chat_id, user_id))
+        return ChatMember(status="member")
+
+    async def create_forum_topic(self, chat_id: int, name: str) -> ForumTopic | None:
+        self.calls.append("create_forum_topic")
+        self.create_topic_calls.append((chat_id, name))
+        return ForumTopic(message_thread_id=11)
+
 
 @pytest.mark.anyio
 async def test_edit_forum_topic_uses_outbox() -> None:
@@ -169,6 +204,93 @@ async def test_edit_forum_topic_uses_outbox() -> None:
     assert bot.topic_calls == [(7, 42, "takopi @main")]
 
 
+@pytest.mark.anyio
+async def test_send_document_uses_outbox() -> None:
+    bot = FakeBot()
+    client = TelegramClient(client=bot, private_chat_rps=0.0, group_chat_rps=0.0)
+
+    result = await client.send_document(
+        chat_id=5,
+        filename="note.txt",
+        content=b"hello",
+        caption="greetings",
+    )
+
+    assert result is not None
+    assert bot.calls == ["send_document"]
+    assert bot.document_calls == [
+        (5, "note.txt", b"hello", None, None, False, "greetings")
+    ]
+    await client.close()
+
+
+@pytest.mark.anyio
+async def test_set_my_commands_uses_outbox() -> None:
+    bot = FakeBot()
+    client = TelegramClient(client=bot, private_chat_rps=0.0, group_chat_rps=0.0)
+
+    commands = [{"command": "ping", "description": "Ping the bot"}]
+    result = await client.set_my_commands(
+        commands,
+        scope={"type": "default"},
+        language_code="en",
+    )
+
+    assert result is True
+    assert bot.calls == ["set_my_commands"]
+    assert bot.command_calls == [(commands, {"type": "default"}, "en")]
+    await client.close()
+
+
+@pytest.mark.anyio
+async def test_answer_callback_query_uses_outbox() -> None:
+    bot = FakeBot()
+    client = TelegramClient(client=bot, private_chat_rps=0.0, group_chat_rps=0.0)
+
+    result = await client.answer_callback_query(
+        callback_query_id="cb-1",
+        text="ok",
+        show_alert=True,
+    )
+
+    assert result is True
+    assert bot.calls == ["answer_callback_query"]
+    assert bot.callback_calls == [("cb-1", "ok", True)]
+    await client.close()
+
+
+@pytest.mark.anyio
+async def test_get_chat_and_member_uses_outbox() -> None:
+    bot = FakeBot()
+    client = TelegramClient(client=bot, private_chat_rps=0.0, group_chat_rps=0.0)
+
+    chat = await client.get_chat(9)
+    member = await client.get_chat_member(9, 42)
+
+    assert chat is not None
+    assert chat.id == 9
+    assert member is not None
+    assert member.status == "member"
+    assert bot.calls == ["get_chat", "get_chat_member"]
+    assert bot.chat_calls == [9]
+    assert bot.chat_member_calls == [(9, 42)]
+    await client.close()
+
+
+@pytest.mark.anyio
+async def test_create_forum_topic_uses_outbox() -> None:
+    bot = FakeBot()
+    client = TelegramClient(client=bot, private_chat_rps=0.0, group_chat_rps=0.0)
+
+    topic = await client.create_forum_topic(3, "status updates")
+
+    assert topic is not None
+    assert topic.message_thread_id == 11
+    assert bot.calls == ["create_forum_topic"]
+    assert bot.create_topic_calls == [(3, "status updates")]
+    await client.close()
+
+
 @pytest.mark.anyio
 async def test_edits_coalesce_latest() -> None:
     class _BlockingBot(FakeBot):