test(conformance): wait for initial shell output during bounded boot
This commit is contained in:
1 parent
d2306b91ad
commit
615ace707e
3 files changed
+55
-2
No files matched your search
@@ -19,6 +19,7 @@ jobs:
|
||||
env:
|
||||
PYTHONUTF8: "1"
|
||||
NEBULA_RESIZE_TRACE: "1"
|
||||
NEBULA_BOOT_TRACE: "1"
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- uses: actions/setup-python@v5
|
||||
|
||||
@@ -1,9 +1,12 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
|
||||
from conformance.harness import ConformanceContext, PROTOCOL_NAME, PROTOCOL_VERSION, path_shape, require
|
||||
|
||||
|
||||
def run(ctx: ConformanceContext) -> dict[str, object]:
|
||||
started = time.monotonic()
|
||||
snapshot = ctx.snapshot()
|
||||
ctx.refresh_targets(snapshot)
|
||||
windows = snapshot.get("windows") or []
|
||||
@@ -47,6 +50,14 @@ def run(ctx: ConformanceContext) -> dict[str, object]:
|
||||
cwd = pane.get("cwd", "")
|
||||
require(path_shape(cwd) != "relative", f"pane cwd is not absolute: {cwd!r}")
|
||||
|
||||
# Runtime discovery and a child PID can exist before a cold shell emits any
|
||||
# text. Finish the bounded boot phase before the echo case submits input;
|
||||
# its own output deadline and single submission remain unchanged.
|
||||
ctx.poll(
|
||||
ctx.read, lambda read: bool(read["text"].strip()),
|
||||
"default shell did not render initial output", ctx.startup_timeout,
|
||||
)
|
||||
|
||||
return {
|
||||
"window_count": 1,
|
||||
"default_tab_present": True,
|
||||
@@ -54,5 +65,5 @@ def run(ctx: ConformanceContext) -> dict[str, object]:
|
||||
"process_tree_available": True,
|
||||
"shell_category": shell,
|
||||
"cwd_shape": path_shape(cwd),
|
||||
"startup_ms": ctx.startup_ms,
|
||||
"startup_ms": ctx.startup_ms + round((time.monotonic() - started) * 1000),
|
||||
}
|
||||
@@ -8,7 +8,7 @@ import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import Mock, patch
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
|
||||
@@ -31,6 +31,47 @@ from conformance.harness import ( # noqa: E402
|
||||
from conformance.macos_launch import capture_screenshot
|
||||
|
||||
|
||||
class BootReadinessTests(unittest.TestCase):
|
||||
def context(self, outputs):
|
||||
from conformance.harness import ConformanceContext, PROTOCOL_NAME, PROTOCOL_VERSION
|
||||
|
||||
pane = {"id": 1, "cwd": r"C:\fixture"}
|
||||
tab = {"panes": [pane]}
|
||||
snapshot = {"protocol_version": PROTOCOL_VERSION, "windows": [{"tabs": [tab]}]}
|
||||
context = SimpleNamespace(
|
||||
snapshot=lambda: snapshot, refresh_targets=lambda value: None,
|
||||
window_id=1, pane_id=1, startup_ms=10, startup_timeout=4,
|
||||
description={"protocol": PROTOCOL_NAME, "protocol_version": PROTOCOL_VERSION,
|
||||
"capabilities": ["runtime.snapshot", "window.close", "tab.new",
|
||||
"tab.rename", "pane.split", "pane.resize", "pane.prompt",
|
||||
"pane.paste", "pane.read", "pane.procs", "pane.send_key"]},
|
||||
detect_shell=lambda: "powershell", api=lambda *args: {"processes": [{"pid": 42}]},
|
||||
tab_for_pane=lambda *args: tab, read=Mock(side_effect=outputs),
|
||||
)
|
||||
context.poll = ConformanceContext.poll.__get__(context)
|
||||
return context
|
||||
|
||||
def test_runtime_discovery_alone_does_not_finish_cold_shell_boot(self):
|
||||
from conformance.cases.boot import run
|
||||
|
||||
context = self.context([{"text": ""}, {"text": " \n"}, {"text": "PS C:\\fixture> "}])
|
||||
with patch("time.sleep"), patch("time.monotonic", side_effect=range(20)):
|
||||
result = run(context)
|
||||
self.assertEqual(context.read.call_count, 3)
|
||||
self.assertTrue(result["default_tab_present"])
|
||||
self.assertGreater(result["startup_ms"], context.startup_ms)
|
||||
|
||||
def test_silent_shell_fails_boot_instead_of_being_marked_ready(self):
|
||||
from conformance.cases.boot import run
|
||||
|
||||
context = self.context(None)
|
||||
context.read.side_effect = None
|
||||
context.read.return_value = {"text": ""}
|
||||
with patch("time.sleep"), patch("time.monotonic", side_effect=range(20)):
|
||||
with self.assertRaisesRegex(ConformanceError, "did not render initial output"):
|
||||
run(context)
|
||||
|
||||
|
||||
class NormalizationTests(unittest.TestCase):
|
||||
def test_normalize_path_handles_drive_and_posix_paths(self) -> None:
|
||||
self.assertEqual(normalize_path(r"C:\Users\Test\nebula"), "/c/Users/Test/nebula")
|
||||
|
||||
Reference in new issue
Block a user