591df2d18e
- 将 Windows/Linux service control 执行器从占位实现推进到可用 - 新增 service control 测试，覆盖 status/start/stop/restart 主路径 - 增强 edge-agent 启动脚本，优先使用包内私有 Python 运行时 - 增强 Windows/Linux 打包脚本，支持携带私有 Python 运行时 - 更新 edge-agent README 与当前进度总结 - 新增 dist 忽略规则，避免打包产物污染仓库
92 lines
3.2 KiB
Python
92 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
from unittest.mock import patch
|
|
|
|
from app.executors.linux_service_executor import LinuxServiceExecutor
|
|
from app.executors.windows_service_executor import WindowsServiceExecutor
|
|
|
|
|
|
class DummyCompletedProcess:
|
|
def __init__(self, stdout: str = "", stderr: str = "", returncode: int = 0) -> None:
|
|
self.stdout = stdout
|
|
self.stderr = stderr
|
|
self.returncode = returncode
|
|
|
|
|
|
def test_windows_service_executor_status_running() -> None:
|
|
query_output = "STATE : 4 RUNNING"
|
|
with patch(
|
|
"app.executors.windows_service_executor.subprocess.run",
|
|
return_value=DummyCompletedProcess(stdout=query_output, returncode=0),
|
|
):
|
|
success, message, data, evidence = WindowsServiceExecutor().execute(
|
|
{"service_name": "Spooler", "action": "status"}
|
|
)
|
|
|
|
assert success is True
|
|
assert message == "service status queried"
|
|
assert data["service_status"] == "RUNNING"
|
|
assert "RUNNING" in evidence["raw_output"]
|
|
|
|
|
|
def test_windows_service_executor_restart() -> None:
|
|
responses = [
|
|
DummyCompletedProcess(stdout="STATE : 4 RUNNING", returncode=0),
|
|
DummyCompletedProcess(stdout="", returncode=0),
|
|
DummyCompletedProcess(stdout="STATE : 1 STOPPED", returncode=0),
|
|
DummyCompletedProcess(stdout="STATE : 1 STOPPED", returncode=0),
|
|
DummyCompletedProcess(stdout="", returncode=0),
|
|
DummyCompletedProcess(stdout="STATE : 4 RUNNING", returncode=0),
|
|
]
|
|
|
|
with patch(
|
|
"app.executors.windows_service_executor.subprocess.run",
|
|
side_effect=responses,
|
|
):
|
|
success, message, data, evidence = WindowsServiceExecutor().execute(
|
|
{"service_name": "Spooler", "action": "restart"}
|
|
)
|
|
|
|
assert success is True
|
|
assert data["service_status"] == "RUNNING"
|
|
assert data["previous_action"] == "stop"
|
|
assert "stop" in evidence
|
|
|
|
|
|
def test_linux_service_executor_status_inactive() -> None:
|
|
with patch(
|
|
"app.executors.linux_service_executor.subprocess.run",
|
|
return_value=DummyCompletedProcess(stdout="inactive\n", returncode=3),
|
|
):
|
|
success, message, data, evidence = LinuxServiceExecutor().execute(
|
|
{"service_name": "nginx", "action": "status"}
|
|
)
|
|
|
|
assert success is False
|
|
assert message == "inactive"
|
|
assert data["service_status"] == "inactive"
|
|
assert "inactive" in evidence["raw_output"]
|
|
|
|
|
|
def test_linux_service_executor_restart_user_scope() -> None:
|
|
responses = [
|
|
DummyCompletedProcess(stdout="", returncode=0),
|
|
DummyCompletedProcess(stdout="active\n", returncode=0),
|
|
]
|
|
|
|
with patch(
|
|
"app.executors.linux_service_executor.subprocess.run",
|
|
side_effect=responses,
|
|
) as mocked_run:
|
|
success, message, data, evidence = LinuxServiceExecutor().execute(
|
|
{"service_name": "demo.service", "action": "restart", "scope": "user"}
|
|
)
|
|
|
|
assert success is True
|
|
assert message == "service status queried"
|
|
assert data["service_status"] == "active"
|
|
assert data["scope"] == "user"
|
|
first_command = mocked_run.call_args_list[0].args[0]
|
|
assert first_command[:2] == ["systemctl", "--user"]
|
|
assert "restart" in first_command
|