kuddev
c8c951bcda
The resize fixture submitted the new pane's command before its shell produced initial output. Reuse the bounded boot precondition before the first sibling probe while preserving immediate post-resize measurements and strict geometry assertions. Cover delayed and silent startup with cleanup.
165 lines
6.9 KiB
Python
165 lines
6.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
SCRIPTS_DIR = Path(__file__).resolve().parents[2]
|
|
if str(SCRIPTS_DIR) not in sys.path:
|
|
sys.path.insert(0, str(SCRIPTS_DIR))
|
|
|
|
from conformance.cases import resize
|
|
from conformance.harness import ConformanceContext, ConformanceError
|
|
|
|
|
|
class ResizeContext:
|
|
window_id = 1
|
|
pane_id = 10
|
|
shell = "bash"
|
|
platform_family = "macos"
|
|
|
|
def __init__(self, artifact_dir: Path, columns: list[int], outputs=None) -> None:
|
|
self.artifact_dir = artifact_dir
|
|
self.columns = iter(columns)
|
|
self.cleanup: list[str] = []
|
|
self.startup_timeout = 3.0
|
|
self.outputs = list(outputs) if outputs is not None else ["fixture prompt"]
|
|
self.sibling_ready = outputs is None
|
|
self.read_count = 0
|
|
self.measured_panes: list[int] = []
|
|
self.resized = False
|
|
self.poll = ConformanceContext.poll.__get__(self)
|
|
|
|
def api(self, method: str, params=None, **kwargs):
|
|
if method == "pane.split":
|
|
return {"action": {"pane_id": 11}}
|
|
if method == "pane.resize":
|
|
self.resized = True
|
|
return {"action": {"ratio": params["ratio"]}}
|
|
if method == "runtime.snapshot":
|
|
return {"windows": [{"id": self.window_id}]}
|
|
if method == "pane.read":
|
|
return {"text": "NEBULA_COLS_3_116"}
|
|
raise AssertionError(method)
|
|
|
|
def measure_columns(self, pane_id: int) -> int:
|
|
if pane_id == 11 and not self.sibling_ready:
|
|
raise AssertionError("columns probe sent before sibling startup")
|
|
self.measured_panes.append(pane_id)
|
|
return next(self.columns)
|
|
|
|
def read(self, pane_id: int):
|
|
if self.resized:
|
|
raise AssertionError("startup polling must not delay post-resize measurements")
|
|
self.read_count += 1
|
|
text = self.outputs[0]
|
|
if len(self.outputs) > 1:
|
|
self.outputs.pop(0)
|
|
self.sibling_ready = bool(text.strip())
|
|
return {"text": text}
|
|
|
|
def best_effort_api(self, method: str, params) -> None:
|
|
self.cleanup.append(method)
|
|
|
|
|
|
class ResizeDiagnosticsTests(unittest.TestCase):
|
|
def test_cold_sibling_renders_output_before_its_first_columns_command(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
ctx = ResizeContext(Path(directory), [57, 56, 80, 33], ["", " \n", "PS C:\\fixture> "])
|
|
with patch("time.sleep"), patch("time.monotonic", side_effect=[0, 0.1, 0.2, 0.3]):
|
|
result = resize.run(ctx)
|
|
self.assertTrue(result["pty_columns_changed"])
|
|
self.assertEqual(ctx.read_count, 3)
|
|
self.assertEqual(ctx.measured_panes, [10, 11, 10, 11])
|
|
self.assertEqual(ctx.cleanup, ["pane.close", "window.focus"])
|
|
|
|
def test_silent_sibling_fails_before_resize_and_is_closed(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
ctx = ResizeContext(Path(directory), [57], [""])
|
|
with patch("time.sleep"), patch("time.monotonic", side_effect=range(20)):
|
|
with self.assertRaisesRegex(ConformanceError, "split shell did not render initial output"):
|
|
resize.run(ctx)
|
|
self.assertFalse(ctx.resized)
|
|
self.assertEqual(ctx.measured_panes, [10])
|
|
self.assertEqual(ctx.cleanup, ["pane.close", "window.focus"])
|
|
|
|
def test_growing_and_shrinking_panes_pass_without_diagnostic_sampling(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
ctx = ResizeContext(Path(directory), [57, 56, 80, 33])
|
|
with patch.object(resize, "_write_failure_diagnostics") as diagnose:
|
|
result = resize.run(ctx)
|
|
self.assertTrue(result["pty_columns_changed"])
|
|
diagnose.assert_not_called()
|
|
self.assertEqual(ctx.cleanup, ["pane.close", "window.focus"])
|
|
|
|
def test_unchanged_spawn_width_still_fails_and_preserves_all_measurements(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
ctx = ResizeContext(Path(directory), [116, 116, 116, 116])
|
|
with self.assertRaisesRegex(ConformanceError, "70% pane did not grow: 116 -> 116"):
|
|
resize.run(ctx)
|
|
details = json.loads((Path(directory) / "resize-failure.json").read_text())
|
|
self.assertEqual(details["measurements"], {
|
|
"source_columns_before": 116,
|
|
"sibling_columns_before": 116,
|
|
"source_columns_after": 116,
|
|
"sibling_columns_after": 116,
|
|
})
|
|
self.assertEqual(set(details["panes"]), {"10", "11"})
|
|
self.assertEqual(ctx.cleanup, ["pane.close", "window.focus"])
|
|
|
|
def test_sibling_and_total_assertions_remain_strict(self) -> None:
|
|
for columns, message in [
|
|
([57, 56, 80, 56], "30% pane did not shrink"),
|
|
([57, 56, 85, 15], "PTY column total changed"),
|
|
]:
|
|
with self.subTest(columns=columns), tempfile.TemporaryDirectory() as directory:
|
|
ctx = ResizeContext(Path(directory), columns)
|
|
with self.assertRaisesRegex(ConformanceError, message):
|
|
resize.run(ctx)
|
|
self.assertEqual(ctx.cleanup, ["pane.close", "window.focus"])
|
|
|
|
def test_diagnostic_failure_cannot_replace_the_resize_failure(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
ctx = ResizeContext(Path(directory), [116] * 4)
|
|
with patch.object(resize, "_write_failure_diagnostics", side_effect=OSError("read-only")), \
|
|
patch("builtins.print"):
|
|
with self.assertRaisesRegex(ConformanceError, "70% pane did not grow"):
|
|
resize.run(ctx)
|
|
self.assertEqual(ctx.cleanup, ["pane.close", "window.focus"])
|
|
|
|
@unittest.skipUnless(sys.platform == "linux", "reads the Linux process PTY")
|
|
def test_kernel_probe_follows_pty_geometry_even_with_stale_columns_environment(self) -> None:
|
|
import fcntl
|
|
import struct
|
|
import termios
|
|
|
|
master, slave = os.openpty()
|
|
child = None
|
|
try:
|
|
child = subprocess.Popen(
|
|
[sys.executable, "-c", "import time; time.sleep(30)"],
|
|
stdin=slave, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
env={**os.environ, "COLUMNS": "116"},
|
|
)
|
|
for columns in (57, 82):
|
|
fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack("HHHH", 30, columns, 0, 0))
|
|
self.assertEqual(
|
|
resize._linux_terminal_size(child.pid),
|
|
{"rows": 30, "columns": columns},
|
|
)
|
|
finally:
|
|
if child is not None:
|
|
child.terminate()
|
|
child.wait(timeout=5)
|
|
os.close(master)
|
|
os.close(slave)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|