Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 11 additions & 4 deletions tests/test_quicker_mcp.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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)
Expand Down
Loading