From e21d3df89babee45f660ad91ae99e60edb23d90b Mon Sep 17 00:00:00 2001 From: Cea <61349137+ceastld@users.noreply.github.com> Date: Mon, 7 Sep 2026 04:51:37 +0800 Subject: [PATCH] =?UTF-8?q?test(mcp):=20=E5=88=86=E7=A6=BB=E8=BF=9B?= =?UTF-8?q?=E7=A8=8B=E5=90=AF=E5=8A=A8=E4=B8=8E=E8=AF=B7=E6=B1=82=E8=B6=85?= =?UTF-8?q?=E6=97=B6=E9=A2=84=E7=AE=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_quicker_mcp.py | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/tests/test_quicker_mcp.py b/tests/test_quicker_mcp.py index c26f12a..96dd2fd 100644 --- a/tests/test_quicker_mcp.py +++ b/tests/test_quicker_mcp.py @@ -16,6 +16,9 @@ ROOT = Path(__file__).resolve().parents[1] PLUGIN_ROOT = ROOT / "plugins/quicker" TEMP_ROOT = ROOT / ".temp" +# A fresh Windows runner can take over ten seconds to start PowerShell/.NET. +# This bounds the test process wait, independently of the transport's HTTP timeout. +PROCESS_TIMEOUT_SECONDS = 30 POWERSHELL = Path(os.environ.get("SystemRoot", "C:/Windows")) / "System32/WindowsPowerShell/v1.0/powershell.exe" if not POWERSHELL.is_file(): POWERSHELL = shutil.which("powershell.exe") @@ -125,7 +128,7 @@ def send(self, value): self.process.stdin.write(raw + "\n") self.process.stdin.flush() - def receive(self, timeout=10): + def receive(self, timeout=PROCESS_TIMEOUT_SECONDS): try: return self.output.get(timeout=timeout) except queue.Empty: @@ -277,12 +280,16 @@ def test_disconnect_after_post_is_not_retried_and_marks_unknown_state(self): self.assertEqual(len(self.host.requests), 1) def test_timeout_is_bounded_and_never_retries(self): + bridge = self.start(timeout=1) + self.host.respond(reply("ready", {})) + self.assertEqual(bridge.call(request("ready")), reply("ready", {})) self.host.respond(reply(1, {}), delay=2) + # Measure the request deadline after startup, not PowerShell cold-start time. start = time.monotonic() - result = self.start(timeout=1).call(request()) + result = bridge.call(request()) self.assert_failure(result, "transport_failed", unknown=True) self.assertLess(time.monotonic() - start, 6) - self.assertEqual(len(self.host.requests), 1) + self.assertEqual([json.loads(item["body"])["id"] for item in self.host.requests], ["ready", 1]) def test_http_500_does_not_expose_response_and_marks_unknown_state(self): self.host.respond(status=500, body=b"sensitive-request-body qk_test_secret_DO_NOT_LOG") @@ -324,7 +331,7 @@ def test_check_mode_is_read_only_sanitized_and_does_not_initialize(self): before = self.settings_path.read_bytes() completed = subprocess.run( **manifest_launch(["-SettingsPath", str(self.settings_path), "-Check"]), - capture_output=True, encoding="utf-8", timeout=10, + capture_output=True, encoding="utf-8", timeout=PROCESS_TIMEOUT_SECONDS, ) self.assertEqual(completed.returncode, 0, completed.stderr) diagnostic = json.loads(completed.stdout)