← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/testtransport.py

  • Committer: John Arbash Meinel
  • Date: 2005-09-15 21:35:53 UTC
  • mfrom: (907.1.57)
  • mto: (1393.2.1)
  • mto: This revision was merged to the branch mainline in revision 1396.
  • Revision ID: john@arbash-meinel.com-20050915213552-a6c83a5ef1e20897
(broken) Transport work is merged in. Tests do not pass yet.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/selftest/testtransport.py
 
1
# Copyright (C) 2004, 2005 by Canonical Ltd
 
2
 
 
3
# This program is free software; you can redistribute it and/or modify
 
4
# it under the terms of the GNU General Public License as published by
 
5
# the Free Software Foundation; either version 2 of the License, or
 
6
# (at your option) any later version.
 
7
 
 
8
# This program is distributed in the hope that it will be useful,
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
11
# GNU General Public License for more details.
 
12
 
 
13
# You should have received a copy of the GNU General Public License
 
14
# along with this program; if not, write to the Free Software
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
16
 
 
17
 
 
18
from bzrlib.selftest import TestCaseInTempDir
 
19
 
 
20
def test_transport(tester, t, readonly=False):
 
21
    """Test a transport object. Basically, it assumes that the
 
22
    Transport object is connected to the current working directory.
 
23
    So that whatever is done through the transport, should show
 
24
    up in the working directory, and vice-versa.
 
25
 
 
26
    This also tests to make sure that the functions work with both
 
27
    generators and lists (assuming iter(list) is effectively a generator)
 
28
    """
 
29
    import tempfile, os
 
30
    from bzrlib.transport.local import LocalTransport
 
31
 
 
32
    # Test has
 
33
    files = ['a', 'b', 'e', 'g']
 
34
    tester.build_tree(files)
 
35
    tester.assertEqual(t.has('a'), True)
 
36
    tester.assertEqual(t.has('c'), False)
 
37
    tester.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
 
38
            [True, True, False, False, True, False, True, False])
 
39
    tester.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),
 
40
            [True, True, False, False, True, False, True, False])
 
41
 
 
42
    # Test get
 
43
    tester.assertEqual(t.get('a').read(), open('a').read())
 
44
    content_f = t.get_multi(files)
 
45
    for path,f in zip(files, content_f):
 
46
        tester.assertEqual(open(path).read(), f.read())
 
47
 
 
48
    content_f = t.get_multi(iter(files))
 
49
    for path,f in zip(files, content_f):
 
50
        tester.assertEqual(open(path).read(), f.read())
 
51
 
 
52
    tester.assertRaises(NoSuchFile, t.get, 'c')
 
53
    try:
 
54
        files = list(t.get_multi(['a', 'b', 'c']))
 
55
    except NoSuchFile:
 
56
        pass
 
57
    else:
 
58
        tester.fail('Failed to raise NoSuchFile for missing file in get_multi')
 
59
    try:
 
60
        files = list(t.get_multi(iter(['a', 'b', 'c', 'e'])))
 
61
    except NoSuchFile:
 
62
        pass
 
63
    else:
 
64
        tester.fail('Failed to raise NoSuchFile for missing file in get_multi')
 
65
 
 
66
    # Test put
 
67
    if readonly:
 
68
        open('c', 'wb').write('some text for c\n')
 
69
    else:
 
70
        t.put('c', 'some text for c\n')
 
71
    tester.assert_(os.path.exists('c'))
 
72
    tester.assertEqual(open('c').read(), 'some text for c\n')
 
73
    tester.assertEqual(t.get('c').read(), 'some text for c\n')
 
74
    # Make sure 'has' is updated
 
75
    tester.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
 
76
            [True, True, True, False, True, False, True, False])
 
77
    if readonly:
 
78
        open('a', 'wb').write('new\ncontents for\na\n')
 
79
        open('d', 'wb').write('contents\nfor d\n')
 
80
    else:
 
81
        # Put also replaces contents
 
82
        tester.assertEqual(t.put_multi([('a', 'new\ncontents for\na\n'),
 
83
                                      ('d', 'contents\nfor d\n')]),
 
84
                         2)
 
85
    tester.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
 
86
            [True, True, True, True, True, False, True, False])
 
87
    tester.assertEqual(open('a').read(), 'new\ncontents for\na\n')
 
88
    tester.assertEqual(open('d').read(), 'contents\nfor d\n')
 
89
 
 
90
    if readonly:
 
91
        open('a', 'wb').write('diff\ncontents for\na\n')
 
92
        open('d', 'wb').write('another contents\nfor d\n')
 
93
    else:
 
94
        tester.assertEqual(
 
95
            t.put_multi(iter([('a', 'diff\ncontents for\na\n'),
 
96
                              ('d', 'another contents\nfor d\n')]))
 
97
                         , 2)
 
98
    tester.assertEqual(open('a').read(), 'diff\ncontents for\na\n')
 
99
    tester.assertEqual(open('d').read(), 'another contents\nfor d\n')
 
100
 
 
101
    if not readonly:
 
102
        tester.assertRaises(NoSuchFile, t.put, 'path/doesnt/exist/c', 'contents')
 
103
 
 
104
    # Test mkdir
 
105
    os.mkdir('dir_a')
 
106
    tester.assertEqual(t.has('dir_a'), True)
 
107
    tester.assertEqual(t.has('dir_b'), False)
 
