Home | History | Annotate | Download | only in test 
      1 """Test script for popen2.py"""
      2 
      3 import warnings
      4 warnings.filterwarnings("ignore", ".*popen2 module is deprecated.*",
      5                         DeprecationWarning)
      6 warnings.filterwarnings("ignore", "os\.popen. is deprecated.*",
      7                         DeprecationWarning)
      8 
      9 import os
     10 import sys
     11 import unittest
     12 import popen2
     13 
     14 from test.test_support import run_unittest, reap_children
     15 
     16 if sys.platform[:4] == 'beos' or sys.platform[:6] == 'atheos':
     17     #  Locks get messed up or something.  Generally we're supposed
     18     #  to avoid mixing "posix" fork & exec with native threads, and
     19     #  they may be right about that after all.
     20     raise unittest.SkipTest("popen2() doesn't work on " + sys.platform)
     21 
     22 # if we don't have os.popen, check that
     23 # we have os.fork.  if not, skip the test
     24 # (by raising an ImportError)
     25 try:
     26     from os import popen
     27     del popen
     28 except ImportError:
     29     from os import fork
     30     del fork
     31 
     32 class Popen2Test(unittest.TestCase):
     33     cmd = "cat"
     34     if os.name == "nt":
     35         cmd = "more"
     36     teststr = "ab cd\n"
     37     # "more" doesn't act the same way across Windows flavors,
     38     # sometimes adding an extra newline at the start or the
     39     # end.  So we strip whitespace off both ends for comparison.
     40     expected = teststr.strip()
     41 
     42     def setUp(self):
     43         popen2._cleanup()
     44         # When the test runs, there shouldn't be any open pipes
     45         self.assertFalse(popen2._active, "Active pipes when test starts" +
     46             repr([c.cmd for c in popen2._active]))
     47 
     48     def tearDown(self):
     49         for inst in popen2._active:
     50             inst.wait()
     51         popen2._cleanup()
     52         self.assertFalse(popen2._active, "popen2._active not empty")
     53         # The os.popen*() API delegates to the subprocess module (on Unix)
     54         import subprocess
     55         for inst in subprocess._active:
     56             inst.wait()
     57         subprocess._cleanup()
     58         self.assertFalse(subprocess._active, "subprocess._active not empty")
     59         reap_children()
     60 
     61     def validate_output(self, teststr, expected_out, r, w, e=None):
     62         w.write(teststr)
     63         w.close()
     64         got = r.read()
     65         self.assertEqual(expected_out, got.strip(), "wrote %r read %r" %
     66                          (teststr, got))
     67 
     68         if e is not None:
     69             got = e.read()
     70             self.assertFalse(got, "unexpected %r on stderr" % got)
     71 
     72     def test_popen2(self):
     73         r, w = popen2.popen2(self.cmd)
     74         self.validate_output(self.teststr, self.expected, r, w)
     75 
     76     def test_popen3(self):
     77         if os.name == 'posix':
     78             r, w, e = popen2.popen3([self.cmd])
     79             self.validate_output(self.teststr, self.expected, r, w, e)
     80 
     81         r, w, e = popen2.popen3(self.cmd)
     82         self.validate_output(self.teststr, self.expected, r, w, e)
     83 
     84     def test_os_popen2(self):
     85         # same test as test_popen2(), but using the os.popen*() API
     86         if os.name == 'posix':
     87             w, r = os.popen2([self.cmd])
     88             self.validate_output(self.teststr, self.expected, r, w)
     89 
     90             w, r = os.popen2(["echo", self.teststr])
     91             got = r.read()
     92             self.assertEqual(got, self.teststr + "\n")
     93 
     94         w, r = os.popen2(self.cmd)
     95         self.validate_output(self.teststr, self.expected, r, w)
     96 
     97     def test_os_popen3(self):
     98         # same test as test_popen3(), but using the os.popen*() API
     99         if os.name == 'posix':
    100             w, r, e = os.popen3([self.cmd])
    101             self.validate_output(self.teststr, self.expected, r, w, e)
    102 
    103             w, r, e = os.popen3(["echo", self.teststr])
    104             got = r.read()
    105             self.assertEqual(got, self.teststr + "\n")
    106             got = e.read()
    107             self.assertFalse(got, "unexpected %r on stderr" % got)
    108 
    109         w, r, e = os.popen3(self.cmd)
    110         self.validate_output(self.teststr, self.expected, r, w, e)
    111 
    112     def test_os_popen4(self):
    113         if os.name == 'posix':
    114             w, r = os.popen4([self.cmd])
    115             self.validate_output(self.teststr, self.expected, r, w)
    116 
    117             w, r = os.popen4(["echo", self.teststr])
    118             got = r.read()
    119             self.assertEqual(got, self.teststr + "\n")
    120 
    121         w, r = os.popen4(self.cmd)
    122         self.validate_output(self.teststr, self.expected, r, w)
    123 
    124 
    125 def test_main():
    126     run_unittest(Popen2Test)
    127 
    128 if __name__ == "__main__":
    129     test_main()
    130