~ubuntu-branches/ubuntu/maverick/python3.1/maverick

« back to all changes in this revision

Viewing changes to Lib/test/test_pipes.py

  • Committer: Bazaar Package Importer
  • Author(s): Matthias Klose
  • Date: 2009-03-23 00:01:27 UTC
  • Revision ID: james.westby@ubuntu.com-20090323000127-5fstfxju4ufrhthq
Tags: upstream-3.1~a1+20090322
ImportĀ upstreamĀ versionĀ 3.1~a1+20090322

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
import pipes
 
2
import os
 
3
import string
 
4
import unittest
 
5
from 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
        open(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
        open(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
        open(TESTFN, 'w').write(d)
 
45
        open(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
        open(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()