Lib/test/test_pipes.py

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