xieyi-weixin-/test_remote_gui_bridge.py

# -*- coding: utf-8 -*-
"""主窗口执行远程指令：要等真实结果再回执，不能点了就报成功。"""

import concurrent.futures
import os
import types
import unittest

os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")

from PySide6.QtCore import QCoreApplication  # noqa: E402

import wechat_gui_qt as gui  # noqa: E402


class FakeThread:
    def __init__(self, alive=True):
        self.alive = alive

    def is_alive(self):
        return self.alive


class FakeHost:
    """只带远程指令用到的那几样状态。"""

    def __init__(self, *, running=False, state="stopped"):
        self._thread = FakeThread(True) if running else None
        self._running = running
        self._status_key = state
        self._protocol_start_pending = False
        self._exit_requested = False
        self._forced_logout_handled = False
        self._protocol_probe = {}
        self.logs = []
        self.after_start = None
        for name in ("_remote_listening", "_settle_remote_command", "_run_remote_command",
                     "_start_after_stop", "_begin_remote_start"):
            setattr(self, name, types.MethodType(getattr(gui.MainWindow, name), self))
        self._resolve_remote = gui.MainWindow._resolve_remote

    def append_log(self, text, tag=""):
        self.logs.append((text, tag))

    def _refresh_remote_status_cache(self):
        pass

    def start_monitoring(self):
        if self.after_start:
            self.after_start(self)

    def stop_monitoring(self):
        self._running = False
        self._status_key = "stopping"


class RemoteCommandExecutionTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.app = QCoreApplication.instance() or QCoreApplication([])

    def execute(self, host, kind, *, finish=None):
        future = concurrent.futures.Future()
        host._run_remote_command({"command": {"type": kind, "created_by": "主管（手机）"}, "future": future})
        for tick in range(200):
            if future.done():
                break
            if finish and tick == 3:
                finish(host)
            QCoreApplication.processEvents()
            QCoreApplication.sendPostedEvents()
            import time
            time.sleep(0.05)
        self.assertTrue(future.done(), "远程指令没有得到结果")
        return future.result()

    def test_start_reports_success_once_the_listener_is_running(self):
        host = FakeHost()

        def start(h):
            h._thread = FakeThread(True)
            h._running = True
            h._status_key = "running"

        host.after_start = start
        result = self.execute(host, "start_listening")
        self.assertTrue(result["ok"])
        self.assertTrue(result["result"]["listening"])
        self.assertIn("远程启动监听", host.logs[0][0])

    def test_start_failure_is_reported_as_failure(self):
        host = FakeHost()

        def fail(h):
            h._status_key = "error"

        host.after_start = fail
        result = self.execute(host, "start_listening")
        self.assertFalse(result["ok"])
        self.assertIn("运行日志", result["error"])

    def test_stop_waits_for_the_worker_to_exit(self):
        host = FakeHost(running=True, state="running")

        def exited(h):
            h._thread.alive = False
            h._status_key = "stopped"

        result = self.execute(host, "stop_listening", finish=exited)
        self.assertTrue(result["ok"])
        self.assertFalse(result["result"]["listening"])
        self.assertEqual(result["result"]["message"], "监听已停止")

    def test_no_op_commands_answer_immediately(self):
        self.assertEqual(self.execute(FakeHost(running=True, state="running"), "start_listening")["result"]["message"],
                         "监听本来就在运行")
        self.assertEqual(self.execute(FakeHost(), "stop_listening")["result"]["message"], "监听本来就是停止状态")

    def test_restart_after_a_failed_run_waits_for_the_protocol_check(self):
        """上一轮以"连接失败"结束：协议检测期间状态还是旧的 error，不能据此判失败。"""
        host = FakeHost(state="error")

        def probe_started(h):
            h._protocol_start_pending = True

        def probe_finished(h):
            h._protocol_start_pending = False
            h._thread = FakeThread(True)
            h._running = True
            h._status_key = "running"

        host.after_start = probe_started
        result = self.execute(host, "start_listening", finish=probe_finished)
        self.assertTrue(result["ok"], result)
        self.assertEqual(result["result"]["message"], "监听已启动")

    def test_start_while_the_previous_run_is_still_stopping(self):
        host = FakeHost(running=True, state="stopping")
        host._running = False          # 正在停止：线程还活着
        old_thread = host._thread
        started = []

        def start(h):
            started.append(True)
            h._thread = FakeThread(True)
            h._running = True
            h._status_key = "running"

        def old_run_exits(h):
            old_thread.alive = False

        host.after_start = start
        result = self.execute(host, "start_listening", finish=old_run_exits)
        self.assertTrue(result["ok"], result)
        self.assertEqual(started, [True], "旧线程退出后才启动，而且只启动一次")

    def test_a_start_that_never_creates_a_thread_fails(self):
        host = FakeHost(state="stopped")       # start_monitoring 什么都没做（例如设置保存失败）
        result = self.execute(host, "start_listening")
        self.assertFalse(result["ok"])

    def test_late_results_after_cancellation_are_ignored(self):
        import concurrent.futures

        future = concurrent.futures.Future()
        future.cancel()
        gui.MainWindow._resolve_remote(future, {"ok": True})
        self.assertTrue(future.cancelled())

    def test_commands_are_refused_while_logging_out(self):
        host = FakeHost()
        host._forced_logout_handled = True
        self.assertFalse(self.execute(host, "start_listening")["ok"])
        self.assertFalse(self.execute(FakeHost(), "format_disk")["ok"])


if __name__ == "__main__":
    unittest.main()