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

#691 ruff format the 6 telegram test files @skills left unformatted (lint job runs ruff format --check). Whitespace/line-wrap only, 0 semantic change; ruff check + format --check now clean, 289 tests on the 6 files green. Fixes the lint red on deffa0c.

Browse Source
This commit is contained in:
AIOSAI
2026-07-11 09:11:14 -07:00
parent deffa0c912
commit 8a606c8c2b
6 changed files with 66 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aipass/skills/lib/telegram/tests/test_botfather_client.py
+3 -1
View File
@@ -318,7 +318,9 @@ class TestBotFatherClientSendAndWait:
# Shrink the timeout to 0 so the while-loop exits immediately
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):
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
src/aipass/skills/lib/telegram/tests/test_log_streamer.py
+14 -4
View File
@@ -248,7 +248,9 @@ class TestSendMessage:
"""Payload should include chat_id, text, and disable_notification."""
mock_resp = self._make_mock_response({"ok": True})
with patch("aipass.skills.lib.telegram.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
@@ -311,7 +313,9 @@ class TestSendMessage:
"""Request should have Content-Type: application/json header."""
mock_resp = self._make_mock_response({"ok": True})
with patch("aipass.skills.lib.telegram.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]
@@ -504,7 +508,10 @@ class TestBaseBotIntegration:
}
# Patch LogStreamer at the import location in base_bot
with patch.object(bot, "send_message"), patch("aipass.skills.lib.telegram.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
@@ -537,7 +544,10 @@ class TestBaseBotIntegration:
}
}
with patch.object(bot, "send_message"), patch("aipass.skills.lib.telegram.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()
src/aipass/skills/lib/telegram/tests/test_mirror_session.py
+9 -3
View File
@@ -127,7 +127,9 @@ class TestLaunchMirrorSession:
"""Returns True without creating if session already exists."""
has_session = MagicMock(returncode=0)
with patch("aipass.skills.lib.telegram.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 +137,9 @@ class TestLaunchMirrorSession:
def test_returns_false_when_tmux_not_found(self):
"""Returns False when tmux is not installed."""
with patch("aipass.skills.lib.telegram.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 +156,9 @@ class TestStartService:
def test_starts_systemd_service(self):
"""Calls systemctl --user start telegram-bot@{bot_id}."""
mock_result = MagicMock(returncode=0)
with patch("aipass.skills.lib.telegram.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
src/aipass/skills/lib/telegram/tests/test_multi_bot.py
+15 -5
View File
@@ -607,7 +607,9 @@ class TestEnsureTmuxSession:
def test_tmux_not_found_returns_false(self, base_bot):
"""FileNotFoundError (tmux not installed) returns False."""
with patch("aipass.skills.lib.telegram.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
@@ -912,7 +914,9 @@ class TestCreateCommand:
self.bot.send_message = MagicMock(return_value={"message_id": 42})
self.chat_id = 12345
@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.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):
@@ -1205,7 +1209,9 @@ class TestStatusWithRegistry:
},
}
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"):
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
@@ -1223,7 +1229,9 @@ class TestStatusWithRegistry:
},
}
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"):
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()
@@ -1315,7 +1323,9 @@ class TestCreateFlowInProcessUpdate:
@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"))
@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"}
src/aipass/skills/lib/telegram/tests/test_scheduler_bot.py
+4 -1
View File
@@ -170,7 +170,10 @@ class TestQueueCommand:
bot = _make_scheduler_bot(tmp_path, _patch_base_bot_deps)
import subprocess as sp
with patch("aipass.skills.lib.telegram.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
src/aipass/skills/lib/telegram/tests/test_status_reset.py
+21 -7
View File
@@ -116,8 +116,12 @@ class TestStatusUptimes:
with (
patch.object(bot, "send_message"),
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),
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 +147,9 @@ class TestStatusUptimes:
# Now check /status — conversation uptime should be near 0
with (
patch.object(bot, "send_message") as mock_send,
patch("aipass.skills.lib.telegram.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 +169,9 @@ class TestStatusUptimes:
with (
patch.object(bot, "send_message") as mock_send,
patch("aipass.skills.lib.telegram.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 +188,9 @@ class TestBuildStatusText:
"""build_status_text renders daemon_uptime when provided."""
def test_includes_daemon_uptime(self):
with patch("aipass.skills.lib.telegram.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 +202,9 @@ class TestBuildStatusText:
assert "Uptime: 0h 5m 0s" in text
def test_omits_daemon_uptime_when_none(self):
with patch("aipass.skills.lib.telegram.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 +215,9 @@ class TestBuildStatusText:
assert "Uptime: 1h 0m 0s" in text
def test_uptime_before_daemon_uptime(self):
with patch("aipass.skills.lib.telegram.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",