test_pipes.py 6.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204
  1. import pipes
  2. import os
  3. import string
  4. import unittest
  5. from test.test_support import TESTFN, run_unittest, unlink, reap_children
  6. if os.name != 'posix':
  7. raise unittest.SkipTest('pipes module only works on posix')
  8. TESTFN2 = TESTFN + "2"
  9. # tr a-z A-Z is not portable, so make the ranges explicit
  10. s_command = 'tr %s %s' % (string.ascii_lowercase, string.ascii_uppercase)
  11. class SimplePipeTests(unittest.TestCase):
  12. def tearDown(self):
  13. for f in (TESTFN, TESTFN2):
  14. unlink(f)
  15. def testSimplePipe1(self):
  16. t = pipes.Template()
  17. t.append(s_command, pipes.STDIN_STDOUT)
  18. f = t.open(TESTFN, 'w')
  19. f.write('hello world #1')
  20. f.close()
  21. with open(TESTFN) as f:
  22. self.assertEqual(f.read(), 'HELLO WORLD #1')
  23. def testSimplePipe2(self):
  24. with open(TESTFN, 'w') as f:
  25. f.write('hello world #2')
  26. t = pipes.Template()
  27. t.append(s_command + ' < $IN > $OUT', pipes.FILEIN_FILEOUT)
  28. t.copy(TESTFN, TESTFN2)
  29. with open(TESTFN2) as f:
  30. self.assertEqual(f.read(), 'HELLO WORLD #2')
  31. def testSimplePipe3(self):
  32. with open(TESTFN, 'w') as f:
  33. f.write('hello world #2')
  34. t = pipes.Template()
  35. t.append(s_command + ' < $IN', pipes.FILEIN_STDOUT)
  36. with t.open(TESTFN, 'r') as f:
  37. self.assertEqual(f.read(), 'HELLO WORLD #2')
  38. def testEmptyPipeline1(self):
  39. # copy through empty pipe
  40. d = 'empty pipeline test COPY'
  41. with open(TESTFN, 'w') as f:
  42. f.write(d)
  43. with open(TESTFN2, 'w') as f:
  44. f.write('')
  45. t=pipes.Template()
  46. t.copy(TESTFN, TESTFN2)
  47. with open(TESTFN2) as f:
  48. self.assertEqual(f.read(), d)
  49. def testEmptyPipeline2(self):
  50. # read through empty pipe
  51. d = 'empty pipeline test READ'
  52. with open(TESTFN, 'w') as f:
  53. f.write(d)
  54. t=pipes.Template()
  55. with t.open(TESTFN, 'r') as f:
  56. self.assertEqual(f.read(), d)
  57. def testEmptyPipeline3(self):
  58. # write through empty pipe
  59. d = 'empty pipeline test WRITE'
  60. t = pipes.Template()
  61. with t.open(TESTFN, 'w') as f:
  62. f.write(d)
  63. with open(TESTFN) as f:
  64. self.assertEqual(f.read(), d)
  65. def testQuoting(self):
  66. safeunquoted = string.ascii_letters + string.digits + '@%_-+=:,./'
  67. unsafe = '"`$\\!'
  68. self.assertEqual(pipes.quote(''), "''")
  69. self.assertEqual(pipes.quote(safeunquoted), safeunquoted)
  70. self.assertEqual(pipes.quote('test file name'), "'test file name'")
  71. for u in unsafe:
  72. self.assertEqual(pipes.quote('test%sname' % u),
  73. "'test%sname'" % u)
  74. for u in unsafe:
  75. self.assertEqual(pipes.quote("test%s'name'" % u),
  76. "'test%s'\"'\"'name'\"'\"''" % u)
  77. def testRepr(self):
  78. t = pipes.Template()
  79. self.assertEqual(repr(t), "<Template instance, steps=[]>")
  80. t.append('tr a-z A-Z', pipes.STDIN_STDOUT)
  81. self.assertEqual(repr(t),
  82. "<Template instance, steps=[('tr a-z A-Z', '--')]>")
  83. def testSetDebug(self):
  84. t = pipes.Template()
  85. t.debug(False)
  86. self.assertEqual(t.debugging, False)
  87. t.debug(True)
  88. self.assertEqual(t.debugging, True)
  89. def testReadOpenSink(self):
  90. # check calling open('r') on a pipe ending with
  91. # a sink raises ValueError
  92. t = pipes.Template()
  93. t.append('boguscmd', pipes.SINK)
  94. self.assertRaises(ValueError, t.open, 'bogusfile', 'r')
  95. def testWriteOpenSource(self):
  96. # check calling open('w') on a pipe ending with
  97. # a source raises ValueError
  98. t = pipes.Template()
  99. t.prepend('boguscmd', pipes.SOURCE)
  100. self.assertRaises(ValueError, t.open, 'bogusfile', 'w')
  101. def testBadAppendOptions(self):
  102. t = pipes.Template()
  103. # try a non-string command
  104. self.assertRaises(TypeError, t.append, 7, pipes.STDIN_STDOUT)
  105. # try a type that isn't recognized
  106. self.assertRaises(ValueError, t.append, 'boguscmd', 'xx')
  107. # shouldn't be able to append a source
  108. self.assertRaises(ValueError, t.append, 'boguscmd', pipes.SOURCE)
  109. # check appending two sinks
  110. t = pipes.Template()
  111. t.append('boguscmd', pipes.SINK)
  112. self.assertRaises(ValueError, t.append, 'boguscmd', pipes.SINK)
  113. # command needing file input but with no $IN
  114. t = pipes.Template()
  115. self.assertRaises(ValueError, t.append, 'boguscmd $OUT',
  116. pipes.FILEIN_FILEOUT)
  117. t = pipes.Template()
  118. self.assertRaises(ValueError, t.append, 'boguscmd',
  119. pipes.FILEIN_STDOUT)
  120. # command needing file output but with no $OUT
  121. t = pipes.Template()
  122. self.assertRaises(ValueError, t.append, 'boguscmd $IN',
  123. pipes.FILEIN_FILEOUT)
  124. t = pipes.Template()
  125. self.assertRaises(ValueError, t.append, 'boguscmd',
  126. pipes.STDIN_FILEOUT)
  127. def testBadPrependOptions(self):
  128. t = pipes.Template()
  129. # try a non-string command
  130. self.assertRaises(TypeError, t.prepend, 7, pipes.STDIN_STDOUT)
  131. # try a type that isn't recognized
  132. self.assertRaises(ValueError, t.prepend, 'tr a-z A-Z', 'xx')
  133. # shouldn't be able to prepend a sink
  134. self.assertRaises(ValueError, t.prepend, 'boguscmd', pipes.SINK)
  135. # check prepending two sources
  136. t = pipes.Template()
  137. t.prepend('boguscmd', pipes.SOURCE)
  138. self.assertRaises(ValueError, t.prepend, 'boguscmd', pipes.SOURCE)
  139. # command needing file input but with no $IN
  140. t = pipes.Template()
  141. self.assertRaises(ValueError, t.prepend, 'boguscmd $OUT',
  142. pipes.FILEIN_FILEOUT)
  143. t = pipes.Template()
  144. self.assertRaises(ValueError, t.prepend, 'boguscmd',
  145. pipes.FILEIN_STDOUT)
  146. # command needing file output but with no $OUT
  147. t = pipes.Template()
  148. self.assertRaises(ValueError, t.prepend, 'boguscmd $IN',
  149. pipes.FILEIN_FILEOUT)
  150. t = pipes.Template()
  151. self.assertRaises(ValueError, t.prepend, 'boguscmd',
  152. pipes.STDIN_FILEOUT)
  153. def testBadOpenMode(self):
  154. t = pipes.Template()
  155. self.assertRaises(ValueError, t.open, 'bogusfile', 'x')
  156. def testClone(self):
  157. t = pipes.Template()
  158. t.append('tr a-z A-Z', pipes.STDIN_STDOUT)
  159. u = t.clone()
  160. self.assertNotEqual(id(t), id(u))
  161. self.assertEqual(t.steps, u.steps)
  162. self.assertNotEqual(id(t.steps), id(u.steps))
  163. self.assertEqual(t.debugging, u.debugging)
  164. def test_main():
  165. run_unittest(SimplePipeTests)
  166. reap_children()
  167. if __name__ == "__main__":
  168. test_main()