I have Flask application that exposed API that can run application in background and can kill it later by specifying the PID. However, for unit testing purpose, after killing the PID and checking if the PID is killed using psutil.pid_exists(pid)
it seems to always returning true. I have checked manually that PID does not exist and run psutil.pid_exists(pid)
on different python console and it returned true. This is causing my test to fail.
In views.py
, I have:
@api.route('/cancel/<pid>', methods=['POST'])
def cancel(pid=None):
try:
os.kill(int(pid), signal.SIGTERM)
data = dict(
message = 'Successfully killed pid ' + pid)
return jsonify(status='success', data=data), 200
except:
data = dict(
message = 'Fail to kill job with pid ' + pid)
return jsonify(status='error', data=data), 400
And in my test:
def test_cancel_job(self):
# run_script will run something in the background and return the PID
jobid, pid, cmd = run_script('fake_db', 'fake_cancel_jobid', 'tests/doubles/child.py')
if not psutil.pid_exists(pid):
raise Exception('Process is not running')
# kill the job and assert it is successful
resp = self.client.post('/api/cancel/' + str(pid))
self.assert200(resp)
# at this point, I have confirmed that the PID has been killed
# but, the line below still get executed
# psutil.pid_exists(pid) returns true
# check pid is really killed
if psutil.pid_exists(pid):
raise Exception('Process {0} still exist'.format(pid))
I'm running on OSX if that make any difference.
Update: I've tried running on the test on my build server (Ubuntu 14.04) and the test failed.
This is my run_script
def run_script(db, jobid, script):
log = SCRIPTS_LOG + jobid + ".log"
if not os.path.exists(SCRIPTS_LOG):
os.makedirs(SCRIPTS_LOG)
with open(log, "w") as output:
cmd = ["nohup", "python", script, db]
p = subprocess.Popen(cmd, stdout=output)
return jobid, p.pid, " ".join(cmd)
and my child.py
#!/usr/bin/env python
import time
import os, sys
if 'TEST_ENV' not in os.environ:
os.environ['TEST_ENV'] = 'Set some env'
try:
os.execv(sys.argv[0], sys.argv)
except Exception, exc:
print "Failed re-exec:", exc
sys.exit(1)
def main(argv):
db = argv[0]
while True:
print 'Running child with params ', db
time.sleep(1)
if __name__ == '__main__':
main(sys.argv[1:])
I added a simple scripts that demonstrate this. https://github.com/shulhi/kill-pid/tree/master