AdamWen95
69 lines
4.3 KiB
Python
69 lines
4.3 KiB
Python
import hashlib
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from console.settings import Settings
|
|
from console.store import Store
|
|
from console.native_plans import build_plan
|
|
from console.native_rpc import handle
|
|
|
|
|
|
class NativeRPCTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
root = Path(self.temp.name)
|
|
source = root / "source"
|
|
source.mkdir()
|
|
entry = source / "test.py"
|
|
entry.write_text("print(1)\n")
|
|
self.settings = Settings(root / "runtime", [{"id": "original", "path": str(source)}])
|
|
self.settings.test_definitions = [{"key": "native", "source_group": "original", "entry": "test.py", "name": "原环境样例"}]
|
|
self.settings.native_catalog = {"targets": [], "profiles": [{"id": "native", "source_group": "original", "entry": "test.py",
|
|
"cwd": str(source), "user": "root", "interpreter": "/usr/bin/python3", "parameters": [], "arguments": [],
|
|
"impact": {"level": "service", "title": "服务切换", "details": ["会停止指定服务"], "requires_confirmation": True}}]}
|
|
self.script = {"id": "a" * 32, "name": "test.py", "description": "", "kind": "external", "source_group": "original",
|
|
"source_path": str(entry), "sha256": hashlib.sha256(entry.read_bytes()).hexdigest(), "language": "python", "version": 1,
|
|
"enabled": True, "status": "ready", "config": {"native_options": {}, "timeout_seconds": 300}}
|
|
self.store = Store(self.settings)
|
|
self.store.save_script(self.script)
|
|
self.plan = build_plan(self.settings, self.script, check_host=False)
|
|
self.request = {"op": "queue", "script_id": self.script["id"], "request_id": "5b0202b7-851d-48ad-a7b9-dce22f000001",
|
|
"plan_token": self.plan["plan_token"], "source_ip": "127.0.0.1"}
|
|
|
|
def tearDown(self):
|
|
self.temp.cleanup()
|
|
|
|
def test_server_requires_explicit_current_confirmation(self):
|
|
with patch("console.native_rpc.native_plans.build_plan", return_value=self.plan):
|
|
with self.assertRaises(ValueError):handle(self.settings, self.store, self.request)
|
|
with self.assertRaises(ValueError):handle(self.settings, self.store, {**self.request, "impact_acknowledged": True, "plan_token": "old"})
|
|
self.assertEqual(self.store.runs(), [])
|
|
run = handle(self.settings, self.store, {**self.request, "impact_acknowledged": True})
|
|
self.assertTrue(run["execution"]["impact_acknowledged"])
|
|
self.assertEqual(run["execution"]["mode"], "native")
|
|
|
|
def test_retry_returns_original_even_if_plan_later_changed(self):
|
|
with patch("console.native_rpc.native_plans.build_plan", return_value=self.plan):
|
|
first = handle(self.settings, self.store, {**self.request, "impact_acknowledged": True})
|
|
with patch("console.native_rpc.native_plans.build_plan", side_effect=ValueError("changed")):
|
|
again = handle(self.settings, self.store, self.request)
|
|
self.assertEqual(first["id"], again["id"])
|
|
self.assertEqual(len(self.store.runs()), 1)
|
|
|
|
def test_failed_preflight_and_unknown_operation_never_queue(self):
|
|
with patch("console.native_rpc.native_plans.build_plan", return_value={**self.plan, "ready": False}):
|
|
with self.assertRaises(ValueError):handle(self.settings, self.store, {**self.request, "impact_acknowledged": True})
|
|
with self.assertRaises(ValueError):handle(self.settings, self.store, {"op": "exec", "command": "id"})
|
|
self.assertEqual(self.store.runs(), [])
|
|
|
|
def test_input_rejected_for_queued_or_noninteractive_run(self):
|
|
run = self.store.queue(self.script, "sample", "127.0.0.1", execution={"mode": "native", "interactive": True})
|
|
with self.assertRaises(ValueError):handle(self.settings, self.store, {"op": "input", "run_id": run["id"], "text": "1"})
|
|
with self.assertRaises(ValueError):handle(self.settings, self.store, {"op": "input", "run_id": "../../bad", "text": "1"})
|
|
|
|
def test_input_channel_rejects_tmux_keys_and_shell_commands(self):
|
|
for text in ("\x02c", "\x1b", "bash", "1\nid", "echo test", ":new-window"):
|
|
with self.assertRaisesRegex(ValueError, "控制键"):
|
|
handle(self.settings, self.store, {"op": "input", "run_id": "a"*32, "text": text})
|