ce299cbb18
- 新增 app_metadata 模型、仓储与服务 - 将默认 edge 验证步骤改为由 app_metadata 驱动生成 - 新增 chat_session / chat_message 会话层模型与 chat service - 新增 demo chat API，支持会话创建、消息发送、任务确认 - 新增最小 Web Demo 页面，形成聊天式演示入口 - 增强任务报告，补充 audit_summary 与更细粒度 task_metrics - 增强 edge-agent 执行器：tcp_probe、日志时间范围过滤、进程指标与更灵活健康检查 - 更新 README 与当前进度总结，MVP 进度推进到约 94%
79 lines
2.8 KiB
Python
79 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
from unittest.mock import patch
|
|
|
|
from app.executors.process_executor import ProcessCheckExecutor
|
|
|
|
|
|
class DummyCompletedProcess:
|
|
def __init__(self, stdout: str) -> None:
|
|
self.stdout = stdout
|
|
|
|
|
|
def test_process_check_executor_windows_match() -> None:
|
|
with (
|
|
patch("app.executors.process_executor.platform.system", return_value="Windows"),
|
|
patch(
|
|
"app.executors.process_executor.subprocess.run",
|
|
return_value=DummyCompletedProcess('"python.exe","1234","Console","1","10,000 K"\n'),
|
|
),
|
|
):
|
|
success, message, data, evidence = ProcessCheckExecutor().execute({"process_name": "python"})
|
|
|
|
assert success is True
|
|
assert message == "process found"
|
|
assert data["matched_count"] == 1
|
|
assert data["memory_rss_kb_total"] == 10000
|
|
assert evidence["matches"][0]["pid"] == 1234
|
|
|
|
|
|
def test_process_check_executor_unix_pid_miss() -> None:
|
|
with (
|
|
patch("app.executors.process_executor.platform.system", return_value="Linux"),
|
|
patch(
|
|
"app.executors.process_executor.subprocess.run",
|
|
return_value=DummyCompletedProcess("1234 python 1.5 2.0 20480 python app.py\n"),
|
|
),
|
|
):
|
|
success, message, data, evidence = ProcessCheckExecutor().execute({"pid": 9999})
|
|
|
|
assert success is False
|
|
assert message == "process not found"
|
|
assert data["matched_count"] == 0
|
|
assert evidence["matches"] == []
|
|
|
|
|
|
def test_process_check_executor_unix_collects_metrics() -> None:
|
|
with (
|
|
patch("app.executors.process_executor.platform.system", return_value="Linux"),
|
|
patch(
|
|
"app.executors.process_executor.subprocess.run",
|
|
return_value=DummyCompletedProcess("1234 python 1.5 2.0 20480 python app.py\n"),
|
|
),
|
|
):
|
|
success, message, data, evidence = ProcessCheckExecutor().execute({"process_name": "python"})
|
|
|
|
assert success is True
|
|
assert message == "process found"
|
|
assert data["matched_count"] == 1
|
|
assert data["cpu_percent_total"] == 1.5
|
|
assert data["memory_rss_kb_total"] == 20480
|
|
assert evidence["matches"][0]["memory_percent"] == 2.0
|
|
|
|
|
|
def test_process_check_executor_command_contains_match() -> None:
|
|
with (
|
|
patch("app.executors.process_executor.platform.system", return_value="Linux"),
|
|
patch(
|
|
"app.executors.process_executor.subprocess.run",
|
|
return_value=DummyCompletedProcess("1234 java 1.5 2.0 20480 java -jar order-service.jar\n"),
|
|
),
|
|
):
|
|
success, message, data, evidence = ProcessCheckExecutor().execute({"command_contains": "order-service"})
|
|
|
|
assert success is True
|
|
assert message == "process found"
|
|
assert data["matched_count"] == 1
|
|
assert data["command_contains"] == "order-service"
|
|
assert "order-service" in evidence["matches"][0]["command"]
|