Python subprocess works from unit-test framework but hangs in application

Question

I am writing a Python wrapper around an SQL database access call and am having trouble getting the actual sub-process to work as it behaves different when called from the unit-test vs. the actual script.

If I comment out the proc.wait() call then the unit-test kicks up a bunch of "Resource Warnings" about subprocesses that are still running. If I uncomment the proc.wait() and proc.stdout.close() and proc.stderr.close() statements then the unit-test works perfectly but the script will occasionally hang on me when run from the command line.

What is different between the unit-test environment and the actual running of the script from the command line that would cause it to either (a) kick up Resource Warnings in the unit-test or (b) cause the actual script to hang when run from the command-line?

Is there some way to eliminate both (a) and (b) problems above?

Thank you for any advice,

Catherine

Subprocess call

def execute_command(command, env=os.environ):

    proc = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, \
                            stderr=subprocess.PIPE, env=env)
    #proc.wait()
    out_string = proc.stdout.read()
    err_string = proc.stderr.read()

    #proc.stdout.close()
    #proc.stderr.close()

    out = [s.decode("utf-8") for s in out_string.split(b"\n") if (len(s) > 0)]
    err = [s.decode("utf-8") for s in err_string.split(b"\n") if (len(s) > 0)]

    #print("Command %s: out=%s, err=%s" % (command, out, err))

    return (out, err)

Unit-test


from filemgr_oco3 import FileMgrOco3
import parameters_oco3 as parameters

import os
import unittest

TESTDIR = "testdata/in"

class FileMgrOco3Test(unittest.TestCase):

    def setUp(self):
        pass

    def tearDown(self):
        pass
            
    def test_lites(self):

        <snip>

        products = ["OCO3_L2DailyFP", "OCO3_L2DailySIF"]
        select = ["Filename", "StartDateTime", "EndDateTime", "CollectionLabel"]
        output = ["Filename", "StartDateTime", "EndDateTime", "CollectionLabel"]
        conditions = {"StartDateTime": (">=", "2020-04-29T00:00:00.000Z"), \
                      "EndDateTime": ("<=", "2020-04-30T00:00:00.000Z")}

        query = FileMgrOco3(products, select, output, conditions, logic="AND")
        result = query.execute_query()

        <snip>

Class definition

class FileMgrOco3(object):

    <snip>

    def execute_query(self, print_debug=False):

        self.validate_query()

        if ("L2Daily" in self.products[0]):
            self.command = f"{self.PCSQUERYBIN} --url {self.FILEMGR_LITES} --sql "
        else:
            self.command = f"{self.PCSQUERYBIN} --url {self.FILEMGR_L1L2} --sql "

        self.from_string = f','.join(self.products)
        self.query_string = f"SELECT {','.join(self.select)}"
        self.output_string = ",".join([f"${o}" for o in self.output])
        self.where_string = self.build_conditions()

        self.command += f'-query "{self.query_string} ' \
                     +  f'FROM {self.from_string} WHERE {self.where_string}" ' \
                     +  f"-outputFormat '{self.output_string}' "

        (self.out, self.err) = python_utility.execute_command(self.command, \
                                                              env=self.environ)

Have you tried this: https://stackoverflow.com/questions/39477003/python-subprocess-popen-hanging — Ghoti, Jul 29 '21 at 22:39

score 1 · Answer 1 · answered Aug 04 '21 at 18:06

The key is replacing the proc.wait(), proc.stdout.read() and proc.stderr.read() calls with a single call to proc.communicate().

The following subprocess code is working perfectly for me:

def execute_command(command, env=os.environ):

    proc = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, \
                            stderr=subprocess.PIPE, env=env)

    (out_string, err_string) = proc.communicate()
    out = [s.decode("utf-8") for s in out_string.split(b"\n") \
          if (len(s) > 0)]
    err = [s.decode("utf-8") for s in err_string.split(b"\n") \
           if (len(s) > 0)]

Python subprocess works from unit-test framework but hangs in application

1 Answers1