diff options
Diffstat (limited to 'testing/mozbase/mozprocess/tests/test_mozprocess.py')
| -rw-r--r-- | testing/mozbase/mozprocess/tests/test_mozprocess.py | 235 |
1 files changed, 235 insertions, 0 deletions
diff --git a/testing/mozbase/mozprocess/tests/test_mozprocess.py b/testing/mozbase/mozprocess/tests/test_mozprocess.py new file mode 100644 index 000000000..bf8ba194c --- /dev/null +++ b/testing/mozbase/mozprocess/tests/test_mozprocess.py @@ -0,0 +1,235 @@ +#!/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() |