sshclient/test_process_monitor.py

"""
进程监控端到端测试：
- 真实 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}"
    print(f"  ✓ 进程表行数: {panel.proc_table.rowCount()}")
    print(f"  ✓ 第 1 行 PID={pid_text} CPU={panel.proc_table.item(0, 2).text()}%")

    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 >/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()