test(bot): bot.py unit tests — 27 cases for _write_command, _hub_on_pending, ApprovalView #task-410
This commit is contained in:
Variet Worker
486
tests/test_bot_unit.py
Normal file
486
tests/test_bot_unit.py
Normal file
@@ -0,0 +1,486 @@
|
||||
"""Unit tests for GravityBot and ApprovalView — hub/bridge routing logic.
|
||||
|
||||
Uses unittest.mock to avoid real Discord/Hub connections.
|
||||
Run: python -m pytest tests/test_bot_unit.py -v
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, MagicMock, patch, PropertyMock
|
||||
|
||||
import sys, os
|
||||
|
||||
# Ensure project root is on path
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||||
|
||||
from bridge import BridgeProtocol, ApprovalRequest, UserResponse
|
||||
|
||||
pytestmark = pytest.mark.anyio
|
||||
|
||||
|
||||
# ─── Helpers ───────────────────────────────────────────────────────────
|
||||
|
||||
def make_request(
|
||||
request_id="req-001",
|
||||
conversation_id="conv-001",
|
||||
command="Run shell command",
|
||||
project_name="test_project",
|
||||
step_type="shell",
|
||||
status="pending",
|
||||
):
|
||||
return ApprovalRequest(
|
||||
request_id=request_id,
|
||||
conversation_id=conversation_id,
|
||||
command=command,
|
||||
description="test description",
|
||||
timestamp=time.time(),
|
||||
project_name=project_name,
|
||||
step_type=step_type,
|
||||
status=status,
|
||||
)
|
||||
|
||||
|
||||
def make_mock_hub(
|
||||
active_count=1,
|
||||
send_response_result=True,
|
||||
):
|
||||
"""Create a mock Hub with async methods."""
|
||||
hub = MagicMock()
|
||||
hub.broadcast_to_project = AsyncMock()
|
||||
hub.send_to_instance = AsyncMock()
|
||||
hub.send_response_to_pending_owner = AsyncMock(return_value=send_response_result)
|
||||
hub.get_active_count = MagicMock(return_value=active_count)
|
||||
return hub
|
||||
|
||||
|
||||
def make_mock_bridge():
|
||||
"""Create a mock BridgeProtocol."""
|
||||
bridge = MagicMock(spec=BridgeProtocol)
|
||||
bridge.write_command = MagicMock()
|
||||
bridge.write_response = MagicMock()
|
||||
return bridge
|
||||
|
||||
|
||||
def make_mock_gateway():
|
||||
"""Create a mock Gateway."""
|
||||
gw = MagicMock()
|
||||
gw.push_command = MagicMock()
|
||||
return gw
|
||||
|
||||
|
||||
def make_bot_stub(hub=None, bridge=None, gateway=None):
|
||||
"""Create a minimal bot-like object with the fields used by _write_command.
|
||||
|
||||
We don't instantiate GravityBot (which requires Discord) — instead we
|
||||
create a simple namespace that has the same attributes the methods read.
|
||||
"""
|
||||
|
||||
class BotStub:
|
||||
pass
|
||||
|
||||
bot = BotStub()
|
||||
bot.hub = hub
|
||||
bot.bridge = bridge or make_mock_bridge()
|
||||
bot.gateway = gateway
|
||||
bot.auto_approve_projects = set()
|
||||
bot._sent_approval_ids = set()
|
||||
bot._approval_messages = {}
|
||||
bot._deferred_ids = {}
|
||||
bot._sent_commands = {}
|
||||
return bot
|
||||
|
||||
|
||||
# ─── _write_command Dual-Write Prevention ──────────────────────────────
|
||||
|
||||
class TestWriteCommand:
|
||||
"""_write_command must use Hub-only OR bridge-only, never both."""
|
||||
|
||||
def _call_write_command(self, bot, project, text, **kwargs):
|
||||
"""Import and call _write_command bound to our stub."""
|
||||
from bot import GravityBot
|
||||
return GravityBot._write_command(bot, project, text, **kwargs)
|
||||
|
||||
async def test_hub_present_uses_ws_broadcast(self):
|
||||
"""Hub connected + no target → broadcast_to_project, skip bridge."""
|
||||
hub = make_mock_hub()
|
||||
bot = make_bot_stub(hub=hub)
|
||||
self._call_write_command(bot, "proj", "hello")
|
||||
|
||||
hub.broadcast_to_project.assert_called_once()
|
||||
bot.bridge.write_command.assert_not_called()
|
||||
|
||||
async def test_hub_present_skips_bridge(self):
|
||||
"""Hub connected → bridge.write_command is NEVER called."""
|
||||
hub = make_mock_hub()
|
||||
bot = make_bot_stub(hub=hub)
|
||||
self._call_write_command(bot, "proj", "hello")
|
||||
|
||||
bot.bridge.write_command.assert_not_called()
|
||||
|
||||
async def test_hub_present_with_target_instance(self):
|
||||
"""Hub connected + target_instance → send_to_instance."""
|
||||
hub = make_mock_hub()
|
||||
bot = make_bot_stub(hub=hub)
|
||||
self._call_write_command(bot, "proj", "hello", target_instance=2)
|
||||
|
||||
hub.send_to_instance.assert_called_once()
|
||||
args = hub.send_to_instance.call_args
|
||||
assert args[0][0] == "proj"
|
||||
assert args[0][1] == 2
|
||||
|
||||
def test_no_hub_uses_bridge(self):
|
||||
"""No Hub → bridge.write_command is called."""
|
||||
bot = make_bot_stub(hub=None)
|
||||
self._call_write_command(bot, "proj", "hello")
|
||||
|
||||
bot.bridge.write_command.assert_called_once()
|
||||
|
||||
def test_no_hub_with_gateway(self):
|
||||
"""No Hub + gateway → both bridge.write_command and gateway.push_command."""
|
||||
gw = make_mock_gateway()
|
||||
bot = make_bot_stub(hub=None, gateway=gw)
|
||||
self._call_write_command(bot, "proj", "hello")
|
||||
|
||||
bot.bridge.write_command.assert_called_once()
|
||||
gw.push_command.assert_called_once()
|
||||
|
||||
def test_no_hub_no_gateway(self):
|
||||
"""No Hub, no gateway → only bridge.write_command."""
|
||||
bot = make_bot_stub(hub=None, gateway=None)
|
||||
self._call_write_command(bot, "proj", "hello")
|
||||
|
||||
bot.bridge.write_command.assert_called_once()
|
||||
|
||||
async def test_message_format_contains_type_and_text(self):
|
||||
"""Hub broadcast message has correct structure."""
|
||||
hub = make_mock_hub()
|
||||
bot = make_bot_stub(hub=hub)
|
||||
self._call_write_command(bot, "proj", "test msg")
|
||||
|
||||
call_args = hub.broadcast_to_project.call_args
|
||||
msg = call_args[0][1] # second positional arg
|
||||
assert msg["type"] == "command"
|
||||
assert msg["data"]["text"] == "test msg"
|
||||
assert "id" in msg["data"]
|
||||
|
||||
|
||||
# ─── _hub_on_pending Routing ───────────────────────────────────────────
|
||||
|
||||
class TestHubOnPending:
|
||||
"""_hub_on_pending routing: dedup, status checks, auto-approve, Discord send."""
|
||||
|
||||
async def _call_hub_on_pending(self, bot, project, data):
|
||||
from bot import GravityBot
|
||||
return await GravityBot._hub_on_pending(bot, project, data)
|
||||
|
||||
async def test_missing_request_id_returns_early(self):
|
||||
"""No request_id in data → silently skip."""
|
||||
bot = make_bot_stub()
|
||||
# Should not raise
|
||||
await self._call_hub_on_pending(bot, "proj", {"status": "pending"})
|
||||
# No crash = pass
|
||||
|
||||
async def test_duplicate_request_id_skipped(self):
|
||||
"""Already in _sent_approval_ids → skip."""
|
||||
bot = make_bot_stub()
|
||||
bot._sent_approval_ids.add("req-dup")
|
||||
bot._get_channel = AsyncMock()
|
||||
|
||||
await self._call_hub_on_pending(bot, "proj", {
|
||||
"request_id": "req-dup", "command": "test",
|
||||
})
|
||||
|
||||
bot._get_channel.assert_not_called()
|
||||
|
||||
async def test_auto_resolved_status_handled(self):
|
||||
"""status=auto_resolved → _handle_auto_resolved."""
|
||||
bot = make_bot_stub()
|
||||
bot._handle_auto_resolved = AsyncMock()
|
||||
|
||||
await self._call_hub_on_pending(bot, "proj", {
|
||||
"request_id": "req-ar",
|
||||
"status": "auto_resolved",
|
||||
})
|
||||
|
||||
bot._handle_auto_resolved.assert_awaited_once_with("req-ar", "auto_resolved")
|
||||
|
||||
async def test_expired_status_handled(self):
|
||||
"""status=expired → _handle_auto_resolved."""
|
||||
bot = make_bot_stub()
|
||||
bot._handle_auto_resolved = AsyncMock()
|
||||
|
||||
await self._call_hub_on_pending(bot, "proj", {
|
||||
"request_id": "req-exp",
|
||||
"status": "expired",
|
||||
})
|
||||
|
||||
bot._handle_auto_resolved.assert_awaited_once_with("req-exp", "expired")
|
||||
|
||||
async def test_auto_approve_project_routes_to_auto_approve(self):
|
||||
"""Project in auto_approve_projects → _auto_approve_via_hub."""
|
||||
hub = make_mock_hub()
|
||||
bot = make_bot_stub(hub=hub)
|
||||
bot.auto_approve_projects = {"proj"}
|
||||
bot._auto_approve_via_hub = AsyncMock()
|
||||
bot._get_channel = AsyncMock()
|
||||
bot._get_instance_header = MagicMock(return_value="")
|
||||
|
||||
await self._call_hub_on_pending(bot, "proj", {
|
||||
"request_id": "req-auto",
|
||||
"command": "Run",
|
||||
"conversation_id": "conv-1",
|
||||
})
|
||||
|
||||
bot._auto_approve_via_hub.assert_awaited_once()
|
||||
# Discord send should NOT have been called directly
|
||||
bot._get_channel.assert_not_called()
|
||||
|
||||
async def test_normal_pending_sends_to_discord(self):
|
||||
"""Normal pending → _get_channel + channel.send with embed + view."""
|
||||
hub = make_mock_hub()
|
||||
bot = make_bot_stub(hub=hub)
|
||||
bot._get_instance_header = MagicMock(return_value="")
|
||||
|
||||
mock_msg = MagicMock()
|
||||
mock_msg.id = 12345
|
||||
mock_channel = AsyncMock()
|
||||
mock_channel.send = AsyncMock(return_value=mock_msg)
|
||||
bot._get_channel = AsyncMock(return_value=mock_channel)
|
||||
|
||||
await self._call_hub_on_pending(bot, "proj", {
|
||||
"request_id": "req-normal",
|
||||
"command": "Run shell command",
|
||||
"conversation_id": "conv-1",
|
||||
})
|
||||
|
||||
mock_channel.send.assert_awaited_once()
|
||||
assert "req-normal" in bot._sent_approval_ids
|
||||
assert bot._approval_messages["req-normal"] == 12345
|
||||
|
||||
async def test_no_channel_skips_gracefully(self):
|
||||
"""No channel found → warning, no crash."""
|
||||
hub = make_mock_hub()
|
||||
bot = make_bot_stub(hub=hub)
|
||||
bot._get_instance_header = MagicMock(return_value="")
|
||||
bot._get_channel = AsyncMock(return_value=None)
|
||||
|
||||
await self._call_hub_on_pending(bot, "proj", {
|
||||
"request_id": "req-nochan",
|
||||
"command": "test",
|
||||
"conversation_id": "conv-1",
|
||||
})
|
||||
|
||||
assert "req-nochan" not in bot._sent_approval_ids
|
||||
|
||||
|
||||
# ─── _auto_approve_via_hub ─────────────────────────────────────────────
|
||||
|
||||
class TestAutoApproveViaHub:
|
||||
"""_auto_approve_via_hub: hub routing with bridge fallback."""
|
||||
|
||||
async def _call_auto_approve(self, bot, request):
|
||||
from bot import GravityBot
|
||||
return await GravityBot._auto_approve_via_hub(bot, request)
|
||||
|
||||
async def test_hub_delivered_skips_bridge(self):
|
||||
"""Hub delivers successfully → bridge.write_response NOT called."""
|
||||
hub = make_mock_hub(send_response_result=True)
|
||||
bot = make_bot_stub(hub=hub)
|
||||
bot._get_channel = AsyncMock(return_value=None)
|
||||
req = make_request()
|
||||
|
||||
await self._call_auto_approve(bot, req)
|
||||
|
||||
hub.send_response_to_pending_owner.assert_awaited_once()
|
||||
bot.bridge.write_response.assert_not_called()
|
||||
assert req.request_id in bot._sent_approval_ids
|
||||
|
||||
async def test_hub_delivery_failed_falls_back_to_bridge(self):
|
||||
"""Hub delivery fails → bridge.write_response called as fallback."""
|
||||
hub = make_mock_hub(send_response_result=False)
|
||||
bot = make_bot_stub(hub=hub)
|
||||
bot._get_channel = AsyncMock(return_value=None)
|
||||
req = make_request()
|
||||
|
||||
await self._call_auto_approve(bot, req)
|
||||
|
||||
hub.send_response_to_pending_owner.assert_awaited_once()
|
||||
bot.bridge.write_response.assert_called_once()
|
||||
|
||||
async def test_no_hub_uses_bridge_directly(self):
|
||||
"""No Hub → bridge.write_response called directly."""
|
||||
bot = make_bot_stub(hub=None)
|
||||
bot._get_channel = AsyncMock(return_value=None)
|
||||
req = make_request()
|
||||
|
||||
await self._call_auto_approve(bot, req)
|
||||
|
||||
bot.bridge.write_response.assert_called_once()
|
||||
|
||||
async def test_sent_approval_ids_updated(self):
|
||||
"""request_id added to _sent_approval_ids regardless of route."""
|
||||
hub = make_mock_hub(send_response_result=True)
|
||||
bot = make_bot_stub(hub=hub)
|
||||
bot._get_channel = AsyncMock(return_value=None)
|
||||
req = make_request(request_id="req-track")
|
||||
|
||||
await self._call_auto_approve(bot, req)
|
||||
|
||||
assert "req-track" in bot._sent_approval_ids
|
||||
|
||||
|
||||
# ─── ApprovalView Hub/Bridge Branching ─────────────────────────────────
|
||||
|
||||
class TestApprovalView:
|
||||
"""ApprovalView approve/reject: Hub-first with bridge fallback.
|
||||
|
||||
Note: approve()/reject() are @discord.ui.button-decorated methods.
|
||||
The decorator turns them into Button objects. To call the underlying
|
||||
coroutine we use view.approve.callback(view, interaction, button).
|
||||
"""
|
||||
|
||||
def _make_interaction(self, user_name="TestUser"):
|
||||
"""Create a mock Discord interaction."""
|
||||
interaction = AsyncMock()
|
||||
interaction.user.display_name = user_name
|
||||
# Embed in message
|
||||
embed = MagicMock()
|
||||
embed.color = None
|
||||
embed.set_footer = MagicMock()
|
||||
interaction.message.embeds = [embed]
|
||||
interaction.response.edit_message = AsyncMock()
|
||||
interaction.response.send_message = AsyncMock()
|
||||
return interaction
|
||||
|
||||
async def _invoke_approve(self, view, interaction):
|
||||
"""Call the approve button's underlying callback.
|
||||
|
||||
discord.ui.button wraps the method into an _ItemCallback.
|
||||
The callback signature is just (interaction) — the Button itself
|
||||
is bound internally via the descriptor.
|
||||
"""
|
||||
await view.approve.callback(interaction)
|
||||
|
||||
async def _invoke_reject(self, view, interaction):
|
||||
"""Call the reject button's underlying callback."""
|
||||
await view.reject.callback(interaction)
|
||||
|
||||
async def test_approve_hub_delivered_skips_bridge(self):
|
||||
"""Approve: hub delivers → bridge NOT called."""
|
||||
from bot import ApprovalView
|
||||
|
||||
hub = make_mock_hub(send_response_result=True)
|
||||
bridge = make_mock_bridge()
|
||||
req = make_request()
|
||||
view = ApprovalView(bridge, req, hub=hub)
|
||||
|
||||
interaction = self._make_interaction()
|
||||
await self._invoke_approve(view, interaction)
|
||||
|
||||
hub.send_response_to_pending_owner.assert_awaited_once()
|
||||
bridge.write_response.assert_not_called()
|
||||
assert view.responded is True
|
||||
|
||||
async def test_approve_hub_failed_uses_bridge(self):
|
||||
"""Approve: hub delivery fails → bridge fallback."""
|
||||
from bot import ApprovalView
|
||||
|
||||
hub = make_mock_hub(send_response_result=False)
|
||||
bridge = make_mock_bridge()
|
||||
req = make_request()
|
||||
view = ApprovalView(bridge, req, hub=hub)
|
||||
|
||||
interaction = self._make_interaction()
|
||||
await self._invoke_approve(view, interaction)
|
||||
|
||||
bridge.write_response.assert_called_once()
|
||||
|
||||
async def test_approve_no_hub_uses_bridge(self):
|
||||
"""Approve: no hub → bridge."""
|
||||
from bot import ApprovalView
|
||||
|
||||
bridge = make_mock_bridge()
|
||||
req = make_request()
|
||||
view = ApprovalView(bridge, req, hub=None)
|
||||
|
||||
interaction = self._make_interaction()
|
||||
await self._invoke_approve(view, interaction)
|
||||
|
||||
bridge.write_response.assert_called_once()
|
||||
|
||||
async def test_reject_hub_delivered(self):
|
||||
"""Reject: hub delivers → bridge NOT called, approved=False."""
|
||||
from bot import ApprovalView
|
||||
|
||||
hub = make_mock_hub(send_response_result=True)
|
||||
bridge = make_mock_bridge()
|
||||
req = make_request()
|
||||
view = ApprovalView(bridge, req, hub=hub)
|
||||
|
||||
interaction = self._make_interaction()
|
||||
await self._invoke_reject(view, interaction)
|
||||
|
||||
call_args = hub.send_response_to_pending_owner.call_args
|
||||
response_msg = call_args[0][1]
|
||||
assert response_msg["data"]["approved"] is False
|
||||
bridge.write_response.assert_not_called()
|
||||
|
||||
async def test_double_response_blocked(self):
|
||||
"""Second response attempt → blocked with ephemeral message."""
|
||||
from bot import ApprovalView
|
||||
|
||||
hub = make_mock_hub(send_response_result=True)
|
||||
bridge = make_mock_bridge()
|
||||
req = make_request()
|
||||
view = ApprovalView(bridge, req, hub=hub)
|
||||
|
||||
interaction1 = self._make_interaction()
|
||||
await self._invoke_approve(view, interaction1)
|
||||
|
||||
interaction2 = self._make_interaction()
|
||||
await self._invoke_approve(view, interaction2)
|
||||
|
||||
interaction2.response.send_message.assert_awaited_once()
|
||||
# Hub should only be called once (first response)
|
||||
assert hub.send_response_to_pending_owner.await_count == 1
|
||||
|
||||
async def test_on_timeout_sends_bridge_reject(self):
|
||||
"""Timeout with no response → bridge.write_response(approved=False)."""
|
||||
from bot import ApprovalView
|
||||
|
||||
bridge = make_mock_bridge()
|
||||
req = make_request()
|
||||
view = ApprovalView(bridge, req, hub=None)
|
||||
|
||||
await view.on_timeout()
|
||||
|
||||
bridge.write_response.assert_called_once()
|
||||
call_args = bridge.write_response.call_args[0][0]
|
||||
assert isinstance(call_args, UserResponse)
|
||||
assert call_args.approved is False
|
||||
|
||||
|
||||
# ─── _get_instance_header ──────────────────────────────────────────────
|
||||
|
||||
class TestGetInstanceHeader:
|
||||
"""Instance header formatting based on active count."""
|
||||
|
||||
def _call(self, bot, project, instance_number):
|
||||
from bot import GravityBot
|
||||
return GravityBot._get_instance_header(bot, project, instance_number)
|
||||
|
||||
def test_no_hub_returns_empty(self):
|
||||
bot = make_bot_stub(hub=None)
|
||||
assert self._call(bot, "proj", 1) == ""
|
||||
|
||||
def test_single_instance_returns_empty(self):
|
||||
hub = make_mock_hub(active_count=1)
|
||||
bot = make_bot_stub(hub=hub)
|
||||
assert self._call(bot, "proj", 1) == ""
|
||||
|
||||
def test_multiple_instances_returns_header(self):
|
||||
hub = make_mock_hub(active_count=3)
|
||||
bot = make_bot_stub(hub=hub)
|
||||
result = self._call(bot, "proj", 2)
|
||||
assert "PC #2" in result
|
||||
Reference in New Issue
Block a user