/usr/lib/python3/dist-packages/pyutilib/subprocess/tests/test_subprocess.py is in python3-pyutilib 5.3.5-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 | import sys
import os
from os.path import abspath, dirname
currdir = dirname(abspath(__file__))+os.sep
import pyutilib.th as unittest
import pyutilib.services
from pyutilib.subprocess import subprocess, SubprocessMngr, timer
from pyutilib.subprocess.processmngr import _peek_available
import six
_mswindows = (sys.platform == 'win32')
try:
import __pypy__
is_pypy = True
except:
is_pypy = False
pyutilib.services.register_executable("memmon")
global_memmon = pyutilib.services.registered_executable('memmon')
pyutilib.services.register_executable("valgrind")
global_valgrind = pyutilib.services.registered_executable('valgrind')
class Test(unittest.TestCase):
def test_foo(self):
if not _mswindows:
foo = SubprocessMngr("ls *py > /tmp/.pyutilib", stdout=subprocess.PIPE, shell=True)
foo.wait()
print("")
if os.path.exists("/tmp/.pyutilib"):
os.remove("/tmp/.pyutilib")
else:
foo = SubprocessMngr("cmd /C \"dir\" > C:/tmp", shell=True)
foo.wait()
print("")
@unittest.skipIf(is_pypy, "Cannot launch python in this test with pypy")
def test_timeout(self):
targetTime = 2
stime = timer()
# On MS Windows, do not run this in a shell. If so, MS Windows has difficulty
# killing the process after the timelimit
print("Subprocess python process")
sys.stdout.flush()
if ' ' in sys.executable:
foo = SubprocessMngr("'" + sys.executable + "' -q -c \"while True: pass\"", shell=not _mswindows)
else:
foo = SubprocessMngr(sys.executable + " -q -c \"while True: pass\"", shell=not _mswindows)
foo.wait(targetTime)
runTime = timer()-stime
print("Ran for %f seconds" % (runTime,))
# timeout should be accurate to 1/10 second
self.assertTrue( runTime <= targetTime + 0.1 )
@unittest.skipIf(_mswindows, "Cannot test the use of 'memmon' on MS Windows")
@unittest.skipIf(sys.platform == 'darwin', "Cannot test the use of 'memmon' on Darwin")
@unittest.skipIf(not global_memmon, "The 'memmon' executable is not available.")
def test_memmon(self):
pyutilib.services.register_executable('ls')
pyutilib.subprocess.run(pyutilib.services.registered_executable('ls').get_path()+' *.py', memmon=True, outfile=currdir+'ls.out')
INPUT = open(currdir+'ls.out','r')
flag = False
for line in INPUT:
flag = line.startswith('memmon:')
if flag:
break
INPUT.close()
if not flag:
self.fail("Failed to properly execute 'memmon' with the 'ls' command")
os.remove(currdir+'ls.out')
@unittest.skipIf(_mswindows, "Cannot test the use of 'valgrind' on MS Windows")
@unittest.skipIf(not global_valgrind, "The 'valgrind' executable is not available.")
def test_valgrind(self):
pyutilib.services.register_executable('ls')
pyutilib.subprocess.run(pyutilib.services.registered_executable('ls').get_path()+' *.py', valgrind=True, outfile=currdir+'valgrind.out')
INPUT = open(currdir+'valgrind.out','r')
flag = False
for line in INPUT:
flag = 'Memcheck' in line
if flag:
break
INPUT.close()
if not flag:
self.fail("Failed to properly execute 'valgrind' with the 'ls' command")
os.remove(currdir+'valgrind.out')
def test_outputfile(self):
pyutilib.subprocess.run([sys.executable, currdir+"tee_script.py"], outfile=currdir+'tee.out')
INPUT = open(currdir+'tee.out')
output = INPUT.read()
INPUT.close()
os.remove(currdir+'tee.out')
if _peek_available:
self.assertEquals( sorted(output.splitlines()),
["Tee Script: ERR","Tee Script: OUT"] )
else:
sys.stderr.write("BEGIN OUTPUT:\n"+output+"END OUTPUT\n")
self.assertTrue( output.splitlines() in
( ["Tee Script: ERR","Tee Script: OUT"],
["Tee Script: OUT","Tee Script: ERR"] ) )
def test_ostream_stringio(self):
script_out = six.StringIO()
output = pyutilib.subprocess.run(
[sys.executable, currdir+"tee_script.py"], ostream=script_out )
if _peek_available:
self.assertEquals( sorted(script_out.getvalue().splitlines()),
["Tee Script: ERR","Tee Script: OUT"] )
else:
sys.stderr.write("BEGIN OUTPUT:\n"+script_out.getvalue()+"END OUTPUT\n")
self.assertTrue( script_out.getvalue().splitlines() in
( ["Tee Script: ERR","Tee Script: OUT"],
["Tee Script: OUT","Tee Script: ERR"] ) )
def test_tee(self):
stream_out = six.StringIO()
script_out = six.StringIO()
pyutilib.misc.setup_redirect(stream_out)
output = pyutilib.subprocess.run(
[sys.executable, currdir+"tee_script.py"],
ostream=script_out, tee=True )
pyutilib.misc.reset_redirect()
# The following is only deterministic if Peek/Select is available
if _peek_available:
self.assertEquals( sorted(stream_out.getvalue().splitlines()),
["Tee Script: ERR","Tee Script: OUT"] )
self.assertEquals( sorted(script_out.getvalue().splitlines()),
["Tee Script: ERR","Tee Script: OUT"] )
else:
sys.stderr.write("BEGIN OUTPUT1:\n"+stream_out.getvalue()+"END OUTPUT1\n")
sys.stderr.write("BEGIN OUTPUT2:\n"+script_out.getvalue()+"END OUTPUT2\n")
self.assertTrue( stream_out.getvalue().splitlines() in
( ["Tee Script: ERR","Tee Script: OUT"],
["Tee Script: OUT","Tee Script: ERR"] ) )
self.assertTrue( script_out.getvalue().splitlines() in
( ["Tee Script: ERR","Tee Script: OUT"],
["Tee Script: OUT","Tee Script: ERR"] ) )
def test_tee_stdout(self):
stream_out = six.StringIO()
script_out = six.StringIO()
pyutilib.misc.setup_redirect(stream_out)
output = pyutilib.subprocess.run(
[sys.executable, currdir+"tee_script.py"],
ostream=script_out, tee=(True, False) )
pyutilib.misc.reset_redirect()
self.assertEquals( stream_out.getvalue().splitlines(),
["Tee Script: OUT"] )
# The following is only deterministic if Peek/Select is available
if _peek_available:
self.assertEquals( sorted(script_out.getvalue().splitlines()),
["Tee Script: ERR","Tee Script: OUT"] )
else:
sys.stderr.write("BEGIN OUTPUT:\n"+script_out.getvalue()+"END OUTPUT\n")
self.assertTrue( script_out.getvalue().splitlines() in
( ["Tee Script: ERR","Tee Script: OUT"],
["Tee Script: OUT","Tee Script: ERR"] ) )
def test_tee_stderr(self):
stream_out = six.StringIO()
script_out = six.StringIO()
pyutilib.misc.setup_redirect(stream_out)
output = pyutilib.subprocess.run(
[sys.executable, currdir+"tee_script.py"],
ostream=script_out, tee=(False, True) )
pyutilib.misc.reset_redirect()
self.assertEquals( stream_out.getvalue().splitlines(),
["Tee Script: ERR"] )
# The following is only deterministic if Peek/Select is available
if _peek_available:
self.assertEquals( sorted(script_out.getvalue().splitlines()),
["Tee Script: ERR","Tee Script: OUT"] )
else:
sys.stderr.write("BEGIN OUTPUT:\n"+script_out.getvalue()+"END OUTPUT\n")
self.assertTrue( script_out.getvalue().splitlines() in
( ["Tee Script: ERR","Tee Script: OUT"],
["Tee Script: OUT","Tee Script: ERR"] ) )
if __name__ == "__main__":
unittest.main()
|