|
19 | 19 |
|
20 | 20 | from tox.execute.api import ExecuteOptions, Outcome
|
21 | 21 | from tox.execute.local_sub_process import SIG_INTERRUPT, LocalSubProcessExecuteInstance, LocalSubProcessExecutor
|
| 22 | +from tox.execute.local_sub_process.read_via_thread_unix import ReadViaThreadUnix |
22 | 23 | from tox.execute.request import ExecuteRequest, StdinSource
|
23 | 24 | from tox.execute.stream import SyncWrite
|
24 | 25 | from tox.report import NamedBytesIO
|
@@ -140,6 +141,41 @@ def test_local_execute_write_a_lot(os_env: dict[str, str]) -> None:
|
140 | 141 | assert outcome.err == expected_err, expected_err[len(outcome.err) :]
|
141 | 142 |
|
142 | 143 |
|
| 144 | +@pytest.mark.skipif(sys.platform == "win32", reason="Unix terminal size test") |
| 145 | +def test_local_execute_terminal_size(os_env: dict[str, str], monkeypatch: MonkeyPatch) -> None: |
| 146 | + """Regression test for #2999 - check terminal size is set correctly in tox subprocess.""" |
| 147 | + import pty |
| 148 | + |
| 149 | + terminal_size = os.terminal_size((84, 42)) |
| 150 | + main, child = pty.openpty() # type: ignore[attr-defined, unused-ignore] |
| 151 | + # Use ReadViaThreadUnix to help with debugging the test itself. |
| 152 | + pipe_out = ReadViaThreadUnix(main, sys.stdout.buffer.write, name="testout", drain=True) # type: ignore[arg-type] |
| 153 | + with pipe_out, monkeypatch.context() as monkey, open(child, "w") as stdout_mock: # noqa: PTH123 |
| 154 | + # Switch stdout with test pty |
| 155 | + monkey.setattr(sys, "stdout", stdout_mock) |
| 156 | + monkey.setenv("COLUMNS", "84") |
| 157 | + monkey.setenv("LINES", "42") |
| 158 | + |
| 159 | + executor = LocalSubProcessExecutor(colored=False) |
| 160 | + request = ExecuteRequest( |
| 161 | + cmd=[sys.executable, "-c", "import os; print(os.get_terminal_size())"], |
| 162 | + cwd=Path(), |
| 163 | + env=os_env, |
| 164 | + stdin=StdinSource.OFF, |
| 165 | + run_id="", |
| 166 | + ) |
| 167 | + out_err = FakeOutErr() |
| 168 | + with executor.call(request, show=False, out_err=out_err.out_err, env=MagicMock()) as status: |
| 169 | + while status.exit_code is None: # pragma: no branch |
| 170 | + status.wait() |
| 171 | + outcome = status.outcome |
| 172 | + assert outcome is not None |
| 173 | + assert bool(outcome), outcome |
| 174 | + expected_out = f"{terminal_size!r}\r\n" |
| 175 | + assert outcome.out == expected_out, expected_out[len(outcome.out) :] |
| 176 | + assert not outcome.err |
| 177 | + |
| 178 | + |
143 | 179 | def test_local_execute_basic_fail(capsys: CaptureFixture, caplog: LogCaptureFixture, monkeypatch: MonkeyPatch) -> None:
|
144 | 180 | monkeypatch.chdir(Path(__file__).parents[3])
|
145 | 181 | caplog.set_level(logging.NOTSET)
|
|
0 commit comments