mkbk
117 lines
5.2 KiB
Python
117 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from owner_voice_pet.config import AppConfig
|
|
from owner_voice_pet.external_adapters import (
|
|
CODEX_COMPUTER_USE_SAFETY_REFERENCE,
|
|
BrowserPlaywrightAdapter,
|
|
ComputerControlProvider,
|
|
OpenInterpreterAdapter,
|
|
is_high_risk_task,
|
|
planned_external_adapters,
|
|
)
|
|
from owner_voice_pet.tool_router import ToolCallRequest, ToolContext, ToolRouter
|
|
|
|
|
|
class ExternalAdaptersTests(unittest.TestCase):
|
|
def test_openinterpreter_disabled_returns_unavailable_status(self) -> None:
|
|
adapter = OpenInterpreterAdapter(enabled=False)
|
|
request = ToolCallRequest("1", "openinterpreter.run", {"task": "列出文件"}, "turn-1")
|
|
|
|
result = adapter.execute(request, ToolContext())
|
|
|
|
self.assertEqual(result.status, "failed")
|
|
self.assertEqual(result.error_code, "OPENINTERPRETER_DISABLED")
|
|
|
|
def test_openinterpreter_missing_command_is_structured(self) -> None:
|
|
adapter = OpenInterpreterAdapter(enabled=True, command="/tmp/owner-missing-openinterpreter")
|
|
request = ToolCallRequest("1", "openinterpreter.run", {"task": "列出文件"}, "turn-1")
|
|
|
|
result = adapter.execute(request, ToolContext())
|
|
|
|
self.assertEqual(result.status, "failed")
|
|
self.assertEqual(result.error_code, "OPENINTERPRETER_UNAVAILABLE")
|
|
|
|
def test_openinterpreter_low_risk_dry_run(self) -> None:
|
|
adapter = OpenInterpreterAdapter(enabled=True, command="python3", dry_run=True)
|
|
request = ToolCallRequest("1", "openinterpreter.run", {"task": "只读检查当前目录"}, "turn-1")
|
|
|
|
result = adapter.execute(request, ToolContext())
|
|
|
|
self.assertEqual(result.status, "success")
|
|
self.assertIn("dry-run Open Interpreter", result.output_text)
|
|
|
|
def test_openinterpreter_high_risk_task_requires_confirmation(self) -> None:
|
|
adapter = OpenInterpreterAdapter(enabled=True, command="python3", dry_run=True)
|
|
request = ToolCallRequest("1", "openinterpreter.run", {"task": "delete files"}, "turn-1")
|
|
|
|
result = adapter.execute(request, ToolContext())
|
|
|
|
self.assertEqual(result.status, "confirmation_required")
|
|
self.assertEqual(result.error_code, "OPENINTERPRETER_HIGH_RISK")
|
|
|
|
def test_playwright_disabled_and_missing_dependency_are_structured(self) -> None:
|
|
request = ToolCallRequest("1", "browser.playwright", {"task": "读取网页标题"}, "turn-1")
|
|
|
|
disabled = BrowserPlaywrightAdapter(enabled=False).execute(request, ToolContext())
|
|
with patch("importlib.util.find_spec", return_value=None):
|
|
missing = BrowserPlaywrightAdapter(enabled=True).execute(request, ToolContext())
|
|
|
|
self.assertEqual(disabled.error_code, "PLAYWRIGHT_DISABLED")
|
|
self.assertEqual(missing.error_code, "PLAYWRIGHT_UNAVAILABLE")
|
|
|
|
def test_playwright_high_risk_task_requires_confirmation_when_available(self) -> None:
|
|
request = ToolCallRequest("1", "browser.playwright", {"task": "purchase item"}, "turn-1")
|
|
with patch("importlib.util.find_spec", return_value=object()):
|
|
result = BrowserPlaywrightAdapter(enabled=True).execute(request, ToolContext())
|
|
|
|
self.assertEqual(result.status, "confirmation_required")
|
|
self.assertEqual(result.error_code, "PLAYWRIGHT_SENSITIVE_ACTION")
|
|
|
|
def test_computer_control_provider_is_reserved_and_unsupported(self) -> None:
|
|
result = ComputerControlProvider(enabled=True).execute(
|
|
ToolCallRequest("1", "computer.control", {"action": "click"}, "turn-1"),
|
|
ToolContext(),
|
|
)
|
|
|
|
self.assertEqual(result.status, "failed")
|
|
self.assertEqual(result.error_code, "COMPUTER_CONTROL_UNSUPPORTED")
|
|
|
|
def test_planned_external_adapters_respect_config_defaults(self) -> None:
|
|
adapters = planned_external_adapters(AppConfig())
|
|
|
|
self.assertFalse(adapters["openinterpreter.run"].enabled)
|
|
self.assertFalse(adapters["browser.playwright"].enabled)
|
|
self.assertFalse(adapters["computer.control"].enabled)
|
|
|
|
def test_tool_router_requires_confirmation_for_browser_sensitive_intent(self) -> None:
|
|
router = ToolRouter({"browser.playwright": BrowserPlaywrightAdapter(enabled=True)})
|
|
request = ToolCallRequest(
|
|
"1",
|
|
"browser.playwright",
|
|
{"task": "读取网页"},
|
|
"turn-1",
|
|
natural_language_intent="购买商品",
|
|
)
|
|
|
|
decision = router.route(request, ToolContext())
|
|
|
|
self.assertEqual(decision.action, "require_confirmation")
|
|
self.assertEqual(decision.risk_level, "high")
|
|
|
|
def test_codex_computer_use_reference_is_policy_only(self) -> None:
|
|
self.assertIn("safety-confirmation", CODEX_COMPUTER_USE_SAFETY_REFERENCE)
|
|
self.assertIn("public", CODEX_COMPUTER_USE_SAFETY_REFERENCE)
|
|
self.assertNotIn("private API", CODEX_COMPUTER_USE_SAFETY_REFERENCE)
|
|
|
|
def test_high_risk_task_classifier_covers_chinese_and_english(self) -> None:
|
|
self.assertTrue(is_high_risk_task("删除下载目录"))
|
|
self.assertTrue(is_high_risk_task("upload account backup"))
|
|
self.assertFalse(is_high_risk_task("read local project summary"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|