```
feat(deploy-control): 添加命令执行异常处理 当subprocess.run执行失败时捕获OSError异常,并抛出带有详细错误信息的ApiError, 提供更好的错误提示和调试支持。 ```
This commit is contained in:
@ -100,14 +100,18 @@ def run_command(args: list[str], *, cwd: Path | None = None, extra_env: dict[str
|
|||||||
env = os.environ.copy()
|
env = os.environ.copy()
|
||||||
if extra_env:
|
if extra_env:
|
||||||
env.update(extra_env)
|
env.update(extra_env)
|
||||||
completed = subprocess.run(
|
try:
|
||||||
args,
|
completed = subprocess.run(
|
||||||
cwd=str(cwd) if cwd else None,
|
args,
|
||||||
env=env,
|
cwd=str(cwd) if cwd else None,
|
||||||
text=True,
|
env=env,
|
||||||
capture_output=True,
|
text=True,
|
||||||
check=False,
|
capture_output=True,
|
||||||
)
|
check=False,
|
||||||
|
)
|
||||||
|
except OSError as exc:
|
||||||
|
command = args[0] if args else "<empty command>"
|
||||||
|
raise ApiError(HTTPStatus.BAD_GATEWAY, f"failed to execute {command}: {exc}") from exc
|
||||||
if completed.returncode != 0:
|
if completed.returncode != 0:
|
||||||
detail = completed.stderr.strip() or completed.stdout.strip() or "command failed"
|
detail = completed.stderr.strip() or completed.stdout.strip() or "command failed"
|
||||||
raise ApiError(HTTPStatus.BAD_GATEWAY, detail)
|
raise ApiError(HTTPStatus.BAD_GATEWAY, detail)
|
||||||
|
|||||||
29
deploy-control/tests/test_command_errors.py
Normal file
29
deploy-control/tests/test_command_errors.py
Normal file
@ -0,0 +1,29 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import importlib.util
|
||||||
|
from http import HTTPStatus
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
|
||||||
|
SERVER_PATH = Path(__file__).resolve().parents[1] / "server.py"
|
||||||
|
|
||||||
|
|
||||||
|
def _load_server_module():
|
||||||
|
spec = importlib.util.spec_from_file_location("deploy_control_server_command_tests", SERVER_PATH)
|
||||||
|
assert spec and spec.loader
|
||||||
|
module = importlib.util.module_from_spec(spec)
|
||||||
|
spec.loader.exec_module(module)
|
||||||
|
return module
|
||||||
|
|
||||||
|
|
||||||
|
def test_run_command_reports_missing_executable_as_bad_gateway(tmp_path: Path) -> None:
|
||||||
|
server = _load_server_module()
|
||||||
|
missing = tmp_path / "missing-command"
|
||||||
|
|
||||||
|
with pytest.raises(server.ApiError) as exc_info:
|
||||||
|
server.run_command([str(missing)])
|
||||||
|
|
||||||
|
assert exc_info.value.status_code == HTTPStatus.BAD_GATEWAY
|
||||||
|
assert str(missing) in exc_info.value.detail
|
||||||
Reference in New Issue
Block a user