1 import pipes

     2 import os

     3 import string

     4 import unittest

     5 from test.test_support import TESTFN, run_unittest, unlink, TestSkipped

     6 

     7 if os.name != 'posix':

     8     raise TestSkipped('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         self.assertEqual(t.open(TESTFN, 'r').read(), 'HELLO WORLD #2')

    40 

    41     def testEmptyPipeline1(self):

    42         # copy through empty pipe

    43         d = 'empty pipeline test COPY'

    44         file(TESTFN, 'w').write(d)

    45         file(TESTFN2, 'w').write('')

    46         t=pipes.Template()

    47         t.copy(TESTFN, TESTFN2)

    48         self.assertEqual(open(TESTFN2).read(), d)

    49 

    50     def testEmptyPipeline2(self):

    51         # read through empty pipe

    52         d = 'empty pipeline test READ'

    53         file(TESTFN, 'w').write(d)

    54         t=pipes.Template()

    55         self.assertEqual(t.open(TESTFN, 'r').read(), d)

    56 

    57     def testEmptyPipeline3(self):

    58         # write through empty pipe

    59         d = 'empty pipeline test WRITE'

    60         t = pipes.Template()

    61         t.open(TESTFN, 'w').write(d)

    62         self.assertEqual(open(TESTFN).read(), d)

    63 

    64     def testQuoting(self):

    65         safeunquoted = string.ascii_letters + string.digits + '!@%_-+=:,./'

    66         unsafe = '"`$\\'

    67 

    68         self.assertEqual(pipes.quote(safeunquoted), safeunquoted)

    69         self.assertEqual(pipes.quote('test file name'), "'test file name'")

    70         for u in unsafe:

    71             self.assertEqual(pipes.quote('test%sname' % u),

    72                               "'test%sname'" % u)

    73         for u in unsafe:

    74             self.assertEqual(pipes.quote("test%s'name'" % u),

    75                               '"test\\%s\'name\'"' % u)

    76 

    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 

    84     def testSetDebug(self):

    85         t = pipes.Template()

    86         t.debug(False)

    87         self.assertEqual(t.debugging, False)

    88         t.debug(True)

    89         self.assertEqual(t.debugging, True)

    90 

    91     def testReadOpenSink(self):

    92         # check calling open('r') on a pipe ending with

    93         # a sink raises ValueError

    94         t = pipes.Template()

    95         t.append('boguscmd', pipes.SINK)

    96         self.assertRaises(ValueError, t.open, 'bogusfile', 'r')

    97 

    98     def testWriteOpenSource(self):

    99         # check calling open('w') on a pipe ending with

   100         # a source raises ValueError

   101         t = pipes.Template()

   102         t.prepend('boguscmd', pipes.SOURCE)

   103         self.assertRaises(ValueError, t.open, 'bogusfile', 'w')

   104 

   105     def testBadAppendOptions(self):

   106         t = pipes.Template()

   107 

   108         # try a non-string command

   109         self.assertRaises(TypeError, t.append, 7, pipes.STDIN_STDOUT)

   110 

   111         # try a type that isn't recognized

   112         self.assertRaises(ValueError, t.append, 'boguscmd', 'xx')

   113 

   114         # shouldn't be able to append a source

   115         self.assertRaises(ValueError, t.append, 'boguscmd', pipes.SOURCE)

   116 

   117         # check appending two sinks

   118         t = pipes.Template()

   119         t.append('boguscmd', pipes.SINK)

   120         self.assertRaises(ValueError, t.append, 'boguscmd', pipes.SINK)

   121 

   122         # command needing file input but with no $IN

   123         t = pipes.Template()

   124         self.assertRaises(ValueError, t.append, 'boguscmd $OUT',

   125                            pipes.FILEIN_FILEOUT)

   126         t = pipes.Template()

   127         self.assertRaises(ValueError, t.append, 'boguscmd',

   128                            pipes.FILEIN_STDOUT)

   129 

   130         # command needing file output but with no $OUT

   131         t = pipes.Template()

   132         self.assertRaises(ValueError, t.append, 'boguscmd $IN',

   133                            pipes.FILEIN_FILEOUT)

   134         t = pipes.Template()

   135         self.assertRaises(ValueError, t.append, 'boguscmd',

   136                            pipes.STDIN_FILEOUT)

   137 

   138 

   139     def testBadPrependOptions(self):

   140         t = pipes.Template()

   141 

   142         # try a non-string command

   143         self.assertRaises(TypeError, t.prepend, 7, pipes.STDIN_STDOUT)

   144 

   145         # try a type that isn't recognized

   146         self.assertRaises(ValueError, t.prepend, 'tr a-z A-Z', 'xx')

   147 

   148         # shouldn't be able to prepend a sink

   149         self.assertRaises(ValueError, t.prepend, 'boguscmd', pipes.SINK)

   150 

   151         # check prepending two sources

   152         t = pipes.Template()

   153         t.prepend('boguscmd', pipes.SOURCE)

   154         self.assertRaises(ValueError, t.prepend, 'boguscmd', pipes.SOURCE)

   155 

   156         # command needing file input but with no $IN

   157         t = pipes.Template()

   158         self.assertRaises(ValueError, t.prepend, 'boguscmd $OUT',

   159                            pipes.FILEIN_FILEOUT)

   160         t = pipes.Template()

   161         self.assertRaises(ValueError, t.prepend, 'boguscmd',

   162                            pipes.FILEIN_STDOUT)

   163 

   164         # command needing file output but with no $OUT

   165         t = pipes.Template()

   166         self.assertRaises(ValueError, t.prepend, 'boguscmd $IN',

   167                            pipes.FILEIN_FILEOUT)

   168         t = pipes.Template()

   169         self.assertRaises(ValueError, t.prepend, 'boguscmd',

   170                            pipes.STDIN_FILEOUT)

   171 

   172     def testBadOpenMode(self):

   173         t = pipes.Template()

   174         self.assertRaises(ValueError, t.open, 'bogusfile', 'x')

   175 

   176     def testClone(self):

   177         t = pipes.Template()

   178         t.append('tr a-z A-Z', pipes.STDIN_STDOUT)

   179 

   180         u = t.clone()

   181         self.assertNotEqual(id(t), id(u))

   182         self.assertEqual(t.steps, u.steps)

   183         self.assertNotEqual(id(t.steps), id(u.steps))

   184         self.assertEqual(t.debugging, u.debugging)

   185 

   186 def test_main():

   187     run_unittest(SimplePipeTests)

   188 

   189 if __name__ == "__main__":

   190     test_main()