221 lines
8.8 KiB
Python
221 lines
8.8 KiB
Python
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from app.services import ops_runtime_executor_service
|
|
from app.services.ops_runtime_executor_service import execute_local_support_action, execute_ssh_action
|
|
|
|
|
|
class OpsRuntimeExecutorServiceTests(unittest.TestCase):
|
|
@patch("app.services.ops_runtime_executor_service.subprocess.run")
|
|
@patch("app.services.ops_runtime_executor_service._load_ssh_secret")
|
|
def test_execute_ssh_action_parses_structured_json_result(self, mock_load_secret, mock_run) -> None:
|
|
mock_load_secret.return_value = {"ssh_password": "", "ssh_private_key": ""}
|
|
mock_run.return_value = subprocess.CompletedProcess(
|
|
args=["ssh"],
|
|
returncode=0,
|
|
stdout='login banner\n{"ok": true, "message": "health snapshot collected", "result": {"checks": {"api": {"state": "active"}}}}\n',
|
|
stderr="",
|
|
)
|
|
|
|
ok, message, data = execute_ssh_action(
|
|
{
|
|
"node_code": "mainland-worker-01",
|
|
"ssh_host": "121.204.244.248",
|
|
"ssh_port": 22,
|
|
"ssh_user": "root",
|
|
},
|
|
"health.snapshot",
|
|
{},
|
|
)
|
|
|
|
self.assertTrue(ok)
|
|
self.assertEqual("health snapshot collected", message)
|
|
self.assertEqual("active", data["checks"]["api"]["state"])
|
|
self.assertEqual("ssh", data["transport"]["executor"])
|
|
self.assertEqual("121.204.244.248", data["transport"]["ssh_host"])
|
|
issued_command = mock_run.call_args.args[0]
|
|
self.assertEqual("ssh", issued_command[0])
|
|
self.assertIn("root@121.204.244.248", issued_command)
|
|
|
|
@patch("app.services.ops_runtime_executor_service.subprocess.run")
|
|
def test_execute_local_support_action_collects_logs_with_payload(self, mock_run) -> None:
|
|
mock_run.return_value = subprocess.CompletedProcess(
|
|
args=["journalctl"],
|
|
returncode=0,
|
|
stdout="log line 1\nlog line 2",
|
|
stderr="",
|
|
)
|
|
|
|
ok, message, data = execute_local_support_action(
|
|
"logs.collect",
|
|
{
|
|
"service_name": "domaincheck-worker",
|
|
"lines": 180,
|
|
},
|
|
)
|
|
|
|
self.assertTrue(ok)
|
|
self.assertEqual("log line 1\nlog line 2", message)
|
|
self.assertEqual("domaincheck-worker", data["service_name"])
|
|
self.assertEqual(180, data["lines"])
|
|
issued_command = mock_run.call_args.args[0]
|
|
self.assertEqual(
|
|
["journalctl", "-u", "domaincheck-worker", "-n", "180", "--no-pager"],
|
|
issued_command,
|
|
)
|
|
|
|
@patch("app.services.ops_runtime_executor_service.subprocess.run")
|
|
def test_execute_local_support_action_falls_back_to_worker_log_file(self, mock_run) -> None:
|
|
mock_run.return_value = subprocess.CompletedProcess(
|
|
args=["journalctl"],
|
|
returncode=1,
|
|
stdout="",
|
|
stderr="No journal files were opened due to insufficient permissions.",
|
|
)
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
log_path = Path(temp_dir) / "detect_worker.log"
|
|
log_path.write_text("fallback-1\nfallback-2\n", encoding="utf-8")
|
|
with patch("app.services.ops_action_executor_core.settings.domain_root", temp_dir):
|
|
ok, message, data = execute_local_support_action(
|
|
"logs.collect",
|
|
{
|
|
"service_name": "domaincheck-worker",
|
|
"lines": 180,
|
|
},
|
|
)
|
|
|
|
self.assertTrue(ok)
|
|
self.assertEqual("fallback-1\nfallback-2", message)
|
|
self.assertEqual("file_fallback", data["collection_source"])
|
|
self.assertTrue(data["fallback_used"])
|
|
|
|
@patch("app.services.ops_runtime_executor_service.subprocess.run")
|
|
def test_execute_local_support_action_restarts_service(self, mock_run) -> None:
|
|
mock_run.return_value = subprocess.CompletedProcess(
|
|
args=["systemctl"],
|
|
returncode=0,
|
|
stdout="",
|
|
stderr="",
|
|
)
|
|
|
|
ok, message, data = execute_local_support_action(
|
|
"service.restart",
|
|
{
|
|
"service_name": "domaincheck-worker",
|
|
},
|
|
)
|
|
|
|
self.assertTrue(ok)
|
|
self.assertEqual("domaincheck-worker restart completed", message)
|
|
self.assertEqual("domaincheck-worker", data["service_name"])
|
|
self.assertEqual("restart", data["systemctl_action"])
|
|
issued_command = mock_run.call_args.args[0]
|
|
self.assertEqual(["systemctl", "restart", "domaincheck-worker"], issued_command)
|
|
|
|
@patch("app.services.ops_runtime_executor_service.subprocess.run")
|
|
@patch("app.services.ops_runtime_executor_service._load_ssh_secret")
|
|
@patch("app.services.ops_runtime_executor_service.paramiko.SSHClient")
|
|
def test_execute_ssh_action_uses_paramiko_for_password_auth(self, mock_ssh_client_cls, mock_load_secret, mock_run) -> None:
|
|
mock_load_secret.return_value = {"ssh_password": "demo-pass", "ssh_private_key": ""}
|
|
mock_client = MagicMock()
|
|
stdout = MagicMock()
|
|
stderr = MagicMock()
|
|
stdout.channel.recv_exit_status.return_value = 0
|
|
stdout.read.return_value = (
|
|
b'{"ok": true, "message": "health snapshot collected", "result": {"checks": {"api": {"state": "active"}}}}\n'
|
|
)
|
|
stderr.read.return_value = b""
|
|
mock_client.exec_command.return_value = (None, stdout, stderr)
|
|
mock_ssh_client_cls.return_value = mock_client
|
|
|
|
ok, message, data = execute_ssh_action(
|
|
{
|
|
"node_code": "mainland-worker-01",
|
|
"ssh_host": "121.204.244.248",
|
|
"ssh_port": 22,
|
|
"ssh_user": "root",
|
|
"auth_mode": "password",
|
|
},
|
|
"health.snapshot",
|
|
{},
|
|
)
|
|
|
|
self.assertTrue(ok)
|
|
self.assertEqual("health snapshot collected", message)
|
|
self.assertEqual("active", data["checks"]["api"]["state"])
|
|
self.assertEqual("password", data["transport"]["auth_mode"])
|
|
mock_client.connect.assert_called_once()
|
|
self.assertFalse(mock_run.called)
|
|
|
|
@patch("app.services.ops_runtime_executor_service.subprocess.run")
|
|
@patch("app.services.ops_runtime_executor_service._load_ssh_secret")
|
|
@patch("app.services.ops_runtime_executor_service.paramiko.SSHClient")
|
|
def test_execute_ssh_action_falls_back_to_password_when_node_auth_mode_is_key_but_only_password_exists(
|
|
self,
|
|
mock_ssh_client_cls,
|
|
mock_load_secret,
|
|
mock_run,
|
|
) -> None:
|
|
mock_load_secret.return_value = {"ssh_password": "demo-pass", "ssh_private_key": ""}
|
|
mock_client = MagicMock()
|
|
stdout = MagicMock()
|
|
stderr = MagicMock()
|
|
stdout.channel.recv_exit_status.return_value = 0
|
|
stdout.read.return_value = (
|
|
b'{"ok": true, "message": "health snapshot collected", "result": {"checks": {"api": {"state": "active"}}}}\n'
|
|
)
|
|
stderr.read.return_value = b""
|
|
mock_client.exec_command.return_value = (None, stdout, stderr)
|
|
mock_ssh_client_cls.return_value = mock_client
|
|
|
|
ok, message, data = execute_ssh_action(
|
|
{
|
|
"node_code": "mainland-controller-01",
|
|
"ssh_host": "121.204.244.188",
|
|
"ssh_port": 22,
|
|
"ssh_user": "root",
|
|
"auth_mode": "key",
|
|
},
|
|
"health.snapshot",
|
|
{},
|
|
)
|
|
|
|
self.assertTrue(ok)
|
|
self.assertEqual("health snapshot collected", message)
|
|
self.assertEqual("password", data["transport"]["auth_mode"])
|
|
mock_client.connect.assert_called_once()
|
|
self.assertFalse(mock_run.called)
|
|
|
|
@patch("app.services.ops_runtime_executor_service._load_ssh_secret")
|
|
def test_execute_ssh_action_returns_graceful_error_when_paramiko_missing_for_password_auth(
|
|
self,
|
|
mock_load_secret,
|
|
) -> None:
|
|
mock_load_secret.return_value = {"ssh_password": "demo-pass", "ssh_private_key": ""}
|
|
|
|
with patch.object(ops_runtime_executor_service, "paramiko", None):
|
|
ok, message, data = execute_ssh_action(
|
|
{
|
|
"node_code": "mainland-controller-01",
|
|
"ssh_host": "121.204.244.188",
|
|
"ssh_port": 22,
|
|
"ssh_user": "root",
|
|
"auth_mode": "password",
|
|
},
|
|
"health.snapshot",
|
|
{},
|
|
)
|
|
|
|
self.assertFalse(ok)
|
|
self.assertIn("paramiko", message)
|
|
self.assertEqual("password", data["transport"]["auth_mode"])
|
|
self.assertEqual("121.204.244.188", data["transport"]["ssh_host"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|