sshclient/test_monitor_nonblock.py

"""
监控 UI 不卡死测试：
- 启动监控
- 5 秒内应该完成 ≥2 次采集（说明没卡死）
- worker 不会重复启动
- 关闭/重新打开工作正常
- _on_refresh_now 不打乱 timer
"""
import os
import sys
import time
import threading

os.environ["QT_QPA_PLATFORM"] = "offscreen"
sys.path.insert(0, os.path.dirname(__file__))

from PyQt5.QtCore import QTimer
from PyQt5.QtWidgets import QApplication

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/6] 连接本机 SSH")
    conn = find_local_ssh()
    if not conn:
        print("  ⚠ 本机 SSH 不可用，跳过")
        return
    print(f"  ✓ {conn.username}@{conn.host}")

    print("[2/6] 创建 MonitorPanel + 启动监控")
    mgr = ConnectionManager()
    mgr.connections["fake"] = conn
    panel = MonitorPanel(mgr)
    panel.set_host("fake")
    panel.btn_toggle.setChecked(True)  # 等价于点「开始监控」
    app.processEvents()
    assert panel.btn_toggle.isChecked()
    assert panel._monitor_conn is not None
    print(f"  ✓ 监控启动, interval={panel._monitor_interval_ms}ms")

    print("[3/6] 5 秒内累计采集次数（要求 ≥2 次且不卡死）")
    # 改用更短的 interval 让测试在 5s 内能采到 2 次以上
    panel._monitor_interval_ms = 1000
    # 用 wrapper 替换 _on_sample 来计数。
    # 关键：_kick_one_sample 里每次创建新 worker 时执行
    #   self.worker.sample_ready.connect(self._on_sample)
    # 此处 self._on_sample 在连接时求值，所以新 worker 会连到 wrapper。
    # 旧的断开/重连方式只作用于当前 worker，新 worker 不经过 counting。
    orig_on_sample = panel._on_sample
    samples = [0]
    def counting(m):
        samples[0] += 1
        orig_on_sample(m)
    panel._on_sample = counting
    # 重启 timer 到 1s（如果 worker #1 还在跑，等它结束后 _on_worker_finished
    # 会创建/重启 timer，此时 _monitor_interval_ms 已是 1000）
    if getattr(panel, "_monitor_timer", None):
        panel._monitor_timer.stop()
        panel._monitor_timer.start(1000)
    # 立即触发一次采集（如果 worker #1 已结束的话）
    panel._kick_one_sample()
    # 跑 5 秒（interval 1s + ~0.4s 采集耗时，应该能采 ~3 次）
    deadline = time.time() + 5.0
    while time.time() < deadline:
        app.processEvents()
        time.sleep(0.05)
    wr = panel.worker.isRunning() if panel.worker else 'N/A'
    timer = getattr(panel, '_monitor_timer', None)
    print(f"  调试: samples={samples[0]} worker.isRunning={wr} timer.isActive={timer.isActive() if timer else 'N/A'}")
    assert samples[0] >= 2, f"5s 内只采了 {samples[0]} 次，太少（卡死了？）"
    print(f"  ✓ 5s 内采集次数: {samples[0]} (期望 ≥2)")

    print("[4/6] 验证 worker 不会重复启动")
    assert panel.worker is not None
    # worker 刚跑完应在 finished 状态
    deadline = time.time() + 2
    while panel.worker and panel.worker.isRunning() and time.time() < deadline:
        app.processEvents()
        time.sleep(0.05)
    assert not panel.worker.isRunning(), "上次的 worker 应该在 finished 状态"
    # 现在再点 refresh_now
    samples[0] = 0
    panel._on_refresh_now()
    app.processEvents()
    time.sleep(0.5)
    app.processEvents()
    assert samples[0] >= 1 or panel.worker.isRunning(), "refresh_now 后没采集到"
    print(f"  ✓ 立即刷新后采集 1 次（worker isRunning={panel.worker.isRunning() if panel.worker else 'N/A'}）")
    # 恢复原始 _on_sample
    panel._on_sample = orig_on_sample

    print("[5/6] 关闭监控（停止 worker）")
    panel.btn_toggle.setChecked(False)
    app.processEvents()
    # 等 worker 退出
    if panel.worker:
        deadline = time.time() + 2
        while panel.worker.isRunning() and time.time() < deadline:
            app.processEvents()
            time.sleep(0.05)
        assert not panel.worker.isRunning(), "关闭后 worker 还在跑"
    assert panel._monitor_timer is None, "关闭后 timer 还在"
    print(f"  ✓ 监控关闭 (worker stopped, timer cleared)")

    print("[6/6] 重新打开监控应能正常恢复")
    panel.btn_toggle.setChecked(True)
    app.processEvents()
    assert panel._monitor_conn is not None
    assert panel._monitor_interval_ms > 0
    # 等 worker 跑完
    deadline = time.time() + 3
    while time.time() < deadline:
        app.processEvents()
        if panel.worker and not panel.worker.isRunning() and panel._proc_data:
            break
        time.sleep(0.05)
    assert panel._proc_data, "重开后没采到数据"
    print(f"  ✓ 重开正常，进程数={len(panel._proc_data)}")

    # 清理
    panel.btn_toggle.setChecked(False)
    app.processEvents()
    conn.disconnect()
    print("\n监控 UI 不卡死测试通过 ✓")


if __name__ == "__main__":
    main()