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 TestSkipped, 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 TestSkipped("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, "_active not empty")

    54         reap_children()

    55 

    56     def validate_output(self, teststr, expected_out, r, w, e=None):

    57         w.write(teststr)

    58         w.close()

    59         got = r.read()

    60         self.assertEquals(expected_out, got.strip(), "wrote %r read %r" %

    61                           (teststr, got))

    62 

    63         if e is not None:

    64             got = e.read()

    65             self.assertFalse(got, "unexpected %r on stderr" % got)

    66 

    67     def test_popen2(self):

    68         r, w = popen2.popen2(self.cmd)

    69         self.validate_output(self.teststr, self.expected, r, w)

    70 

    71     def test_popen3(self):

    72         if os.name == 'posix':

    73             r, w, e = popen2.popen3([self.cmd])

    74             self.validate_output(self.teststr, self.expected, r, w, e)

    75 

    76         r, w, e = popen2.popen3(self.cmd)

    77         self.validate_output(self.teststr, self.expected, r, w, e)

    78 

    79     def test_os_popen2(self):

    80         # same test as test_popen2(), but using the os.popen*() API

    81         w, r = os.popen2(self.cmd)

    82         self.validate_output(self.teststr, self.expected, r, w)

    83 

    84     def test_os_popen3(self):

    85         # same test as test_popen3(), but using the os.popen*() API

    86         if os.name == 'posix':

    87             w, r, e = os.popen3([self.cmd])

    88             self.validate_output(self.teststr, self.expected, r, w, e)

    89 

    90         w, r, e = os.popen3(self.cmd)

    91         self.validate_output(self.teststr, self.expected, r, w, e)

    92 

    93 

    94 def test_main():

    95     run_unittest(Popen2Test)

    96 

    97 if __name__ == "__main__":

    98     test_main()