DXR is a code search and navigation tool aimed at making sense of large projects. It supports full-text and regex searches as well as structural queries.

Mercurial (dcc6d7a0dc00)

VCS Links

Line Code
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95
import unittest
import subprocess
import sys
from mozprocess.processhandler import ProcessReader, StoreOutput

def run_python(str_code, stdout=subprocess.PIPE, stderr=subprocess.PIPE):
    cmd = [sys.executable, '-c', str_code]
    return subprocess.Popen(cmd, stdout=stdout, stderr=stderr)

class TestProcessReader(unittest.TestCase):
    def setUp(self):
        self.out = StoreOutput()
        self.err = StoreOutput()
        self.finished = False
        def on_finished():
            self.finished = True
        self.timeout = False
        def on_timeout():
            self.timeout = True
        self.reader = ProcessReader(stdout_callback=self.out,
                                    stderr_callback=self.err,
                                    finished_callback=on_finished,
                                    timeout_callback=on_timeout)

    def test_stdout_callback(self):
        proc = run_python('print 1; print 2')
        self.reader.start(proc)
        self.reader.thread.join()

        self.assertEqual(self.out.output, ['1', '2'])
        self.assertEqual(self.err.output, [])

    def test_stderr_callback(self):
        proc = run_python('import sys; sys.stderr.write("hello world\\n")')
        self.reader.start(proc)
        self.reader.thread.join()

        self.assertEqual(self.out.output, [])
        self.assertEqual(self.err.output, ['hello world'])

    def test_stdout_and_stderr_callbacks(self):
        proc = run_python('import sys; sys.stderr.write("hello world\\n"); print 1; print 2')
        self.reader.start(proc)
        self.reader.thread.join()

        self.assertEqual(self.out.output, ['1', '2'])
        self.assertEqual(self.err.output, ['hello world'])

    def test_finished_callback(self):
        self.assertFalse(self.finished)
        proc = run_python('')
        self.reader.start(proc)
        self.reader.thread.join()
        self.assertTrue(self.finished)

    def test_timeout(self):
        self.reader.timeout = 0.05
        self.assertFalse(self.timeout)
        proc = run_python('import time; time.sleep(0.1)')
        self.reader.start(proc)
        self.reader.thread.join()
        self.assertTrue(self.timeout)
        self.assertFalse(self.finished)

    def test_output_timeout(self):
        self.reader.output_timeout = 0.05
        self.assertFalse(self.timeout)
        proc = run_python('import time; time.sleep(0.1)')
        self.reader.start(proc)
        self.reader.thread.join()
        self.assertTrue(self.timeout)
        self.assertFalse(self.finished)

    def test_read_without_eol(self):
        proc = run_python('import sys; sys.stdout.write("1")')
        self.reader.start(proc)
        self.reader.thread.join()
        self.assertEqual(self.out.output, ['1'])

    def test_read_with_strange_eol(self):
        proc = run_python('import sys; sys.stdout.write("1\\r\\r\\r\\n")')
        self.reader.start(proc)
        self.reader.thread.join()
        self.assertEqual(self.out.output, ['1'])

    def test_mixed_stdout_stderr(self):
        proc = run_python('import sys; sys.stderr.write("hello world\\n"); print 1; print 2', stderr=subprocess.STDOUT)
        self.reader.start(proc)
        self.reader.thread.join()

        self.assertEqual(sorted(self.out.output), sorted(['1', '2', 'hello world']))
        self.assertEqual(self.err.output, [])

if __name__ == '__main__':
    unittest.main()