michael@0: #!/usr/bin/env python
michael@0: 
michael@0: import io
michael@0: import os
michael@0: import unittest
michael@0: import proctest
michael@0: from mozprocess import processhandler
michael@0: 
michael@0: here = os.path.dirname(os.path.abspath(__file__))
michael@0: 
michael@0: class ProcTestOutput(proctest.ProcTest):
michael@0:     """ Class to test operations related to output handling """
michael@0: 
michael@0:     def test_process_output_nonewline(self):
michael@0:         """
michael@0:         Process is started, outputs data with no newline
michael@0:         """
michael@0:         p = processhandler.ProcessHandler([self.python, "procnonewline.py"],
michael@0:                                           cwd=here)
michael@0: 
michael@0:         p.run()
michael@0:         p.processOutput(timeout=5)
michael@0:         p.wait()
michael@0: 
michael@0:         detected, output = proctest.check_for_process("procnonewline.py")
michael@0:         self.determine_status(detected,
michael@0:                               output,
michael@0:                               p.proc.returncode,
michael@0:                               p.didTimeout,
michael@0:                               False,
michael@0:                               ())
michael@0: 
michael@0:     def test_stream_process_output(self):
michael@0:         """
michael@0:         Process output stream does not buffer
michael@0:         """
michael@0:         expected = '\n'.join([str(n) for n in range(0,10)])
michael@0: 
michael@0:         stream = io.BytesIO()
michael@0:         buf = io.BufferedRandom(stream)
michael@0: 
michael@0:         p = processhandler.ProcessHandler([self.python, "proccountfive.py"],
michael@0:                                           cwd=here,
michael@0:                                           stream=buf)
michael@0: 
michael@0:         p.run()
michael@0:         p.wait()
michael@0:         for i in range(5, 10):
michael@0:             stream.write(str(i)+'\n')
michael@0: 
michael@0:         buf.flush()
michael@0:         self.assertEquals(stream.getvalue().strip(), expected)
michael@0: 
michael@0:         # make sure mozprocess doesn't close the stream
michael@0:         # since mozprocess didn't create it
michael@0:         self.assertFalse(buf.closed)
michael@0:         buf.close()
michael@0: 
michael@0:         detected, output = proctest.check_for_process("proccountfive.py")
michael@0:         self.determine_status(detected,
michael@0:                               output,
michael@0:                               p.proc.returncode,
michael@0:                               p.didTimeout,
michael@0:                               False,
michael@0:                               ())
michael@0: 
michael@0: if __name__ == '__main__':
michael@0:     unittest.main()