diff --git a/pyproject.toml b/pyproject.toml index 03bc41f..f73f93f 100644 --- a/pyproject.toml +++ b/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"] diff --git a/src/takopi/config_watch.py b/src/takopi/config_watch.py index 6151dbe..3a0740c 100644 --- a/src/takopi/config_watch.py +++ b/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 diff --git a/src/takopi/logging.py b/src/takopi/logging.py index a1f87d6..8f0e30f 100644 --- a/src/takopi/logging.py +++ b/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 diff --git a/tests/test_onboarding_helpers.py b/tests/test_onboarding_helpers.py index f00ff7c..8d6071a 100644 --- a/tests/test_onboarding_helpers.py +++ b/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: diff --git a/tests/test_telegram_queue.py b/tests/test_telegram_queue.py index 0330729..d2138e8 100644 --- a/tests/test_telegram_queue.py +++ b/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,16 +69,18 @@ class FakeBot(BotClient): disable_notification: bool | None = False, caption: str | None = None, ) -> Message | None: - _ = ( - chat_id, - filename, - content, - reply_to_message_id, - message_thread_id, - disable_notification, - caption, - ) self.calls.append("send_document") + self.document_calls.append( + ( + chat_id, + filename, + content, + reply_to_message_id, + message_thread_id, + disable_notification, + caption, + ) + ) 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):