idds/tests/subprocess_test.py

0001 import subprocess
0002
0003 process = subprocess.Popen(['/usr/bin/ps', '-ef'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf-8')
0004
0005 # Read the output and error streams asynchronously
0006 output_reader = process.stdout
0007 error_reader = process.stderr
0008
0009 output = ""
0010 error = ""
0011
0012 while True:
0013     # Read from the output stream
0014     output_chunk = output_reader.read()
0015     if output_chunk:
0016         output += output_chunk
0017     else:
0018         break
0019
0020 while True:
0021     # Read from the error stream
0022     error_chunk = error_reader.read()
0023     if error_chunk:
0024         error += error_chunk
0025     else:
0026         break
0027
0028 stdout, stderr = process.communicate()
0029 print("stdout, stderr")
0030 print(stdout)
0031 print(stderr)
0032
0033 output = output + stdout
0034 error = error + stderr
0035 # Wait for the process to finish and get the return code
0036 return_code = process.wait()
0037
0038 # Decode the output and error streams
0039 # output = output.decode('utf-8')
0040 # error = error.decode('utf-8')
0041
0042 # Print the output, error, and return code
0043 print("Output:", output)
0044 print("Error:", error)
0045 print("Return Code:", return_code)