tor-browser

The Tor Browser
git clone https://git.dasho.dev/tor-browser.git
Log | Files | Refs | README | LICENSE

test_process_reader.py (3675B)

      1 import subprocess
      2 import sys
      3 import unittest
      4 
      5 import mozunit
      6 from mozprocess.processhandler import ProcessReader, StoreOutput
      7 
      8 
      9 def run_python(str_code, stdout=subprocess.PIPE, stderr=subprocess.PIPE):
     10    cmd = [sys.executable, "-c", str_code]
     11    return subprocess.Popen(cmd, stdout=stdout, stderr=stderr)
     12 
     13 
     14 class TestProcessReader(unittest.TestCase):
     15    def setUp(self):
     16        self.out = StoreOutput()
     17        self.err = StoreOutput()
     18        self.finished = False
     19 
     20        def on_finished():
     21            self.finished = True
     22 
     23        self.timeout = False
     24 
     25        def on_timeout():
     26            self.timeout = True
     27 
     28        self.reader = ProcessReader(
     29            stdout_callback=self.out,
     30            stderr_callback=self.err,
     31            finished_callback=on_finished,
     32            timeout_callback=on_timeout,
     33        )
     34 
     35    def test_stdout_callback(self):
     36        proc = run_python("print(1); print(2)")
     37        self.reader.start(proc)
     38        self.reader.thread.join()
     39 
     40        self.assertEqual([x.decode("utf8") for x in self.out.output], ["1", "2"])
     41        self.assertEqual(self.err.output, [])
     42 
     43    def test_stderr_callback(self):
     44        proc = run_python('import sys; sys.stderr.write("hello world\\n")')
     45        self.reader.start(proc)
     46        self.reader.thread.join()
     47 
     48        self.assertEqual(self.out.output, [])
     49        self.assertEqual([x.decode("utf8") for x in self.err.output], ["hello world"])
     50 
     51    def test_stdout_and_stderr_callbacks(self):
     52        proc = run_python(
     53            'import sys; sys.stderr.write("hello world\\n"); print(1); print(2)'
     54        )
     55        self.reader.start(proc)
     56        self.reader.thread.join()
     57 
     58        self.assertEqual([x.decode("utf8") for x in self.out.output], ["1", "2"])
     59        self.assertEqual([x.decode("utf8") for x in self.err.output], ["hello world"])
     60 
     61    def test_finished_callback(self):
     62        self.assertFalse(self.finished)
     63        proc = run_python("")
     64        self.reader.start(proc)
     65        self.reader.thread.join()
     66        self.assertTrue(self.finished)
     67 
     68    def test_timeout(self):
     69        self.reader.timeout = 0.05
     70        self.assertFalse(self.timeout)
     71        proc = run_python("import time; time.sleep(0.1)")
     72        self.reader.start(proc)
     73        self.reader.thread.join()
     74        self.assertTrue(self.timeout)
     75        self.assertFalse(self.finished)
     76 
     77    def test_output_timeout(self):
     78        self.reader.output_timeout = 0.05
     79        self.assertFalse(self.timeout)
     80        proc = run_python("import time; time.sleep(0.1)")
     81        self.reader.start(proc)
     82        self.reader.thread.join()
     83        self.assertTrue(self.timeout)
     84        self.assertFalse(self.finished)
     85 
     86    def test_read_without_eol(self):
     87        proc = run_python('import sys; sys.stdout.write("1")')
     88        self.reader.start(proc)
     89        self.reader.thread.join()
     90        self.assertEqual([x.decode("utf8") for x in self.out.output], ["1"])
     91 
     92    def test_read_with_strange_eol(self):
     93        proc = run_python('import sys; sys.stdout.write("1\\r\\r\\r\\n")')
     94        self.reader.start(proc)
     95        self.reader.thread.join()
     96        self.assertEqual([x.decode("utf8") for x in self.out.output], ["1"])
     97 
     98    def test_mixed_stdout_stderr(self):
     99        proc = run_python(
    100            'import sys; sys.stderr.write("hello world\\n"); print(1); print(2)',
    101            stderr=subprocess.STDOUT,
    102        )
    103        self.reader.start(proc)
    104        self.reader.thread.join()
    105 
    106        self.assertEqual(
    107            sorted([x.decode("utf8") for x in self.out.output]),
    108            sorted(["1", "2", "hello world"]),
    109        )
    110        self.assertEqual(self.err.output, [])
    111 
    112 
    113 if __name__ == "__main__":
    114    mozunit.main()