test_popen2.py 4.2 KB

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