mirror of
https://github.com/SCST-project/scst.git
synced 2026-10-22 06:02:33 +00:00
scripts: Share analyzer process and raw-log handling
Provide common process, progress and report helpers for the upcoming analyzer drivers. Preserve raw output bytes and command failure statuses, keep progress separate from captured diagnostics, and terminate the owned process group when an invocation is interrupted. Forward valid inherited Make jobserver pipes without taking their tokens. Document named-log truncation explicitly: a later run replaces that named report unless the caller requests preservation. Import the process and terminal fixtures and add raw-log coverage. No analyzer-specific filtering or diagnostic exclusions are introduced.
This commit is contained in:
Gleb Chesnokov
@@ -92,6 +92,18 @@ They cannot be combined with a diff base. Both a path and its resolved symlink
|
||||
destination must be in scope. Results are sorted and deduplicated.
|
||||
An empty selection means there are no eligible inputs, not a completed scan.
|
||||
|
||||
## Shared process and log helpers
|
||||
|
||||
`lint_common.py` handles process-group cleanup, exit statuses and Make's
|
||||
parallel jobserver. Drivers must check command results or use `check=True`.
|
||||
The entry-point helper supports command tracing with `V=1` or `--verbose`.
|
||||
Progress is separate from captured analyzer output.
|
||||
|
||||
Drivers save complete stdout/stderr under `.cache/lint/`. The helpers preserve
|
||||
raw bytes without filtering. Reusing a log name replaces the previous report
|
||||
by default; callers can preserve it with `truncate=False`. There is no automatic
|
||||
log archive.
|
||||
|
||||
## Dependencies and tooling tests
|
||||
|
||||
Shared analyzer tooling requires Linux, Git and Python 3.10+, using only the
|
||||
|
||||
@@ -0,0 +1,172 @@
|
||||
"""Process and report helpers for the analyzer drivers."""
|
||||
|
||||
import contextlib
|
||||
import os
|
||||
import re
|
||||
import shlex
|
||||
import shutil
|
||||
import signal
|
||||
import stat
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
PROGRESS_INTERVAL = 5
|
||||
SPINNER_INTERVAL = 0.1
|
||||
SPINNER_FRAMES = "|/-\\"
|
||||
VERBOSE = os.environ.get("V", "0") not in {"", "0"}
|
||||
|
||||
|
||||
def show_log(path, label="Full log"):
|
||||
print(f" {label}: {Path(path).resolve()}", file=sys.stderr, flush=True)
|
||||
|
||||
|
||||
class Progress:
|
||||
"""Animate only a quiet child; redirected output gets periodic plain lines."""
|
||||
|
||||
def __init__(self, label, *, quiet=True):
|
||||
self.label = label
|
||||
self.started = time.monotonic()
|
||||
self.animated = (bool(label) and quiet and sys.stderr.isatty()
|
||||
and os.environ.get("TERM") != "dumb")
|
||||
self.interval = SPINNER_INTERVAL if self.animated else PROGRESS_INTERVAL
|
||||
self.frame = 0
|
||||
self.visible = False
|
||||
|
||||
def update(self):
|
||||
elapsed = int(time.monotonic() - self.started)
|
||||
if self.animated:
|
||||
elapsed -= elapsed % PROGRESS_INTERVAL
|
||||
frame = SPINNER_FRAMES[self.frame % len(SPINNER_FRAMES)]
|
||||
text = f" {frame} {self.label} {elapsed // 60:02}:{elapsed % 60:02}"
|
||||
width = shutil.get_terminal_size().columns
|
||||
print("\r\033[2K" + text[:max(1, width - 1)], end="", file=sys.stderr, flush=True)
|
||||
self.frame += 1
|
||||
else:
|
||||
print(f" {self.label}: running ({elapsed}s elapsed)", file=sys.stderr, flush=True)
|
||||
self.visible = True
|
||||
|
||||
def clear(self):
|
||||
if self.animated and self.visible:
|
||||
print("\r\033[2K", end="", file=sys.stderr, flush=True)
|
||||
|
||||
def finish(self, status):
|
||||
self.clear()
|
||||
if self.visible:
|
||||
elapsed = int(time.monotonic() - self.started)
|
||||
print(f" {self.label}: finished ({elapsed}s, exit {status})",
|
||||
file=sys.stderr, flush=True)
|
||||
|
||||
|
||||
def jobserver_fds(env):
|
||||
"""Forward an inherited pipe jobserver without acquiring its tokens."""
|
||||
matches = re.findall(r"--jobserver-(?:auth|fds)=(\S+)", env.get("MAKEFLAGS", ""))
|
||||
if not matches or not re.fullmatch(r"\d+,\d+", matches[-1]):
|
||||
return ()
|
||||
descriptors = tuple(map(int, matches[-1].split(",")))
|
||||
try:
|
||||
for descriptor in descriptors:
|
||||
if (not os.get_inheritable(descriptor)
|
||||
or not stat.S_ISFIFO(os.fstat(descriptor).st_mode)):
|
||||
return ()
|
||||
except OSError:
|
||||
return ()
|
||||
return descriptors
|
||||
|
||||
|
||||
def start(command, *, env=None, **kwargs):
|
||||
environment = os.environ if env is None else env
|
||||
if VERBOSE:
|
||||
print(" $ " + shlex.join(map(str, command)), file=sys.stderr, flush=True)
|
||||
return subprocess.Popen(command, env=environment, start_new_session=True,
|
||||
pass_fds=jobserver_fds(environment), **kwargs)
|
||||
|
||||
|
||||
def stop(process):
|
||||
"""Stop the owned process group, including children of an exited driver."""
|
||||
with contextlib.suppress(ProcessLookupError):
|
||||
os.killpg(process.pid, signal.SIGTERM)
|
||||
try:
|
||||
process.wait(timeout=2)
|
||||
except subprocess.TimeoutExpired:
|
||||
pass
|
||||
with contextlib.suppress(ProcessLookupError):
|
||||
os.killpg(process.pid, signal.SIGKILL)
|
||||
process.wait()
|
||||
|
||||
|
||||
def exit_status(returncode):
|
||||
return returncode if returncode >= 0 else 128 - returncode
|
||||
|
||||
|
||||
def run(command, *, input=None, check=False, progress=None, **kwargs):
|
||||
if input is not None:
|
||||
kwargs["stdin"] = subprocess.PIPE
|
||||
process = start(command, **kwargs)
|
||||
indicator = Progress(progress, quiet=kwargs.get("stdout") is not None
|
||||
and kwargs.get("stderr") is not None)
|
||||
try:
|
||||
if indicator.animated:
|
||||
indicator.update()
|
||||
while True:
|
||||
try:
|
||||
stdout, stderr = process.communicate(
|
||||
input, timeout=indicator.interval if progress else None)
|
||||
break
|
||||
except subprocess.TimeoutExpired:
|
||||
# communicate retains pending input and output across retries.
|
||||
input = None
|
||||
indicator.update()
|
||||
except BaseException:
|
||||
indicator.clear()
|
||||
stop(process)
|
||||
raise
|
||||
result = subprocess.CompletedProcess(command, exit_status(process.returncode), stdout, stderr)
|
||||
indicator.finish(result.returncode)
|
||||
if check:
|
||||
result.check_returncode()
|
||||
return result
|
||||
|
||||
|
||||
def log_path(root, name, *, truncate=True):
|
||||
path = root / ".cache/lint" / name
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
if truncate:
|
||||
path.write_bytes(b"")
|
||||
return path
|
||||
|
||||
|
||||
def copy_output(path, output=None):
|
||||
destination = output if output is not None else sys.stderr.buffer
|
||||
with Path(path).open("rb") as stream:
|
||||
shutil.copyfileobj(stream, destination)
|
||||
destination.flush()
|
||||
|
||||
|
||||
class Interrupted(Exception):
|
||||
def __init__(self, signum):
|
||||
self.signum = signum
|
||||
|
||||
|
||||
def entrypoint(main):
|
||||
global VERBOSE
|
||||
VERBOSE = VERBOSE or "--verbose" in sys.argv[1:]
|
||||
sys.argv[1:] = [arg for arg in sys.argv[1:] if arg != "--verbose"]
|
||||
|
||||
def interrupt(signum, _frame):
|
||||
raise Interrupted(signum)
|
||||
|
||||
signal.signal(signal.SIGTERM, interrupt)
|
||||
try:
|
||||
return main()
|
||||
except Interrupted as error:
|
||||
return 128 + error.signum
|
||||
except KeyboardInterrupt:
|
||||
return 130
|
||||
except subprocess.CalledProcessError as error:
|
||||
print(f"Command failed ({error.returncode}): {error.cmd}", file=sys.stderr)
|
||||
return error.returncode
|
||||
except (OSError, ValueError) as error:
|
||||
print(str(error), file=sys.stderr)
|
||||
return 2
|
||||
@@ -0,0 +1,279 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Check process, progress and raw-log contracts shared by analyzer drivers."""
|
||||
|
||||
import contextlib
|
||||
import io
|
||||
import os
|
||||
import pty
|
||||
import selectors
|
||||
import signal
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.dont_write_bytecode = True
|
||||
SCRIPTS = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(SCRIPTS))
|
||||
from lint_common import Progress, copy_output, jobserver_fds, log_path, run # noqa: E402
|
||||
|
||||
|
||||
class LogTest(unittest.TestCase):
|
||||
def test_log_creation_append_and_reset_preserve_raw_bytes(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
log = log_path(root, "fixture.log")
|
||||
self.assertEqual(log, root / ".cache/lint/fixture.log")
|
||||
self.assertEqual(log.read_bytes(), b"")
|
||||
raw = b"diagnostic\x00\xff\r\n"
|
||||
log.write_bytes(raw)
|
||||
self.assertEqual(log_path(root, "fixture.log", truncate=False), log)
|
||||
output = io.BytesIO()
|
||||
copy_output(log, output)
|
||||
self.assertEqual(output.getvalue(), raw)
|
||||
self.assertEqual(log.read_bytes(), raw)
|
||||
self.assertEqual(log_path(root, "fixture.log"), log)
|
||||
self.assertEqual(log.read_bytes(), b"")
|
||||
|
||||
|
||||
class ProcessTest(unittest.TestCase):
|
||||
def test_progress_is_live_and_preserves_piped_data_and_failure(self):
|
||||
for check in (False, True):
|
||||
with self.subTest(check=check), tempfile.TemporaryDirectory() as directory:
|
||||
release = Path(directory) / "release"
|
||||
|
||||
class Progress(io.StringIO):
|
||||
def write(self, text):
|
||||
size = super().write(text)
|
||||
if self.getvalue().count("running (") >= 2:
|
||||
release.touch()
|
||||
return size
|
||||
|
||||
script = """
|
||||
import os, sys, time
|
||||
from pathlib import Path
|
||||
payload = sys.stdin.buffer.read()
|
||||
deadline = time.monotonic() + 5
|
||||
while not Path(sys.argv[1]).exists():
|
||||
if time.monotonic() > deadline:
|
||||
sys.exit(99)
|
||||
time.sleep(0.01)
|
||||
os.write(1, payload)
|
||||
os.write(2, b'diagnostic\\x00\\xff')
|
||||
sys.exit(7)
|
||||
"""
|
||||
payload = b"input\x00\xff" * 20000
|
||||
progress = Progress()
|
||||
with patch("lint_common.PROGRESS_INTERVAL", 0.02), \
|
||||
contextlib.redirect_stderr(progress):
|
||||
try:
|
||||
result = run([sys.executable, "-c", script, str(release)],
|
||||
input=payload, stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE, progress="Fixture", check=check)
|
||||
except subprocess.CalledProcessError as error:
|
||||
self.assertTrue(check)
|
||||
result = error
|
||||
else:
|
||||
self.assertFalse(check)
|
||||
self.assertEqual(result.returncode, 7)
|
||||
self.assertEqual(result.stdout, payload)
|
||||
self.assertEqual(result.stderr, b"diagnostic\x00\xff")
|
||||
self.assertGreaterEqual(progress.getvalue().count("Fixture: running ("), 2)
|
||||
|
||||
def test_fast_process_does_not_print_periodic_progress(self):
|
||||
progress = io.StringIO()
|
||||
with contextlib.redirect_stderr(progress):
|
||||
result = run([sys.executable, "-c", "pass"], progress="Fixture")
|
||||
self.assertEqual(result.returncode, 0)
|
||||
self.assertEqual(progress.getvalue(), "")
|
||||
|
||||
def test_terminal_spinner_is_live_and_separate_from_child_output(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
release = Path(directory) / "release"
|
||||
child = """
|
||||
import os, sys, time
|
||||
from pathlib import Path
|
||||
deadline = time.monotonic() + 5
|
||||
while not Path(sys.argv[1]).exists():
|
||||
if time.monotonic() > deadline:
|
||||
sys.exit(99)
|
||||
time.sleep(0.01)
|
||||
os.write(1, b'raw\\x00\\xff')
|
||||
os.write(2, b'diagnostic\\n')
|
||||
sys.exit(7)
|
||||
"""
|
||||
script = """
|
||||
import subprocess, sys
|
||||
sys.path.insert(0, sys.argv[1])
|
||||
from lint_common import entrypoint, run
|
||||
def main():
|
||||
result = run([sys.executable, '-c', sys.argv[2], sys.argv[3]],
|
||||
stdout=subprocess.PIPE, stderr=subprocess.PIPE, progress='Fixture')
|
||||
sys.stdout.buffer.write(result.stdout + result.stderr)
|
||||
return result.returncode
|
||||
sys.exit(entrypoint(main))
|
||||
"""
|
||||
master, slave = pty.openpty()
|
||||
process = subprocess.Popen(
|
||||
[sys.executable, "-B", "-c", script, str(SCRIPTS), child, str(release)],
|
||||
stdout=subprocess.PIPE, stderr=slave, env=dict(os.environ, TERM="xterm"))
|
||||
os.close(slave)
|
||||
display = b""
|
||||
try:
|
||||
with selectors.DefaultSelector() as selector:
|
||||
selector.register(master, selectors.EVENT_READ)
|
||||
deadline = time.monotonic() + 5
|
||||
while display.count(b"\r\x1b[2K") < 2:
|
||||
self.assertGreater(deadline - time.monotonic(), 0, display)
|
||||
if selector.select(timeout=0.1):
|
||||
display += os.read(master, 4096)
|
||||
self.assertIsNone(process.poll())
|
||||
release.touch()
|
||||
stdout, _ = process.communicate(timeout=5)
|
||||
while selector.select(timeout=0.1):
|
||||
try:
|
||||
chunk = os.read(master, 4096)
|
||||
except OSError:
|
||||
break # A closed PTY reports EIO on Linux.
|
||||
if not chunk:
|
||||
break
|
||||
display += chunk
|
||||
self.assertEqual(process.returncode, 7, display)
|
||||
self.assertEqual(stdout, b"raw\x00\xffdiagnostic\n")
|
||||
self.assertIn(b"| Fixture", display)
|
||||
self.assertIn(b"/ Fixture", display)
|
||||
self.assertIn(b"exit 7)", display)
|
||||
self.assertNotIn(b"running (", display)
|
||||
finally:
|
||||
release.touch()
|
||||
if process.poll() is None:
|
||||
process.terminate()
|
||||
process.communicate(timeout=5)
|
||||
os.close(master)
|
||||
|
||||
def test_plain_progress_updates_every_five_seconds(self):
|
||||
stream = io.StringIO()
|
||||
with contextlib.redirect_stderr(stream), \
|
||||
patch("lint_common.time.monotonic", side_effect=(10, 15)):
|
||||
progress = Progress("Fixture")
|
||||
self.assertEqual(progress.interval, 5)
|
||||
progress.update()
|
||||
self.assertEqual(stream.getvalue(), " Fixture: running (5s elapsed)\n")
|
||||
|
||||
def test_dumb_terminal_does_not_receive_escape_sequences(self):
|
||||
stream = io.StringIO()
|
||||
with patch.object(stream, "isatty", return_value=True), \
|
||||
patch.dict(os.environ, TERM="dumb"), contextlib.redirect_stderr(stream):
|
||||
progress = Progress("Fixture")
|
||||
progress.update()
|
||||
progress.finish(0)
|
||||
self.assertNotIn("\r", stream.getvalue())
|
||||
self.assertNotIn("\033", stream.getvalue())
|
||||
|
||||
def test_interrupted_spinner_clears_its_line_and_stops_child(self):
|
||||
stream = io.StringIO()
|
||||
with patch.object(stream, "isatty", return_value=True), \
|
||||
patch.dict(os.environ, TERM="xterm"), contextlib.redirect_stderr(stream), \
|
||||
patch("lint_common.start") as start, patch("lint_common.stop") as stop:
|
||||
start.return_value.communicate.side_effect = KeyboardInterrupt
|
||||
with self.assertRaises(KeyboardInterrupt):
|
||||
run(["fixture"], stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
||||
progress="Fixture")
|
||||
stop.assert_called_once_with(start.return_value)
|
||||
self.assertTrue(stream.getvalue().endswith("\r\033[2K"))
|
||||
|
||||
def test_inherited_pipe_jobserver_survives_python_driver(self):
|
||||
reader, writer = os.pipe()
|
||||
try:
|
||||
os.set_inheritable(reader, True)
|
||||
os.set_inheritable(writer, True)
|
||||
env = dict(os.environ, MAKEFLAGS=f"-j --jobserver-auth={reader},{writer}")
|
||||
script = ("import os,sys; r,w=map(int,sys.argv[1:]); "
|
||||
"os.write(w,b'token'); assert os.read(r,5)==b'token'")
|
||||
result = run([sys.executable, "-c", script, str(reader), str(writer)],
|
||||
env=env, stderr=subprocess.PIPE)
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
# FIFO jobservers must not forward a superseded pipe specification.
|
||||
env["MAKEFLAGS"] += " --jobserver-auth=fifo:/tmp/make-test"
|
||||
self.assertEqual(jobserver_fds(env), ())
|
||||
finally:
|
||||
os.close(reader)
|
||||
os.close(writer)
|
||||
|
||||
def test_stale_jobserver_numbers_do_not_forward_reused_log_descriptors(self):
|
||||
with tempfile.TemporaryFile() as first, tempfile.TemporaryFile() as second:
|
||||
env = {"MAKEFLAGS": f"--jobserver-auth={first.fileno()},{second.fileno()}"}
|
||||
self.assertEqual(jobserver_fds(env), ())
|
||||
|
||||
def test_child_signal_keeps_shell_exit_status_and_raw_log(self):
|
||||
with tempfile.TemporaryFile() as output:
|
||||
result = run([sys.executable, "-c",
|
||||
"import os,signal; os.write(1,b'out\\xff\\n'); "
|
||||
"os.write(2,b'err\\r\\n'); os.kill(os.getpid(),signal.SIGTERM)"],
|
||||
stdout=output, stderr=subprocess.STDOUT, progress="Fixture")
|
||||
self.assertEqual(result.returncode, 143)
|
||||
output.seek(0)
|
||||
self.assertEqual(output.read(), b"out\xff\nerr\r\n")
|
||||
|
||||
def test_terminating_driver_also_stops_child_and_grandchild(self):
|
||||
with tempfile.TemporaryDirectory(prefix="scst-lint-signal.") as directory:
|
||||
root = Path(directory)
|
||||
grandchild = root / "grandchild.py"
|
||||
grandchild.write_text("""
|
||||
import signal, sys, time
|
||||
from pathlib import Path
|
||||
root = Path(sys.argv[1])
|
||||
def stop(signum, frame):
|
||||
(root / 'grandchild-stopped').touch()
|
||||
sys.exit(0)
|
||||
signal.signal(signal.SIGTERM, stop)
|
||||
(root / 'grandchild-ready').touch()
|
||||
while True:
|
||||
time.sleep(1)
|
||||
""")
|
||||
child = root / "child.py"
|
||||
child.write_text("""
|
||||
import signal, subprocess, sys
|
||||
from pathlib import Path
|
||||
root = Path(sys.argv[1])
|
||||
process = subprocess.Popen([sys.executable, str(root / 'grandchild.py'), str(root)])
|
||||
stopping = False
|
||||
def stop(signum, frame):
|
||||
global stopping
|
||||
stopping = True
|
||||
signal.signal(signal.SIGTERM, stop)
|
||||
(root / 'child-ready').touch()
|
||||
process.wait()
|
||||
if stopping:
|
||||
(root / 'child-stopped').touch()
|
||||
""")
|
||||
script = ("import sys; sys.path.insert(0, sys.argv[1]); "
|
||||
"from lint_common import entrypoint, run; "
|
||||
"sys.exit(entrypoint(lambda: run([sys.executable,sys.argv[2],"
|
||||
"sys.argv[3]], progress='Fixture').returncode))")
|
||||
process = subprocess.Popen([sys.executable, "-B", "-c", script,
|
||||
str(SCRIPTS), str(child), str(root)],
|
||||
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
try:
|
||||
deadline = time.monotonic() + 5
|
||||
while (not all((root / name).exists() for name in
|
||||
("child-ready", "grandchild-ready")) and process.poll() is None):
|
||||
if time.monotonic() > deadline:
|
||||
self.fail("grandchild did not start")
|
||||
time.sleep(0.01)
|
||||
process.send_signal(signal.SIGTERM)
|
||||
stdout, stderr = process.communicate(timeout=5)
|
||||
self.assertEqual(process.returncode, 143, stdout + stderr)
|
||||
self.assertTrue((root / "child-stopped").exists())
|
||||
self.assertTrue((root / "grandchild-stopped").exists())
|
||||
finally:
|
||||
if process.poll() is None:
|
||||
process.send_signal(signal.SIGTERM)
|
||||
process.communicate(timeout=5)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user