mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-09-07 16:36:53 +08:00
fix(system): improve error reporting for subprocess failures and add tests for output handling
This commit is contained in:
+14
-1
@@ -47,7 +47,20 @@ class SystemUtils:
|
|||||||
output = result.stdout + result.stderr
|
output = result.stdout + result.stderr
|
||||||
return True, output
|
return True, output
|
||||||
except subprocess.CalledProcessError as e:
|
except subprocess.CalledProcessError as e:
|
||||||
error_message = f"命令:{' '.join(pip_command)},执行失败,错误信息:{e.stderr.strip()}"
|
stdout = (e.stdout or "").strip()
|
||||||
|
stderr = (e.stderr or "").strip()
|
||||||
|
# 不同命令/兼容层可能把失败原因写入 stdout,失败时需要同时保留两路输出。
|
||||||
|
output_parts = []
|
||||||
|
if stdout:
|
||||||
|
output_parts.append(f"标准输出:{stdout}")
|
||||||
|
if stderr:
|
||||||
|
output_parts.append(f"错误输出:{stderr}")
|
||||||
|
if not output_parts:
|
||||||
|
output_parts.append("无标准输出或错误输出")
|
||||||
|
error_message = (
|
||||||
|
f"命令:{' '.join(pip_command)},执行失败,"
|
||||||
|
f"返回码:{e.returncode},{'; '.join(output_parts)}"
|
||||||
|
)
|
||||||
return False, error_message
|
return False, error_message
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
error_message = f"未知错误,命令:{' '.join(pip_command)},错误:{str(e)}"
|
error_message = f"未知错误,命令:{' '.join(pip_command)},错误:{str(e)}"
|
||||||
|
|||||||
@@ -0,0 +1,44 @@
|
|||||||
|
import subprocess
|
||||||
|
from unittest import TestCase
|
||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
from app.utils.system import SystemUtils
|
||||||
|
|
||||||
|
|
||||||
|
class SystemUtilsTest(TestCase):
|
||||||
|
|
||||||
|
def test_execute_with_subprocess_keeps_stdout_when_command_fails(self):
|
||||||
|
"""
|
||||||
|
命令失败时如果原因只写入 stdout,也需要回传给调用方用于错误提示。
|
||||||
|
"""
|
||||||
|
error = subprocess.CalledProcessError(
|
||||||
|
returncode=1,
|
||||||
|
cmd=["pip", "check"],
|
||||||
|
output="demo requires pkg>=2, but you have pkg 1\n",
|
||||||
|
stderr="",
|
||||||
|
)
|
||||||
|
|
||||||
|
with patch("app.utils.system.subprocess.run", side_effect=error):
|
||||||
|
success, message = SystemUtils.execute_with_subprocess(["pip", "check"])
|
||||||
|
|
||||||
|
self.assertFalse(success)
|
||||||
|
self.assertIn("返回码:1", message)
|
||||||
|
self.assertIn("标准输出:demo requires pkg>=2, but you have pkg 1", message)
|
||||||
|
|
||||||
|
def test_execute_with_subprocess_reports_empty_failure_output(self):
|
||||||
|
"""
|
||||||
|
命令失败且没有任何输出时,给出明确占位信息,避免错误原因看起来被截断。
|
||||||
|
"""
|
||||||
|
error = subprocess.CalledProcessError(
|
||||||
|
returncode=2,
|
||||||
|
cmd=["pip", "check"],
|
||||||
|
output="",
|
||||||
|
stderr="",
|
||||||
|
)
|
||||||
|
|
||||||
|
with patch("app.utils.system.subprocess.run", side_effect=error):
|
||||||
|
success, message = SystemUtils.execute_with_subprocess(["pip", "check"])
|
||||||
|
|
||||||
|
self.assertFalse(success)
|
||||||
|
self.assertIn("返回码:2", message)
|
||||||
|
self.assertIn("无标准输出或错误输出", message)
|
||||||
Reference in New Issue
Block a user