Your Name
5d83348db3
- core/monitor.py: add fast metrics script (0.3s vs 1.4s sleep), collect() now defaults to fast=True for responsive UI - ui/workers.py: MonitorWorker changed from loop mode to single-shot (QTimer re-arms next cycle on finished signal, avoids paramiko blocking in infinite loop) - ui/widgets.py: remove debug print from _kick_one_sample - test_monitor_nonblock.py: fix sample counting by wrapping _on_sample instead of disconnecting signals (old approach missed new workers) - test_process_monitor.py: assertions account for MAX_RENDER_ROWS=200, fix MEM% column index (4 not 3) in sort test
233 lines
9.5 KiB
Python
233 lines
9.5 KiB
Python
"""
|
|
进程监控端到端测试：
|
|
- 真实 SSH 连接
|
|
- SystemMonitor.collect() 解析进程统计 + 进程列表
|
|
- MonitorPanel 在真实数据下渲染进程表
|
|
- 搜索 / 排序 / 杀进程（用 echo $$ 自己的 PID 测试）
|
|
"""
|
|
import os
|
|
import sys
|
|
import time
|
|
|
|
os.environ["QT_QPA_PLATFORM"] = "offscreen"
|
|
sys.path.insert(0, os.path.dirname(__file__))
|
|
|
|
from PyQt5.QtWidgets import QApplication
|
|
from PyQt5.QtCore import Qt, QTimer
|
|
|
|
from core.ssh_client import SSHConnection
|
|
from core.monitor import SystemMonitor
|
|
from core.manager import ConnectionManager
|
|
from ui.widgets import MonitorPanel
|
|
|
|
|
|
def find_local_ssh():
|
|
for pwd in ("sshclient_test_pwd_2026", "testpass", ""):
|
|
c = SSHConnection("127.0.0.1", 22, "root", pwd, timeout=5)
|
|
ok, _ = c.connect()
|
|
if ok:
|
|
return c
|
|
return None
|
|
|
|
|
|
def main():
|
|
app = QApplication(sys.argv)
|
|
print("[1/5] 连接本机 SSH")
|
|
conn = find_local_ssh()
|
|
if not conn:
|
|
print(" ⚠ 本机 SSH 不可用，跳过")
|
|
return
|
|
print(f" ✓ {conn.username}@{conn.host}")
|
|
|
|
print("[2/5] SystemMonitor 采集进程数据")
|
|
m = SystemMonitor.collect(conn)
|
|
assert m["proc_total"] > 50, f"应有大量进程，实际 {m['proc_total']}"
|
|
assert m["proc_running"] >= 1, "至少应有 1 个运行中进程"
|
|
assert m["proc_sleep"] > 0, "应有睡眠进程"
|
|
assert len(m["processes"]) > 0
|
|
# 验证至少一条进程包含完整字段
|
|
p = m["processes"][0]
|
|
for k in ("pid", "user", "pcpu", "pmem", "vsz", "rss", "stat", "etime", "time", "comm"):
|
|
assert k in p, f"缺字段 {k}"
|
|
assert p["pid"] > 0
|
|
assert p["comm"], "comm 字段不应为空"
|
|
print(f" ✓ 进程 {m['proc_total']} 个 (运行 {m['proc_running']} / 睡眠 {m['proc_sleep']} / 僵尸 {m['proc_zombie']})")
|
|
print(f" ✓ TOP1: PID={p['pid']} USER={p['user']} CPU={p['pcpu']}% MEM={p['pmem']}% STAT={p['stat']} COMM={p['comm'][:50]}")
|
|
|
|
print("[3/5] MonitorPanel 渲染进程表")
|
|
mgr = ConnectionManager()
|
|
mgr.connections["fake"] = conn
|
|
panel = MonitorPanel(mgr)
|
|
panel.set_host("fake")
|
|
panel._proc_data = m["processes"]
|
|
panel._apply_proc_filter()
|
|
# 扁平视图受 MAX_RENDER_ROWS 限制
|
|
expected_rows = min(len(m["processes"]), panel.MAX_RENDER_ROWS)
|
|
assert panel.proc_table.rowCount() == expected_rows, \
|
|
f"行数 {panel.proc_table.rowCount()} != 预期 {expected_rows} (总 {len(m['processes'])}, 上限 {panel.MAX_RENDER_ROWS})"
|
|
# 验证第一列 (PID) 是数字
|
|
pid_text = panel.proc_table.item(0, 0).text()
|
|
assert pid_text.isdigit(), f"PID 列不是数字: {pid_text!r}"
|
|
# 验证 PPID 列存在
|
|
ppid_text = panel.proc_table.item(0, 1).text()
|
|
assert ppid_text.lstrip("-").isdigit() or ppid_text == "-", f"PPID 错: {ppid_text!r}"
|
|
# 验证 PRI/NI 列
|
|
pri_text = panel.proc_table.item(0, 7).text()
|
|
assert pri_text.lstrip("-").isdigit(), f"PRI 错: {pri_text!r}"
|
|
print(f" ✓ 进程表行数: {panel.proc_table.rowCount()}")
|
|
print(f" ✓ 第 1 行 PID={pid_text} PPID={ppid_text} CPU={panel.proc_table.item(0, 3).text()}% PRI={pri_text}")
|
|
|
|
# TOP 5
|
|
assert panel.proc_top_cpu.count() == 5
|
|
assert panel.proc_top_mem.count() == 5
|
|
top_cpu_first = panel.proc_top_cpu.item(0).text()
|
|
top_mem_first = panel.proc_top_mem.item(0).text()
|
|
assert "%" in top_cpu_first and "%" in top_mem_first
|
|
print(f" ✓ TOP CPU: {top_cpu_first[:60]}")
|
|
print(f" ✓ TOP MEM: {top_mem_first[:60]}")
|
|
|
|
# 树形切换
|
|
panel.proc_view_combo.setCurrentIndex(1)
|
|
app.processEvents()
|
|
# 第一行应该有缩进符号（树根没有符号，但深度=0）
|
|
first_cell = panel.proc_table.item(0, 0).text()
|
|
assert "├─" in first_cell or first_cell == "", f"树形第 1 行异常: {first_cell!r}"
|
|
# 列数应该 13（树形多 1 列）
|
|
assert panel.proc_table.columnCount() == 13
|
|
print(f" ✓ 树形视图生效 (第 1 行: {first_cell!r}, 13 列)")
|
|
# 切回扁平
|
|
panel.proc_view_combo.setCurrentIndex(0)
|
|
app.processEvents()
|
|
assert panel.proc_table.columnCount() == 12
|
|
print(f" ✓ 切回扁平 (12 列)")
|
|
|
|
# CPU 过滤
|
|
panel.proc_min_cpu.setValue(10)
|
|
app.processEvents()
|
|
after_min = panel.proc_table.rowCount()
|
|
if after_min > 0:
|
|
# 验证所有行的 CPU 列都 >= 10
|
|
for r in range(min(after_min, 10)):
|
|
cpu_val = float(panel.proc_table.item(r, 3).text())
|
|
assert cpu_val >= 10, f"CPU 过滤失败: 行 {r} CPU={cpu_val}"
|
|
print(f" ✓ CPU>=10% 过滤后剩 {after_min} 行，全部 CPU≥10%")
|
|
panel.proc_min_cpu.setValue(0)
|
|
app.processEvents()
|
|
assert panel.proc_table.rowCount() == expected_rows
|
|
print(f" ✓ CPU 过滤清空恢复 {panel.proc_table.rowCount()} 行")
|
|
|
|
print("[4/5] 搜索过滤")
|
|
# 找一个常见的进程名
|
|
sample = m["processes"][0]
|
|
comm_first = sample["comm"].split()[0] if sample["comm"] else ""
|
|
if not comm_first:
|
|
# 退而求其次用 sshd
|
|
comm_first = "sshd"
|
|
panel.proc_search.setText(comm_first)
|
|
app.processEvents()
|
|
shown = panel.proc_table.rowCount()
|
|
assert shown > 0, f"按 '{comm_first}' 过滤后应至少 1 行"
|
|
assert shown < len(m["processes"]), f"过滤后应少于总数: {shown}/{len(m['processes'])}"
|
|
print(f" ✓ 搜索 '{comm_first}' 过滤后剩 {shown}/{len(m['processes'])} 行")
|
|
|
|
# 清空搜索
|
|
panel.proc_search.setText("")
|
|
app.processEvents()
|
|
assert panel.proc_table.rowCount() == expected_rows, "清空搜索应恢复"
|
|
print(f" ✓ 清空搜索恢复全部 {panel.proc_table.rowCount()} 行")
|
|
|
|
# 测试按内存排序
|
|
panel.proc_sort_combo.setCurrentIndex(1) # 按 MEM 降序
|
|
app.processEvents()
|
|
first = panel.proc_table.item(0, 4).text() # MEM% 在第 4 列
|
|
last = panel.proc_table.item(panel.proc_table.rowCount() - 1, 4).text()
|
|
assert float(first) >= float(last), f"按 MEM 排序失败: {first} < {last}"
|
|
print(f" ✓ 按 MEM 排序: 第 1 行 {first}% > 最后行 {last}%")
|
|
|
|
# 改回按 CPU 排序
|
|
panel.proc_sort_combo.setCurrentIndex(0)
|
|
app.processEvents()
|
|
|
|
print("[5/5] 杀进程（用 echo 自己的 PID 测试 kill 流程）")
|
|
# 启动一个 sleep 进程，立刻拿到 PID。disown + 全部 fd 重定向确保彻底脱离当前 shell。
|
|
# 在子 shell 里后台启动 sleep，父 shell 立即退出后 sleep 由 init 接管
|
|
print(" -> exec_command for sleep")
|
|
code, out, err = conn.exec_command(
|
|
"( sleep 60 </dev/null >/dev/null 2>&1 & echo $! )",
|
|
timeout=10,
|
|
)
|
|
print(f" -> got: out={out!r} err={err!r}")
|
|
pid = None
|
|
for line in out.splitlines():
|
|
line = line.strip()
|
|
if line.isdigit():
|
|
pid = int(line)
|
|
break
|
|
print(f" -> pid={pid}")
|
|
if pid:
|
|
# 确认进程存在
|
|
code, out, _ = conn.exec_command(f"kill -0 {pid} && echo alive", timeout=5)
|
|
assert "alive" in out, f"PID {pid} 不存在"
|
|
print(f" ✓ 启动测试进程 PID={pid}")
|
|
|
|
# 通过 panel._proc_kill 直接调（模拟右键 → 杀进程 流程）
|
|
# 用 monkeypatch 把 QMessageBox.question/information/critical 替换为 no-op + 自动 Yes
|
|
from PyQt5.QtWidgets import QMessageBox
|
|
from ui import widgets as widgets_module
|
|
orig_question = widgets_module.QMessageBox.question
|
|
orig_info = widgets_module.QMessageBox.information
|
|
orig_crit = widgets_module.QMessageBox.critical
|
|
|
|
def fake_question(*args, **kwargs):
|
|
return QMessageBox.Yes
|
|
|
|
def fake_info(*args, **kwargs):
|
|
return QMessageBox.Ok
|
|
|
|
def fake_crit(*args, **kwargs):
|
|
return QMessageBox.Ok
|
|
widgets_module.QMessageBox.question = fake_question
|
|
widgets_module.QMessageBox.information = fake_info
|
|
widgets_module.QMessageBox.critical = fake_crit
|
|
|
|
try:
|
|
# 把这个 PID 加到 _proc_data，模拟它出现在列表里
|
|
sample_proc = {
|
|
"pid": pid, "user": "root", "pcpu": 0.0, "pmem": 0.0,
|
|
"vsz": 0, "rss": 0, "stat": "S", "etime": 1, "time": 0,
|
|
"comm": "sleep 60",
|
|
}
|
|
panel._proc_data.insert(0, sample_proc)
|
|
panel._apply_proc_filter()
|
|
# 找到这一行
|
|
found_row = None
|
|
for row in range(panel.proc_table.rowCount()):
|
|
if panel.proc_table.item(row, 0).data(Qt.UserRole) == pid:
|
|
found_row = row
|
|
break
|
|
assert found_row is not None, f"PID {pid} 未在表中"
|
|
print(f" ✓ 进程在表中位于第 {found_row} 行")
|
|
# 调用 _proc_kill（会弹消息框，monkeypatch 让其通过）
|
|
panel._proc_kill(pid, "sleep 60", force=False)
|
|
print(f" ✓ _proc_kill 调用返回")
|
|
finally:
|
|
widgets_module.QMessageBox.question = orig_question
|
|
widgets_module.QMessageBox.information = orig_info
|
|
widgets_module.QMessageBox.critical = orig_crit
|
|
|
|
# 验证进程已死
|
|
time.sleep(0.5)
|
|
code, out, _ = conn.exec_command(f"kill -0 {pid} 2>&1; echo exit=$?", timeout=5)
|
|
assert "exit=1" in out or "No such" in out or "exit=0" not in out, \
|
|
f"PID {pid} 应该已死，但: {out!r}"
|
|
print(f" ✓ PID {pid} 已被杀 (远程: {out.strip()})")
|
|
else:
|
|
print(" ⚠ 无法获取测试 PID，跳过杀进程步骤")
|
|
|
|
conn.disconnect()
|
|
print("\n进程监控端到端测试通过 ✓")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|