1#!/usr/bin/env python3
2"""Linux PTY regression: python3 internal/app/testdata/broker_terminal_repro.py /absolute/vev
3
4Build the real binary first. Keeps the PTY open (no UI-driver/EOF shortcut),
5uses an implicit long VEV_ENV_ROOT, and cleans up only its isolated processes.
6"""
7import fcntl
8import hashlib
9import os
10import pathlib
11import pty
12import select
13import signal
14import struct
15import sys
16import tempfile
17import termios
18import time
19
20binary = os.path.abspath(sys.argv[1])
21with tempfile.TemporaryDirectory(prefix="vev-pty-") as temporary:
22 cwd = pathlib.Path(temporary) / ("long-worktree-" * 8)
23 cwd.mkdir()
24 root = cwd / ".dev"
25 runtime = root / "test/runtime/vev"
26 short = pathlib.Path("/tmp/vev-%d-%s" % (
27 os.getuid(), hashlib.sha256(str(runtime).encode()).hexdigest()))
28 pid = None
29 fd = None
30
31 def read_until(predicate, timeout):
32 output = b""
33 deadline = time.monotonic() + timeout
34 while time.monotonic() < deadline:
35 if select.select([fd], [], [], 0.1)[0]:
36 try:
37 output += os.read(fd, 65536)
38 except OSError:
39 break
40 if predicate(output):
41 return output
42 raise AssertionError(repr(output[-2000:]))
43
44 def start():
45 global pid, fd
46 pid, fd = pty.fork()
47 if pid == 0:
48 os.chdir(cwd)
49 os.environ.pop("VEV", None)
50 os.environ.pop("VEV_ENV_ROOT", None)
51 os.environ["VEV_ENV"] = "test"
52 os.environ["TERM"] = "xterm-256color"
53 os.environ["SHELL"] = "/bin/sh"
54 os.execv(binary, [binary])
55 fcntl.ioctl(fd, termios.TIOCSWINSZ, struct.pack("HHHH", 24, 100, 0, 0))
56
57 try:
58 start()
59 # The shell command proves client -> broker -> daemon -> PTY, not
60 # merely that the broker socket accepted a connection.
61 read_until(lambda data: b"?25h" in data, 15)
62 os.write(fd, b"printf 'VEV_PTY_%s_OK\\n' REPRO\n")
63 output = read_until(lambda data: b"VEV_PTY_REPRO_OK" in data, 10)
64 assert b"broker connection lost" not in output
65 print("PASS: long implicit development root reaches a real shell")
66 os.kill(pid, signal.SIGTERM)
67 os.waitpid(pid, 0)
68 pid = None
69 os.close(fd)
70 fd = None
71
72 # An insecure configuration fails before any service is adopted.
73 # Raw-mode Ctrl-C must still end the process with stdin held open.
74 (root / "test/config/vev").chmod(0o755)
75 start()
76 read_until(lambda data: b"broker connection lost" in data, 10)
77 os.write(fd, b"\x03")
78 deadline = time.monotonic() + 5
79 while os.waitpid(pid, os.WNOHANG)[0] == 0:
80 if time.monotonic() >= deadline:
81 raise AssertionError("Ctrl-C failed with stdin held open")
82 time.sleep(0.01)
83 pid = None
84 print("PASS: Ctrl-C exits a failed connection without terminal EOF")
85 finally:
86 if pid is not None:
87 os.kill(pid, signal.SIGKILL)
88 os.waitpid(pid, 0)
89 if fd is not None:
90 os.close(fd)
91 # Detached roles inherit the resolved root; never kill by binary name.
92 for process in pathlib.Path("/proc").iterdir():
93 if not process.name.isdigit():
94 continue
95 try:
96 environment = (process / "environ").read_bytes().split(b"\0")
97 if ("VEV_ENV_ROOT=" + str(root)).encode() in environment:
98 os.kill(int(process.name), signal.SIGKILL)
99 except (OSError, ProcessLookupError):
100 pass
101 import shutil
102 shutil.rmtree(short, ignore_errors=True)