""" 进程监控端到端测试： - 真实 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() assert panel.proc_table.rowCount() == len(m["processes"]) # 验证第一列 (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() == len(m["processes"]) 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() == len(m["processes"]), "清空搜索应恢复" print(f" ✓ 清空搜索恢复全部 {panel.proc_table.rowCount()} 行") # 测试按内存排序 panel.proc_sort_combo.setCurrentIndex(1) # 按 MEM 降序 app.processEvents() first = panel.proc_table.item(0, 3).text() # MEM% 在第 3 列 last = panel.proc_table.item(panel.proc_table.rowCount() - 1, 3).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 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()