• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1"""Test script for popen2.py"""
2
3import warnings
4warnings.filterwarnings("ignore", ".*popen2 module is deprecated.*",
5                        DeprecationWarning)
6warnings.filterwarnings("ignore", "os\.popen. is deprecated.*",
7                        DeprecationWarning)
8
9import os
10import sys
11import unittest
12import popen2
13
14from test.test_support import run_unittest, reap_children
15
16if 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)
25try:
26    from os import popen
27    del popen
28except ImportError:
29    from os import fork
30    del fork
31
32class 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
125def test_main():
126    run_unittest(Popen2Test)
127
128if __name__ == "__main__":
129    test_main()
130