108
 
 
109
    if readonly:
 
110
        os.mkdir('dir_b')
 
111
    else:
 
112
        t.mkdir('dir_b')
 
113
    tester.assertEqual(t.has('dir_b'), True)
 
114
    tester.assert_(os.path.isdir('dir_b'))
 
115
 
 
116
    if readonly:
 
117
        os.mkdir('dir_c')
 
118
        os.mkdir('dir_d')
 
119
    else:
 
120
        t.mkdir_multi(['dir_c', 'dir_d'])
 
121
    tester.assertEqual(list(t.has_multi(['dir_a', 'dir_b', 'dir_c', 'dir_d', 'dir_e', 'dir_b'])),
 
122
            [True, True, True, True, False, True])
 
123
    for d in ['dir_a', 'dir_b', 'dir_c', 'dir_d']:
 
124
        tester.assert_(os.path.isdir(d))
 
125
 
 
126
    if not readonly:
 
127
        tester.assertRaises(NoSuchFile, t.mkdir, 'path/doesnt/exist')
 
128
        tester.assertRaises(FileExists, t.mkdir, 'dir_a') # Creating a directory again should fail
 
129
 
 
130
    # Make sure the transport recognizes when a
 
131
    # directory is created by other means
 
132
    # Caching Transports will fail, because dir_e was already seen not
 
133
    # to exist. So instead, we will search for a new directory
 
134
    #os.mkdir('dir_e')
 
135
    #if not readonly:
 
136
    #    tester.assertRaises(FileExists, t.mkdir, 'dir_e')
 
137
 
 
138
    os.mkdir('dir_f')
 
139
    if not readonly:
 
140
        tester.assertRaises(FileExists, t.mkdir, 'dir_f')
 
141
 
 
142
    # Test get/put in sub-directories
 
143
    if readonly:
 
144
        open('dir_a/a', 'wb').write('contents of dir_a/a')
 
145
        open('dir_b/b', 'wb').write('contents of dir_b/b')
 
146
    else:
 
147
        tester.assertEqual(
 
148
            t.put_multi([('dir_a/a', 'contents of dir_a/a'),
 
149
                         ('dir_b/b', 'contents of dir_b/b')])
 
150
                      , 2)
 
151
    for f in ('dir_a/a', 'dir_b/b'):
 
152
        tester.assertEqual(t.get(f).read(), open(f).read())
 
153
 
 
154
    # Test copy_to
 
155
    dtmp = tempfile.mkdtemp(dir='.', prefix='test-transport-')
 
156
    dtmp_base = os.path.basename(dtmp)
 
157
    local_t = LocalTransport(dtmp)
 
158
 
 
159
    files = ['a', 'b', 'c', 'd']
 
160
    t.copy_to(files, local_t)
 
161
    for f in files:
 
162
        tester.assertEquals(open(f).read(), open(os.path.join(dtmp_base, f)).read())
 
163
 
 
164
    # TODO: Test append
 
165
    # TODO: Make sure all entries support file-like objects as well as strings.
 
166
 
 
167
class LocalTransportTest(TestCaseInTempDir):
 
168
    def test_local_transport(self):
 
169
        from bzrlib.transport.local import LocalTransport
 
170
 
 
171
        t = LocalTransport('.')
 
172
        test_transport(self, t)
 
173
 
 
174
class HttpServer(object):
 
175
    """This just encapsulates spawning and stopping
 
176
    an httpserver.
 
177
    """
 
178
    def __init__(self):
 
179
        """This just spawns a separate process to serve files from
 
180
        this directory. Call the .stop() function to kill the
 
181
        process.
 
182
        """
 
183
        from BaseHTTPServer import HTTPServer
 
184
        from SimpleHTTPServer import SimpleHTTPRequestHandler
 
185
        import os
 
186
        if hasattr(os, 'fork'):
 
187
            self.pid = os.fork()
 
188
            if self.pid != 0:
 
189
                return
 
190
        else: # How do we handle windows, which doesn't have fork?
 
191
            raise NotImplementedError('At present HttpServer cannot fork on Windows')
 
192
 
 
193
            # We might be able to do something like os.spawn() for the
 
194
            # python executable, and give it a simple script to run.
 
195
            # but then how do we kill it?
 
196
 
 
197
        try:
 
198
            self.s = HTTPServer(('', 9999), SimpleHTTPRequestHandler)
 
199
            # TODO: Is there something nicer than killing the server when done?
 
200
            self.s.serve_forever()
 
201
        except KeyboardInterrupt:
 
202
            pass
 
203
        os._exit(0)
 
204
 
 
205
    def stop(self):
 
206
        import os
 
207
        if self.pid is None:
 
208
            return
 
209
        if hasattr(os, 'kill'):
 
210
            import signal
 
211
            os.kill(self.pid, signal.SIGINT)
 
212
            os.waitpid(self.pid, 0)
 
213
            self.pid = None
 
214
        else:
 
215
            raise NotImplementedError('At present HttpServer cannot stop on Windows')
 
216
 
 
217
class HttpTransportTest(TestCaseInTempDir):
 
218
    def test_http_transport(self):
 
219
        from bzrlib.transport.http import HttpTransport
 
220
 
 
221
        s = HttpServer()
 
222
 
 
223
        t = HttpTransport('http://localhost:9999/')
 
224
        test_transport(self, t, readonly=True)
 
225
 
 
226
        s.stop()
 
227