| #!/usr/bin/env python |
| |
| # This Source Code Form is subject to the terms of the Mozilla Public |
| # License, v. 2.0. If a copy of the MPL was not distributed with this file, |
| # You can obtain one at http://mozilla.org/MPL/2.0/. |
| |
| import os |
| import subprocess |
| import sys |
| import unittest |
| import proctest |
| from mozprocess import processhandler |
| |
| here = os.path.dirname(os.path.abspath(__file__)) |
| |
| def make_proclaunch(aDir): |
| """ |
| Makes the proclaunch executable. |
| Params: |
| aDir - the directory in which to issue the make commands |
| Returns: |
| the path to the proclaunch executable that is generated |
| """ |
| |
| if sys.platform == "win32": |
| exepath = os.path.join(aDir, "proclaunch.exe") |
| else: |
| exepath = os.path.join(aDir, "proclaunch") |
| |
| # remove the launcher, if it already exists |
| # otherwise, if the make fails you may not notice |
| if os.path.exists(exepath): |
| os.remove(exepath) |
| |
| # Ideally make should take care of both calls through recursion, but since it doesn't, |
| # on windows anyway (to file?), let's just call out both targets explicitly. |
| for command in [["make", "-C", "iniparser"], |
| ["make"]]: |
| process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=aDir) |
| stdout, stderr = process.communicate() |
| if process.returncode: |
| # SomethingBadHappen; print all the things |
| print "%s: exit %d" % (command, process.returncode) |
| print "stdout:\n%s" % stdout |
| print "stderr:\n%s" % stderr |
| raise subprocess.CalledProcessError(process.returncode, command, stdout) |
| |
| # ensure the launcher now exists |
| if not os.path.exists(exepath): |
| raise AssertionError("proclaunch executable '%s' does not exist (sys.platform=%s)" % (exepath, sys.platform)) |
| return exepath |
| |
| |
| class ProcTest(proctest.ProcTest): |
| |
| # whether to remove created files on exit |
| cleanup = os.environ.get('CLEANUP', 'true').lower() in ('1', 'true') |
| |
| @classmethod |
| def setUpClass(cls): |
| cls.proclaunch = make_proclaunch(here) |
| |
| @classmethod |
| def tearDownClass(cls): |
| del cls.proclaunch |
| if not cls.cleanup: |
| return |
| files = [('proclaunch',), |
| ('proclaunch.exe',), |
| ('iniparser', 'dictionary.o'), |
| ('iniparser', 'iniparser.lib'), |
| ('iniparser', 'iniparser.o'), |
| ('iniparser', 'libiniparser.a'), |
| ('iniparser', 'libiniparser.so.0'), |
| ] |
| files = [os.path.join(here, *path) for path in files] |
| errors = [] |
| for path in files: |
| if os.path.exists(path): |
| try: |
| os.remove(path) |
| except OSError as e: |
| errors.append(str(e)) |
| if errors: |
| raise OSError("Error(s) encountered tearing down %s.%s:\n%s" % (cls.__module__, cls.__name__, '\n'.join(errors))) |
| |
| def test_process_normal_finish(self): |
| """Process is started, runs to completion while we wait for it""" |
| |
| p = processhandler.ProcessHandler([self.proclaunch, "process_normal_finish.ini"], |
| cwd=here) |
| p.run() |
| p.wait() |
| |
| self.determine_status(p) |
| |
| def test_commandline_no_args(self): |
| """Command line is reported correctly when no arguments are specified""" |
| p = processhandler.ProcessHandler(self.proclaunch, cwd=here) |
| self.assertEqual(p.commandline, self.proclaunch) |
| |
| def test_commandline_overspecified(self): |
| """Command line raises an exception when the arguments are specified ambiguously""" |
| err = None |
| try: |
| processhandler.ProcessHandler([self.proclaunch, "process_normal_finish.ini"], |
| args=["1", "2", "3"], |
| cwd=here) |
| except TypeError, e: |
| err = e |
| |
| self.assertTrue(err) |
| |
| def test_commandline_from_list(self): |
| """Command line is reported correctly when command and arguments are specified in a list""" |
| p = processhandler.ProcessHandler([self.proclaunch, "process_normal_finish.ini"], |
| cwd=here) |
| self.assertEqual(p.commandline, self.proclaunch + ' process_normal_finish.ini') |
| |
| def test_commandline_over_specified(self): |
| """Command line raises an exception when the arguments are specified ambiguously""" |
| err = None |
| try: |
| processhandler.ProcessHandler([self.proclaunch, "process_normal_finish.ini"], |
| args=["1", "2", "3"], |
| cwd=here) |
| except TypeError, e: |
| err = e |
| |
| self.assertTrue(err) |
| |
| def test_commandline_from_args(self): |
| """Command line is reported correctly when arguments are specified in a dedicated list""" |
| p = processhandler.ProcessHandler(self.proclaunch, |
| args=["1", "2", "3"], |
| cwd=here) |
| self.assertEqual(p.commandline, self.proclaunch + ' 1 2 3') |
| |
| def test_process_wait(self): |
| """Process is started runs to completion while we wait indefinitely""" |
| |
| p = processhandler.ProcessHandler([self.proclaunch, |
| "process_waittimeout_10s.ini"], |
| cwd=here) |
| p.run() |
| p.wait() |
| |
| self.determine_status(p) |
| |
| def test_process_timeout(self): |
| """ Process is started, runs but we time out waiting on it |
| to complete |
| """ |
| p = processhandler.ProcessHandler([self.proclaunch, "process_waittimeout.ini"], |
| cwd=here) |
| p.run(timeout=10) |
| p.wait() |
| |
| self.determine_status(p, False, ['returncode', 'didtimeout']) |
| |
| def test_process_timeout_no_kill(self): |
| """ Process is started, runs but we time out waiting on it |
| to complete. Process should not be killed. |
| """ |
| p = None |
| def timeout_handler(): |
| self.assertEqual(p.proc.poll(), None) |
| p.kill() |
| p = processhandler.ProcessHandler([self.proclaunch, "process_waittimeout.ini"], |
| cwd=here, |
| onTimeout=(timeout_handler,), |
| kill_on_timeout=False) |
| p.run(timeout=1) |
| p.wait() |
| self.assertTrue(p.didTimeout) |
| |
| self.determine_status(p, False, ['returncode', 'didtimeout']) |
| |
| def test_process_waittimeout(self): |
| """ |
| Process is started, then wait is called and times out. |
| Process is still running and didn't timeout |
| """ |
| p = processhandler.ProcessHandler([self.proclaunch, |
| "process_waittimeout_10s.ini"], |
| cwd=here) |
| |
| p.run() |
| p.wait(timeout=5) |
| |
| self.determine_status(p, True, ()) |
| |
| def test_process_waitnotimeout(self): |
| """ Process is started, runs to completion before our wait times out |
| """ |
| p = processhandler.ProcessHandler([self.proclaunch, |
| "process_waittimeout_10s.ini"], |
| cwd=here) |
| p.run(timeout=30) |
| p.wait() |
| |
| self.determine_status(p) |
| |
| def test_process_kill(self): |
| """Process is started, we kill it""" |
| |
| p = processhandler.ProcessHandler([self.proclaunch, "process_normal_finish.ini"], |
| cwd=here) |
| p.run() |
| p.kill() |
| |
| self.determine_status(p) |
| |
| def test_process_output_twice(self): |
| """ |
| Process is started, then processOutput is called a second time explicitly |
| """ |
| p = processhandler.ProcessHandler([self.proclaunch, |
| "process_waittimeout_10s.ini"], |
| cwd=here) |
| |
| p.run() |
| p.processOutput(timeout=5) |
| p.wait() |
| |
| self.determine_status(p, False, ()) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
