125 lines
4.0 KiB
Python
125 lines
4.0 KiB
Python
#! /usr/bin/env python
|
|
"""Test script for popen2.py"""
|
|
|
|
import warnings
|
|
warnings.filterwarnings("ignore", ".*popen2 module is deprecated.*",
|
|
DeprecationWarning)
|
|
warnings.filterwarnings("ignore", "os\.popen. is deprecated.*",
|
|
DeprecationWarning)
|
|
|
|
import os
|
|
import sys
|
|
import unittest
|
|
import popen2
|
|
|
|
from test.test_support import TestSkipped, run_unittest, reap_children
|
|
|
|
if sys.platform[:4] == 'beos' or sys.platform[:6] == 'atheos':
|
|
# Locks get messed up or something. Generally we're supposed
|
|
# to avoid mixing "posix" fork & exec with native threads, and
|
|
# they may be right about that after all.
|
|
raise TestSkipped("popen2() doesn't work on " + sys.platform)
|
|
|
|
# if we don't have os.popen, check that
|
|
# we have os.fork. if not, skip the test
|
|
# (by raising an ImportError)
|
|
try:
|
|
from os import popen
|
|
del popen
|
|
except ImportError:
|
|
from os import fork
|
|
del fork
|
|
|
|
class Popen2Test(unittest.TestCase):
|
|
cmd = "cat"
|
|
if os.name == "nt":
|
|
cmd = "more"
|
|
teststr = "ab cd\n"
|
|
# "more" doesn't act the same way across Windows flavors,
|
|
# sometimes adding an extra newline at the start or the
|
|
# end. So we strip whitespace off both ends for comparison.
|
|
expected = teststr.strip()
|
|
|
|
def setUp(self):
|
|
popen2._cleanup()
|
|
# When the test runs, there shouldn't be any open pipes
|
|
self.assertFalse(popen2._active, "Active pipes when test starts" +
|
|
repr([c.cmd for c in popen2._active]))
|
|
|
|
def tearDown(self):
|
|
for inst in popen2._active:
|
|
inst.wait()
|
|
popen2._cleanup()
|
|
self.assertFalse(popen2._active, "_active not empty")
|
|
reap_children()
|
|
|
|
def validate_output(self, teststr, expected_out, r, w, e=None):
|
|
w.write(teststr)
|
|
w.close()
|
|
got = r.read()
|
|
self.assertEquals(expected_out, got.strip(), "wrote %r read %r" %
|
|
(teststr, got))
|
|
|
|
if e is not None:
|
|
got = e.read()
|
|
self.assertFalse(got, "unexpected %r on stderr" % got)
|
|
|
|
def test_popen2(self):
|
|
r, w = popen2.popen2(self.cmd)
|
|
self.validate_output(self.teststr, self.expected, r, w)
|
|
|
|
def test_popen3(self):
|
|
if os.name == 'posix':
|
|
r, w, e = popen2.popen3([self.cmd])
|
|
self.validate_output(self.teststr, self.expected, r, w, e)
|
|
|
|
r, w, e = popen2.popen3(self.cmd)
|
|
self.validate_output(self.teststr, self.expected, r, w, e)
|
|
|
|
def test_os_popen2(self):
|
|
# same test as test_popen2(), but using the os.popen*() API
|
|
if os.name == 'posix':
|
|
w, r = os.popen2([self.cmd])
|
|
self.validate_output(self.teststr, self.expected, r, w)
|
|
|
|
w, r = os.popen2(["echo", self.teststr])
|
|
got = r.read()
|
|
self.assertEquals(got, self.teststr + "\n")
|
|
|
|
w, r = os.popen2(self.cmd)
|
|
self.validate_output(self.teststr, self.expected, r, w)
|
|
|
|
def test_os_popen3(self):
|
|
# same test as test_popen3(), but using the os.popen*() API
|
|
if os.name == 'posix':
|
|
w, r, e = os.popen3([self.cmd])
|
|
self.validate_output(self.teststr, self.expected, r, w, e)
|
|
|
|
w, r, e = os.popen3(["echo", self.teststr])
|
|
got = r.read()
|
|
self.assertEquals(got, self.teststr + "\n")
|
|
got = e.read()
|
|
self.assertFalse(got, "unexpected %r on stderr" % got)
|
|
|
|
w, r, e = os.popen3(self.cmd)
|
|
self.validate_output(self.teststr, self.expected, r, w, e)
|
|
|
|
def test_os_popen4(self):
|
|
if os.name == 'posix':
|
|
w, r = os.popen4([self.cmd])
|
|
self.validate_output(self.teststr, self.expected, r, w)
|
|
|
|
w, r = os.popen4(["echo", self.teststr])
|
|
got = r.read()
|
|
self.assertEquals(got, self.teststr + "\n")
|
|
|
|
w, r = os.popen4(self.cmd)
|
|
self.validate_output(self.teststr, self.expected, r, w)
|
|
|
|
|
|
def test_main():
|
|
run_unittest(Popen2Test)
|
|
|
|
if __name__ == "__main__":
|
|
test_main()
|