summaryrefslogtreecommitdiff
path: root/lib
diff options
context:
space:
mode:
authorMassimiliano Culpo <massimiliano.culpo@gmail.com>2024-08-16 13:14:58 +0200
committerHarmen Stoppels <harmenstoppels@gmail.com>2024-08-16 14:22:59 +0200
commit4f0e336ed09408809df1b570422a64b3e02e9eb2 (patch)
tree79744b0162b4f04d75178b382b0591f2201cda39 /lib
parent64774f30155f701fb1961e11e38ea3af6d8513b3 (diff)
downloadspack-4f0e336ed09408809df1b570422a64b3e02e9eb2.tar.gz
spack-4f0e336ed09408809df1b570422a64b3e02e9eb2.tar.bz2
spack-4f0e336ed09408809df1b570422a64b3e02e9eb2.tar.xz
spack-4f0e336ed09408809df1b570422a64b3e02e9eb2.zip
Remove "test_foreground_background"
Diffstat (limited to 'lib')
-rw-r--r--lib/spack/spack/test/llnl/util/tty/log.py345
1 files changed, 0 insertions, 345 deletions
diff --git a/lib/spack/spack/test/llnl/util/tty/log.py b/lib/spack/spack/test/llnl/util/tty/log.py
index ea03fd2689..61879fc410 100644
--- a/lib/spack/spack/test/llnl/util/tty/log.py
+++ b/lib/spack/spack/test/llnl/util/tty/log.py
@@ -4,19 +4,13 @@
# SPDX-License-Identifier: (Apache-2.0 OR MIT)
import contextlib
-import multiprocessing
-import os
-import signal
import sys
-import time
from types import ModuleType
from typing import Optional
import pytest
-import llnl.util.lang as lang
import llnl.util.tty.log as log
-import llnl.util.tty.pty as pty
from spack.util.executable import which
@@ -173,342 +167,3 @@ def test_log_subproc_and_echo_output_capfd(capfd, tmpdir):
print("logged")
assert capfd.readouterr()[0] == "echo\n"
-
-
-#
-# Tests below use a pseudoterminal to test llnl.util.tty.log
-#
-def simple_logger(**kwargs):
- """Mock logger (minion) process for testing log.keyboard_input."""
- running = [True]
-
- def handler(signum, frame):
- running[0] = False
-
- signal.signal(signal.SIGUSR1, handler)
-
- log_path = kwargs["log_path"]
- with log.log_output(log_path):
- while running[0]:
- print("line")
- time.sleep(1e-3)
-
-
-def mock_shell_fg(proc, ctl, **kwargs):
- """PseudoShell controller function for test_foreground_background."""
- ctl.fg()
- ctl.status()
- ctl.wait_enabled()
-
- os.kill(proc.pid, signal.SIGUSR1)
-
-
-def mock_shell_fg_no_termios(proc, ctl, **kwargs):
- """PseudoShell controller function for test_foreground_background."""
- ctl.fg()
- ctl.status()
- ctl.wait_disabled_fg()
-
- os.kill(proc.pid, signal.SIGUSR1)
-
-
-def mock_shell_bg(proc, ctl, **kwargs):
- """PseudoShell controller function for test_foreground_background."""
- ctl.bg()
- ctl.status()
- ctl.wait_disabled()
-
- os.kill(proc.pid, signal.SIGUSR1)
-
-
-def mock_shell_tstp_cont(proc, ctl, **kwargs):
- """PseudoShell controller function for test_foreground_background."""
- ctl.tstp()
- ctl.wait_stopped()
-
- ctl.cont()
- ctl.wait_running()
-
- os.kill(proc.pid, signal.SIGUSR1)
-
-
-def mock_shell_tstp_tstp_cont(proc, ctl, **kwargs):
- """PseudoShell controller function for test_foreground_background."""
- ctl.tstp()
- ctl.wait_stopped()
-
- ctl.tstp()
- ctl.wait_stopped()
-
- ctl.cont()
- ctl.wait_running()
-
- os.kill(proc.pid, signal.SIGUSR1)
-
-
-def mock_shell_tstp_tstp_cont_cont(proc, ctl, **kwargs):
- """PseudoShell controller function for test_foreground_background."""
- ctl.tstp()
- ctl.wait_stopped()
-
- ctl.tstp()
- ctl.wait_stopped()
-
- ctl.cont()
- ctl.wait_running()
-
- ctl.cont()
- ctl.wait_running()
-
- os.kill(proc.pid, signal.SIGUSR1)
-
-
-def mock_shell_bg_fg(proc, ctl, **kwargs):
- """PseudoShell controller function for test_foreground_background."""
- ctl.bg()
- ctl.status()
- ctl.wait_disabled()
-
- ctl.fg()
- ctl.status()
- ctl.wait_enabled()
-
- os.kill(proc.pid, signal.SIGUSR1)
-
-
-def mock_shell_bg_fg_no_termios(proc, ctl, **kwargs):
- """PseudoShell controller function for test_foreground_background."""
- ctl.bg()
- ctl.status()
- ctl.wait_disabled()
-
- ctl.fg()
- ctl.status()
- ctl.wait_disabled_fg()
-
- os.kill(proc.pid, signal.SIGUSR1)
-
-
-def mock_shell_fg_bg(proc, ctl, **kwargs):
- """PseudoShell controller function for test_foreground_background."""
- ctl.fg()
- ctl.status()
- ctl.wait_enabled()
-
- ctl.bg()
- ctl.status()
- ctl.wait_disabled()
-
- os.kill(proc.pid, signal.SIGUSR1)
-
-
-def mock_shell_fg_bg_no_termios(proc, ctl, **kwargs):
- """PseudoShell controller function for test_foreground_background."""
- ctl.fg()
- ctl.status()
- ctl.wait_disabled_fg()
-
- ctl.bg()
- ctl.status()
- ctl.wait_disabled()
-
- os.kill(proc.pid, signal.SIGUSR1)
-
-
-@contextlib.contextmanager
-def no_termios():
- saved = log.termios
- log.termios = None
- try:
- yield
- finally:
- log.termios = saved
-
-
-@pytest.mark.skipif(not which("ps"), reason="requires ps utility")
-@pytest.mark.skipif(not termios, reason="requires termios support")
-@pytest.mark.parametrize(
- "test_fn,termios_on_or_off",
- [
- # tests with termios
- (mock_shell_fg, lang.nullcontext),
- (mock_shell_bg, lang.nullcontext),
- (mock_shell_bg_fg, lang.nullcontext),
- (mock_shell_fg_bg, lang.nullcontext),
- (mock_shell_tstp_cont, lang.nullcontext),
- (mock_shell_tstp_tstp_cont, lang.nullcontext),
- (mock_shell_tstp_tstp_cont_cont, lang.nullcontext),
- # tests without termios
- (mock_shell_fg_no_termios, no_termios),
- (mock_shell_bg, no_termios),
- (mock_shell_bg_fg_no_termios, no_termios),
- (mock_shell_fg_bg_no_termios, no_termios),
- (mock_shell_tstp_cont, no_termios),
- (mock_shell_tstp_tstp_cont, no_termios),
- (mock_shell_tstp_tstp_cont_cont, no_termios),
- ],
-)
-@pytest.mark.skip(reason="Fails almost consistently when run with coverage and xdist")
-def test_foreground_background(test_fn, termios_on_or_off, tmpdir):
- """Functional tests for foregrounding and backgrounding a logged process.
-
- This ensures that things like SIGTTOU are not raised and that
- terminal settings are corrected on foreground/background and on
- process stop and start.
-
- """
- shell = pty.PseudoShell(test_fn, simple_logger)
- log_path = str(tmpdir.join("log.txt"))
-
- # run the shell test
- with termios_on_or_off():
- shell.start(log_path=log_path, debug=True)
- exitcode = shell.join()
-
- # processes completed successfully
- assert exitcode == 0
-
- # assert log was created
- assert os.path.exists(log_path)
-
-
-def synchronized_logger(**kwargs):
- """Mock logger (minion) process for testing log.keyboard_input.
-
- This logger synchronizes with the parent process to test that 'v' can
- toggle output. It is used in ``test_foreground_background_output`` below.
-
- """
- running = [True]
-
- def handler(signum, frame):
- running[0] = False
-
- signal.signal(signal.SIGUSR1, handler)
-
- log_path = kwargs["log_path"]
- write_lock = kwargs["write_lock"]
- v_lock = kwargs["v_lock"]
-
- sys.stderr.write(os.getcwd() + "\n")
- with log.log_output(log_path) as logger:
- with logger.force_echo():
- print("forced output")
-
- while running[0]:
- with write_lock:
- if v_lock.acquire(False): # non-blocking acquire
- print("off")
- v_lock.release()
- else:
- print("on") # lock held; v is toggled on
- time.sleep(1e-2)
-
-
-def mock_shell_v_v(proc, ctl, **kwargs):
- """Controller function for test_foreground_background_output."""
- write_lock = kwargs["write_lock"]
- v_lock = kwargs["v_lock"]
-
- ctl.fg()
- ctl.wait_enabled()
- time.sleep(0.1)
-
- write_lock.acquire() # suspend writing
- v_lock.acquire() # enable v lock
- ctl.write(b"v") # toggle v on stdin
- time.sleep(0.1)
- write_lock.release() # resume writing
-
- time.sleep(0.1)
-
- write_lock.acquire() # suspend writing
- ctl.write(b"v") # toggle v on stdin
- time.sleep(0.1)
- v_lock.release() # disable v lock
- write_lock.release() # resume writing
- time.sleep(0.1)
-
- os.kill(proc.pid, signal.SIGUSR1)
-
-
-def mock_shell_v_v_no_termios(proc, ctl, **kwargs):
- """Controller function for test_foreground_background_output."""
- write_lock = kwargs["write_lock"]
- v_lock = kwargs["v_lock"]
-
- ctl.fg()
- ctl.wait_disabled_fg()
- time.sleep(0.1)
-
- write_lock.acquire() # suspend writing
- v_lock.acquire() # enable v lock
- ctl.write(b"v\n") # toggle v on stdin
- time.sleep(0.1)
- write_lock.release() # resume writing
-
- time.sleep(0.1)
-
- write_lock.acquire() # suspend writing
- ctl.write(b"v\n") # toggle v on stdin
- time.sleep(0.1)
- v_lock.release() # disable v lock
- write_lock.release() # resume writing
- time.sleep(0.1)
-
- os.kill(proc.pid, signal.SIGUSR1)
-
-
-@pytest.mark.skipif(not which("ps"), reason="requires ps utility")
-@pytest.mark.skipif(not termios, reason="requires termios support")
-@pytest.mark.parametrize(
- "test_fn,termios_on_or_off",
- [(mock_shell_v_v, lang.nullcontext), (mock_shell_v_v_no_termios, no_termios)],
-)
-@pytest.mark.skip(reason="Fails almost consistently when run with coverage and xdist")
-def test_foreground_background_output(test_fn, capfd, termios_on_or_off, tmpdir):
- """Tests hitting 'v' toggles output, and that force_echo works."""
- if sys.version_info >= (3, 8) and sys.platform == "darwin" and termios_on_or_off == no_termios:
- return
-
- shell = pty.PseudoShell(test_fn, synchronized_logger)
- log_path = str(tmpdir.join("log.txt"))
-
- # Locks for synchronizing with minion
- write_lock = multiprocessing.Lock() # must be held by minion to write
- v_lock = multiprocessing.Lock() # held while controller is in v mode
-
- with termios_on_or_off():
- shell.start(write_lock=write_lock, v_lock=v_lock, debug=True, log_path=log_path)
-
- exitcode = shell.join()
- out, err = capfd.readouterr()
- print(err) # will be shown if something goes wrong
- print(out)
-
- # processes completed successfully
- assert exitcode == 0
-
- # split output into lines
- output = out.strip().split("\n")
-
- # also get lines of log file
- assert os.path.exists(log_path)
- with open(log_path) as logfile:
- log_data = logfile.read().strip().split("\n")
-
- # Controller and minion process coordinate with locks such that the
- # minion writes "off" when echo is off, and "on" when echo is on. The
- # output should contain mostly "on" lines, but may contain "off"
- # lines if the controller is slow. The important thing to observe
- # here is that we started seeing 'on' in the end.
- assert ["forced output", "on"] == lang.uniq(output) or [
- "forced output",
- "off",
- "on",
- ] == lang.uniq(output)
-
- # log should be off for a while, then on, then off
- assert ["forced output", "off", "on", "off"] == lang.uniq(log_data) and log_data.count(
- "off"
- ) > 2 # ensure some "off" lines were omitted