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