bpo-26228: Fix pty EOF handling (GH-12049) · python/cpython@81ab8db (original) (raw)

`@@ -5,8 +5,9 @@

5

`import_module('termios')

6

7

`import errno

8

import pty

9

8

`import os

9

import pty

10

import tty

10

11

`import sys

11

12

`import select

12

13

`import signal

`@@ -123,12 +124,6 @@ def handle_sig(self, sig, frame):

123

124

125

`@staticmethod

125

126

`def handle_sighup(signum, frame):

126

bpo-38547: if the process is the session leader, os.close(master_fd)

127

of "master_fd, slave_name = pty.master_open()" raises SIGHUP

128

signal: just ignore the signal.

129

130

NOTE: the above comment is from an older version of the test;

131

master_open() is not being used anymore.

132

127

`pass

133

128

134

129

`@expectedFailureIfStdinIsTTY

`@@ -190,13 +185,6 @@ def test_openpty(self):

190

185

`self.assertEqual(_get_term_winsz(slave_fd), new_stdin_winsz,

191

186

`"openpty() failed to set slave window size")

192

187

193

Solaris requires reading the fd before anything is returned.

194

My guess is that since we open and close the slave fd

195

in master_open(), we need to read the EOF.

196

197

NOTE: the above comment is from an older version of the test;

198

master_open() is not being used anymore.

199

-

200

188

`# Ensure the fd is non-blocking in case there's nothing to read.

201

189

`blocking = os.get_blocking(master_fd)

202

190

`try:

`@@ -324,22 +312,40 @@ def test_master_read(self):

324

312

325

313

`self.assertEqual(data, b"")

326

314

315

def test_spawn_doesnt_hang(self):

316

pty.spawn([sys.executable, '-c', 'print("hi there")'])

317

+

327

318

`class SmallPtyTests(unittest.TestCase):

328

319

`"""These tests don't spawn children or hang."""

329

320

330

321

`def setUp(self):

331

322

`self.orig_stdin_fileno = pty.STDIN_FILENO

332

323

`self.orig_stdout_fileno = pty.STDOUT_FILENO

324

self.orig_pty_close = pty.close

325

self.orig_pty__copy = pty._copy

326

self.orig_pty_fork = pty.fork

333

327

`self.orig_pty_select = pty.select

328

self.orig_pty_setraw = pty.setraw

329

self.orig_pty_tcgetattr = pty.tcgetattr

330

self.orig_pty_tcsetattr = pty.tcsetattr

331

self.orig_pty_waitpid = pty.waitpid

334

332

`self.fds = [] # A list of file descriptors to close.

335

333

`self.files = []

336

334

`self.select_rfds_lengths = []

337

335

`self.select_rfds_results = []

336

self.tcsetattr_mode_setting = None

338

337

339

338

`def tearDown(self):

340

339

`pty.STDIN_FILENO = self.orig_stdin_fileno

341

340

`pty.STDOUT_FILENO = self.orig_stdout_fileno

341

pty.close = self.orig_pty_close

342

pty._copy = self.orig_pty__copy

343

pty.fork = self.orig_pty_fork

342

344

`pty.select = self.orig_pty_select

345

pty.setraw = self.orig_pty_setraw

346

pty.tcgetattr = self.orig_pty_tcgetattr

347

pty.tcsetattr = self.orig_pty_tcsetattr

348

pty.waitpid = self.orig_pty_waitpid

343

349

`for file in self.files:

344

350

`try:

345

351

`file.close()

`@@ -367,6 +373,14 @@ def _mock_select(self, rfds, wfds, xfds, timeout=0):

367

373

`self.assertEqual(self.select_rfds_lengths.pop(0), len(rfds))

368

374

`return self.select_rfds_results.pop(0), [], []

369

375

376

def _make_mock_fork(self, pid):

377

def mock_fork():

378

return (pid, 12)

379

return mock_fork

380

+

381

def _mock_tcsetattr(self, fileno, opt, mode):

382

self.tcsetattr_mode_setting = mode

383

+

370

384

`def test__copy_to_each(self):

371

385

`"""Test the normal data case on both master_fd and stdin."""

372

386

`read_from_stdout_fd, mock_stdout_fd = self._pipe()

`@@ -407,20 +421,41 @@ def test__copy_eof_on_all(self):

407

421

`socketpair[1].close()

408

422

`os.close(write_to_stdin_fd)

409

423

410

Expect two select calls, the last one will cause IndexError

411

424

`pty.select = self._mock_select

412

425

`self.select_rfds_lengths.append(2)

413

426

`self.select_rfds_results.append([mock_stdin_fd, masters[0]])

414

427

`# We expect that both fds were removed from the fds list as they

415

428

`# both encountered an EOF before the second select call.

416

429

`self.select_rfds_lengths.append(0)

417

430

418

with self.assertRaises(IndexError):

419

pty._copy(masters[0])

431

We expect the function to return without error.

432

self.assertEqual(pty._copy(masters[0]), None)

433

+

434

def test__restore_tty_mode_normal_return(self):

435

"""Test that spawn resets the tty mode no when _copy returns normally."""

436

+

437

PID 1 is returned from mocked fork to run the parent branch

438

of code

439

pty.fork = self._make_mock_fork(1)

440

+

441

status_sentinel = object()

442

pty.waitpid = lambda _1, _2: [None, status_sentinel]

443

pty.close = lambda _: None

444

+

445

pty._copy = lambda _1, _2, _3: None

446

+

447

mode_sentinel = object()

448

pty.tcgetattr = lambda fd: mode_sentinel

449

pty.tcsetattr = self._mock_tcsetattr

450

pty.setraw = lambda _: None

451

+

452

self.assertEqual(pty.spawn([]), status_sentinel, "pty.waitpid process status not returned by pty.spawn")

453

self.assertEqual(self.tcsetattr_mode_setting, mode_sentinel, "pty.tcsetattr not called with original mode value")

420

454

421

455

422

456

`def tearDownModule():

423

457

`reap_children()

424

458

459

+

425

460

`if name == "main":

426

461

`unittest.main()