This is a bit of a tricky one ... I am writing a test suite for a Python project - running on Linux. As part of the tests, I have to fire up various commands (shell commands, small CLI tools, Perl scripts, etc). Some of those commands can hang or crash or loop forever and must be killed after a timeout.

A subset of those commands can also start an arbitrary number of sub-processes of their own, which must also be killed after the timeout (keywords: process groups and setsid). While the test suite runs with normal user privileges, some of the commands require super user privileges, i.e. sudo. If I have to use sudo, I sometimes need to preserve the original user's virtual environment, i.e. make it accessible for the super user.

Sometimes, I push my tests to a CI server, where sudo is not restricted / protected, but sometimes, I also want to test it on my local OS where sudo is password-protected. My test suite must therefore allow me to enter my super user password manually if required in a local test.

My (mostly working) solution is a routine named run_command (plus two helper routines named __kill_proc__ and __get_pid__). So far, it looks like this:

import os
import signal
import subprocess

import psutil # get it with "pip install psutil"

def run_command(cmd_list, return_output = False, sudo = False, sudo_env = False, timeout = None, setsid = False):

    cmd_prefix = []

    if sudo:
        if setsid:
            cmd_prefix.append('-b') # equivalent to "setsid", will spawn new process group including sudo
        if sudo_env: # preserve the user's virtual env for super user
            cmd_prefix.append('%s=%s' % ('VIRTUAL_ENV', os.environ['VIRTUAL_ENV']))
            cmd_prefix.append('%s=%s:%s' % ('PATH', os.path.join(os.environ['VIRTUAL_ENV'], 'bin'), os.environ['PATH']))
    elif setsid:
        cmd_prefix.append('setsid') # TODO untested codepath

    full_cmd = cmd_prefix + cmd_list

    proc = subprocess.Popen(
        full_cmd, stdout = subprocess.PIPE, stderr = subprocess.PIPE

    timeout_alert = ''
    if timeout is not None:
            outs, errs = proc.communicate(timeout = timeout) # If command does not time out, this works just fine
        except subprocess.TimeoutExpired:
            timeout_alert = '\n\nCOMMAND TIMED OUT AND WAS KILLED!'
            if setsid:
                kill_pid = __get_pid__(full_cmd) # proc.pid will deliver wrong pid!
                kill_pid = proc.pid
            __kill_proc__(kill_pid, k_signal = signal.SIGINT, entire_group = setsid, sudo = sudo)
            outs, errs = (proc.stdout.read(), proc.stderr.read()) # PROBLEM: outs and errs are *always* empty! Buffer issue?
            # outs, errs = proc.communicate() # Makes no difference here, outs and errs are always empty this way, too.
        outs, errs = proc.communicate()

    if return_output:
        return (not bool(proc.returncode), outs.decode('utf-8'), errs.decode('utf-8') + timeout_alert)
    return not bool(proc.returncode)

def __kill_proc__(pid, k_signal = signal.SIGINT, entire_group = False, sudo = False):

    if not sudo:
        if entire_group:
            os.killpg(os.getpgid(pid), k_signal)
            os.kill(pid, k_signal)
        if entire_group:
            run_command(['kill', '-%d' % k_signal, '--', '-%d' % os.getpgid(pid)], sudo = sudo)
            run_command(['kill', '-%d' % k_signal, '%d' % pid], sudo = sudo)

def __get_pid__(cmd_line_list):

    for pid in psutil.pids():
        proc = psutil.Process(pid)
        if cmd_line_list == proc.cmdline():
            return proc.pid
    raise # TODO some error ...

The most troubling bit for me is when I have to start a new process group with sudo and it times out (I have discussed the current design of how I start a process group with sudo here on SE Unix & Linux). It works, but it has a fundamental problem: If the sub-process (-group) times out and I kill it, I lose ALL output from this process group and I can not figure out what is going on. I commented the lines in question.

I understand that the above code is a really ugly collection of bad hacks ... I am seriously interested whether this is a good approach or not, how I could possibly improve it or how I should re-implement it.

(I am wondering whether it is better to run the test suite altogether with super user privileges and reduce the privilege level for individual routines or sub-processes, which could cause a whole new set of problems.)

  • 1
    \$\begingroup\$ Why did you name your functions __function_name__? \$\endgroup\$ Dec 18, 2017 at 18:54
  • 1
    \$\begingroup\$ @MrGrj I have organized my testing infrastructure for this particular project into a module ('projectname_libtest'), which is imported by the actual test files (test_*.py, auto-discovered by pytest). In this context, I want ensure that I do not import or use those "internal" helper routines by accident from/at another place ... Only run_command is intended to be "exposed". \$\endgroup\$
    – s-m-e
    Dec 18, 2017 at 19:12
  • \$\begingroup\$ stackoverflow.com/questions/34337840/… \$\endgroup\$ Apr 22, 2018 at 23:50


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Browse other questions tagged or ask your own question.