feat: open Athena tool architecture
This commit is contained in:
@@ -85,12 +85,32 @@ class OperatorTests(unittest.TestCase):
|
||||
with self.assertRaises(RuntimeError):
|
||||
self.module.execute({"ticket": proposal["ticket"], "confirmation": proposal["required_confirmation"]})
|
||||
|
||||
def test_remote_access_and_power_operations_do_not_exist(self):
|
||||
def test_structured_power_operations_do_not_exist(self):
|
||||
self.assertNotIn("shell", self.module.ALLOWED_OPERATIONS)
|
||||
for operation in ("shutdown", "reboot", "ssh", "network", "command"):
|
||||
with self.assertRaises(ValueError):
|
||||
self.module.normalise_operation(operation, {})
|
||||
|
||||
def test_general_terminal_runs_normal_work(self):
|
||||
calls = []
|
||||
self.module.run = lambda argv, **kwargs: calls.append((argv, kwargs)) or {
|
||||
"argv": argv, "exit_code": 0, "output": "ok"
|
||||
}
|
||||
result = self.module.terminal({"command": "docker ps", "cwd": str(self.stack)})
|
||||
self.assertEqual(result["exit_code"], 0)
|
||||
self.assertEqual(calls[0][0], ["/bin/bash", "-lc", "docker ps"])
|
||||
self.assertEqual(result["reachability_guard"], "active")
|
||||
|
||||
def test_general_terminal_blocks_reachability_changes(self):
|
||||
blocked = (
|
||||
"shutdown -h now", "systemctl restart ssh", "iptables -F",
|
||||
"ip route del default", "umount /data", "nano /etc/ssh/sshd_config",
|
||||
"docker restart mike-ai-wireguard-gateway",
|
||||
)
|
||||
for command in blocked:
|
||||
with self.subTest(command=command), self.assertRaises(PermissionError):
|
||||
self.module.terminal({"command": command, "cwd": str(self.stack)})
|
||||
|
||||
def test_wireguard_gateway_cannot_be_stopped(self):
|
||||
with self.assertRaises(PermissionError):
|
||||
self.module.normalise_operation("container_action", {"action": "stop", "containers": ["mike-ai-wireguard-gateway"]})
|
||||
|
||||
Reference in New Issue
Block a user