1.1 --- /dev/null Thu Jan 01 00:00:00 1970 +0000
1.2 +++ b/testing/mozbase/mozprocess/tests/test_mozprocess_output.py Wed Dec 31 06:09:35 2014 +0100
1.3 @@ -0,0 +1,68 @@
1.4 +#!/usr/bin/env python
1.5 +
1.6 +import io
1.7 +import os
1.8 +import unittest
1.9 +import proctest
1.10 +from mozprocess import processhandler
1.11 +
1.12 +here = os.path.dirname(os.path.abspath(__file__))
1.13 +
1.14 +class ProcTestOutput(proctest.ProcTest):
1.15 + """ Class to test operations related to output handling """
1.16 +
1.17 + def test_process_output_nonewline(self):
1.18 + """
1.19 + Process is started, outputs data with no newline
1.20 + """
1.21 + p = processhandler.ProcessHandler([self.python, "procnonewline.py"],
1.22 + cwd=here)
1.23 +
1.24 + p.run()
1.25 + p.processOutput(timeout=5)
1.26 + p.wait()
1.27 +
1.28 + detected, output = proctest.check_for_process("procnonewline.py")
1.29 + self.determine_status(detected,
1.30 + output,
1.31 + p.proc.returncode,
1.32 + p.didTimeout,
1.33 + False,
1.34 + ())
1.35 +
1.36 + def test_stream_process_output(self):
1.37 + """
1.38 + Process output stream does not buffer
1.39 + """
1.40 + expected = '\n'.join([str(n) for n in range(0,10)])
1.41 +
1.42 + stream = io.BytesIO()
1.43 + buf = io.BufferedRandom(stream)
1.44 +
1.45 + p = processhandler.ProcessHandler([self.python, "proccountfive.py"],
1.46 + cwd=here,
1.47 + stream=buf)
1.48 +
1.49 + p.run()
1.50 + p.wait()
1.51 + for i in range(5, 10):
1.52 + stream.write(str(i)+'\n')
1.53 +
1.54 + buf.flush()
1.55 + self.assertEquals(stream.getvalue().strip(), expected)
1.56 +
1.57 + # make sure mozprocess doesn't close the stream
1.58 + # since mozprocess didn't create it
1.59 + self.assertFalse(buf.closed)
1.60 + buf.close()
1.61 +
1.62 + detected, output = proctest.check_for_process("proccountfive.py")
1.63 + self.determine_status(detected,
1.64 + output,
1.65 + p.proc.returncode,
1.66 + p.didTimeout,
1.67 + False,
1.68 + ())
1.69 +
1.70 +if __name__ == '__main__':
1.71 + unittest.main()
