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