slaguru666/AIPass
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

#691 telegram tests CI-safe: fully-qualified imports + hermetic network-block fixture. 16 test files bare 'from apps.handlers' -> 'aipass.skills.lib.telegram.apps.handlers' (+ patch targets) — bare 'apps' collided with other branches' apps at full-repo collection -> ~16 collection errors -> CI red. Verify caught ~11 tests hitting live api.telegram.org (base_bot->set_bot_commands->urlopen, never ran in CI before); added session autouse _block_network conftest fixture patching urlopen on 4 telegram modules (bare+fq, guarded) so live calls fail loud not hang. Test-infra only, product byte-unchanged, 0 assertion changes. Full-repo collect 0 errors (11028); telegram 663 pass/0 fail/0 hang. devpulse independently re-verified.

Browse Source
This commit is contained in:
AIOSAI
2026-07-11 09:06:34 -07:00
parent c244b7bf3f
commit deffa0c912
19 changed files with 439 additions and 386 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+17
View File
@@ -34,6 +34,23 @@ PyPI version — not the changelog header.
### Fixed
- **Telegram skill tests made CI-safe — full-repo collection + hermeticity (issue #691).**
The 16 test files under `skills/lib/telegram/tests/` imported handlers via bare
`from apps.handlers…`, which collided with other branches' `apps` packages during
full-repo CI collection (~16 ImportError collection errors → CI red on Linux +
Windows). Converted to fully-qualified `aipass.skills.lib.telegram.apps.handlers.*`
imports (and matching `mock.patch` targets). Verifying that fix surfaced a second
problem the imports had exposed: ~11 tests reached the live Telegram API
(`base_bot.run → _set_command_menu → set_bot_commands → urlopen`) — they had never
run in CI before because they failed at collection. Added a session-scoped autouse
`_block_network` conftest fixture that patches `urlopen` on the four network-using
telegram modules (both bare and fully-qualified import paths, each guarded) so any
test attempting a live HTTP call fails loud instead of hanging. Test-infra only —
product code byte-unchanged, no telegram behavior change. Full-repo collection now
0 errors (11,028 tests); telegram suite 663 passed / 0 failed / 0 hangs, fully
hermetic. Coverage intact (zero assertion changes — purely import/patch-target
rewiring).
- **`aipass install` from a throwaway path can no longer hijack the machine-wide
`AIPASS_HOME` (issue #688).** A probe install run from a `/tmp` scratchpad had
rewritten `~/.claude/settings.json` `env.AIPASS_HOME`, silently pointing every
src/aipass/skills/lib/telegram/tests/conftest.py
+51 -7
View File
@@ -9,10 +9,10 @@
"""
Telegram skill test configuration.
Sets up sys.path so that both aipass.* (installed package) and the local
apps.handlers.* namespace are importable from tests without a full pip install.
Also stubs the optional telethon dependency and redirects Prax logger output
to a temp dir so test runs don't pollute production log files.
Sets up sys.path so that aipass.* (installed package) is importable from tests
without a full pip install. Also stubs the optional telethon dependency and
redirects Prax logger output to a temp dir so test runs don't pollute production
log files.
"""
import os
@@ -28,9 +28,10 @@ if "AIPASS_TEST_LOG_DIR" not in os.environ:
import pytest
# sys.path setup is intentional test infrastructure — both entries are needed:
# _src_root → resolves aipass.* installed-package imports
# _skill_root → resolves the local apps.handlers.* namespace used by all tests
# sys.path setup:
# _src_root → resolves aipass.* installed-package imports (test imports)
# _skill_root → resolves bare 'apps.handlers' lazy imports inside handler.py
# (product code, exercised at test runtime — not at collection)
_src_root = Path(__file__).resolve().parents[5]
if str(_src_root) not in sys.path:
sys.path.insert(0, str(_src_root))
@@ -88,6 +89,49 @@ def _redirect_prax_logs(tmp_path_factory):
direct_mod._direct_loggers.clear()
class _NetworkBlockedError(Exception):
"""Raised when a test attempts a real network call."""
def __init__(self):
super().__init__("NETWORK BLOCKED: test attempted a live HTTP call. Mock urlopen or the calling function.")
def _blocked_urlopen(*args, **kwargs):
raise _NetworkBlockedError()
_URLOPEN_TARGETS = [
"aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen",
"aipass.skills.lib.telegram.apps.handlers.bot_factory.urlopen",
"aipass.skills.lib.telegram.apps.handlers.notifier.urlopen",
"aipass.skills.lib.telegram.apps.handlers.log_streamer.urlopen",
"apps.handlers.base_bot.urlopen",
"apps.handlers.bot_factory.urlopen",
"apps.handlers.notifier.urlopen",
"apps.handlers.log_streamer.urlopen",
]
@pytest.fixture(autouse=True, scope="session")
def _block_network():
"""Block all outbound HTTP in tests. Any test hitting the real network fails loud."""
from unittest.mock import patch
patches = []
for target in _URLOPEN_TARGETS:
try:
p = patch(target, side_effect=_blocked_urlopen)
p.start()
patches.append(p)
except (ModuleNotFoundError, AttributeError):
pass
yield
for p in patches:
p.stop()
@pytest.fixture
def temp_test_dir() -> Generator[Path, None, None]:
"""Creates temporary directory for testing, cleans up after."""
src/aipass/skills/lib/telegram/tests/test_attach_only.py
+7 -7
View File
@@ -23,15 +23,15 @@ from unittest.mock import patch, MagicMock
import pytest
from apps.handlers.base_bot import BaseBot # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.base_bot import BaseBot
@pytest.fixture
def _patch_base_bot_deps(tmp_path):
patches = [
patch("apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("apps.handlers.base_bot.signal.signal"),
patch("apps.handlers.base_bot.atexit.register"),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.signal.signal"),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.atexit.register"),
]
for p in patches:
p.start()
@@ -43,7 +43,7 @@ def _patch_base_bot_deps(tmp_path):
def _make_bot(tmp_path, _patch_base_bot_deps, attach_only=False, shared_session=None):
workdir = tmp_path / "workdir"
workdir.mkdir(exist_ok=True)
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
bot = BaseBot(
bot_id="mirror_test",
bot_token="123:FAKETOKEN",
@@ -253,7 +253,7 @@ class TestAttachOnlyLock:
patch.object(bot, "clean_stale_pending"),
patch.object(bot, "_load_offset", return_value=0),
patch.object(bot, "poll_updates", side_effect=KeyboardInterrupt),
patch("apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path),
):
bot.run()
@@ -273,7 +273,7 @@ class TestAttachOnlyLock:
patch.object(bot, "clean_stale_pending"),
patch.object(bot, "_load_offset", return_value=0),
patch.object(bot, "poll_updates", side_effect=KeyboardInterrupt),
patch("apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path),
):
bot.run()
src/aipass/skills/lib/telegram/tests/test_bot_registry.py
+1 -1
View File
@@ -8,7 +8,7 @@ No external dependencies beyond pytest.
import json
import pytest
from apps.handlers import bot_registry # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers import bot_registry
# =============================================
src/aipass/skills/lib/telegram/tests/test_botfather_client.py
+37 -37
View File
@@ -17,7 +17,7 @@ import asyncio
from pathlib import Path
from unittest.mock import patch, MagicMock, AsyncMock
from apps.handlers.botfather_client import (
from aipass.skills.lib.telegram.apps.handlers.botfather_client import (
_load_telethon_config,
check_telethon_setup,
_format_display_name,
@@ -36,7 +36,7 @@ from apps.handlers.botfather_client import (
class TestLoadTelethonConfig:
"""Test _load_telethon_config: secret loading, JSON parsing, validation."""
@patch("apps.handlers.botfather_client._get_secret")
@patch("aipass.skills.lib.telegram.apps.handlers.botfather_client._get_secret")
def test_returns_config_when_valid(self, mock_get_secret):
"""Returns config dict when secret store has valid api_id and api_hash."""
mock_get_secret.return_value = {"api_id": 12345, "api_hash": "abc123def"}
@@ -46,7 +46,7 @@ class TestLoadTelethonConfig:
assert result["api_hash"] == "abc123def"
mock_get_secret.assert_called_once_with("telethon_config")
@patch("apps.handlers.botfather_client._get_secret")
@patch("aipass.skills.lib.telegram.apps.handlers.botfather_client._get_secret")
def test_raises_when_secret_missing(self, mock_get_secret):
"""Raises RuntimeError when the secret doesn't exist."""
mock_get_secret.return_value = None
@@ -55,7 +55,7 @@ class TestLoadTelethonConfig:
with pytest.raises(RuntimeError, match="not found in secrets store"):
_load_telethon_config()
@patch("apps.handlers.botfather_client._get_secret")
@patch("aipass.skills.lib.telegram.apps.handlers.botfather_client._get_secret")
def test_raises_when_api_id_missing(self, mock_get_secret):
"""Raises RuntimeError when api_id is missing from config."""
mock_get_secret.return_value = {"api_hash": "abc123def"}
@@ -64,7 +64,7 @@ class TestLoadTelethonConfig:
with pytest.raises(RuntimeError, match="missing api_id or api_hash"):
_load_telethon_config()
@patch("apps.handlers.botfather_client._get_secret")
@patch("aipass.skills.lib.telegram.apps.handlers.botfather_client._get_secret")
def test_coerces_api_id_from_string(self, mock_get_secret):
"""Coerces api_id to int when provided as a string."""
mock_get_secret.return_value = {"api_id": "99999", "api_hash": "xyz789"}
@@ -82,16 +82,16 @@ class TestLoadTelethonConfig:
class TestCheckTelethonSetup:
"""Test check_telethon_setup: checks library, config, session file."""
@patch("apps.handlers.botfather_client._load_telethon_config")
@patch("aipass.skills.lib.telegram.apps.handlers.botfather_client._load_telethon_config")
def test_returns_ready_when_all_in_place(self, mock_load_config, tmp_path, monkeypatch):
"""Returns (True, 'ready') when Telethon is available, config valid, session exists."""
monkeypatch.setattr("apps.handlers.botfather_client.TELETHON_AVAILABLE", True)
monkeypatch.setattr("aipass.skills.lib.telegram.apps.handlers.botfather_client.TELETHON_AVAILABLE", True)
mock_load_config.return_value = {"api_id": 12345, "api_hash": "abc123"}
# Create session file at the new path
session_path = tmp_path / ".telethon"
monkeypatch.setattr(
"apps.handlers.botfather_client.SESSION_PATH",
"aipass.skills.lib.telegram.apps.handlers.botfather_client.SESSION_PATH",
session_path,
)
session_file = Path(str(session_path) + ".session")
@@ -103,15 +103,15 @@ class TestCheckTelethonSetup:
def test_returns_false_when_telethon_not_available(self, monkeypatch):
"""Returns (False, ...) when TELETHON_AVAILABLE is False."""
monkeypatch.setattr("apps.handlers.botfather_client.TELETHON_AVAILABLE", False)
monkeypatch.setattr("aipass.skills.lib.telegram.apps.handlers.botfather_client.TELETHON_AVAILABLE", False)
ready, reason = check_telethon_setup()
assert ready is False
assert "not installed" in reason.lower() or "telethon" in reason.lower()
@patch("apps.handlers.botfather_client._load_telethon_config")
@patch("aipass.skills.lib.telegram.apps.handlers.botfather_client._load_telethon_config")
def test_returns_false_when_config_not_in_secrets(self, mock_load_config, monkeypatch):
"""Returns (False, ...) when secret store has no telethon config."""
monkeypatch.setattr("apps.handlers.botfather_client.TELETHON_AVAILABLE", True)
monkeypatch.setattr("aipass.skills.lib.telegram.apps.handlers.botfather_client.TELETHON_AVAILABLE", True)
mock_load_config.side_effect = RuntimeError(
"Telethon config not found in secrets store (telegram/telethon_config)"
)
@@ -119,24 +119,24 @@ class TestCheckTelethonSetup:
assert ready is False
assert "not found" in reason.lower()
@patch("apps.handlers.botfather_client._load_telethon_config")
@patch("aipass.skills.lib.telegram.apps.handlers.botfather_client._load_telethon_config")
def test_returns_false_when_config_invalid(self, mock_load_config, monkeypatch):
"""Returns (False, ...) when config exists but is invalid (missing fields)."""
monkeypatch.setattr("apps.handlers.botfather_client.TELETHON_AVAILABLE", True)
monkeypatch.setattr("aipass.skills.lib.telegram.apps.handlers.botfather_client.TELETHON_AVAILABLE", True)
mock_load_config.side_effect = RuntimeError("Telethon config incomplete — missing api_id or api_hash")
ready, reason = check_telethon_setup()
assert ready is False
assert "missing api_id or api_hash" in reason
@patch("apps.handlers.botfather_client._load_telethon_config")
@patch("aipass.skills.lib.telegram.apps.handlers.botfather_client._load_telethon_config")
def test_returns_false_when_session_file_missing(self, mock_load_config, tmp_path, monkeypatch):
"""Returns (False, ...) when session file doesn't exist."""
monkeypatch.setattr("apps.handlers.botfather_client.TELETHON_AVAILABLE", True)
monkeypatch.setattr("aipass.skills.lib.telegram.apps.handlers.botfather_client.TELETHON_AVAILABLE", True)
mock_load_config.return_value = {"api_id": 12345, "api_hash": "abc123"}
session_path = tmp_path / ".telethon"
monkeypatch.setattr(
"apps.handlers.botfather_client.SESSION_PATH",
"aipass.skills.lib.telegram.apps.handlers.botfather_client.SESSION_PATH",
session_path,
)
# Do NOT create the session file
@@ -202,7 +202,7 @@ class TestBotFatherClientConnect:
mock_client_instance.get_me.return_value = MagicMock(first_name="TestUser", id=123)
with patch(
"apps.handlers.botfather_client._telethon_check",
"aipass.skills.lib.telegram.apps.handlers.botfather_client._telethon_check",
create=True,
):
with patch(
@@ -299,7 +299,7 @@ class TestBotFatherClientSendAndWait:
# Use a real time base so asyncio loop isn't disrupted.
# MESSAGE_TIMEOUT is 30s; the mock response arrives immediately,
# so the while-loop condition is satisfied on the first iteration.
with patch("apps.handlers.botfather_client.asyncio.sleep", new_callable=AsyncMock):
with patch("aipass.skills.lib.telegram.apps.handlers.botfather_client.asyncio.sleep", new_callable=AsyncMock):
result = asyncio.run(client._send_and_wait(entity, "/newbot"))
assert result == "Please choose a name for your bot."
@@ -317,8 +317,8 @@ class TestBotFatherClientSendAndWait:
mock_telethon.get_messages.return_value = [mock_msg]
# Shrink the timeout to 0 so the while-loop exits immediately
with patch("apps.handlers.botfather_client.MESSAGE_TIMEOUT", 0):
with patch("apps.handlers.botfather_client.asyncio.sleep", new_callable=AsyncMock):
with patch("aipass.skills.lib.telegram.apps.handlers.botfather_client.MESSAGE_TIMEOUT", 0):
with patch("aipass.skills.lib.telegram.apps.handlers.botfather_client.asyncio.sleep", new_callable=AsyncMock):
result = asyncio.run(client._send_and_wait(entity, "/newbot"))
assert result is None
@@ -344,7 +344,7 @@ class TestBotFatherClientSendAndWait:
mock_msg.text = "Response after flood wait"
mock_telethon.get_messages.return_value = [mock_msg]
with patch("apps.handlers.botfather_client.asyncio.sleep", new_callable=AsyncMock):
with patch("aipass.skills.lib.telegram.apps.handlers.botfather_client.asyncio.sleep", new_callable=AsyncMock):
with patch("telethon.errors.FloodWaitError", MockFloodWaitError, create=True):
with patch("telethon.errors.RPCError", MockRPCError, create=True):
result = asyncio.run(client._send_and_wait(entity, "/newbot"))
@@ -601,7 +601,7 @@ class TestCreateBotViaBotfather:
def test_raises_when_setup_not_ready(self, monkeypatch):
"""Raises RuntimeError when check_telethon_setup says not ready."""
monkeypatch.setattr(
"apps.handlers.botfather_client.check_telethon_setup",
"aipass.skills.lib.telegram.apps.handlers.botfather_client.check_telethon_setup",
lambda: (False, "Telethon not installed"),
)
import pytest
@@ -612,7 +612,7 @@ class TestCreateBotViaBotfather:
def test_raises_when_config_load_fails(self, monkeypatch):
"""Raises RuntimeError when _load_telethon_config raises."""
monkeypatch.setattr(
"apps.handlers.botfather_client.check_telethon_setup",
"aipass.skills.lib.telegram.apps.handlers.botfather_client.check_telethon_setup",
lambda: (True, "ready"),
)
@@ -620,7 +620,7 @@ class TestCreateBotViaBotfather:
raise RuntimeError("Telethon config not found in secrets store (telegram/telethon_config)")
monkeypatch.setattr(
"apps.handlers.botfather_client._load_telethon_config",
"aipass.skills.lib.telegram.apps.handlers.botfather_client._load_telethon_config",
_raise,
)
import pytest
@@ -631,11 +631,11 @@ class TestCreateBotViaBotfather:
def test_returns_result_on_success(self, monkeypatch):
"""Returns result dict on successful flow."""
monkeypatch.setattr(
"apps.handlers.botfather_client.check_telethon_setup",
"aipass.skills.lib.telegram.apps.handlers.botfather_client.check_telethon_setup",
lambda: (True, "ready"),
)
monkeypatch.setattr(
"apps.handlers.botfather_client._load_telethon_config",
"aipass.skills.lib.telegram.apps.handlers.botfather_client._load_telethon_config",
lambda: {"api_id": 12345, "api_hash": "abc123"},
)
@@ -661,7 +661,7 @@ class TestCreateBotViaBotfather:
mock_client.disconnect = mock_disconnect
with patch(
"apps.handlers.botfather_client.BotFatherClient",
"aipass.skills.lib.telegram.apps.handlers.botfather_client.BotFatherClient",
return_value=mock_client,
):
result = create_bot_via_botfather("dev_central")
@@ -673,11 +673,11 @@ class TestCreateBotViaBotfather:
def test_handles_connection_failure(self, monkeypatch):
"""Returns None when connection fails."""
monkeypatch.setattr(
"apps.handlers.botfather_client.check_telethon_setup",
"aipass.skills.lib.telegram.apps.handlers.botfather_client.check_telethon_setup",
lambda: (True, "ready"),
)
monkeypatch.setattr(
"apps.handlers.botfather_client._load_telethon_config",
"aipass.skills.lib.telegram.apps.handlers.botfather_client._load_telethon_config",
lambda: {"api_id": 12345, "api_hash": "abc123"},
)
@@ -693,7 +693,7 @@ class TestCreateBotViaBotfather:
mock_client.disconnect = mock_disconnect
with patch(
"apps.handlers.botfather_client.BotFatherClient",
"aipass.skills.lib.telegram.apps.handlers.botfather_client.BotFatherClient",
return_value=mock_client,
):
result = create_bot_via_botfather("dev_central")
@@ -703,11 +703,11 @@ class TestCreateBotViaBotfather:
def test_handles_botfather_failure(self, monkeypatch):
"""Returns None when BotFather automation fails."""
monkeypatch.setattr(
"apps.handlers.botfather_client.check_telethon_setup",
"aipass.skills.lib.telegram.apps.handlers.botfather_client.check_telethon_setup",
lambda: (True, "ready"),
)
monkeypatch.setattr(
"apps.handlers.botfather_client._load_telethon_config",
"aipass.skills.lib.telegram.apps.handlers.botfather_client._load_telethon_config",
lambda: {"api_id": 12345, "api_hash": "abc123"},
)
@@ -727,7 +727,7 @@ class TestCreateBotViaBotfather:
mock_client.disconnect = mock_disconnect
with patch(
"apps.handlers.botfather_client.BotFatherClient",
"aipass.skills.lib.telegram.apps.handlers.botfather_client.BotFatherClient",
return_value=mock_client,
):
result = create_bot_via_botfather("dev_central")
@@ -737,11 +737,11 @@ class TestCreateBotViaBotfather:
def test_handles_existing_event_loop(self, monkeypatch):
"""Handles the edge case where an event loop is already running."""
monkeypatch.setattr(
"apps.handlers.botfather_client.check_telethon_setup",
"aipass.skills.lib.telegram.apps.handlers.botfather_client.check_telethon_setup",
lambda: (True, "ready"),
)
monkeypatch.setattr(
"apps.handlers.botfather_client._load_telethon_config",
"aipass.skills.lib.telegram.apps.handlers.botfather_client._load_telethon_config",
lambda: {"api_id": 12345, "api_hash": "abc123"},
)
@@ -767,7 +767,7 @@ class TestCreateBotViaBotfather:
mock_client.disconnect = mock_disconnect
with patch(
"apps.handlers.botfather_client.BotFatherClient",
"aipass.skills.lib.telegram.apps.handlers.botfather_client.BotFatherClient",
return_value=mock_client,
):
# Simulate an already-running event loop by patching get_running_loop
@@ -776,7 +776,7 @@ class TestCreateBotViaBotfather:
mock_loop.is_running.return_value = True
with patch(
"apps.handlers.botfather_client.asyncio.get_running_loop",
"aipass.skills.lib.telegram.apps.handlers.botfather_client.asyncio.get_running_loop",
return_value=mock_loop,
):
result = create_bot_via_botfather("flow")
src/aipass/skills/lib/telegram/tests/test_handler_routing.py
+1 -9
View File
@@ -1,16 +1,8 @@
"""Tests for handler.py action routing."""
import sys
from pathlib import Path
from unittest.mock import patch
# Ensure skill root is on path so handler can import apps.handlers.*
_skill_root = Path(__file__).resolve().parents[1]
if str(_skill_root) not in sys.path:
sys.path.insert(0, str(_skill_root))
from handler import run, _ok, _err, _ACTIONS, _DISPATCH
from aipass.skills.lib.telegram.handler import run, _ok, _err, _ACTIONS, _DISPATCH
class TestHelpers:
src/aipass/skills/lib/telegram/tests/test_heartbeat_delivered.py
+9 -9
View File
@@ -25,15 +25,15 @@ import time
import pytest
from unittest.mock import patch
from apps.handlers.base_bot import BaseBot # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.base_bot import BaseBot
@pytest.fixture
def _patch_base_bot_deps(tmp_path):
patches = [
patch("apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("apps.handlers.base_bot.signal.signal"),
patch("apps.handlers.base_bot.atexit.register"),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.signal.signal"),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.atexit.register"),
]
for p in patches:
p.start()
@@ -44,7 +44,7 @@ def _patch_base_bot_deps(tmp_path):
def _make_bot(tmp_path, _patch_base_bot_deps, pending_dir=None):
pdir = pending_dir or tmp_path
with patch("apps.handlers.base_bot.PENDING_DIR", pdir):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", pdir):
workdir = tmp_path / "workdir"
workdir.mkdir(exist_ok=True)
bot = BaseBot(
@@ -121,7 +121,7 @@ class TestHeartbeatStopsOnDelivered:
with (
patch.object(bot, "edit_message", side_effect=fake_edit),
patch.object(bot, "_tmux_session_exists", return_value=True),
patch("apps.handlers.base_bot.HEARTBEAT_INTERVAL", 0.1),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.HEARTBEAT_INTERVAL", 0.1),
):
bot._start_heartbeat(42, 999)
time.sleep(0.5)
@@ -144,7 +144,7 @@ class TestHeartbeatStopsOnDelivered:
with (
patch.object(bot, "edit_message", side_effect=fake_edit),
patch.object(bot, "_tmux_session_exists", return_value=True),
patch("apps.handlers.base_bot.HEARTBEAT_INTERVAL", 0.1),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.HEARTBEAT_INTERVAL", 0.1),
):
bot._start_heartbeat(42, 999)
time.sleep(0.5)
@@ -161,7 +161,7 @@ class TestHeartbeatStopsOnDelivered:
with (
patch.object(bot, "edit_message") as mock_edit,
patch.object(bot, "_tmux_session_exists", return_value=True),
patch("apps.handlers.base_bot.HEARTBEAT_INTERVAL", 0.1),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.HEARTBEAT_INTERVAL", 0.1),
):
bot._start_heartbeat(42, 999)
time.sleep(0.4)
@@ -236,7 +236,7 @@ class TestReplyNotClobbered:
with (
patch.object(bot, "edit_message") as mock_edit,
patch.object(bot, "_tmux_session_exists", return_value=True),
patch("apps.handlers.base_bot.HEARTBEAT_INTERVAL", 0.1),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.HEARTBEAT_INTERVAL", 0.1),
):
bot._start_heartbeat(42, 999)
time.sleep(0.4)
src/aipass/skills/lib/telegram/tests/test_log_streamer.py
+28 -28
View File
@@ -18,7 +18,7 @@ import threading
import pytest
from unittest.mock import patch, MagicMock
from apps.handlers.log_streamer import ( # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.log_streamer import (
LogStreamer,
TELEGRAM_MAX_LENGTH,
)
@@ -31,7 +31,7 @@ from apps.handlers.log_streamer import ( # type: ignore[import-not-found]
def _make_streamer(logs_dir, tmp_path, branch_name: str = "api") -> LogStreamer:
"""Build a LogStreamer with SYSTEM_LOGS_DIR redirected to tmp_path."""
with patch("apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
s = LogStreamer(
bot_token="123:FAKETOKEN",
chat_id=999888,
@@ -149,7 +149,7 @@ class TestPositionTracking:
with open(log_file, "a", encoding="utf-8") as f:
f.write("new_line_3\nnew_line_4\n")
with patch("apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
new_lines = streamer_with_files._read_new_lines()
assert "new_line_3" in new_lines
@@ -157,7 +157,7 @@ class TestPositionTracking:
def test_no_new_content_returns_empty(self, streamer_with_files, logs_dir):
"""If nothing was appended, should return empty list."""
with patch("apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
new_lines = streamer_with_files._read_new_lines()
assert new_lines == []
@@ -167,7 +167,7 @@ class TestPositionTracking:
# Simulate rotation: overwrite with smaller content
log_file.write_text("rotated\n", encoding="utf-8")
with patch("apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
new_lines = streamer_with_files._read_new_lines()
assert "rotated" in new_lines
@@ -177,7 +177,7 @@ class TestPositionTracking:
new_file = logs_dir / "api_new_module.log"
new_file.write_text("discovered_line\n", encoding="utf-8")
with patch("apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
new_lines = streamer_with_files._read_new_lines()
assert "discovered_line" in new_lines
@@ -189,7 +189,7 @@ class TestPositionTracking:
with open(non_matching, "a", encoding="utf-8") as f:
f.write("should_be_ignored\n")
with patch("apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
new_lines = streamer_with_files._read_new_lines()
assert "should_be_ignored" not in new_lines
@@ -199,7 +199,7 @@ class TestPositionTracking:
log_file = logs_dir / "api_main.log"
log_file.unlink()
with patch("apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
# Should not raise
new_lines = streamer_with_files._read_new_lines()
@@ -215,7 +215,7 @@ class TestPositionTracking:
with open(log_file, "a", encoding="utf-8") as f:
f.write("extra\n")
with patch("apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
streamer_with_files._read_new_lines()
assert streamer_with_files.log_positions[file_path] > initial_pos
@@ -223,7 +223,7 @@ class TestPositionTracking:
def test_system_logs_dir_missing_returns_empty(self, streamer, tmp_path):
"""If SYSTEM_LOGS_DIR does not exist, _get_log_files returns empty."""
missing_dir = tmp_path / "does_not_exist"
with patch("apps.handlers.log_streamer.SYSTEM_LOGS_DIR", missing_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.SYSTEM_LOGS_DIR", missing_dir):
new_lines = streamer._read_new_lines()
assert new_lines == []
@@ -248,7 +248,7 @@ class TestSendMessage:
"""Payload should include chat_id, text, and disable_notification."""
mock_resp = self._make_mock_response({"ok": True})
with patch("apps.handlers.log_streamer.urlopen", return_value=mock_resp) as mock_urlopen:
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.urlopen", return_value=mock_resp) as mock_urlopen:
streamer._send_message("hello world")
# Verify the request was made
@@ -264,7 +264,7 @@ class TestSendMessage:
"""Should return True when Telegram responds with ok=True."""
mock_resp = self._make_mock_response({"ok": True})
with patch("apps.handlers.log_streamer.urlopen", return_value=mock_resp):
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.urlopen", return_value=mock_resp):
result = streamer._send_message("test")
assert result is True
@@ -273,14 +273,14 @@ class TestSendMessage:
"""Should return False and not crash on URLError."""
from urllib.error import URLError
with patch("apps.handlers.log_streamer.urlopen", side_effect=URLError("fail")):
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.urlopen", side_effect=URLError("fail")):
result = streamer._send_message("test")
assert result is False
def test_returns_false_on_generic_exception(self, streamer):
"""Should return False on any unexpected exception."""
with patch("apps.handlers.log_streamer.urlopen", side_effect=RuntimeError("boom")):
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.urlopen", side_effect=RuntimeError("boom")):
result = streamer._send_message("test")
assert result is False
@@ -290,8 +290,8 @@ class TestSendMessage:
from urllib.error import URLError
with (
patch("apps.handlers.log_streamer.urlopen", side_effect=URLError("network")),
patch("apps.handlers.log_streamer.logger") as mock_logger,
patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.urlopen", side_effect=URLError("network")),
patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.logger") as mock_logger,
):
streamer._send_message("test")
@@ -302,7 +302,7 @@ class TestSendMessage:
"""Should return False when Telegram responds with ok=False."""
mock_resp = self._make_mock_response({"ok": False})
with patch("apps.handlers.log_streamer.urlopen", return_value=mock_resp):
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.urlopen", return_value=mock_resp):
result = streamer._send_message("test")
assert result is False
@@ -311,7 +311,7 @@ class TestSendMessage:
"""Request should have Content-Type: application/json header."""
mock_resp = self._make_mock_response({"ok": True})
with patch("apps.handlers.log_streamer.urlopen", return_value=mock_resp) as mock_urlopen:
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.urlopen", return_value=mock_resp) as mock_urlopen:
streamer._send_message("test")
req = mock_urlopen.call_args[0][0]
@@ -420,7 +420,7 @@ class TestStartStop:
streamer.start()
first_thread = streamer._thread
with patch("apps.handlers.log_streamer.logger") as mock_logger:
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.logger") as mock_logger:
streamer.start()
mock_logger.warning.assert_called_once()
@@ -466,9 +466,9 @@ class TestBaseBotIntegration:
def _patch_base_bot_deps(self, tmp_path):
"""Patch heavy BaseBot dependencies to allow lightweight instantiation."""
patches = [
patch("apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("apps.handlers.base_bot.signal.signal"),
patch("apps.handlers.base_bot.atexit.register"),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.signal.signal"),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.atexit.register"),
]
for p in patches:
p.start()
@@ -478,7 +478,7 @@ class TestBaseBotIntegration:
def test_streamer_starts_on_first_message_with_branch_name(self, tmp_path, _patch_base_bot_deps):
"""When branch_name is set, LogStreamer should start on first message."""
from apps.handlers.base_bot import BaseBot # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.base_bot import BaseBot
workdir = tmp_path / "workdir"
workdir.mkdir()
@@ -504,7 +504,7 @@ class TestBaseBotIntegration:
}
# Patch LogStreamer at the import location in base_bot
with patch.object(bot, "send_message"), patch("apps.handlers.base_bot.LogStreamer") as MockStreamer:
with patch.object(bot, "send_message"), patch("aipass.skills.lib.telegram.apps.handlers.base_bot.LogStreamer") as MockStreamer:
mock_instance = MagicMock()
MockStreamer.return_value = mock_instance
@@ -515,7 +515,7 @@ class TestBaseBotIntegration:
def test_streamer_not_started_when_branch_name_is_none(self, tmp_path, _patch_base_bot_deps):
"""When branch_name is None (base bot), LogStreamer should NOT be created."""
from apps.handlers.base_bot import BaseBot # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.base_bot import BaseBot
workdir = tmp_path / "workdir"
workdir.mkdir()
@@ -537,13 +537,13 @@ class TestBaseBotIntegration:
}
}
with patch.object(bot, "send_message"), patch("apps.handlers.base_bot.LogStreamer") as MockStreamer:
with patch.object(bot, "send_message"), patch("aipass.skills.lib.telegram.apps.handlers.base_bot.LogStreamer") as MockStreamer:
bot.process_update(fake_update)
MockStreamer.assert_not_called()
def test_cleanup_stops_streamer(self, tmp_path, _patch_base_bot_deps):
"""BaseBot._cleanup should call stop() on the LogStreamer if it exists."""
from apps.handlers.base_bot import BaseBot # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.base_bot import BaseBot
workdir = tmp_path / "workdir"
workdir.mkdir()
@@ -566,7 +566,7 @@ class TestBaseBotIntegration:
def test_cleanup_safe_when_no_streamer(self, tmp_path, _patch_base_bot_deps):
"""BaseBot._cleanup should work fine when _log_streamer is None."""
from apps.handlers.base_bot import BaseBot # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.base_bot import BaseBot
workdir = tmp_path / "workdir"
workdir.mkdir()
src/aipass/skills/lib/telegram/tests/test_mirror_session.py
+35 -35
View File
@@ -24,8 +24,8 @@ from unittest.mock import MagicMock, patch
import pytest
from apps.handlers.base_bot import BaseBot # type: ignore[import-not-found]
from apps.handlers.bot_factory import ( # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.base_bot import BaseBot
from aipass.skills.lib.telegram.apps.handlers.bot_factory import (
launch_mirror_session,
start_service,
)
@@ -39,9 +39,9 @@ from apps.handlers.bot_factory import ( # type: ignore[import-not-found]
@pytest.fixture
def _patch_base_bot_deps(tmp_path):
patches = [
patch("apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("apps.handlers.base_bot.signal.signal"),
patch("apps.handlers.base_bot.atexit.register"),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.signal.signal"),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.atexit.register"),
]
for p in patches:
p.start()
@@ -53,7 +53,7 @@ def _patch_base_bot_deps(tmp_path):
def _make_bot(tmp_path, _patch_base_bot_deps, attach_only=False, shared_session=None):
workdir = tmp_path / "workdir"
workdir.mkdir(exist_ok=True)
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
bot = BaseBot(
bot_id="mirror_test",
bot_token="123:FAKETOKEN",
@@ -83,7 +83,7 @@ class TestLaunchMirrorSession:
side = [no_session, ok, ok, ok]
with (
patch("apps.handlers.bot_factory.subprocess.run", side_effect=side),
patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.subprocess.run", side_effect=side),
patch("time.sleep"),
):
result = launch_mirror_session(
@@ -110,7 +110,7 @@ class TestLaunchMirrorSession:
return result
with (
patch("apps.handlers.bot_factory.subprocess.run", side_effect=_track),
patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.subprocess.run", side_effect=_track),
patch("time.sleep"),
):
launch_mirror_session(
@@ -127,7 +127,7 @@ class TestLaunchMirrorSession:
"""Returns True without creating if session already exists."""
has_session = MagicMock(returncode=0)
with patch("apps.handlers.bot_factory.subprocess.run", return_value=has_session) as mock_run:
with patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.subprocess.run", return_value=has_session) as mock_run:
result = launch_mirror_session(session_name="telegram-api", bot_id="api", work_dir="/tmp/test")
assert result is True
@@ -135,7 +135,7 @@ class TestLaunchMirrorSession:
def test_returns_false_when_tmux_not_found(self):
"""Returns False when tmux is not installed."""
with patch("apps.handlers.bot_factory.subprocess.run", side_effect=FileNotFoundError):
with patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.subprocess.run", side_effect=FileNotFoundError):
result = launch_mirror_session(session_name="telegram-api", bot_id="api", work_dir="/tmp/test")
assert result is False
@@ -152,7 +152,7 @@ class TestStartService:
def test_starts_systemd_service(self):
"""Calls systemctl --user start telegram-bot@{bot_id}."""
mock_result = MagicMock(returncode=0)
with patch("apps.handlers.bot_factory.subprocess.run", return_value=mock_result) as mock_run:
with patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.subprocess.run", return_value=mock_result) as mock_run:
result = start_service("api")
assert result is True
@@ -163,7 +163,7 @@ class TestStartService:
def test_returns_false_on_failure(self):
"""Returns False when systemctl returns non-zero."""
mock_result = MagicMock(returncode=1, stderr="unit not found")
with patch("apps.handlers.bot_factory.subprocess.run", return_value=mock_result):
with patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.subprocess.run", return_value=mock_result):
assert start_service("api") is False
def test_returns_false_on_timeout(self):
@@ -171,7 +171,7 @@ class TestStartService:
import subprocess
err = subprocess.TimeoutExpired(cmd="", timeout=10)
with patch("apps.handlers.bot_factory.subprocess.run", side_effect=err):
with patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.subprocess.run", side_effect=err):
assert start_service("api") is False
@@ -189,20 +189,20 @@ class TestCreateBotMirror:
bot_info = {"username": "test_bot", "id": 123}
branch_info = {"name": "api", "path": "/home/test/api"}
patches = [
patch("apps.handlers.bot_factory.validate_token", return_value=bot_info),
patch("apps.handlers.bot_factory.validate_branch", return_value=branch_info),
patch("apps.handlers.bot_factory.get_bot", return_value=None),
patch("apps.handlers.bot_factory.get_bot_by_branch", return_value=None),
patch("apps.handlers.bot_factory.ensure_registry"),
patch("apps.handlers.bot_factory._api_set_secret"),
patch("apps.handlers.bot_factory.register_bot", return_value=True),
patch("apps.handlers.bot_factory.set_bot_commands"),
patch("apps.handlers.bot_factory.build_botfather_commands", return_value=[]),
patch("apps.handlers.bot_factory.enable_service", return_value=True),
patch("apps.handlers.bot_factory.start_bot_process", return_value=True),
patch("apps.handlers.bot_factory.launch_mirror_session", return_value=True),
patch("apps.handlers.bot_factory.start_service", return_value=True),
patch("apps.handlers.bot_factory._BOT_CONFIG_DIR", Path("/tmp/test_bots")),
patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.validate_token", return_value=bot_info),
patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.validate_branch", return_value=branch_info),
patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.get_bot", return_value=None),
patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.get_bot_by_branch", return_value=None),
patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.ensure_registry"),
patch("aipass.skills.lib.telegram.apps.handlers.bot_factory._api_set_secret"),
patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.register_bot", return_value=True),
patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.set_bot_commands"),
patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.build_botfather_commands", return_value=[]),
patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.enable_service", return_value=True),
patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.start_bot_process", return_value=True),
patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.launch_mirror_session", return_value=True),
patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.start_service", return_value=True),
patch("aipass.skills.lib.telegram.apps.handlers.bot_factory._BOT_CONFIG_DIR", Path("/tmp/test_bots")),
]
mocks = {}
started = []
@@ -217,9 +217,9 @@ class TestCreateBotMirror:
def test_config_includes_mirror_fields(self, _mock_create_deps, tmp_path):
"""Config written with shared_session, attach_only, chat_id."""
from apps.handlers.bot_factory import create_bot # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.bot_factory import create_bot
with patch("apps.handlers.bot_factory._BOT_CONFIG_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.bot_factory._BOT_CONFIG_DIR", tmp_path):
result = create_bot(
bot_id="api",
bot_token="123:FAKE",
@@ -239,9 +239,9 @@ class TestCreateBotMirror:
def test_launches_mirror_session_when_attach_only(self, _mock_create_deps, tmp_path):
"""launch_mirror_session called when shared_session + attach_only."""
from apps.handlers.bot_factory import create_bot # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.bot_factory import create_bot
with patch("apps.handlers.bot_factory._BOT_CONFIG_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.bot_factory._BOT_CONFIG_DIR", tmp_path):
create_bot(
bot_id="api",
bot_token="123:FAKE",
@@ -254,9 +254,9 @@ class TestCreateBotMirror:
def test_starts_via_systemd_when_mirror(self, _mock_create_deps, tmp_path):
"""Mirror bot started via start_service, not start_bot_process."""
from apps.handlers.bot_factory import create_bot # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.bot_factory import create_bot
with patch("apps.handlers.bot_factory._BOT_CONFIG_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.bot_factory._BOT_CONFIG_DIR", tmp_path):
create_bot(
bot_id="api",
bot_token="123:FAKE",
@@ -270,9 +270,9 @@ class TestCreateBotMirror:
def test_starts_via_popen_when_not_mirror(self, _mock_create_deps, tmp_path):
"""Non-mirror bot still uses start_bot_process."""
from apps.handlers.bot_factory import create_bot # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.bot_factory import create_bot
with patch("apps.handlers.bot_factory._BOT_CONFIG_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.bot_factory._BOT_CONFIG_DIR", tmp_path):
create_bot(
bot_id="api",
bot_token="123:FAKE",
src/aipass/skills/lib/telegram/tests/test_monitor.py
+19 -19
View File
@@ -24,7 +24,7 @@ from pathlib import Path
import pytest
from unittest.mock import patch, MagicMock
from apps.handlers.log_streamer import LogStreamer # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.log_streamer import LogStreamer
# =============================================
@@ -37,9 +37,9 @@ def _patch_base_bot_deps(tmp_path):
"""Patch heavy BaseBot dependencies to allow lightweight instantiation."""
sub_file = tmp_path / "monitor_sub.json"
patches = [
patch("apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("apps.handlers.base_bot.signal.signal"),
patch("apps.handlers.base_bot.atexit.register"),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.signal.signal"),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.atexit.register"),
]
for p in patches:
p.start()
@@ -50,7 +50,7 @@ def _patch_base_bot_deps(tmp_path):
def _make_bot(tmp_path, _patch_base_bot_deps):
"""Create a BaseBot with monitor subscription redirected to tmp_path."""
from apps.handlers.base_bot import BaseBot # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.base_bot import BaseBot
workdir = tmp_path / "workdir"
workdir.mkdir()
@@ -81,7 +81,7 @@ class TestSubscribePersists:
sub_file: Path = _patch_base_bot_deps
with (
patch.object(bot, "send_message"),
patch("apps.handlers.base_bot.LogStreamer") as MockStreamer,
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.LogStreamer") as MockStreamer,
):
MockStreamer.return_value = MagicMock()
bot._monitor_subscribe(42, "default")
@@ -109,7 +109,7 @@ class TestSubscribePersists:
bot = _make_bot(tmp_path, _patch_base_bot_deps)
with (
patch.object(bot, "send_message"),
patch("apps.handlers.base_bot.LogStreamer") as MockStreamer,
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.LogStreamer") as MockStreamer,
):
mock_instance = MagicMock()
MockStreamer.return_value = mock_instance
@@ -130,7 +130,7 @@ class TestSubscribePersists:
bot = _make_bot(tmp_path, _patch_base_bot_deps)
with (
patch.object(bot, "send_message") as mock_send,
patch("apps.handlers.base_bot.LogStreamer", return_value=MagicMock()),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.LogStreamer", return_value=MagicMock()),
):
bot._monitor_subscribe(42, "default")
mock_send.assert_called_once()
@@ -144,7 +144,7 @@ class TestSubscribePersists:
with (
patch.object(bot, "send_message"),
patch("apps.handlers.base_bot.LogStreamer", return_value=MagicMock()),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.LogStreamer", return_value=MagicMock()),
):
bot._monitor_subscribe(42, "all")
old_streamer.stop.assert_called_once()
@@ -176,7 +176,7 @@ class TestBootMonitor:
sub_file.write_text(json.dumps({"chat_id": 42, "mode": "default"}))
with patch("apps.handlers.base_bot.LogStreamer") as MockStreamer:
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.LogStreamer") as MockStreamer:
mock_instance = MagicMock()
MockStreamer.return_value = mock_instance
@@ -195,7 +195,7 @@ class TestBootMonitor:
def test_boot_noop_when_no_subscription(self, tmp_path, _patch_base_bot_deps):
bot = _make_bot(tmp_path, _patch_base_bot_deps)
# No file written — subscription absent
with patch("apps.handlers.base_bot.LogStreamer") as MockStreamer:
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.LogStreamer") as MockStreamer:
bot._boot_monitor()
MockStreamer.assert_not_called()
assert bot._monitor_streamer is None
@@ -205,7 +205,7 @@ class TestBootMonitor:
sub_file: Path = _patch_base_bot_deps
sub_file.write_text("{}")
with patch("apps.handlers.base_bot.LogStreamer") as MockStreamer:
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.LogStreamer") as MockStreamer:
bot._boot_monitor()
MockStreamer.assert_not_called()
@@ -216,7 +216,7 @@ class TestBootMonitor:
sub_file.write_text(json.dumps({"chat_id": 99, "mode": "all"}))
with patch("apps.handlers.base_bot.LogStreamer") as MockStreamer:
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.LogStreamer") as MockStreamer:
MockStreamer.return_value = MagicMock()
bot._boot_monitor()
MockStreamer.assert_called_once_with(
@@ -240,14 +240,14 @@ class TestLevelFilter:
def streamer_default(self, tmp_path):
logs_dir = tmp_path / "system_logs"
logs_dir.mkdir()
with patch("apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
return LogStreamer("tok", 1, "x", system_wide=True, level_filter="default")
@pytest.fixture
def streamer_all(self, tmp_path):
logs_dir = tmp_path / "system_logs"
logs_dir.mkdir()
with patch("apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
return LogStreamer("tok", 1, "x", system_wide=True, level_filter="all")
def test_default_keeps_warning(self, streamer_default):
@@ -306,7 +306,7 @@ class TestSystemWideGlob:
(logs_dir / "prax_main.log").write_text("b\n")
(logs_dir / "trigger_events.log").write_text("c\n")
with patch("apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
s = LogStreamer("tok", 1, "monitor", system_wide=True, level_filter="all")
assert len(s.log_positions) == 3
@@ -317,7 +317,7 @@ class TestSystemWideGlob:
(logs_dir / "api_main.log").write_text("a\n")
(logs_dir / "prax_main.log").write_text("b\n")
with patch("apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
s = LogStreamer("tok", 1, "api", system_wide=False, level_filter="all")
assert len(s.log_positions) == 1
@@ -331,13 +331,13 @@ class TestSystemWideGlob:
f1.write_text("")
f2.write_text("")
with patch("apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
s = LogStreamer("tok", 1, "monitor", system_wide=True, level_filter="all")
f1.write_text("api line\n")
f2.write_text("prax line\n")
with patch("apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.log_streamer.SYSTEM_LOGS_DIR", logs_dir):
lines = s._read_new_lines()
assert "api line" in lines
src/aipass/skills/lib/telegram/tests/test_multi_bot.py
+130 -130
View File
@@ -33,8 +33,8 @@ import time
import pytest
from unittest.mock import patch, MagicMock
from apps.handlers.base_bot import BaseBot # type: ignore[import-not-found]
from apps.handlers.branch_plugin import BranchPlugin # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.base_bot import BaseBot
from aipass.skills.lib.telegram.apps.handlers.branch_plugin import BranchPlugin
# =============================================
@@ -47,7 +47,7 @@ def base_bot(tmp_path):
"""Create a BaseBot instance with PENDING_DIR pointed at tmp_path."""
workdir = tmp_path / "workdir"
workdir.mkdir()
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
bot = BaseBot(
bot_id="test_bot",
bot_token="123:FAKETOKEN",
@@ -65,7 +65,7 @@ def base_bot_open(tmp_path):
"""Create a BaseBot with an empty allowlist (allows everyone)."""
workdir = tmp_path / "workdir"
workdir.mkdir()
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
bot = BaseBot(
bot_id="open_bot",
bot_token="456:FAKETOKEN",
@@ -82,7 +82,7 @@ def branch_bot(tmp_path):
"""Create a BranchPlugin instance."""
workdir = tmp_path / "workdir"
workdir.mkdir()
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
bot = BranchPlugin(
branch_name="dev_central",
bot_id="dev_central",
@@ -141,7 +141,7 @@ class TestBaseBotInit:
assert base_bot.custom_commands == {}
def test_custom_commands_set(self, tmp_path):
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
bot = BaseBot(
bot_id="cmd_bot",
bot_token="t",
@@ -151,7 +151,7 @@ class TestBaseBotInit:
assert bot.custom_commands == {"ping": "Pong!"}
def test_allowed_user_ids_none_becomes_empty_list(self, tmp_path):
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
bot = BaseBot(
bot_id="none_bot",
bot_token="t",
@@ -318,14 +318,14 @@ class TestWritePendingFile:
"""Test pending file creation with correct JSON content."""
def test_write_creates_file(self, base_bot, tmp_path):
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch.object(base_bot, "_get_transcript_line_count", return_value=42):
result = base_bot.write_pending_file(chat_id=12345, message_id=100, processing_message_id=101)
assert result is True
assert base_bot.pending_file.exists()
def test_write_correct_json_content(self, base_bot, tmp_path):
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch.object(base_bot, "_get_transcript_line_count", return_value=10):
base_bot.write_pending_file(chat_id=12345, message_id=100, processing_message_id=101)
data = json.loads(base_bot.pending_file.read_text())
@@ -339,14 +339,14 @@ class TestWritePendingFile:
assert isinstance(data["timestamp"], float)
def test_write_with_none_processing_id(self, base_bot, tmp_path):
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch.object(base_bot, "_get_transcript_line_count", return_value=0):
base_bot.write_pending_file(chat_id=12345, message_id=100, processing_message_id=None)
data = json.loads(base_bot.pending_file.read_text())
assert data["processing_message_id"] is None
def test_write_includes_work_dir(self, base_bot, tmp_path):
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch.object(base_bot, "_get_transcript_line_count", return_value=0):
base_bot.write_pending_file(chat_id=1, message_id=1)
data = json.loads(base_bot.pending_file.read_text())
@@ -361,7 +361,7 @@ class TestWritePendingFile:
class TestHeartbeat:
"""Test heartbeat thread updates the processing message with elapsed time."""
@patch("apps.handlers.base_bot.HEARTBEAT_INTERVAL", 0.1)
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.HEARTBEAT_INTERVAL", 0.1)
def test_heartbeat_calls_edit_message(self, base_bot, tmp_path):
"""Verify heartbeat updates the message with elapsed time text."""
# Create the pending file so heartbeat doesn't exit early
@@ -384,7 +384,7 @@ class TestHeartbeat:
assert first_call[0][1] == 101 # message_id
assert "Processing..." in first_call[0][2]
@patch("apps.handlers.base_bot.HEARTBEAT_INTERVAL", 0.1)
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.HEARTBEAT_INTERVAL", 0.1)
def test_heartbeat_stops_when_pending_removed(self, base_bot, tmp_path):
"""Heartbeat should exit when pending file is removed."""
# Create then immediately remove pending file
@@ -431,30 +431,30 @@ class TestVerifyConnection:
mock_resp.__exit__ = MagicMock(return_value=False)
return mock_resp
@patch("apps.handlers.base_bot.urlopen")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen")
def test_verify_success(self, mock_urlopen, base_bot):
mock_urlopen.return_value = self._make_mock_response({"ok": True, "result": {"username": "test_bot"}})
assert base_bot.verify_connection() is True
mock_urlopen.assert_called_once()
@patch("apps.handlers.base_bot.urlopen")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen")
def test_verify_api_rejected(self, mock_urlopen, base_bot):
mock_urlopen.return_value = self._make_mock_response({"ok": False, "description": "Unauthorized"})
assert base_bot.verify_connection() is False
@patch("apps.handlers.base_bot.urlopen")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen")
def test_verify_network_error(self, mock_urlopen, base_bot):
from urllib.error import URLError
mock_urlopen.side_effect = URLError("Connection refused")
assert base_bot.verify_connection() is False
@patch("apps.handlers.base_bot.urlopen")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen")
def test_verify_unexpected_exception(self, mock_urlopen, base_bot):
mock_urlopen.side_effect = RuntimeError("unexpected")
assert base_bot.verify_connection() is False
@patch("apps.handlers.base_bot.urlopen")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen")
def test_verify_uses_correct_url(self, mock_urlopen, base_bot):
mock_urlopen.return_value = self._make_mock_response({"ok": True, "result": {"username": "test_bot"}})
base_bot.verify_connection()
@@ -478,27 +478,27 @@ class TestSendMessage:
mock_resp.__exit__ = MagicMock(return_value=False)
return mock_resp
@patch("apps.handlers.base_bot.urlopen")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen")
def test_send_success(self, mock_urlopen, base_bot):
mock_urlopen.return_value = self._make_mock_response({"ok": True, "result": {"message_id": 42}})
result = base_bot.send_message(12345, "Hello!")
assert result == {"message_id": 42}
@patch("apps.handlers.base_bot.urlopen")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen")
def test_send_returns_none_on_failure(self, mock_urlopen, base_bot):
mock_urlopen.side_effect = RuntimeError("fail")
result = base_bot.send_message(12345, "Hello!")
assert result is None
@patch("apps.handlers.base_bot.urlopen")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen")
def test_send_retries_on_failure(self, mock_urlopen, base_bot):
"""send_message retries up to 3 times."""
mock_urlopen.side_effect = RuntimeError("fail")
with patch("apps.handlers.base_bot.time.sleep"):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.time.sleep"):
base_bot.send_message(12345, "Hello!")
assert mock_urlopen.call_count == 3
@patch("apps.handlers.base_bot.urlopen")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen")
def test_send_with_reply_to(self, mock_urlopen, base_bot):
mock_urlopen.return_value = self._make_mock_response({"ok": True, "result": {"message_id": 43}})
result = base_bot.send_message(12345, "reply", reply_to=10)
@@ -507,7 +507,7 @@ class TestSendMessage:
sent_data = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8"))
assert sent_data["reply_to_message_id"] == 10
@patch("apps.handlers.base_bot.urlopen")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen")
def test_send_uses_correct_url(self, mock_urlopen, base_bot):
mock_urlopen.return_value = self._make_mock_response({"ok": True, "result": {"message_id": 1}})
base_bot.send_message(12345, "test")
@@ -526,25 +526,25 @@ class TestEditMessage:
mock_resp.__exit__ = MagicMock(return_value=False)
return mock_resp
@patch("apps.handlers.base_bot.urlopen")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen")
def test_edit_success(self, mock_urlopen, base_bot):
mock_urlopen.return_value = self._make_mock_response({"ok": True})
result = base_bot.edit_message(12345, 42, "Updated text")
assert result is True
@patch("apps.handlers.base_bot.urlopen")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen")
def test_edit_failure(self, mock_urlopen, base_bot):
mock_urlopen.return_value = self._make_mock_response({"ok": False, "description": "Message not modified"})
result = base_bot.edit_message(12345, 42, "Same text")
assert result is False
@patch("apps.handlers.base_bot.urlopen")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen")
def test_edit_exception(self, mock_urlopen, base_bot):
mock_urlopen.side_effect = RuntimeError("network error")
result = base_bot.edit_message(12345, 42, "text")
assert result is False
@patch("apps.handlers.base_bot.urlopen")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen")
def test_edit_sends_correct_payload(self, mock_urlopen, base_bot):
mock_urlopen.return_value = self._make_mock_response({"ok": True})
base_bot.edit_message(12345, 42, "new text")
@@ -553,7 +553,7 @@ class TestEditMessage:
assert sent_data["message_id"] == 42
assert sent_data["text"] == "new text"
@patch("apps.handlers.base_bot.urlopen")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen")
def test_edit_uses_correct_url(self, mock_urlopen, base_bot):
mock_urlopen.return_value = self._make_mock_response({"ok": True})
base_bot.edit_message(12345, 42, "x")
@@ -569,8 +569,8 @@ class TestEditMessage:
class TestEnsureTmuxSession:
"""Test tmux session creation with mocked subprocess."""
@patch("apps.handlers.base_bot.time.sleep")
@patch("apps.handlers.base_bot.subprocess.run")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.time.sleep")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.subprocess.run")
def test_session_exists_returns_true(self, mock_run, mock_sleep, base_bot):
"""If tmux session already exists, return True without creating."""
# has-session returns 0 (session exists)
@@ -581,8 +581,8 @@ class TestEnsureTmuxSession:
mock_run.assert_called_once()
assert "has-session" in mock_run.call_args[0][0]
@patch("apps.handlers.base_bot.time.sleep")
@patch("apps.handlers.base_bot.subprocess.run")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.time.sleep")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.subprocess.run")
def test_no_session_returns_false(self, mock_run, mock_sleep, base_bot):
"""When no session exists, bot returns False (never spawns own brain)."""
mock_run.return_value = MagicMock(returncode=1)
@@ -596,7 +596,7 @@ class TestEnsureTmuxSession:
def test_session_refuses_nonexistent_work_dir(self, tmp_path):
"""When work_dir doesn't exist, ensure_tmux_session returns False."""
bad_dir = tmp_path / "nonexistent"
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
bot = BaseBot(
bot_id="bad_dir_bot",
bot_token="t",
@@ -607,7 +607,7 @@ class TestEnsureTmuxSession:
def test_tmux_not_found_returns_false(self, base_bot):
"""FileNotFoundError (tmux not installed) returns False."""
with patch("apps.handlers.base_bot.subprocess.run", side_effect=FileNotFoundError("tmux")):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.subprocess.run", side_effect=FileNotFoundError("tmux")):
result = base_bot.ensure_tmux_session()
assert result is False
@@ -620,15 +620,15 @@ class TestEnsureTmuxSession:
class TestInjectMessage:
"""Test tmux send-keys injection with mocked subprocess."""
@patch("apps.handlers.base_bot.time.sleep")
@patch("apps.handlers.base_bot.subprocess.run")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.time.sleep")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.subprocess.run")
def test_inject_success(self, mock_run, mock_sleep, base_bot):
mock_run.return_value = MagicMock(returncode=0)
result = base_bot.inject_message("hello world")
assert result is True
@patch("apps.handlers.base_bot.time.sleep")
@patch("apps.handlers.base_bot.subprocess.run")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.time.sleep")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.subprocess.run")
def test_inject_calls_send_keys_with_text_then_enter(self, mock_run, mock_sleep, base_bot):
mock_run.return_value = MagicMock(returncode=0)
base_bot.inject_message("test message")
@@ -647,16 +647,16 @@ class TestInjectMessage:
assert "send-keys" in second_cmd
assert "Enter" in second_cmd
@patch("apps.handlers.base_bot.time.sleep")
@patch("apps.handlers.base_bot.subprocess.run")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.time.sleep")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.subprocess.run")
def test_inject_uses_correct_session_name(self, mock_run, mock_sleep, base_bot):
mock_run.return_value = MagicMock(returncode=0)
base_bot.inject_message("x")
first_cmd = mock_run.call_args_list[0][0][0]
assert base_bot.session_name in first_cmd
@patch("apps.handlers.base_bot.time.sleep")
@patch("apps.handlers.base_bot.subprocess.run")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.time.sleep")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.subprocess.run")
def test_inject_failure(self, mock_run, mock_sleep, base_bot):
import subprocess as sp
@@ -664,8 +664,8 @@ class TestInjectMessage:
result = base_bot.inject_message("hello")
assert result is False
@patch("apps.handlers.base_bot.time.sleep")
@patch("apps.handlers.base_bot.subprocess.run")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.time.sleep")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.subprocess.run")
def test_inject_sleeps_between_commands(self, mock_run, mock_sleep, base_bot):
"""Verify there's a delay between sending text and pressing Enter."""
mock_run.return_value = MagicMock(returncode=0)
@@ -689,20 +689,20 @@ class TestPollUpdates:
mock_resp.__exit__ = MagicMock(return_value=False)
return mock_resp
@patch("apps.handlers.base_bot.urlopen")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen")
def test_poll_returns_updates(self, mock_urlopen, base_bot):
updates = [{"update_id": 1, "message": {"text": "hi"}}]
mock_urlopen.return_value = self._make_mock_response({"ok": True, "result": updates})
result = base_bot.poll_updates(0)
assert result == updates
@patch("apps.handlers.base_bot.urlopen")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen")
def test_poll_returns_empty_on_error(self, mock_urlopen, base_bot):
mock_urlopen.return_value = self._make_mock_response({"ok": False, "description": "Bad Request"})
result = base_bot.poll_updates(0)
assert result == []
@patch("apps.handlers.base_bot.urlopen")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen")
def test_poll_returns_empty_on_exception(self, mock_urlopen, base_bot):
from urllib.error import URLError
@@ -710,7 +710,7 @@ class TestPollUpdates:
result = base_bot.poll_updates(0)
assert result == []
@patch("apps.handlers.base_bot.urlopen")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen")
def test_poll_includes_offset_in_url(self, mock_urlopen, base_bot):
mock_urlopen.return_value = self._make_mock_response({"ok": True, "result": []})
base_bot.poll_updates(42)
@@ -791,7 +791,7 @@ class TestProcessUpdate:
},
}
with patch.object(base_bot, "handle_message") as mock_handle:
with patch("apps.handlers.base_bot.parse_command", return_value=None):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.parse_command", return_value=None):
base_bot.process_update(update)
mock_handle.assert_called_once()
@@ -901,7 +901,7 @@ class TestCreateCommand:
@pytest.fixture(autouse=True)
def setup_bot(self, tmp_path):
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
self.bot = BaseBot(
bot_id="test",
bot_token="123:FAKE",
@@ -912,9 +912,9 @@ class TestCreateCommand:
self.bot.send_message = MagicMock(return_value={"message_id": 42})
self.chat_id = 12345
@patch("apps.handlers.base_bot.check_telethon_setup", return_value=(False, "not configured"))
@patch("apps.handlers.base_bot.get_bot_by_branch", return_value=None)
@patch("apps.handlers.base_bot.validate_branch")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.check_telethon_setup", return_value=(False, "not configured"))
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.get_bot_by_branch", return_value=None)
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.validate_branch")
def test_create_valid_branch(self, mock_validate, mock_get_bot, mock_telethon):
mock_validate.return_value = {"name": "dev_central", "path": "/home/aipass/dev_central"}
self.bot._handle_create_command(self.chat_id, "chat dev_central")
@@ -936,15 +936,15 @@ class TestCreateCommand:
msg = self.bot.send_message.call_args[0][1]
assert "Usage" in msg
@patch("apps.handlers.base_bot.validate_branch", return_value=None)
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.validate_branch", return_value=None)
def test_create_branch_not_found(self, mock_validate):
self.bot._handle_create_command(self.chat_id, "chat nonexistent")
self.bot.send_message.assert_called_once()
msg = self.bot.send_message.call_args[0][1]
assert "not found" in msg
@patch("apps.handlers.base_bot.get_bot_by_branch")
@patch("apps.handlers.base_bot.validate_branch")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.get_bot_by_branch")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.validate_branch")
def test_create_branch_already_has_bot(self, mock_validate, mock_get_bot):
mock_validate.return_value = {"name": "dev_central", "path": "/tmp"}
mock_get_bot.return_value = {"bot_id": "dev_central", "username": "dc_bot"}
@@ -953,8 +953,8 @@ class TestCreateCommand:
msg = self.bot.send_message.call_args[0][1]
assert "already has a bot" in msg
@patch("apps.handlers.base_bot.get_bot_by_branch", return_value=None)
@patch("apps.handlers.base_bot.validate_branch")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.get_bot_by_branch", return_value=None)
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.validate_branch")
def test_create_sets_state_with_branch_info(self, mock_validate, mock_get_bot):
mock_validate.return_value = {"name": "flow", "path": "/home/aipass/flow"}
self.bot._handle_create_command(self.chat_id, "chat @flow")
@@ -971,8 +971,8 @@ class TestCreateCommand:
msg = self.bot.send_message.call_args[0][1]
assert "Usage" in msg
@patch("apps.handlers.base_bot.get_bot_by_branch", return_value=None)
@patch("apps.handlers.base_bot.validate_branch")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.get_bot_by_branch", return_value=None)
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.validate_branch")
def test_create_strips_at_symbol(self, mock_validate, mock_get_bot):
"""Branch name should have @ stripped before lookup."""
mock_validate.return_value = {"name": "seed", "path": "/home/aipass/seed"}
@@ -990,7 +990,7 @@ class TestCreateToken:
@pytest.fixture(autouse=True)
def setup_bot(self, tmp_path):
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
self.bot = BaseBot(
bot_id="test",
bot_token="123:FAKE",
@@ -1009,8 +1009,8 @@ class TestCreateToken:
"started_at": started_at or time.time(),
}
@patch("apps.handlers.base_bot.create_bot")
@patch("apps.handlers.base_bot.validate_token")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.create_bot")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.validate_token")
def test_valid_token_creates_bot(self, mock_validate_token, mock_create_bot):
self._set_create_state()
mock_validate_token.return_value = {"username": "my_new_bot"}
@@ -1036,7 +1036,7 @@ class TestCreateToken:
msg = self.bot.send_message.call_args[0][1]
assert "doesn't look like a valid" in msg
@patch("apps.handlers.base_bot.validate_token", return_value=None)
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.validate_token", return_value=None)
def test_token_validation_fails(self, mock_validate_token):
self._set_create_state()
self.bot._handle_create_token(self.chat_id, "123456789:ABCdefGHIjklMNOpqr")
@@ -1044,8 +1044,8 @@ class TestCreateToken:
msg = self.bot.send_message.call_args[0][1]
assert "validation failed" in msg.lower()
@patch("apps.handlers.base_bot.create_bot", return_value=None)
@patch("apps.handlers.base_bot.validate_token")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.create_bot", return_value=None)
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.validate_token")
def test_bot_creation_fails(self, mock_validate_token, mock_create_bot):
self._set_create_state()
mock_validate_token.return_value = {"username": "test_bot"}
@@ -1062,8 +1062,8 @@ class TestCreateToken:
assert "expired" in msg.lower()
assert self.chat_id not in self.bot._create_state
@patch("apps.handlers.base_bot.create_bot")
@patch("apps.handlers.base_bot.validate_token")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.create_bot")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.validate_token")
def test_state_cleared_after_success(self, mock_validate_token, mock_create_bot):
self._set_create_state()
mock_validate_token.return_value = {"username": "new_bot"}
@@ -1071,8 +1071,8 @@ class TestCreateToken:
self.bot._handle_create_token(self.chat_id, "123456789:ABCdefGHIjklMNOpqr")
assert self.chat_id not in self.bot._create_state
@patch("apps.handlers.base_bot.create_bot", return_value=None)
@patch("apps.handlers.base_bot.validate_token")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.create_bot", return_value=None)
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.validate_token")
def test_state_cleared_after_failure(self, mock_validate_token, mock_create_bot):
"""State should be cleared even when bot creation fails (after token validated)."""
self._set_create_state()
@@ -1092,7 +1092,7 @@ class TestCancelCommand:
@pytest.fixture(autouse=True)
def setup_bot(self, tmp_path):
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
self.bot = BaseBot(
bot_id="test",
bot_token="123:FAKE",
@@ -1119,7 +1119,7 @@ class TestCancelCommand:
"message_id": 1,
},
}
with patch("apps.handlers.base_bot.parse_command", return_value=("cancel", "")):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.parse_command", return_value=("cancel", "")):
self.bot.process_update(update)
assert self.chat_id not in self.bot._create_state
msg = self.bot.send_message.call_args[0][1]
@@ -1135,7 +1135,7 @@ class TestCancelCommand:
"message_id": 1,
},
}
with patch("apps.handlers.base_bot.parse_command", return_value=("cancel", "")):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.parse_command", return_value=("cancel", "")):
self.bot.process_update(update)
msg = self.bot.send_message.call_args[0][1]
assert "Nothing to cancel" in msg
@@ -1162,7 +1162,7 @@ class TestCancelCommand:
"message_id": 1,
},
}
with patch("apps.handlers.base_bot.parse_command", return_value=("cancel", "")):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.parse_command", return_value=("cancel", "")):
self.bot.process_update(update)
assert self.chat_id not in self.bot._create_state
assert other_chat in self.bot._create_state
@@ -1178,7 +1178,7 @@ class TestStatusWithRegistry:
@pytest.fixture(autouse=True)
def setup_bot(self, tmp_path):
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
self.bot = BaseBot(
bot_id="test",
bot_token="123:FAKE",
@@ -1189,7 +1189,7 @@ class TestStatusWithRegistry:
self.bot.send_message = MagicMock(return_value={"message_id": 42})
self.chat_id = 12345
@patch("apps.handlers.base_bot.registry_list_bots")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.registry_list_bots")
def test_status_includes_registry(self, mock_list_bots):
mock_list_bots.return_value = [
{"bot_id": "dev_central", "username": "dc_bot", "status": "running", "branch_name": "dev_central"},
@@ -1204,14 +1204,14 @@ class TestStatusWithRegistry:
"message_id": 1,
},
}
with patch("apps.handlers.base_bot.parse_command", return_value=("status", "")):
with patch("apps.handlers.base_bot.build_status_text", return_value="Bot Status"):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.parse_command", return_value=("status", "")):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.build_status_text", return_value="Bot Status"):
self.bot.process_update(update)
msg = self.bot.send_message.call_args[0][1]
assert "Registered Bots" in msg
assert "dc_bot" in msg
@patch("apps.handlers.base_bot.registry_list_bots", return_value=[])
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.registry_list_bots", return_value=[])
def test_status_empty_registry(self, mock_list_bots):
update = {
"update_id": 1,
@@ -1222,13 +1222,13 @@ class TestStatusWithRegistry:
"message_id": 1,
},
}
with patch("apps.handlers.base_bot.parse_command", return_value=("status", "")):
with patch("apps.handlers.base_bot.build_status_text", return_value="Bot Status"):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.parse_command", return_value=("status", "")):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.build_status_text", return_value="Bot Status"):
self.bot.process_update(update)
msg = self.bot.send_message.call_args[0][1]
assert "none" in msg.lower()
@patch("apps.handlers.base_bot.registry_list_bots")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.registry_list_bots")
def test_build_registry_status_with_bots(self, mock_list_bots):
mock_list_bots.return_value = [
{"bot_id": "seed", "username": "seed_bot", "status": "running", "branch_name": "seed"},
@@ -1239,13 +1239,13 @@ class TestStatusWithRegistry:
assert "seed_bot" in result
assert "running" in result
@patch("apps.handlers.base_bot.registry_list_bots")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.registry_list_bots")
def test_build_registry_status_exception(self, mock_list_bots):
mock_list_bots.side_effect = RuntimeError("DB error")
result = self.bot._build_registry_status()
assert result == ""
@patch("apps.handlers.base_bot.registry_list_bots", return_value=[])
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.registry_list_bots", return_value=[])
def test_build_registry_status_empty(self, mock_list_bots):
result = self.bot._build_registry_status()
assert result == "Registered Bots: none"
@@ -1261,7 +1261,7 @@ class TestGetCustomCommands:
@pytest.fixture(autouse=True)
def setup_bot(self, tmp_path):
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
self.bot = BaseBot(
bot_id="test",
bot_token="123:FAKE",
@@ -1291,7 +1291,7 @@ class TestCreateFlowInProcessUpdate:
@pytest.fixture(autouse=True)
def setup_bot(self, tmp_path):
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
self.bot = BaseBot(
bot_id="test",
bot_token="123:FAKE",
@@ -1313,9 +1313,9 @@ class TestCreateFlowInProcessUpdate:
},
}
@patch("apps.handlers.base_bot.get_bot_by_branch", return_value=None)
@patch("apps.handlers.base_bot.validate_branch")
@patch("apps.handlers.base_bot.parse_command", return_value=("create", "chat test_branch"))
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.get_bot_by_branch", return_value=None)
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.validate_branch")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.parse_command", return_value=("create", "chat test_branch"))
def test_process_update_create_command_routed(self, mock_parse, mock_validate, mock_get_bot):
"""Sending /create chat test_branch should call _handle_create_command."""
mock_validate.return_value = {"name": "test_branch", "path": "/tmp"}
@@ -1323,8 +1323,8 @@ class TestCreateFlowInProcessUpdate:
self.bot.process_update(self._make_update("/create chat test_branch"))
mock_method.assert_called_once_with(self.chat_id, "chat test_branch")
@patch("apps.handlers.base_bot.create_bot")
@patch("apps.handlers.base_bot.validate_token")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.create_bot")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.validate_token")
def test_process_update_token_paste_during_create_flow(self, mock_validate_token, mock_create_bot):
"""When _create_state is active, non-command text routes to _handle_create_token."""
self.bot._create_state[self.chat_id] = {
@@ -1341,7 +1341,7 @@ class TestCreateFlowInProcessUpdate:
def test_process_update_no_token_paste_without_state(self):
"""Without _create_state, non-command text routes to handle_message."""
with patch.object(self.bot, "handle_message") as mock_handle:
with patch("apps.handlers.base_bot.parse_command", return_value=None):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.parse_command", return_value=None):
self.bot.process_update(self._make_update("just regular text"))
mock_handle.assert_called_once()
@@ -1352,12 +1352,12 @@ class TestCreateFlowInProcessUpdate:
"branch_path": "/tmp",
"started_at": time.time(),
}
with patch("apps.handlers.base_bot.parse_command", return_value=("cancel", "")):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.parse_command", return_value=("cancel", "")):
self.bot.process_update(self._make_update("/cancel"))
assert self.chat_id not in self.bot._create_state
@patch("apps.handlers.base_bot.create_bot")
@patch("apps.handlers.base_bot.validate_token")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.create_bot")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.validate_token")
def test_command_during_create_flow_not_treated_as_token(self, mock_validate_token, mock_create_bot):
"""Even with _create_state active, /commands should not be routed to _handle_create_token."""
self.bot._create_state[self.chat_id] = {
@@ -1365,7 +1365,7 @@ class TestCreateFlowInProcessUpdate:
"branch_path": "/tmp",
"started_at": time.time(),
}
with patch("apps.handlers.base_bot.parse_command", return_value=("cancel", "")):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.parse_command", return_value=("cancel", "")):
with patch.object(self.bot, "_handle_create_token") as mock_token:
self.bot.process_update(self._make_update("/cancel"))
mock_token.assert_not_called()
@@ -1381,7 +1381,7 @@ class TestCreateAutomated:
@pytest.fixture(autouse=True)
def setup_bot(self, tmp_path):
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
self.bot = BaseBot(
bot_id="test",
bot_token="123:FAKE",
@@ -1392,8 +1392,8 @@ class TestCreateAutomated:
self.bot.send_message = MagicMock(return_value={"message_id": 42})
self.chat_id = 12345
@patch("apps.handlers.base_bot.create_bot")
@patch("apps.handlers.base_bot.create_bot_via_botfather")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.create_bot")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.create_bot_via_botfather")
def test_automated_flow_succeeds(self, mock_bf_create, mock_create_bot):
"""Automated flow: sends progress, calls BotFather, registers, sends success."""
mock_bf_create.return_value = {
@@ -1423,7 +1423,7 @@ class TestCreateAutomated:
allowed_user_ids=[111],
)
@patch("apps.handlers.base_bot.create_bot_via_botfather")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.create_bot_via_botfather")
def test_automated_flow_botfather_fails_falls_back_to_manual(self, mock_bf_create):
"""When BotFather automation fails, falls back to manual mode (sets _create_state)."""
mock_bf_create.return_value = None
@@ -1441,8 +1441,8 @@ class TestCreateAutomated:
assert state["branch_name"] == "flow"
assert state["branch_path"] == "/home/aipass/flow"
@patch("apps.handlers.base_bot.create_bot", return_value=None)
@patch("apps.handlers.base_bot.create_bot_via_botfather")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.create_bot", return_value=None)
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.create_bot_via_botfather")
def test_automated_flow_registration_fails(self, mock_bf_create, mock_create_bot):
"""When BotFather succeeds but bot_factory.create_bot fails, sends error."""
mock_bf_create.return_value = {
@@ -1460,9 +1460,9 @@ class TestCreateAutomated:
# Should NOT set _create_state (no manual fallback after registration failure)
assert self.chat_id not in self.bot._create_state
@patch("apps.handlers.base_bot.get_bot_by_branch", return_value=None)
@patch("apps.handlers.base_bot.validate_branch")
@patch("apps.handlers.base_bot.check_telethon_setup")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.get_bot_by_branch", return_value=None)
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.validate_branch")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.check_telethon_setup")
def test_create_command_uses_automated_when_telethon_ready(self, mock_check, mock_validate, mock_get_bot):
"""_handle_create_command uses automated path when Telethon is ready."""
mock_check.return_value = (True, "ready")
@@ -1472,9 +1472,9 @@ class TestCreateAutomated:
self.bot._handle_create_command(self.chat_id, "chat dev_central")
mock_automated.assert_called_once_with(self.chat_id, "dev_central", "/home/aipass/dev_central")
@patch("apps.handlers.base_bot.get_bot_by_branch", return_value=None)
@patch("apps.handlers.base_bot.validate_branch")
@patch("apps.handlers.base_bot.check_telethon_setup")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.get_bot_by_branch", return_value=None)
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.validate_branch")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.check_telethon_setup")
def test_create_command_uses_manual_when_telethon_not_ready(self, mock_check, mock_validate, mock_get_bot):
"""_handle_create_command uses manual path when Telethon is not ready."""
mock_check.return_value = (False, "Telethon not installed")
@@ -1489,9 +1489,9 @@ class TestCreateAutomated:
state = self.bot._create_state[self.chat_id]
assert state["branch_name"] == "flow"
@patch("apps.handlers.base_bot.get_bot_by_branch", return_value=None)
@patch("apps.handlers.base_bot.validate_branch")
@patch("apps.handlers.base_bot.check_telethon_setup")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.get_bot_by_branch", return_value=None)
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.validate_branch")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.check_telethon_setup")
def test_manual_fallback_message_shows_reason(self, mock_check, mock_validate, mock_get_bot):
"""Manual fallback message includes the reason automation is unavailable."""
mock_check.return_value = (False, "Telethon library not installed. Run: pip install telethon")
@@ -1503,8 +1503,8 @@ class TestCreateAutomated:
assert "Telethon library not installed" in msg
assert "Falling back to manual token flow" in msg
@patch("apps.handlers.base_bot.create_bot")
@patch("apps.handlers.base_bot.create_bot_via_botfather")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.create_bot")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.create_bot_via_botfather")
def test_automated_success_message_contains_service_info(self, mock_bf_create, mock_create_bot):
"""Success message includes systemd service name and start command."""
mock_bf_create.return_value = {
@@ -1520,8 +1520,8 @@ class TestCreateAutomated:
assert "telegram-bot@memory_bank" in success_msg
assert "systemctl" in success_msg
@patch("apps.handlers.base_bot.create_bot")
@patch("apps.handlers.base_bot.create_bot_via_botfather")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.create_bot")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.create_bot_via_botfather")
def test_automated_flow_passes_allowed_user_ids(self, mock_bf_create, mock_create_bot):
"""Automated flow passes the base bot's allowed_user_ids to create_bot."""
mock_bf_create.return_value = {
@@ -1551,7 +1551,7 @@ class TestSharedSession:
self.workdir = tmp_path / "workdir"
self.workdir.mkdir()
self.tmp_path = tmp_path
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
self.bot = BaseBot(
bot_id="dev_central",
bot_token="123:FAKETOKEN",
@@ -1571,8 +1571,8 @@ class TestSharedSession:
"""Default session_name is still telegram-{bot_id} until shared session found."""
assert self.bot.session_name == "telegram-dev_central"
@patch("apps.handlers.base_bot.time.sleep")
@patch("apps.handlers.base_bot.subprocess.run")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.time.sleep")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.subprocess.run")
def test_ensure_attaches_to_shared_session(self, mock_run, mock_sleep):
"""When shared session exists, bot attaches to it."""
mock_run.return_value = MagicMock(returncode=0)
@@ -1584,8 +1584,8 @@ class TestSharedSession:
mock_run.assert_called_once()
assert mock_run.call_args[0][0] == ["tmux", "has-session", "-t", "pc"]
@patch("apps.handlers.base_bot.time.sleep")
@patch("apps.handlers.base_bot.subprocess.run")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.time.sleep")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.subprocess.run")
def test_ensure_returns_false_when_shared_missing(self, mock_run, mock_sleep):
"""When shared session doesn't exist and no presence, returns False (no spawn)."""
mock_run.return_value = MagicMock(returncode=1)
@@ -1596,7 +1596,7 @@ class TestSharedSession:
for call in calls:
assert "new-session" not in call
@patch("apps.handlers.base_bot.subprocess.run")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.subprocess.run")
def test_inject_uses_shared_session_name(self, mock_run):
"""After attaching, inject_message sends to the shared session."""
self.bot.session_name = "pc"
@@ -1618,8 +1618,8 @@ class TestSharedSession:
assert self.bot._using_shared_session is False
assert self.bot.session_name == "telegram-dev_central"
@patch("apps.handlers.base_bot.time.sleep")
@patch("apps.handlers.base_bot.subprocess.run")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.time.sleep")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.subprocess.run")
def test_pending_file_has_shared_session_name(self, mock_run, mock_sleep):
"""Pending file session_name reflects the shared session."""
self.bot.session_name = "pc"
@@ -1635,7 +1635,7 @@ class TestSharedSession:
"""Bot without shared_session behaves exactly as before."""
workdir = tmp_path / "normal"
workdir.mkdir()
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
bot = BaseBot(
bot_id="flow",
bot_token="456:FAKE",
@@ -1645,8 +1645,8 @@ class TestSharedSession:
assert bot._using_shared_session is False
assert bot.session_name == "telegram-flow"
@patch("apps.handlers.base_bot.time.sleep")
@patch("apps.handlers.base_bot.subprocess.run")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.time.sleep")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.subprocess.run")
def test_reattaches_after_new_command(self, mock_run, mock_sleep):
"""After /new detaches, next ensure_tmux_session reattaches to shared session."""
# Simulate attached to shared session
@@ -1679,7 +1679,7 @@ class TestLockPidReuse:
self.workdir = tmp_path / "workdir"
self.workdir.mkdir()
self.tmp_path = tmp_path
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
self.bot = BaseBot(
bot_id="vera",
bot_token="123:FAKETOKEN",
@@ -1701,7 +1701,7 @@ class TestLockPidReuse:
assert self.bot._check_lock() is False
assert not self.bot._lock_file.exists()
@patch("apps.handlers.base_bot.os.kill")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.os.kill")
def test_alive_pid_same_bot_returns_true(self, mock_kill):
"""Live PID running this bot returns True (lock held)."""
mock_kill.return_value = None # PID alive
@@ -1716,7 +1716,7 @@ class TestLockPidReuse:
assert self.bot._check_lock() is True
assert self.bot._lock_file.exists() # Lock preserved
@patch("apps.handlers.base_bot.os.kill")
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.os.kill")
def test_alive_pid_different_bot_cleans_lock(self, mock_kill):
"""Live PID running a DIFFERENT bot cleans stale lock (PID reuse)."""
mock_kill.return_value = None # PID alive
src/aipass/skills/lib/telegram/tests/test_multibot_config.py
+43 -43
View File
@@ -12,8 +12,8 @@ from unittest.mock import patch, MagicMock
import pytest
# Modules under test
from apps.handlers import config as tg_config
from apps.handlers.telegram_standards import (
from aipass.skills.lib.telegram.apps.handlers import config as tg_config
from aipass.skills.lib.telegram.apps.handlers.telegram_standards import (
STANDARD_COMMANDS,
PROCESSING_MSG,
parse_command,
@@ -23,7 +23,7 @@ from apps.handlers.telegram_standards import (
build_status_text,
build_botfather_commands,
)
from apps.handlers import bot_operations
from aipass.skills.lib.telegram.apps.handlers import bot_operations
# =============================================
@@ -75,7 +75,7 @@ def sample_bots() -> list[dict]:
class TestLoadBotConfig:
"""Tests for config.load_bot_config (via _get_secret)."""
@patch("apps.handlers.config._get_secret")
@patch("aipass.skills.lib.telegram.apps.handlers.config._get_secret")
def test_load_valid_config(self, mock_get_secret: MagicMock, valid_bot_config: dict) -> None:
"""Valid config returned from _get_secret loads correctly."""
mock_get_secret.return_value = valid_bot_config
@@ -88,7 +88,7 @@ class TestLoadBotConfig:
assert result["branch_name"] == "dev_central"
mock_get_secret.assert_called_once_with("dev_central")
@patch("apps.handlers.config._get_secret")
@patch("aipass.skills.lib.telegram.apps.handlers.config._get_secret")
def test_load_missing_file(self, mock_get_secret: MagicMock) -> None:
"""Returns None when secret not found."""
mock_get_secret.return_value = None
@@ -97,7 +97,7 @@ class TestLoadBotConfig:
assert result is None
mock_get_secret.assert_called_once_with("nonexistent_bot")
@patch("apps.handlers.config._get_secret")
@patch("aipass.skills.lib.telegram.apps.handlers.config._get_secret")
def test_load_corrupt_json(self, mock_get_secret: MagicMock) -> None:
"""Returns None when _get_secret returns None (e.g., invalid JSON from subprocess)."""
mock_get_secret.return_value = None
@@ -105,7 +105,7 @@ class TestLoadBotConfig:
result = tg_config.load_bot_config("broken")
assert result is None
@patch("apps.handlers.config._get_secret")
@patch("aipass.skills.lib.telegram.apps.handlers.config._get_secret")
def test_load_non_dict_json(self, mock_get_secret: MagicMock) -> None:
"""Returns None when _get_secret returns None (non-dict JSON is filtered by _get_secret)."""
# _get_secret already filters non-dict responses and returns None
@@ -123,7 +123,7 @@ class TestLoadBotConfig:
class TestListBotConfigs:
"""Tests for config.list_bot_configs (via in-process secrets API)."""
@patch("apps.handlers.config._api_list_secrets")
@patch("aipass.skills.lib.telegram.apps.handlers.config._api_list_secrets")
def test_list_returns_bot_ids(self, mock_list: MagicMock) -> None:
"""Returns list of bot_ids from the secrets API."""
mock_list.return_value = ["dev_central", "assistant", "scheduler"]
@@ -136,7 +136,7 @@ class TestListBotConfigs:
assert len(result) == 3
mock_list.assert_called_once_with("telegram")
@patch("apps.handlers.config._api_list_secrets")
@patch("aipass.skills.lib.telegram.apps.handlers.config._api_list_secrets")
def test_list_returns_empty_on_failure(self, mock_list: MagicMock) -> None:
"""Returns empty list when the secrets API raises."""
mock_list.side_effect = RuntimeError("connection failed")
@@ -144,7 +144,7 @@ class TestListBotConfigs:
result = tg_config.list_bot_configs()
assert result == []
@patch("apps.handlers.config._api_list_secrets")
@patch("aipass.skills.lib.telegram.apps.handlers.config._api_list_secrets")
def test_list_returns_empty_when_no_secrets(self, mock_list: MagicMock) -> None:
"""Returns empty list when no secrets exist."""
mock_list.return_value = []
@@ -152,7 +152,7 @@ class TestListBotConfigs:
result = tg_config.list_bot_configs()
assert result == []
@patch("apps.handlers.config._api_list_secrets")
@patch("aipass.skills.lib.telegram.apps.handlers.config._api_list_secrets")
def test_list_returns_empty_on_unexpected_error(self, mock_list: MagicMock) -> None:
"""Returns empty list on unexpected exception."""
mock_list.side_effect = OSError("disk error")
@@ -352,7 +352,7 @@ class TestHandleStandardCommand:
assert "@assistant" in result[1]
assert "fresh" in result[1].lower() or "cleared" in result[1].lower()
@patch("apps.handlers.telegram_standards._tmux_session_exists")
@patch("aipass.skills.lib.telegram.apps.handlers.telegram_standards._tmux_session_exists")
def test_status_returns_status_text(self, mock_tmux: MagicMock) -> None:
"""The 'status' command returns status text string."""
mock_tmux.return_value = True
@@ -476,7 +476,7 @@ class TestBuildWelcomeText:
class TestBuildStatusText:
"""Tests for telegram_standards.build_status_text."""
@patch("apps.handlers.telegram_standards._tmux_session_exists")
@patch("aipass.skills.lib.telegram.apps.handlers.telegram_standards._tmux_session_exists")
def test_active_session(self, mock_tmux: MagicMock) -> None:
"""Active tmux session shows 'Active' state."""
mock_tmux.return_value = True
@@ -488,7 +488,7 @@ class TestBuildStatusText:
assert "@dev_central" in result
assert "telegram-dev_central" in result
@patch("apps.handlers.telegram_standards._tmux_session_exists")
@patch("aipass.skills.lib.telegram.apps.handlers.telegram_standards._tmux_session_exists")
def test_inactive_session(self, mock_tmux: MagicMock) -> None:
"""Inactive tmux session shows 'Inactive' state."""
mock_tmux.return_value = False
@@ -498,7 +498,7 @@ class TestBuildStatusText:
)
assert "Inactive" in result
@patch("apps.handlers.telegram_standards._tmux_session_exists")
@patch("aipass.skills.lib.telegram.apps.handlers.telegram_standards._tmux_session_exists")
def test_optional_fields_included(self, mock_tmux: MagicMock) -> None:
"""Optional fields (uptime, message_count, chat_id) appear when provided."""
mock_tmux.return_value = True
@@ -513,7 +513,7 @@ class TestBuildStatusText:
assert "Messages: 99" in result
assert "Chat ID: 12345" in result
@patch("apps.handlers.telegram_standards._tmux_session_exists")
@patch("aipass.skills.lib.telegram.apps.handlers.telegram_standards._tmux_session_exists")
def test_optional_fields_omitted(self, mock_tmux: MagicMock) -> None:
"""Optional fields are not shown when not provided."""
mock_tmux.return_value = True
@@ -753,7 +753,7 @@ class TestFormatBotTable:
class TestGetStatusAndGetAllBots:
"""Tests for bot_operations.get_status and get_all_bots."""
@patch("apps.handlers.bot_operations.get_bot")
@patch("aipass.skills.lib.telegram.apps.handlers.bot_operations.get_bot")
def test_get_status_specific_bot(self, mock_get_bot: MagicMock) -> None:
"""get_status with bot_id delegates to get_bot."""
mock_get_bot.return_value = {"bot_id": "dev_central", "status": "active"}
@@ -763,7 +763,7 @@ class TestGetStatusAndGetAllBots:
assert result[0]["bot_id"] == "dev_central"
mock_get_bot.assert_called_once_with("dev_central")
@patch("apps.handlers.bot_operations.get_bot")
@patch("aipass.skills.lib.telegram.apps.handlers.bot_operations.get_bot")
def test_get_status_bot_not_found(self, mock_get_bot: MagicMock) -> None:
"""get_status returns empty list when bot not found."""
mock_get_bot.return_value = None
@@ -771,7 +771,7 @@ class TestGetStatusAndGetAllBots:
result = bot_operations.get_status("nonexistent")
assert result == []
@patch("apps.handlers.bot_operations.list_bots")
@patch("aipass.skills.lib.telegram.apps.handlers.bot_operations.list_bots")
def test_get_status_all_bots(self, mock_list_bots: MagicMock) -> None:
"""get_status with no bot_id delegates to list_bots."""
mock_list_bots.return_value = [
@@ -783,7 +783,7 @@ class TestGetStatusAndGetAllBots:
assert len(result) == 2
mock_list_bots.assert_called_once()
@patch("apps.handlers.bot_operations.list_bots")
@patch("aipass.skills.lib.telegram.apps.handlers.bot_operations.list_bots")
def test_get_all_bots(self, mock_list_bots: MagicMock) -> None:
"""get_all_bots delegates to list_bots."""
expected = [{"bot_id": "x"}, {"bot_id": "y"}]
@@ -802,11 +802,11 @@ class TestGetStatusAndGetAllBots:
class TestCreateBotRoundTrip:
"""Prove GAP1 is closed: create_bot persists config that load_bot_config reads."""
@patch("apps.handlers.bot_factory.start_bot_process", return_value=True)
@patch("apps.handlers.bot_factory.enable_service", return_value=True)
@patch("apps.handlers.bot_factory.set_bot_commands", return_value=True)
@patch("apps.handlers.bot_factory.validate_token")
@patch("apps.handlers.bot_factory.ensure_registry")
@patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.start_bot_process", return_value=True)
@patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.enable_service", return_value=True)
@patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.set_bot_commands", return_value=True)
@patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.validate_token")
@patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.ensure_registry")
def test_create_then_load_roundtrip(
self,
mock_ensure_registry,
@@ -818,7 +818,7 @@ class TestCreateBotRoundTrip:
monkeypatch,
):
"""After create_bot, load_bot_config returns the persisted config."""
from apps.handlers import bot_factory, config as tg_config
from aipass.skills.lib.telegram.apps.handlers import bot_factory, config as tg_config
mock_validate_token.return_value = {"username": "test_bot", "id": 123}
@@ -836,9 +836,9 @@ class TestCreateBotRoundTrip:
monkeypatch.setattr(bot_factory, "_api_set_secret", fake_set_secret)
monkeypatch.setattr(tg_config, "_api_get_secret", fake_get_secret)
monkeypatch.setattr("apps.handlers.bot_registry.REGISTRY_DIR", tmp_path / "state")
monkeypatch.setattr("aipass.skills.lib.telegram.apps.handlers.bot_registry.REGISTRY_DIR", tmp_path / "state")
monkeypatch.setattr(
"apps.handlers.bot_registry.REGISTRY_FILE",
"aipass.skills.lib.telegram.apps.handlers.bot_registry.REGISTRY_FILE",
tmp_path / "state" / "_registry.json",
)
@@ -857,8 +857,8 @@ class TestCreateBotRoundTrip:
assert loaded["bot_token"] == "111:AAA-test-token"
assert loaded["allowed_user_ids"] == [42]
@patch("apps.handlers.bot_factory.validate_token")
@patch("apps.handlers.bot_factory.ensure_registry")
@patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.validate_token")
@patch("aipass.skills.lib.telegram.apps.handlers.bot_factory.ensure_registry")
def test_create_fails_loud_on_set_secret_error(
self,
mock_ensure_registry,
@@ -867,7 +867,7 @@ class TestCreateBotRoundTrip:
monkeypatch,
):
"""create_bot returns None and logs error if set_secret raises."""
from apps.handlers import bot_factory
from aipass.skills.lib.telegram.apps.handlers import bot_factory
mock_validate_token.return_value = {"username": "test_bot", "id": 123}
monkeypatch.setattr(bot_factory, "_BOT_CONFIG_DIR", tmp_path)
@@ -894,11 +894,11 @@ class TestCommandMenuSync:
def test_menu_and_help_have_same_commands(self):
"""The command names in build_botfather_commands match those in build_help_text."""
from apps.handlers.telegram_standards import (
from aipass.skills.lib.telegram.apps.handlers.telegram_standards import (
build_botfather_commands,
build_help_text,
)
from apps.handlers.base_bot import BaseBot
from aipass.skills.lib.telegram.apps.handlers.base_bot import BaseBot
from unittest.mock import MagicMock
mock_self = MagicMock(spec=BaseBot)
@@ -918,8 +918,8 @@ class TestCommandMenuSync:
def test_help_contains_enriched_descriptions(self):
"""The /help text includes the enriched descriptions."""
from apps.handlers.telegram_standards import build_help_text
from apps.handlers.base_bot import BaseBot
from aipass.skills.lib.telegram.apps.handlers.telegram_standards import build_help_text
from aipass.skills.lib.telegram.apps.handlers.base_bot import BaseBot
from unittest.mock import MagicMock
mock_self = MagicMock(spec=BaseBot)
@@ -936,15 +936,15 @@ class TestCommandMenuSync:
def test_help_footer_updated(self):
"""The /help footer uses the enriched text."""
from apps.handlers.telegram_standards import build_help_text
from aipass.skills.lib.telegram.apps.handlers.telegram_standards import build_help_text
help_text = build_help_text()
assert "Just send any message to talk to me" in help_text
def test_create_bot_uses_single_source(self):
"""create_bot calls set_bot_commands with build_botfather_commands output."""
from apps.handlers import bot_factory
from apps.handlers.telegram_standards import build_botfather_commands
from aipass.skills.lib.telegram.apps.handlers import bot_factory
from aipass.skills.lib.telegram.apps.handlers.telegram_standards import build_botfather_commands
expected = build_botfather_commands()
with patch.object(bot_factory, "set_bot_commands") as mock_set:
@@ -963,11 +963,11 @@ class TestCommandMenuSync:
class TestBaseBotStartupMenu:
"""Verify base_bot sets command menu on startup."""
@patch("apps.handlers.base_bot.set_bot_commands", return_value=True)
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.set_bot_commands", return_value=True)
def test_set_command_menu_called_on_startup(self, mock_set_commands):
"""_set_command_menu calls set_bot_commands with merged commands."""
from apps.handlers.base_bot import BaseBot
from apps.handlers.telegram_standards import build_botfather_commands
from aipass.skills.lib.telegram.apps.handlers.base_bot import BaseBot
from aipass.skills.lib.telegram.apps.handlers.telegram_standards import build_botfather_commands
bot = BaseBot.__new__(BaseBot)
bot.bot_token = "123:ABC"
@@ -981,10 +981,10 @@ class TestBaseBotStartupMenu:
expected = build_botfather_commands(custom_commands=bot.get_custom_commands())
assert actual_commands == expected
@patch("apps.handlers.base_bot.set_bot_commands", return_value=True)
@patch("aipass.skills.lib.telegram.apps.handlers.base_bot.set_bot_commands", return_value=True)
def test_menu_includes_custom_commands(self, mock_set_commands):
"""Menu includes /create and /cancel from get_custom_commands."""
from apps.handlers.base_bot import BaseBot
from aipass.skills.lib.telegram.apps.handlers.base_bot import BaseBot
bot = BaseBot.__new__(BaseBot)
bot.bot_token = "123:ABC"
src/aipass/skills/lib/telegram/tests/test_multibot_integration.py
+7 -7
View File
@@ -13,7 +13,7 @@ import time
import pytest
from unittest.mock import patch, MagicMock
from apps.handlers import tmux_manager as tg_tmux
from aipass.skills.lib.telegram.apps.handlers import tmux_manager as tg_tmux
try:
from aipass.hooks.apps.handlers.notification import telegram_response as tg_hook
@@ -67,7 +67,7 @@ def _make_subprocess_result(returncode=0, stdout="", stderr=""):
class TestTmuxSessionExists:
"""Tests for tmux_manager.session_exists."""
@patch("apps.handlers.tmux_manager.subprocess.run")
@patch("aipass.skills.lib.telegram.apps.handlers.tmux_manager.subprocess.run")
def test_session_exists_returns_true(self, mock_run):
"""session_exists returns True when tmux has-session succeeds."""
mock_run.return_value = _make_subprocess_result(returncode=0)
@@ -78,7 +78,7 @@ class TestTmuxSessionExists:
capture_output=True,
)
@patch("apps.handlers.tmux_manager.subprocess.run")
@patch("aipass.skills.lib.telegram.apps.handlers.tmux_manager.subprocess.run")
def test_session_exists_returns_false(self, mock_run):
"""session_exists returns False when tmux has-session fails."""
mock_run.return_value = _make_subprocess_result(returncode=1)
@@ -89,7 +89,7 @@ class TestTmuxSessionExists:
class TestTmuxKillSession:
"""Tests for tmux_manager.kill_session."""
@patch("apps.handlers.tmux_manager.subprocess.run")
@patch("aipass.skills.lib.telegram.apps.handlers.tmux_manager.subprocess.run")
def test_kill_session_returns_true(self, mock_run):
"""kill_session returns True when session exists and is killed."""
mock_run.side_effect = [
@@ -101,7 +101,7 @@ class TestTmuxKillSession:
kill_call = mock_run.call_args_list[1]
assert kill_call[0][0] == ["tmux", "kill-session", "-t", "telegram-dev_central"]
@patch("apps.handlers.tmux_manager.subprocess.run")
@patch("aipass.skills.lib.telegram.apps.handlers.tmux_manager.subprocess.run")
def test_kill_session_returns_true_when_not_exists(self, mock_run):
"""kill_session returns True when session doesn't exist (nothing to kill)."""
mock_run.return_value = _make_subprocess_result(returncode=1) # has-session: not found
@@ -112,7 +112,7 @@ class TestTmuxKillSession:
class TestTmuxListSessions:
"""Tests for tmux_manager.list_sessions."""
@patch("apps.handlers.tmux_manager.subprocess.run")
@patch("aipass.skills.lib.telegram.apps.handlers.tmux_manager.subprocess.run")
def test_list_sessions_filters_telegram_prefix(self, mock_run):
"""list_sessions returns only branch names from telegram-* sessions."""
mock_run.return_value = _make_subprocess_result(
@@ -123,7 +123,7 @@ class TestTmuxListSessions:
assert result == ["dev_central", "flow"]
@patch("apps.handlers.tmux_manager.subprocess.run")
@patch("aipass.skills.lib.telegram.apps.handlers.tmux_manager.subprocess.run")
def test_list_sessions_returns_empty_on_failure(self, mock_run):
"""list_sessions returns [] when tmux command fails."""
mock_run.return_value = _make_subprocess_result(returncode=1)
src/aipass/skills/lib/telegram/tests/test_poll_and_gate.py
+4 -4
View File
@@ -15,15 +15,15 @@ Tests cover:
import pytest
from unittest.mock import patch, MagicMock
from apps.handlers.base_bot import BaseBot # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.base_bot import BaseBot
@pytest.fixture
def _patch_base_bot_deps(tmp_path):
patches = [
patch("apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("apps.handlers.base_bot.signal.signal"),
patch("apps.handlers.base_bot.atexit.register"),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.signal.signal"),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.atexit.register"),
]
for p in patches:
p.start()
src/aipass/skills/lib/telegram/tests/test_presence_pointer.py
+16 -16
View File
@@ -26,7 +26,7 @@ from unittest.mock import MagicMock, patch
import pytest
from apps.handlers.base_bot import BaseBot # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.base_bot import BaseBot
# =============================================
@@ -38,9 +38,9 @@ from apps.handlers.base_bot import BaseBot # type: ignore[import-not-found]
def _patch_base_bot_deps(tmp_path):
"""Patch signal and atexit for safe BaseBot construction."""
patches = [
patch("apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("apps.handlers.base_bot.signal.signal"),
patch("apps.handlers.base_bot.atexit.register"),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.signal.signal"),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.atexit.register"),
]
for p in patches:
p.start()
@@ -53,7 +53,7 @@ def _make_bot(tmp_path, _patch_base_bot_deps, branch_name="devpulse"):
"""Create a BaseBot with test defaults."""
workdir = tmp_path / "workdir"
workdir.mkdir(exist_ok=True)
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
bot = BaseBot(
bot_id="cc_test",
bot_token="123:FAKETOKEN",
@@ -106,7 +106,7 @@ class TestDiscoverCcSession:
bot = _make_bot(tmp_path, _patch_base_bot_deps)
sessions_dir = tmp_path / "sessions"
_write_cc_session(sessions_dir, os.getpid(), str(bot.work_dir.resolve()))
with patch("apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir):
result = bot._discover_cc_session()
assert result is not None
assert result["pid"] == os.getpid()
@@ -116,7 +116,7 @@ class TestDiscoverCcSession:
"""Returns None when the sessions directory does not exist."""
bot = _make_bot(tmp_path, _patch_base_bot_deps)
fake_dir = tmp_path / "no_such_dir"
with patch("apps.handlers.base_bot.CC_SESSIONS_DIR", fake_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.CC_SESSIONS_DIR", fake_dir):
result = bot._discover_cc_session()
assert result is None
@@ -125,7 +125,7 @@ class TestDiscoverCcSession:
bot = _make_bot(tmp_path, _patch_base_bot_deps)
sessions_dir = tmp_path / "sessions"
_write_cc_session(sessions_dir, os.getpid(), "/some/other/dir")
with patch("apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir):
result = bot._discover_cc_session()
assert result is None
@@ -134,7 +134,7 @@ class TestDiscoverCcSession:
bot = _make_bot(tmp_path, _patch_base_bot_deps)
sessions_dir = tmp_path / "sessions"
_write_cc_session(sessions_dir, 99999999, str(bot.work_dir.resolve()))
with patch("apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir):
result = bot._discover_cc_session()
assert result is None
@@ -147,7 +147,7 @@ class TestDiscoverCcSession:
_write_cc_session(sessions_dir, my_pid, cwd, session_id="old", started_at=100)
_write_cc_session(sessions_dir, my_pid + 1, cwd, session_id="new", started_at=200)
with (
patch("apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir),
patch.object(BaseBot, "_is_pid_alive", return_value=True),
):
result = bot._discover_cc_session()
@@ -161,7 +161,7 @@ class TestDiscoverCcSession:
sessions_dir.mkdir(parents=True)
(sessions_dir / "readme.txt").write_text("ignore me")
(sessions_dir / "notapid.json").write_text("{}")
with patch("apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir):
result = bot._discover_cc_session()
assert result is None
@@ -171,7 +171,7 @@ class TestDiscoverCcSession:
sessions_dir = tmp_path / "sessions"
sessions_dir.mkdir(parents=True)
(sessions_dir / "12345.json").write_text("not json!")
with patch("apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir):
result = bot._discover_cc_session()
assert result is None
@@ -181,7 +181,7 @@ class TestDiscoverCcSession:
real_dir.mkdir()
link = tmp_path / "link_workdir"
link.symlink_to(real_dir)
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
bot = BaseBot(
bot_id="sym_test",
bot_token="t",
@@ -190,7 +190,7 @@ class TestDiscoverCcSession:
)
sessions_dir = tmp_path / "sessions"
_write_cc_session(sessions_dir, os.getpid(), str(real_dir))
with patch("apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir):
result = bot._discover_cc_session()
assert result is not None
@@ -438,7 +438,7 @@ class TestEnsureWithCcDiscovery:
"""CC session found but no tmux pane → falls through to Strategy 2."""
workdir = tmp_path / "workdir"
workdir.mkdir(exist_ok=True)
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
bot = BaseBot(
bot_id="fb_test",
bot_token="t",
@@ -461,7 +461,7 @@ class TestEnsureWithCcDiscovery:
"""When CC discovery returns None, falls back to shared_session config."""
workdir = tmp_path / "workdir"
workdir.mkdir(exist_ok=True)
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
bot = BaseBot(
bot_id="fb_test",
bot_token="t",
src/aipass/skills/lib/telegram/tests/test_response_router.py
+1 -1
View File
@@ -25,7 +25,7 @@ from pathlib import Path
import pytest
from unittest.mock import MagicMock
import apps.handlers.response_router as response_router # type: ignore[import-not-found]
import aipass.skills.lib.telegram.apps.handlers.response_router as response_router
# =============================================
src/aipass/skills/lib/telegram/tests/test_scheduler_bot.py
+10 -10
View File
@@ -15,7 +15,7 @@ import json
import pytest
from unittest.mock import patch, MagicMock
from apps.handlers.scheduler_bot import SchedulerBot, QUEUE_CMD # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.scheduler_bot import SchedulerBot, QUEUE_CMD
# =============================================
@@ -63,9 +63,9 @@ EMPTY_QUEUE = {"generated_at": "2026-06-25T15:00:00Z", "count": 0, "jobs": []}
def _patch_base_bot_deps(tmp_path):
"""Patch heavy BaseBot dependencies to allow lightweight instantiation."""
patches = [
patch("apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("apps.handlers.base_bot.signal.signal"),
patch("apps.handlers.base_bot.atexit.register"),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.signal.signal"),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.atexit.register"),
]
for p in patches:
p.start()
@@ -148,7 +148,7 @@ class TestQueueCommand:
mock_result.returncode = 0
mock_result.stdout = json.dumps(SAMPLE_QUEUE)
with patch("apps.handlers.scheduler_bot.subprocess.run", return_value=mock_result):
with patch("aipass.skills.lib.telegram.apps.handlers.scheduler_bot.subprocess.run", return_value=mock_result):
data = bot._fetch_queue()
assert data is not None
@@ -161,7 +161,7 @@ class TestQueueCommand:
mock_result.returncode = 1
mock_result.stderr = "error"
with patch("apps.handlers.scheduler_bot.subprocess.run", return_value=mock_result):
with patch("aipass.skills.lib.telegram.apps.handlers.scheduler_bot.subprocess.run", return_value=mock_result):
data = bot._fetch_queue()
assert data is None
@@ -170,7 +170,7 @@ class TestQueueCommand:
bot = _make_scheduler_bot(tmp_path, _patch_base_bot_deps)
import subprocess as sp
with patch("apps.handlers.scheduler_bot.subprocess.run", side_effect=sp.TimeoutExpired(QUEUE_CMD, 15)):
with patch("aipass.skills.lib.telegram.apps.handlers.scheduler_bot.subprocess.run", side_effect=sp.TimeoutExpired(QUEUE_CMD, 15)):
data = bot._fetch_queue()
assert data is None
@@ -199,7 +199,7 @@ class TestNoTmux:
bot = _make_scheduler_bot(tmp_path, _patch_base_bot_deps)
with (
patch.object(bot, "send_message") as mock_send,
patch("apps.handlers.base_bot.BaseBot.handle_file") as mock_parent_file,
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.BaseBot.handle_file") as mock_parent_file,
):
bot.handle_file(42, {"message_id": 1, "document": {"file_id": "abc"}})
mock_parent_file.assert_not_called()
@@ -325,9 +325,9 @@ class TestSecretLoading:
def test_missing_secret_returns_none(self):
"""When get_secret returns None, load_bot_config returns None — bot won't start."""
from apps.handlers.config import load_bot_config # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.config import load_bot_config
with patch("apps.handlers.config._get_secret", return_value=None):
with patch("aipass.skills.lib.telegram.apps.handlers.config._get_secret", return_value=None):
config = load_bot_config("scheduler")
assert config is None
src/aipass/skills/lib/telegram/tests/test_status_reset.py
+12 -12
View File
@@ -22,16 +22,16 @@ import time
import pytest
from unittest.mock import patch
from apps.handlers.base_bot import BaseBot # type: ignore[import-not-found]
from apps.handlers.telegram_standards import build_status_text # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.base_bot import BaseBot
from aipass.skills.lib.telegram.apps.handlers.telegram_standards import build_status_text
@pytest.fixture
def _patch_base_bot_deps(tmp_path):
patches = [
patch("apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("apps.handlers.base_bot.signal.signal"),
patch("apps.handlers.base_bot.atexit.register"),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.signal.signal"),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.atexit.register"),
]
for p in patches:
p.start()
@@ -116,8 +116,8 @@ class TestStatusUptimes:
with (
patch.object(bot, "send_message"),
patch("apps.handlers.base_bot.build_status_text", wraps=build_status_text) as mock_build,
patch("apps.handlers.telegram_standards._tmux_session_exists", return_value=True),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.build_status_text", wraps=build_status_text) as mock_build,
patch("aipass.skills.lib.telegram.apps.handlers.telegram_standards._tmux_session_exists", return_value=True),
):
bot._dispatch_command(42, ("status", ""))
@@ -143,7 +143,7 @@ class TestStatusUptimes:
# Now check /status — conversation uptime should be near 0
with (
patch.object(bot, "send_message") as mock_send,
patch("apps.handlers.telegram_standards._tmux_session_exists", return_value=True),
patch("aipass.skills.lib.telegram.apps.handlers.telegram_standards._tmux_session_exists", return_value=True),
):
bot._dispatch_command(42, ("status", ""))
@@ -163,7 +163,7 @@ class TestStatusUptimes:
with (
patch.object(bot, "send_message") as mock_send,
patch("apps.handlers.telegram_standards._tmux_session_exists", return_value=True),
patch("aipass.skills.lib.telegram.apps.handlers.telegram_standards._tmux_session_exists", return_value=True),
):
bot._dispatch_command(42, ("status", ""))
@@ -180,7 +180,7 @@ class TestBuildStatusText:
"""build_status_text renders daemon_uptime when provided."""
def test_includes_daemon_uptime(self):
with patch("apps.handlers.telegram_standards._tmux_session_exists", return_value=True):
with patch("aipass.skills.lib.telegram.apps.handlers.telegram_standards._tmux_session_exists", return_value=True):
text = build_status_text(
session_name="telegram-base",
branch_name="base",
@@ -192,7 +192,7 @@ class TestBuildStatusText:
assert "Uptime: 0h 5m 0s" in text
def test_omits_daemon_uptime_when_none(self):
with patch("apps.handlers.telegram_standards._tmux_session_exists", return_value=True):
with patch("aipass.skills.lib.telegram.apps.handlers.telegram_standards._tmux_session_exists", return_value=True):
text = build_status_text(
session_name="telegram-base",
branch_name="base",
@@ -203,7 +203,7 @@ class TestBuildStatusText:
assert "Uptime: 1h 0m 0s" in text
def test_uptime_before_daemon_uptime(self):
with patch("apps.handlers.telegram_standards._tmux_session_exists", return_value=True):
with patch("aipass.skills.lib.telegram.apps.handlers.telegram_standards._tmux_session_exists", return_value=True):
text = build_status_text(
session_name="telegram-base",
branch_name="base",
src/aipass/skills/lib/telegram/tests/test_streaming.py
+11 -11
View File
@@ -27,7 +27,7 @@ from unittest.mock import MagicMock, patch
import pytest
from apps.handlers.base_bot import BaseBot # type: ignore[import-not-found]
from aipass.skills.lib.telegram.apps.handlers.base_bot import BaseBot
# =============================================
@@ -39,9 +39,9 @@ from apps.handlers.base_bot import BaseBot # type: ignore[import-not-found]
def _patch_deps(tmp_path):
"""Patch signal and atexit for safe BaseBot construction."""
patches = [
patch("apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("apps.handlers.base_bot.signal.signal"),
patch("apps.handlers.base_bot.atexit.register"),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.signal.signal"),
patch("aipass.skills.lib.telegram.apps.handlers.base_bot.atexit.register"),
]
for p in patches:
p.start()
@@ -54,7 +54,7 @@ def _make_bot(tmp_path, _patch_deps, stream=False):
"""Create a BaseBot with test defaults."""
workdir = tmp_path / "workdir"
workdir.mkdir(exist_ok=True)
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
bot = BaseBot(
bot_id="stream_test",
bot_token="123:FAKETOKEN",
@@ -274,7 +274,7 @@ class TestStreamEdit:
mock_resp.__enter__ = MagicMock(return_value=mock_resp)
mock_resp.__exit__ = MagicMock(return_value=False)
with patch("apps.handlers.base_bot.urlopen", return_value=mock_resp):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen", return_value=mock_resp):
ok, retry = bot._stream_edit(123, 456, "hello")
assert ok is True
assert retry == 0.0
@@ -295,7 +295,7 @@ class TestStreamEdit:
err = HTTPError("url", 429, "Too Many Requests", None, None) # type: ignore[arg-type]
err.read = MagicMock(return_value=body)
with patch("apps.handlers.base_bot.urlopen", side_effect=err):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen", side_effect=err):
ok, retry = bot._stream_edit(123, 456, "hello")
assert ok is False
assert retry == 15.0
@@ -315,14 +315,14 @@ class TestStreamEdit:
err = HTTPError("url", 400, "Bad Request", None, None) # type: ignore[arg-type]
err.read = MagicMock(return_value=body)
with patch("apps.handlers.base_bot.urlopen", side_effect=err):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen", side_effect=err):
ok, retry = bot._stream_edit(123, 456, "same text")
assert ok is True
assert retry == 0.0
def test_other_error(self, tmp_path, _patch_deps):
bot = _make_bot(tmp_path, _patch_deps)
with patch("apps.handlers.base_bot.urlopen", side_effect=ConnectionError("fail")):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.urlopen", side_effect=ConnectionError("fail")):
ok, retry = bot._stream_edit(123, 456, "hello")
assert ok is False
assert retry == 0.0
@@ -338,7 +338,7 @@ class TestPendingFileStreaming:
def test_no_streaming_key_when_off(self, tmp_path, _patch_deps):
bot = _make_bot(tmp_path, _patch_deps, stream=False)
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
bot.pending_file = tmp_path / "bot-stream_test.json"
bot.write_pending_file(123, 1, 2)
data = json.loads(bot.pending_file.read_text(encoding="utf-8"))
@@ -346,7 +346,7 @@ class TestPendingFileStreaming:
def test_streaming_true_when_on(self, tmp_path, _patch_deps):
bot = _make_bot(tmp_path, _patch_deps, stream=True)
with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
with patch("aipass.skills.lib.telegram.apps.handlers.base_bot.PENDING_DIR", tmp_path):
bot.pending_file = tmp_path / "bot-stream_test.json"
bot.write_pending_file(123, 1, 2)
data = json.loads(bot.pending_file.read_text(encoding="utf-8"))