|
| 1 | +"""How a render behaves with and without a terminal attached. |
| 2 | +
|
| 3 | +The TUI needs a person at a terminal: it reads keys from stdin, draws on stderr, |
| 4 | +and the console summary goes to stdout. Started from a script or an agent |
| 5 | +(`codeplain module.plain > render.log 2>&1 &`), stdin is at end-of-file. |
| 6 | +Textual's input thread then reads EOF in a tight loop - millions of reads a |
| 7 | +second - so the process pins a CPU core for the whole render, and a stopped |
| 8 | +render waits for an Enter nobody can press. One such render stayed alive for |
| 9 | +an hour and a half at ~100% CPU until it was killed. |
| 10 | +
|
| 11 | +So without a terminal codeplain runs as --headless does, and a stopped render |
| 12 | +exits non-zero. With a terminal the TUI is kept, and waiting for Enter must |
| 13 | +stay idle. |
| 14 | +""" |
| 15 | + |
| 16 | +import fcntl |
| 17 | +import os |
| 18 | +import struct |
| 19 | +import subprocess |
| 20 | +import sys |
| 21 | +import termios |
| 22 | +import textwrap |
| 23 | +import threading |
| 24 | +import time |
| 25 | +from argparse import Namespace |
| 26 | +from contextlib import ExitStack |
| 27 | +from unittest.mock import MagicMock, patch |
| 28 | + |
| 29 | +import pytest |
| 30 | + |
| 31 | +import plain2code |
| 32 | + |
| 33 | +REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) |
| 34 | + |
| 35 | + |
| 36 | +class FakeStream: |
| 37 | + def __init__(self, is_terminal: bool): |
| 38 | + self._is_terminal = is_terminal |
| 39 | + |
| 40 | + def isatty(self) -> bool: |
| 41 | + return self._is_terminal |
| 42 | + |
| 43 | + |
| 44 | +class TestAttachedToTerminal: |
| 45 | + @pytest.fixture |
| 46 | + def streams(self, monkeypatch): |
| 47 | + def set_streams(stdin=True, stdout=True, stderr=True): |
| 48 | + for name, value in (("stdin", stdin), ("stdout", stdout), ("stderr", stderr)): |
| 49 | + monkeypatch.setattr(sys, name, value if value is None else FakeStream(value)) |
| 50 | + |
| 51 | + return set_streams |
| 52 | + |
| 53 | + def test_all_three_streams_on_a_terminal(self, streams): |
| 54 | + streams() |
| 55 | + |
| 56 | + assert plain2code.attached_to_terminal() is True |
| 57 | + |
| 58 | + @pytest.mark.parametrize("stream", ["stdin", "stdout", "stderr"]) |
| 59 | + def test_any_stream_off_the_terminal(self, streams, stream): |
| 60 | + """stdin carries the keys, stderr the TUI, stdout the summary. Losing any |
| 61 | + of them leaves a TUI nobody can drive or see.""" |
| 62 | + streams(**{stream: False}) |
| 63 | + |
| 64 | + assert plain2code.attached_to_terminal() is False |
| 65 | + |
| 66 | + def test_a_closed_stream(self, streams): |
| 67 | + """sys.stdin is None when the process has no stdin at all.""" |
| 68 | + streams(stdin=None) |
| 69 | + |
| 70 | + assert plain2code.attached_to_terminal() is False |
| 71 | + |
| 72 | + |
| 73 | +def make_args(): |
| 74 | + return Namespace( |
| 75 | + version=False, |
| 76 | + status=False, |
| 77 | + full_plain=False, |
| 78 | + dry_run=False, |
| 79 | + api_key="test-key", |
| 80 | + api="https://api.example.com", |
| 81 | + filename="project.plain", |
| 82 | + template_dir=None, |
| 83 | + build_folder="build", |
| 84 | + render_range=None, |
| 85 | + render_from=None, |
| 86 | + force_render=False, |
| 87 | + headless=False, |
| 88 | + replay_with=None, |
| 89 | + log_to_file=False, |
| 90 | + log_file_name="codeplain.log", |
| 91 | + render_conformance_tests=False, |
| 92 | + conformance_tests_script=None, |
| 93 | + unittests_script=None, |
| 94 | + prepare_environment_script=None, |
| 95 | + ) |
| 96 | + |
| 97 | + |
| 98 | +def run_render(succeeded, terminal, render_state=None, render_choices=None): |
| 99 | + """Drive the real main() and render() up to the module renderer. |
| 100 | +
|
| 101 | + The module renderer stands in for the state machine: it decides how the |
| 102 | + render ended. Everything the decision under test does not touch - the API, |
| 103 | + parsing, logging, telemetry - is stubbed. |
| 104 | +
|
| 105 | + Returns the mocks for the two TUIs and the module renderer, the args main() |
| 106 | + rendered with, and the exit status it asked for (None when it returned |
| 107 | + normally). |
| 108 | + """ |
| 109 | + args = make_args() |
| 110 | + module_renderer = MagicMock() |
| 111 | + |
| 112 | + def make_module_renderer(codeplainAPI, plain_module, render_choice, render_range, args, run_state, *rest, **kw): |
| 113 | + module_renderer.render_module.side_effect = lambda: run_state.set_render_succeeded(succeeded) |
| 114 | + return module_renderer |
| 115 | + |
| 116 | + progress_tui = MagicMock() |
| 117 | + progress_tui.return_value.run.side_effect = lambda: progress_tui.call_args.kwargs["on_ready"]() |
| 118 | + choice_tui = MagicMock() |
| 119 | + |
| 120 | + patches = [ |
| 121 | + patch("plain2code.attached_to_terminal", return_value=terminal), |
| 122 | + patch("plain2code.parse_arguments", return_value=args), |
| 123 | + patch("plain2code.file_utils.get_template_directories", return_value=["templates"]), |
| 124 | + patch("plain2code.plain_modules.PlainModule", return_value=MagicMock(all_required_modules=[])), |
| 125 | + patch("plain2code.setup_logging", return_value="INFO"), |
| 126 | + patch("plain2code.initialize_telemetry"), |
| 127 | + patch("plain2code.stop_narrating"), |
| 128 | + patch("plain2code.print_exit_summary"), |
| 129 | + patch("plain2code.dump_crash_logs"), |
| 130 | + patch("plain2code.capture_crash"), |
| 131 | + patch("plain2code.codeplain_api.CodeplainAPI"), |
| 132 | + patch("plain2code._check_connection", return_value="user@codeplain.ai"), |
| 133 | + patch("plain2code.signal.signal"), |
| 134 | + patch("plain2code.warn_if_acceptance_tests_without_conformance_script"), |
| 135 | + patch("plain2code.get_plain_module_render_state", return_value=render_state), |
| 136 | + patch("plain2code.get_render_choices", return_value=render_choices), |
| 137 | + patch("plain2code.ModuleRenderer", side_effect=make_module_renderer), |
| 138 | + patch("plain2code.Plain2CodeTUI", progress_tui), |
| 139 | + patch("plain2code.PlainModuleRenderChoiceTUI", choice_tui), |
| 140 | + patch.object(plain2code.console, "quiet", plain2code.console.quiet), |
| 141 | + ] |
| 142 | + |
| 143 | + exit_status = None |
| 144 | + with ExitStack() as stack: |
| 145 | + for each in patches: |
| 146 | + stack.enter_context(each) |
| 147 | + try: |
| 148 | + plain2code.main() |
| 149 | + except SystemExit as e: |
| 150 | + exit_status = e.code |
| 151 | + |
| 152 | + return Namespace( |
| 153 | + progress_tui=progress_tui, |
| 154 | + choice_tui=choice_tui, |
| 155 | + module_renderer=module_renderer, |
| 156 | + exit_status=exit_status, |
| 157 | + args=args, |
| 158 | + ) |
| 159 | + |
| 160 | + |
| 161 | +class TestRenderWithoutATerminal: |
| 162 | + def test_the_tui_never_starts(self): |
| 163 | + result = run_render(succeeded=False, terminal=False) |
| 164 | + |
| 165 | + result.progress_tui.assert_not_called() |
| 166 | + result.module_renderer.render_module.assert_called_once() |
| 167 | + |
| 168 | + def test_a_stopped_render_exits_non_zero(self): |
| 169 | + """The status --headless gives, so a script can tell it stopped.""" |
| 170 | + result = run_render(succeeded=False, terminal=False) |
| 171 | + |
| 172 | + assert result.exit_status == 1 |
| 173 | + |
| 174 | + def test_a_completed_render_exits_cleanly(self): |
| 175 | + result = run_render(succeeded=True, terminal=False) |
| 176 | + |
| 177 | + assert result.exit_status is None |
| 178 | + |
| 179 | + def test_says_why_it_runs_without_the_tui(self, caplog): |
| 180 | + caplog.set_level("INFO", logger="codeplain") |
| 181 | + |
| 182 | + run_render(succeeded=True, terminal=False) |
| 183 | + |
| 184 | + assert "No terminal attached" in caplog.text |
| 185 | + |
| 186 | + def test_never_asks_where_to_resume(self): |
| 187 | + """The choice prompt is a TUI too, and would wait for a key just the same.""" |
| 188 | + destructive = MagicMock(is_destructive=True) |
| 189 | + choices = {"restart": destructive, "other": MagicMock(), "quit": MagicMock()} |
| 190 | + |
| 191 | + result = run_render(succeeded=True, terminal=False, render_state=MagicMock(), render_choices=choices) |
| 192 | + |
| 193 | + result.choice_tui.assert_not_called() |
| 194 | + |
| 195 | + |
| 196 | +class TestRenderWithATerminal: |
| 197 | + def test_the_tui_still_runs(self): |
| 198 | + result = run_render(succeeded=True, terminal=True) |
| 199 | + |
| 200 | + result.progress_tui.assert_called_once() |
| 201 | + assert result.args.headless is False |
| 202 | + |
| 203 | + def test_says_nothing_about_a_missing_terminal(self, caplog): |
| 204 | + caplog.set_level("INFO", logger="codeplain") |
| 205 | + |
| 206 | + run_render(succeeded=True, terminal=True) |
| 207 | + |
| 208 | + assert "No terminal attached" not in caplog.text |
| 209 | + |
| 210 | + |
| 211 | +IDLE_SECONDS = 2.0 |
| 212 | + |
| 213 | +# Drives the real TUI to the stopped state, then measures the CPU the process |
| 214 | +# spends over IDLE_SECONDS while it waits for Enter. |
| 215 | +STOPPED_TUI = textwrap.dedent(""" |
| 216 | + import os, sys, threading, time |
| 217 | +
|
| 218 | + from event_bus import EventBus |
| 219 | + from plain2code_events import RenderFailed |
| 220 | + from plain2code_state import RunState |
| 221 | + from tui.plain2code_tui import Plain2CodeTUI |
| 222 | +
|
| 223 | + report_path, idle_seconds = sys.argv[1], float(sys.argv[2]) |
| 224 | + event_bus = EventBus() |
| 225 | +
|
| 226 | + def stop_then_measure(): |
| 227 | + event_bus.publish(RenderFailed(error_message="unit tests still failing")) |
| 228 | + start = time.process_time() |
| 229 | + time.sleep(idle_seconds) |
| 230 | + with open(report_path + ".tmp", "w") as report: |
| 231 | + report.write(str(time.process_time() - start)) |
| 232 | + os.replace(report_path + ".tmp", report_path) |
| 233 | +
|
| 234 | + Plain2CodeTUI( |
| 235 | + event_bus=event_bus, |
| 236 | + run_state=RunState(spec_filename="x.plain"), |
| 237 | + on_ready=threading.Thread(target=stop_then_measure, daemon=True).start, |
| 238 | + render_id="test", |
| 239 | + unittests_script=None, |
| 240 | + conformance_tests_script=None, |
| 241 | + prepare_environment_script=None, |
| 242 | + state_machine_version="0.0.0", |
| 243 | + css_path="styles.css", |
| 244 | + ).run() |
| 245 | + """) |
| 246 | + |
| 247 | + |
| 248 | +@pytest.mark.skipif(sys.platform == "win32", reason="needs a POSIX pseudo-terminal") |
| 249 | +def test_a_stopped_render_on_a_terminal_waits_for_enter_without_spinning(tmp_path): |
| 250 | + controller, terminal = os.openpty() |
| 251 | + fcntl.ioctl(terminal, termios.TIOCSWINSZ, struct.pack("HHHH", 30, 100, 0, 0)) |
| 252 | + report = tmp_path / "cpu_seconds" |
| 253 | + |
| 254 | + process = subprocess.Popen( |
| 255 | + [sys.executable, "-c", STOPPED_TUI, str(report), str(IDLE_SECONDS)], |
| 256 | + stdin=terminal, |
| 257 | + stdout=terminal, |
| 258 | + stderr=terminal, |
| 259 | + cwd=REPO_ROOT, |
| 260 | + env={**os.environ, "TERM": "xterm-256color"}, |
| 261 | + start_new_session=True, |
| 262 | + ) |
| 263 | + os.close(terminal) |
| 264 | + |
| 265 | + def drain(): |
| 266 | + # The TUI blocks once the terminal buffer fills, so keep reading it. |
| 267 | + try: |
| 268 | + while os.read(controller, 65536): |
| 269 | + pass |
| 270 | + except OSError: |
| 271 | + pass |
| 272 | + |
| 273 | + threading.Thread(target=drain, daemon=True).start() |
| 274 | + |
| 275 | + try: |
| 276 | + deadline = time.monotonic() + 60 |
| 277 | + while not report.exists(): |
| 278 | + assert process.poll() is None, "the TUI exited before Enter was pressed" |
| 279 | + assert time.monotonic() < deadline, "the TUI never reached the stopped state" |
| 280 | + time.sleep(0.1) |
| 281 | + cpu_seconds = float(report.read_text()) |
| 282 | + |
| 283 | + os.write(controller, b"\r") |
| 284 | + exit_status = process.wait(timeout=15) |
| 285 | + finally: |
| 286 | + if process.poll() is None: |
| 287 | + process.kill() |
| 288 | + process.wait() |
| 289 | + os.close(controller) |
| 290 | + |
| 291 | + # A busy loop burns about one CPU-second per second of waiting. |
| 292 | + assert cpu_seconds < IDLE_SECONDS / 4 |
| 293 | + assert exit_status == 0 |
0 commit comments