chore: don't use vendored copy of Subplot lib/runcmd.py

Use the one provided by Subplot. It has improvements.

Sponsored-by: author

2 files changed, 1 insertions, 247 deletions

diff --git a/subplot.md b/subplot.md
index 6e1a4d4..9a01900 100644
--- a/subplot.md
+++ b/subplot.md
@@ -73,4 +73,5 @@ bindings:
 - subplot.yaml
 functions:
 - subplot.py
+- lib/runcmd.py
 ...
diff --git a/subplot/runcmd.py b/subplot/runcmd.py
deleted file mode 100644
index 2bf153c..0000000
--- a/subplot/runcmd.py
+++ /dev/null
@@ -1,247 +0,0 @@
-import logging
-import os
-import re
-import shlex
-import subprocess
-
-
-#
-# Helper functions.
-#
-
-# Get exit code or other stored data about the latest command run by
-# runcmd_run.
-
-
-def _runcmd_get(ctx, name):
-    ns = ctx.declare("_runcmd")
-    return ns[name]
-
-
-def runcmd_get_exit_code(ctx):
-    return _runcmd_get(ctx, "exit")
-
-
-def runcmd_get_stdout(ctx):
-    return _runcmd_get(ctx, "stdout")
-
-
-def runcmd_get_stdout_raw(ctx):
-    return _runcmd_get(ctx, "stdout.raw")
-
-
-def runcmd_get_stderr(ctx):
-    return _runcmd_get(ctx, "stderr")
-
-
-def runcmd_get_stderr_raw(ctx):
-    return _runcmd_get(ctx, "stderr.raw")
-
-
-def runcmd_get_argv(ctx):
-    return _runcmd_get(ctx, "argv")
-
-
-# Run a command, given an argv and other arguments for subprocess.Popen.
-#
-# This is meant to be a helper function, not bound directly to a step. The
-# stdout, stderr, and exit code are stored in the "_runcmd" namespace in the
-# ctx context.
-def runcmd_run(ctx, argv, **kwargs):
-    ns = ctx.declare("_runcmd")
-
-    env = dict(os.environ)
-    for key, arg in kwargs.pop("env", {}).items():
-        env[key] = arg
-
-    pp = ns.get("path-prefix")
-    if pp:
-        env["PATH"] = pp + ":" + env["PATH"]
-
-    logging.debug(f"runcmd_run")
-    logging.debug(f"  argv: {argv}")
-    logging.debug(f"  env: {env}")
-    p = subprocess.Popen(
-        argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env, **kwargs
-    )
-    stdout, stderr = p.communicate("")
-    ns["argv"] = argv
-    ns["stdout.raw"] = stdout
-    ns["stderr.raw"] = stderr
-    ns["stdout"] = stdout.decode("utf-8")
-    ns["stderr"] = stderr.decode("utf-8")
-    ns["exit"] = p.returncode
-    logging.debug(f"  ctx: {ctx}")
-    logging.debug(f"  ns: {ns}")
-
-
-# Step: prepend srcdir to PATH whenever runcmd runs a command.
-def runcmd_helper_srcdir_path(ctx):
-    srcdir = globals()["srcdir"]
-    runcmd_prepend_to_path(ctx, srcdir)
-
-
-# Step: This creates a helper script.
-def runcmd_helper_script(ctx, filename=None):
-    get_file = globals()["get_file"]
-    with open(filename, "wb") as f:
-        f.write(get_file(filename))
-
-
-#
-# Step functions for running commands.
-#
-
-
-def runcmd_prepend_to_path(ctx, dirname=None):
-    ns = ctx.declare("_runcmd")
-    pp = ns.get("path-prefix", "")
-    if pp:
-        pp = f"{pp}:{dirname}"
-    else:
-        pp = dirname
-    ns["path-prefix"] = pp
-
-
-def runcmd_step(ctx, argv0=None, args=None):
-    runcmd_try_to_run(ctx, argv0=argv0, args=args)
-    runcmd_exit_code_is_zero(ctx)
-
-
-def runcmd_try_to_run(ctx, argv0=None, args=None):
-    argv = [shlex.quote(argv0)] + shlex.split(args)
-    runcmd_run(ctx, argv)
-
-
-#
-# Step functions for examining exit codes.
-#
-
-
-def runcmd_exit_code_is_zero(ctx):
-    runcmd_exit_code_is(ctx, exit=0)
-
-
-def runcmd_exit_code_is(ctx, exit=None):
-    assert_eq = globals()["assert_eq"]
-    assert_eq(runcmd_get_exit_code(ctx), int(exit))
-
-
-def runcmd_exit_code_is_nonzero(ctx):
-    runcmd_exit_code_is_not(ctx, exit=0)
-
-
-def runcmd_exit_code_is_not(ctx, exit=None):
-    assert_ne = globals()["assert_ne"]
-    assert_ne(runcmd_get_exit_code(ctx), int(exit))
-
-
-#
-# Step functions and helpers for examining output in various ways.
-#
-
-
-def runcmd_stdout_is(ctx, text=None):
-    _runcmd_output_is(runcmd_get_stdout(ctx), text)
-
-
-def runcmd_stdout_isnt(ctx, text=None):
-    _runcmd_output_isnt(runcmd_get_stdout(ctx), text)
-
-
-def runcmd_stderr_is(ctx, text=None):
-    _runcmd_output_is(runcmd_get_stderr(ctx), text)
-
-
-def runcmd_stderr_isnt(ctx, text=None):
-    _runcmd_output_isnt(runcmd_get_stderr(ctx), text)
-
-
-def _runcmd_output_is(actual, wanted):
-    assert_eq = globals()["assert_eq"]
-    wanted = bytes(wanted, "utf8").decode("unicode_escape")
-    logging.debug("_runcmd_output_is:")
-    logging.debug(f"  actual: {actual!r}")
-    logging.debug(f"  wanted: {wanted!r}")
-    assert_eq(actual, wanted)
-
-
-def _runcmd_output_isnt(actual, wanted):
-    assert_ne = globals()["assert_ne"]
-    wanted = bytes(wanted, "utf8").decode("unicode_escape")
-    logging.debug("_runcmd_output_isnt:")
-    logging.debug(f"  actual: {actual!r}")
-    logging.debug(f"  wanted: {wanted!r}")
-    assert_ne(actual, wanted)
-
-
-def runcmd_stdout_contains(ctx, text=None):
-    _runcmd_output_contains(runcmd_get_stdout(ctx), text)
-
-
-def runcmd_stdout_doesnt_contain(ctx, text=None):
-    _runcmd_output_doesnt_contain(runcmd_get_stdout(ctx), text)
-
-
-def runcmd_stderr_contains(ctx, text=None):
-    _runcmd_output_contains(runcmd_get_stderr(ctx), text)
-
-
-def runcmd_stderr_doesnt_contain(ctx, text=None):
-    _runcmd_output_doesnt_contain(runcmd_get_stderr(ctx), text)
-
-
-def _runcmd_output_contains(actual, wanted):
-    assert_eq = globals()["assert_eq"]
-    wanted = bytes(wanted, "utf8").decode("unicode_escape")
-    logging.debug("_runcmd_output_contains:")
-    logging.debug(f"  actual: {actual!r}")
-    logging.debug(f"  wanted: {wanted!r}")
-    assert_eq(wanted in actual, True)
-
-
-def _runcmd_output_doesnt_contain(actual, wanted):
-    assert_ne = globals()["assert_ne"]
-    wanted = bytes(wanted, "utf8").decode("unicode_escape")
-    logging.debug("_runcmd_output_doesnt_contain:")
-    logging.debug(f"  actual: {actual!r}")
-    logging.debug(f"  wanted: {wanted!r}")
-    assert_ne(wanted in actual, True)
-
-
-def runcmd_stdout_matches_regex(ctx, regex=None):
-    _runcmd_output_matches_regex(runcmd_get_stdout(ctx), regex)
-
-
-def runcmd_stdout_doesnt_match_regex(ctx, regex=None):
-    _runcmd_output_doesnt_match_regex(runcmd_get_stdout(ctx), regex)
-
-
-def runcmd_stderr_matches_regex(ctx, regex=None):
-    _runcmd_output_matches_regex(runcmd_get_stderr(ctx), regex)
-
-
-def runcmd_stderr_doesnt_match_regex(ctx, regex=None):
-    _runcmd_output_doesnt_match_regex(runcmd_get_stderr(ctx), regex)
-
-
-def _runcmd_output_matches_regex(actual, regex):
-    assert_ne = globals()["assert_ne"]
-    r = re.compile(regex)
-    m = r.search(actual)
-    logging.debug("_runcmd_output_matches_regex:")
-    logging.debug(f"  actual: {actual!r}")
-    logging.debug(f"  regex: {regex!r}")
-    logging.debug(f"  match: {m}")
-    assert_ne(m, None)
-
-
-def _runcmd_output_doesnt_match_regex(actual, regex):
-    assert_eq = globals()["assert_eq"]
-    r = re.compile(regex)
-    m = r.search(actual)
-    logging.debug("_runcmd_output_doesnt_match_regex:")
-    logging.debug(f"  actual: {actual!r}")
-    logging.debug(f"  regex: {regex!r}")
-    logging.debug(f"  match: {m}")
-    assert_eq(m, None)