Fix Bash activity completion after failed commands (#36)
This commit is contained in:
Samuel Onoja authored and GitHub committed
1 parent
305f49a62f
commit
f0620e317b
2 files changed
+129
-60
No files matched your search
@@ -17,58 +17,128 @@ ROOT = Path(__file__).resolve().parents[2]
|
||||
class ActivityHooks(unittest.TestCase):
|
||||
def test_real_interactive_commands_publish_start_and_exit_status(self):
|
||||
for shell in ('bash', 'fish'):
|
||||
if not shutil.which(shell):
|
||||
continue
|
||||
with self.subTest(shell=shell), tempfile.TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
publisher = root / 'feresectl'
|
||||
publisher.write_text('#!/usr/bin/env python3\nimport json, os, sys\n'
|
||||
'with open(os.environ["ACTIVITY_TEST_LOG"], "a") as f:\n'
|
||||
' f.write(json.dumps(sys.argv[1:]) + "\\n")\n')
|
||||
publisher.chmod(0o755)
|
||||
log = root / 'calls'
|
||||
startup = root / 'startup'
|
||||
dependency = f'source "{ROOT}/contrib/activities/bash-preexec/bash-preexec.sh"\n' if shell == 'bash' else ''
|
||||
startup.write_text(dependency + f'source "{ROOT}/contrib/activities/ferese.{shell}"\n')
|
||||
master, slave = pty.openpty()
|
||||
env = dict(os.environ, HOME=directory, XDG_CONFIG_HOME=directory, HISTFILE='/dev/null',
|
||||
PATH=str(root) + ':' + os.environ['PATH'], ACTIVITY_TEST_LOG=str(log), TERM='dumb', PS1='ACTIVITY_PROMPT> ')
|
||||
command = (['bash', '--noprofile', '--rcfile', str(startup), '-i'] if shell == 'bash' else
|
||||
['fish', '--no-config', '-i', '--init-command', f'function fish_prompt; printf "ACTIVITY_PROMPT> "; end; source {startup}'])
|
||||
process = subprocess.Popen(command, env=env, stdin=slave, stdout=slave, stderr=slave,
|
||||
start_new_session=True, preexec_fn=lambda: fcntl.ioctl(0, termios.TIOCSCTTY, 0))
|
||||
os.close(slave)
|
||||
trace = b''
|
||||
def wait_for(predicate):
|
||||
nonlocal trace
|
||||
deadline = time.monotonic() + 5
|
||||
while time.monotonic() < deadline:
|
||||
if select.select([master], [], [], .05)[0]:
|
||||
trace += os.read(master, 65536)
|
||||
if predicate():
|
||||
return
|
||||
self.fail(trace.decode(errors='replace'))
|
||||
def calls():
|
||||
return [json.loads(line) for line in log.read_text().splitlines()] if log.exists() else []
|
||||
if shutil.which(shell):
|
||||
with self.subTest(shell=shell):
|
||||
self.check_interactive_commands(shell)
|
||||
|
||||
def test_bash_prompt_composition_and_delayed_installation(self):
|
||||
for prompt in ('scalar', 'array'):
|
||||
for installed in (False, True):
|
||||
for extdebug in (False, True):
|
||||
with self.subTest(prompt=prompt, installed=installed, extdebug=extdebug):
|
||||
self.check_interactive_commands('bash', prompt, installed, extdebug)
|
||||
|
||||
def test_bash_publisher_failure_does_not_skip_the_next_command(self):
|
||||
self.check_interactive_commands('bash', 'array', extdebug=True, publisher_status=71)
|
||||
|
||||
def check_interactive_commands(self, shell, prompt=None, installed=False, extdebug=False, publisher_status=0):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
publisher = root / 'feresectl'
|
||||
publisher.write_text('#!/usr/bin/env python3\nimport json, os, sys\n'
|
||||
'with open(os.environ["ACTIVITY_TEST_LOG"], "a") as f:\n'
|
||||
' f.write(json.dumps(sys.argv[1:]) + "\\n")\n'
|
||||
f'sys.exit({publisher_status})\n')
|
||||
publisher.chmod(0o755)
|
||||
log = root / 'calls'
|
||||
statuses = root / 'statuses'
|
||||
startup = root / 'startup'
|
||||
source_hook = f'source "{ROOT}/contrib/activities/ferese.{shell}"\n'
|
||||
setup = ''
|
||||
if shell == 'bash':
|
||||
if prompt:
|
||||
setup += ('_activity_prompt() { printf "%s\\n" "$?" >> "$ACTIVITY_TEST_STATUS"; }\n'
|
||||
'_activity_precmd() { return 7; }\n'
|
||||
'precmd_functions=(_activity_precmd)\n')
|
||||
setup += ('PROMPT_COMMAND=_activity_prompt\n' if prompt == 'scalar' else
|
||||
'PROMPT_COMMAND=(_activity_prompt :)\n')
|
||||
setup += f'source "{ROOT}/contrib/activities/bash-preexec/bash-preexec.sh"\n'
|
||||
if extdebug:
|
||||
setup += 'shopt -s extdebug\n'
|
||||
if not installed:
|
||||
setup += source_hook * 2
|
||||
startup.write_text(setup)
|
||||
master, slave = pty.openpty()
|
||||
env = dict(os.environ, HOME=directory, XDG_CONFIG_HOME=directory, HISTFILE='/dev/null',
|
||||
PATH=str(root) + ':' + os.environ['PATH'], ACTIVITY_TEST_LOG=str(log),
|
||||
ACTIVITY_TEST_STATUS=str(statuses), TERM='dumb', PS1='ACTIVITY_PROMPT> ')
|
||||
command = (['bash', '--noprofile', '--rcfile', str(startup), '-i'] if shell == 'bash' else
|
||||
['fish', '--no-config', '-i', '--init-command', f'function fish_prompt; printf "ACTIVITY_PROMPT> "; end; source {startup}'])
|
||||
process = subprocess.Popen(command, env=env, stdin=slave, stdout=slave, stderr=slave,
|
||||
start_new_session=True, preexec_fn=lambda: fcntl.ioctl(0, termios.TIOCSCTTY, 0))
|
||||
os.close(slave)
|
||||
trace = b''
|
||||
def wait_for(predicate):
|
||||
nonlocal trace
|
||||
deadline = time.monotonic() + 5
|
||||
while time.monotonic() < deadline:
|
||||
if select.select([master], [], [], .05)[0]:
|
||||
trace += os.read(master, 65536)
|
||||
if predicate():
|
||||
return
|
||||
self.fail(trace.decode(errors='replace'))
|
||||
def calls():
|
||||
return [json.loads(line) for line in log.read_text().splitlines()] if log.exists() else []
|
||||
def sleep_is_foreground():
|
||||
try:
|
||||
# Wait for the first prompt before sending an interactive command.
|
||||
wait_for(lambda: b'# ' in trace or b'$ ' in trace or b'> ' in trace)
|
||||
os.write(master, b'false\n')
|
||||
wait_for(lambda: any(call[:2] == ['activity', 'end'] for call in calls()))
|
||||
records = calls()
|
||||
start = next(call for call in records if call[:2] == ['activity', 'hook'])
|
||||
end = next(call for call in records if call[:2] == ['activity', 'end'])
|
||||
self.assertEqual(start[start.index('--command') + 1], 'false')
|
||||
self.assertEqual(end[2], start[2])
|
||||
self.assertEqual(end[end.index('--result') + 1], 'fail')
|
||||
finally:
|
||||
process.terminate()
|
||||
try:
|
||||
process.wait(timeout=2)
|
||||
except subprocess.TimeoutExpired:
|
||||
process.kill()
|
||||
process.wait()
|
||||
os.close(master)
|
||||
return (Path('/proc') / str(os.tcgetpgrp(master)) / 'comm').read_text().strip() == 'sleep'
|
||||
except FileNotFoundError:
|
||||
return False
|
||||
def next_prompt(command, interrupt=False, publish=True):
|
||||
previous = trace.count(b'ACTIVITY_PROMPT> ')
|
||||
before = len(calls())
|
||||
os.write(master, command.encode() + b'\n')
|
||||
if interrupt:
|
||||
wait_for(sleep_is_foreground)
|
||||
os.write(master, b'\x03')
|
||||
if command and publish:
|
||||
wait_for(lambda: len(calls()) >= before + 2)
|
||||
wait_for(lambda: trace.count(b'ACTIVITY_PROMPT> ') > previous)
|
||||
try:
|
||||
wait_for(lambda: b'ACTIVITY_PROMPT> ' in trace)
|
||||
if installed:
|
||||
# Let bash-preexec finish its delayed installation before sourcing Ferese.
|
||||
next_prompt(source_hook.strip() + '; ' + source_hook.strip(), publish=False)
|
||||
self.assertEqual(calls(), [], 'startup and prompt callbacks are not activities')
|
||||
seen = set()
|
||||
commands = [('false', 'fail', 1)]
|
||||
if shell == 'bash':
|
||||
commands += [
|
||||
("printf 'ACTIVITY_STATUS=%s\\n' \"$?\"", 'ok', 0),
|
||||
('true', 'ok', 0),
|
||||
('true | false', 'fail', 1),
|
||||
('false | true', 'ok', 0),
|
||||
('sleep 30', 'cancel', 130),
|
||||
("sh -c 'exit 143'", 'cancel', 143),
|
||||
('true', 'ok', 0),
|
||||
]
|
||||
for command, result, status in commands:
|
||||
before = len(calls())
|
||||
next_prompt(command, interrupt=command == 'sleep 30')
|
||||
records = calls()[before:]
|
||||
self.assertEqual(len(records), 2, records)
|
||||
start, end = records
|
||||
self.assertEqual(start[:2], ['activity', 'hook'])
|
||||
self.assertEqual(start[start.index('--command') + 1], command)
|
||||
self.assertNotIn(start[2], seen)
|
||||
seen.add(start[2])
|
||||
self.assertEqual(end[:3], ['activity', 'cancel' if result == 'cancel' else 'end', start[2]])
|
||||
if result != 'cancel':
|
||||
self.assertEqual(end[end.index('--result') + 1], result)
|
||||
if command.startswith("printf 'ACTIVITY_STATUS="):
|
||||
self.assertIn(b'\r\nACTIVITY_STATUS=1\r\n', trace)
|
||||
if prompt:
|
||||
self.assertEqual(int(statuses.read_text().splitlines()[-1]), status)
|
||||
next_prompt('')
|
||||
self.assertEqual(calls()[before:], records, 'an empty prompt must not publish another event')
|
||||
finally:
|
||||
process.terminate()
|
||||
try:
|
||||
process.wait(timeout=2)
|
||||
except subprocess.TimeoutExpired:
|
||||
process.kill()
|
||||
process.wait()
|
||||
os.close(master)
|
||||
|
||||
@unittest.skipUnless(shutil.which("fish"), "fish unavailable")
|
||||
def test_fish_preserves_status_and_does_not_execute_metadata(self):
|
||||
|
||||
Reference in new issue
Block a user