forked from Rockachopa/Timmy-time-dashboard
9d78eb31d1
* polish: streamline nav, extract inline styles, improve tablet UX - Restructure desktop nav from 8+ flat links + overflow dropdown into 5 grouped dropdowns (Core, Agents, Intel, System, More) matching the mobile menu structure to reduce decision fatigue - Extract all inline styles from mission_control.html and base.html notification elements into mission-control.css with semantic classes - Replace JS-built innerHTML with secure DOM construction in notification loader and chat history - Add CONNECTING state to connection indicator (amber) instead of showing OFFLINE before WebSocket connects - Add tablet breakpoint (1024px) with larger touch targets for Apple Pencil / stylus use and safe-area padding for iPad toolbar - Add active-link highlighting in desktop dropdown menus - Rename "Mission Control" page title to "System Overview" to disambiguate from the chat home page - Add "Home — Timmy Time" page title to index.html https://claude.ai/code/session_015uPUoKyYa8M2UAcyk5Gt6h * fix(security): move auth-gate credentials to environment variables Hardcoded username, password, and HMAC secret in auth-gate.py replaced with os.environ lookups. Startup now refuses to run if any variable is unset. Added AUTH_GATE_SECRET/USER/PASS to .env.example. https://claude.ai/code/session_015uPUoKyYa8M2UAcyk5Gt6h * refactor(tooling): migrate from black+isort+bandit to ruff Replace three separate linting/formatting tools with a single ruff invocation. Updates tox.ini (lint, format, pre-push, pre-commit envs), .pre-commit-config.yaml, and CI workflow. Fixes all ruff errors including unused imports, missing raise-from, and undefined names. Ruff config maps existing bandit skips to equivalent S-rules. https://claude.ai/code/session_015uPUoKyYa8M2UAcyk5Gt6h --------- Co-authored-by: Claude <noreply@anthropic.com>
275 lines
9.4 KiB
Python
275 lines
9.4 KiB
Python
"""Tests for Discord action confirmation system using native Agno RunOutput.
|
|
|
|
Covers tool safety classification, formatting, impact levels,
|
|
and the confirmation flow in _handle_message.
|
|
"""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _format_action_description (imported from tool_safety)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFormatActionDescription:
|
|
def test_shell_command_string(self):
|
|
from integrations.chat_bridge.vendors.discord import _format_action_description
|
|
|
|
desc = _format_action_description("shell", {"command": "ls -la /tmp"})
|
|
assert "ls -la /tmp" in desc
|
|
|
|
def test_shell_command_list(self):
|
|
from integrations.chat_bridge.vendors.discord import _format_action_description
|
|
|
|
desc = _format_action_description("shell", {"args": ["mkdir", "-p", "/tmp/test"]})
|
|
assert "mkdir -p /tmp/test" in desc
|
|
|
|
def test_write_file(self):
|
|
from integrations.chat_bridge.vendors.discord import _format_action_description
|
|
|
|
desc = _format_action_description(
|
|
"write_file", {"file_name": "/tmp/foo.md", "contents": "hello world"}
|
|
)
|
|
assert "/tmp/foo.md" in desc
|
|
assert "11 chars" in desc
|
|
|
|
def test_python_code(self):
|
|
from integrations.chat_bridge.vendors.discord import _format_action_description
|
|
|
|
desc = _format_action_description("python", {"code": "print(42)"})
|
|
assert "print(42)" in desc
|
|
|
|
def test_unknown_tool(self):
|
|
from integrations.chat_bridge.vendors.discord import _format_action_description
|
|
|
|
desc = _format_action_description("custom_tool", {"key": "value"})
|
|
assert "custom_tool" in desc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _get_impact_level (imported from tool_safety)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGetImpactLevel:
|
|
def test_high_impact(self):
|
|
from integrations.chat_bridge.vendors.discord import _get_impact_level
|
|
|
|
assert _get_impact_level("shell") == "high"
|
|
assert _get_impact_level("python") == "high"
|
|
|
|
def test_medium_impact(self):
|
|
from integrations.chat_bridge.vendors.discord import _get_impact_level
|
|
|
|
assert _get_impact_level("write_file") == "medium"
|
|
assert _get_impact_level("aider") == "medium"
|
|
|
|
def test_low_impact(self):
|
|
from integrations.chat_bridge.vendors.discord import _get_impact_level
|
|
|
|
assert _get_impact_level("web_search") == "low"
|
|
assert _get_impact_level("unknown") == "low"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool safety classification
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestToolSafety:
|
|
def test_shell_requires_confirmation(self):
|
|
from timmy.tool_safety import requires_confirmation
|
|
|
|
assert requires_confirmation("shell") is True
|
|
|
|
def test_python_requires_confirmation(self):
|
|
from timmy.tool_safety import requires_confirmation
|
|
|
|
assert requires_confirmation("python") is True
|
|
|
|
def test_write_file_requires_confirmation(self):
|
|
from timmy.tool_safety import requires_confirmation
|
|
|
|
assert requires_confirmation("write_file") is True
|
|
|
|
def test_read_file_is_safe(self):
|
|
from timmy.tool_safety import requires_confirmation
|
|
|
|
assert requires_confirmation("read_file") is False
|
|
|
|
def test_calculator_is_safe(self):
|
|
from timmy.tool_safety import requires_confirmation
|
|
|
|
assert requires_confirmation("calculator") is False
|
|
|
|
def test_web_search_is_safe(self):
|
|
from timmy.tool_safety import requires_confirmation
|
|
|
|
assert requires_confirmation("web_search") is False
|
|
|
|
def test_unknown_tool_requires_confirmation(self):
|
|
from timmy.tool_safety import requires_confirmation
|
|
|
|
assert requires_confirmation("unknown_tool") is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _handle_message confirmation flow (native Agno RunOutput)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _mock_paused_run(tool_name="shell", tool_args=None, content="I will create the dir."):
|
|
"""Create a mock RunOutput for a paused run needing tool confirmation."""
|
|
tool_args = tool_args or {"args": ["mkdir", "/tmp/test"]}
|
|
|
|
te = MagicMock()
|
|
te.tool_name = tool_name
|
|
te.tool_args = tool_args
|
|
|
|
req = MagicMock()
|
|
req.needs_confirmation = True
|
|
req.tool_execution = te
|
|
|
|
run = MagicMock()
|
|
run.content = content
|
|
run.status = "PAUSED"
|
|
run.active_requirements = [req]
|
|
return run
|
|
|
|
|
|
def _mock_completed_run(content="Hello! How can I help?"):
|
|
"""Create a mock RunOutput for a completed (no tool) run."""
|
|
run = MagicMock()
|
|
run.content = content
|
|
run.status = "COMPLETED"
|
|
run.active_requirements = []
|
|
return run
|
|
|
|
|
|
class TestHandleMessageConfirmation:
|
|
@pytest.mark.asyncio
|
|
async def test_dangerous_tool_sends_confirmation(self, monkeypatch):
|
|
"""When Agno pauses for tool confirmation, should send confirmation prompt."""
|
|
from integrations.chat_bridge.vendors.discord import DiscordVendor
|
|
|
|
vendor = DiscordVendor()
|
|
|
|
# Mock chat_with_tools returning a paused RunOutput
|
|
paused_run = _mock_paused_run()
|
|
monkeypatch.setattr(
|
|
"integrations.chat_bridge.vendors.discord.chat_with_tools",
|
|
lambda msg, sid=None: paused_run,
|
|
)
|
|
|
|
vendor._client = MagicMock()
|
|
vendor._client.user = MagicMock()
|
|
vendor._client.user.id = 12345
|
|
|
|
message = MagicMock()
|
|
message.content = "create a directory"
|
|
message.channel = MagicMock()
|
|
message.channel.guild = MagicMock()
|
|
|
|
monkeypatch.setattr(vendor, "_get_or_create_thread", AsyncMock(return_value=None))
|
|
|
|
ctx = AsyncMock()
|
|
ctx.__aenter__ = AsyncMock(return_value=None)
|
|
ctx.__aexit__ = AsyncMock(return_value=False)
|
|
message.channel.typing = MagicMock(return_value=ctx)
|
|
message.channel.send = AsyncMock()
|
|
|
|
# Mock approvals
|
|
mock_item = MagicMock()
|
|
mock_item.id = "test-approval-id-1234"
|
|
monkeypatch.setattr(
|
|
"timmy.approvals.create_item",
|
|
lambda **kwargs: mock_item,
|
|
)
|
|
|
|
vendor._send_confirmation = AsyncMock()
|
|
|
|
await vendor._handle_message(message)
|
|
|
|
# Should have called _send_confirmation for the shell tool
|
|
vendor._send_confirmation.assert_called_once()
|
|
call_args = vendor._send_confirmation.call_args
|
|
assert call_args[0][1] == "shell" # tool_name
|
|
assert call_args[0][3] == "test-approval-id-1234" # approval_id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_tool_calls_sends_normal_response(self, monkeypatch):
|
|
"""When Agno returns a completed run, should send text directly."""
|
|
from integrations.chat_bridge.vendors.discord import DiscordVendor
|
|
|
|
vendor = DiscordVendor()
|
|
|
|
completed_run = _mock_completed_run()
|
|
monkeypatch.setattr(
|
|
"integrations.chat_bridge.vendors.discord.chat_with_tools",
|
|
lambda msg, sid=None: completed_run,
|
|
)
|
|
|
|
vendor._client = MagicMock()
|
|
vendor._client.user = MagicMock()
|
|
vendor._client.user.id = 12345
|
|
|
|
message = MagicMock()
|
|
message.content = "hello"
|
|
message.channel = MagicMock()
|
|
message.channel.guild = MagicMock()
|
|
monkeypatch.setattr(vendor, "_get_or_create_thread", AsyncMock(return_value=None))
|
|
|
|
ctx = AsyncMock()
|
|
ctx.__aenter__ = AsyncMock(return_value=None)
|
|
ctx.__aexit__ = AsyncMock(return_value=False)
|
|
message.channel.typing = MagicMock(return_value=ctx)
|
|
message.channel.send = AsyncMock()
|
|
|
|
await vendor._handle_message(message)
|
|
|
|
# Should send the text response directly (no confirmation)
|
|
message.channel.send.assert_called()
|
|
sent_text = message.channel.send.call_args_list[-1][0][0]
|
|
assert "Hello" in sent_text
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_confirmation_disabled_via_config(self, monkeypatch):
|
|
"""When discord_confirm_actions=False, no confirmation prompts sent."""
|
|
from config import settings
|
|
from integrations.chat_bridge.vendors.discord import DiscordVendor
|
|
|
|
monkeypatch.setattr(settings, "discord_confirm_actions", False)
|
|
|
|
vendor = DiscordVendor()
|
|
|
|
paused_run = _mock_paused_run()
|
|
monkeypatch.setattr(
|
|
"integrations.chat_bridge.vendors.discord.chat_with_tools",
|
|
lambda msg, sid=None: paused_run,
|
|
)
|
|
|
|
vendor._client = MagicMock()
|
|
vendor._client.user = MagicMock()
|
|
vendor._client.user.id = 12345
|
|
|
|
message = MagicMock()
|
|
message.content = "do something"
|
|
message.channel = MagicMock()
|
|
message.channel.guild = MagicMock()
|
|
monkeypatch.setattr(vendor, "_get_or_create_thread", AsyncMock(return_value=None))
|
|
|
|
ctx = AsyncMock()
|
|
ctx.__aenter__ = AsyncMock(return_value=None)
|
|
ctx.__aexit__ = AsyncMock(return_value=False)
|
|
message.channel.typing = MagicMock(return_value=ctx)
|
|
message.channel.send = AsyncMock()
|
|
|
|
vendor._send_confirmation = AsyncMock()
|
|
|
|
await vendor._handle_message(message)
|
|
|
|
# Should NOT call _send_confirmation
|
|
vendor._send_confirmation.assert_not_called()
|