~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

  • Committer: Martin Pool
  • Date: 2005-07-11 04:53:07 UTC
  • Revision ID: mbp@sourcefrog.net-20050711045307-2b38378d043dc25c
- Refactor weave calculation of inclusions

Show diffs side-by-side

added added

removed removed

Lines of Context:
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
 
import os
19
 
from cStringIO import StringIO
20
 
 
21
 
from bzrlib.errors import (NoSuchFile, FileExists,
22
 
                           TransportNotPossible, ConnectionError)
23
 
from bzrlib.tests import TestCase, TestCaseInTempDir
24
 
from bzrlib.tests.HTTPTestUtil import TestCaseWithWebserver
25
 
from bzrlib.transport import memory, urlescape
26
 
 
27
 
 
28
 
def _append(fn, txt):
29
 
    """Append the given text (file-like object) to the supplied filename."""
30
 
    f = open(fn, 'ab')
31
 
    f.write(txt)
32
 
    f.flush()
33
 
    f.close()
34
 
    del f
35
 
 
36
 
class TestTransport(TestCase):
37
 
    """Test the non transport-concrete class functionality."""
38
 
 
39
 
    def test_urlescape(self):
40
 
        self.assertEqual('%25', urlescape('%'))
41
 
 
42
 
 
43
 
class TestTransportMixIn(object):
44
 
    """Subclass this, and it will provide a series of tests for a Transport.
45
 
    It assumes that the Transport object is connected to the 
46
 
    current working directory.  So that whatever is done 
47
 
    through the transport, should show up in the working 
48
 
    directory, and vice-versa.
49
 
 
50
 
    This also tests to make sure that the functions work with both
51
 
    generators and lists (assuming iter(list) is effectively a generator)
52
 
    """
53
 
    readonly = False
54
 
    def get_transport(self):
55
 
        """Children should override this to return the Transport object.
56
 
        """
57
 
        raise NotImplementedError
58
 
 
59
 
    def assertListRaises(self, excClass, func, *args, **kwargs):
60
 
        """Many transport functions can return generators this makes sure
61
 
        to wrap them in a list() call to make sure the whole generator
62
 
        is run, and that the proper exception is raised.
63
 
        """
64
 
        try:
65
 
            list(func(*args, **kwargs))
66
 
        except excClass:
67
 
            return
68
 
        else:
69
 
            if hasattr(excClass,'__name__'): excName = excClass.__name__
70
 
            else: excName = str(excClass)
71
 
            raise self.failureException, "%s not raised" % excName
72
 
 
73
 
    def test_has(self):
74
 
        t = self.get_transport()
75
 
 
76
 
        files = ['a', 'b', 'e', 'g', '%']
77
 
        self.build_tree(files)
78
 
        self.assertEqual(True, t.has('a'))
79
 
        self.assertEqual(False, t.has('c'))
80
 
        self.assertEqual(True, t.has(urlescape('%')))
81
 
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
82
 
                [True, True, False, False, True, False, True, False])
83
 
        self.assertEqual(True, t.has_any(['a', 'b', 'c']))
84
 
        self.assertEqual(False, t.has_any(['c', 'd', 'f', urlescape('%%')]))
85
 
        self.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),
86
 
                [True, True, False, False, True, False, True, False])
87
 
        self.assertEqual(False, t.has_any(['c', 'c', 'c']))
88
 
        self.assertEqual(True, t.has_any(['b', 'b', 'b']))
89
 
 
90
 
    def test_get(self):
91
 
        t = self.get_transport()
92
 
 
93
 
        files = ['a', 'b', 'e', 'g']
94
 
        self.build_tree(files)
95
 
        self.assertEqual(open('a', 'rb').read(), t.get('a').read())
96
 
        content_f = t.get_multi(files)
97
 
        for path,f in zip(files, content_f):
98
 
            self.assertEqual(open(path).read(), f.read())
99
 
 
100
 
        content_f = t.get_multi(iter(files))
101
 
        for path,f in zip(files, content_f):
102
 
            self.assertEqual(f.read(), open(path).read())
103
 
 
104
 
        self.assertRaises(NoSuchFile, t.get, 'c')
105
 
        self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
106
 
        self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))
107
 
 
108
 
    def test_put(self):
109
 
        t = self.get_transport()
110
 
 
111
 
        if self.readonly:
112
 
            self.assertRaises(TransportNotPossible,
113
 
                    t.put, 'a', 'some text for a\n')
114
 
            open('a', 'wb').write('some text for a\n')
115
 
        else:
116
 
            t.put('a', 'some text for a\n')
117
 
        self.assert_(os.path.exists('a'))
118
 
        self.check_file_contents('a', 'some text for a\n')
119
 
        self.assertEqual(t.get('a').read(), 'some text for a\n')
120
 
        # Make sure 'has' is updated
121
 
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
122
 
                [True, False, False, False, False])
123
 
        if self.readonly:
124
 
            self.assertRaises(TransportNotPossible,
125
 
                    t.put_multi,
126
 
                    [('a', 'new\ncontents for\na\n'),
127
 
                        ('d', 'contents\nfor d\n')])
128
 
            open('a', 'wb').write('new\ncontents for\na\n')
129
 
            open('d', 'wb').write('contents\nfor d\n')
130
 
        else:
131
 
            # Put also replaces contents
132
 
            self.assertEqual(t.put_multi([('a', 'new\ncontents for\na\n'),
133
 
                                          ('d', 'contents\nfor d\n')]),
134
 
                             2)
135
 
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
136
 
                [True, False, False, True, False])
137
 
        self.check_file_contents('a', 'new\ncontents for\na\n')
138
 
        self.check_file_contents('d', 'contents\nfor d\n')
139
 
 
140
 
        if self.readonly:
141
 
            self.assertRaises(TransportNotPossible,
142
 
                t.put_multi, iter([('a', 'diff\ncontents for\na\n'),
143
 
                                  ('d', 'another contents\nfor d\n')]))
144
 
            open('a', 'wb').write('diff\ncontents for\na\n')
145
 
            open('d', 'wb').write('another contents\nfor d\n')
146
 
        else:
147
 
            self.assertEqual(
148
 
                t.put_multi(iter([('a', 'diff\ncontents for\na\n'),
149
 
                                  ('d', 'another contents\nfor d\n')]))
150
 
                             , 2)
151
 
        self.check_file_contents('a', 'diff\ncontents for\na\n')
152
 
        self.check_file_contents('d', 'another contents\nfor d\n')
153
 
 
154
 
        if self.readonly:
155
 
            self.assertRaises(TransportNotPossible,
156
 
                    t.put, 'path/doesnt/exist/c', 'contents')
157
 
        else:
158
 
            self.assertRaises(NoSuchFile,
159
 
                    t.put, 'path/doesnt/exist/c', 'contents')
160
 
 
161
 
    def test_put_file(self):
162
 
        t = self.get_transport()
163
 
 
164
 
        # Test that StringIO can be used as a file-like object with put
165
 
        f1 = StringIO('this is a string\nand some more stuff\n')
166
 
        if self.readonly:
167
 
            open('f1', 'wb').write(f1.read())
168
 
        else:
169
 
            t.put('f1', f1)
170
 
 
171
 
        del f1
172
 
 
173
 
        self.check_file_contents('f1', 
174
 
                'this is a string\nand some more stuff\n')
175
 
 
176
 
        f2 = StringIO('here is some text\nand a bit more\n')
177
 
        f3 = StringIO('some text for the\nthird file created\n')
178
 
 
179
 
        if self.readonly:
180
 
            open('f2', 'wb').write(f2.read())
181
 
            open('f3', 'wb').write(f3.read())
182
 
        else:
183
 
            t.put_multi([('f2', f2), ('f3', f3)])
184
 
 
185
 
        del f2, f3
186
 
 
187
 
        self.check_file_contents('f2', 'here is some text\nand a bit more\n')
188
 
        self.check_file_contents('f3', 'some text for the\nthird file created\n')
189
 
 
190
 
        # Test that an actual file object can be used with put
191
 
        f4 = open('f1', 'rb')
192
 
        if self.readonly:
193
 
            open('f4', 'wb').write(f4.read())
194
 
        else:
195
 
            t.put('f4', f4)
196
 
 
197
 
        del f4
198
 
 
199
 
        self.check_file_contents('f4', 
200
 
                'this is a string\nand some more stuff\n')
201
 
 
202
 
        f5 = open('f2', 'rb')
203
 
        f6 = open('f3', 'rb')
204
 
        if self.readonly:
205
 
            open('f5', 'wb').write(f5.read())
206
 
            open('f6', 'wb').write(f6.read())
207
 
        else:
208
 
            t.put_multi([('f5', f5), ('f6', f6)])
209
 
 
210
 
        del f5, f6
211
 
 
212
 
        self.check_file_contents('f5', 'here is some text\nand a bit more\n')
213
 
        self.check_file_contents('f6', 'some text for the\nthird file created\n')
214
 
 
215
 
    def test_mkdir(self):
216
 
        t = self.get_transport()
217
 
 
218
 
        # Test mkdir
219
 
        os.mkdir('dir_a')
220
 
        self.assertEqual(t.has('dir_a'), True)
221
 
        self.assertEqual(t.has('dir_b'), False)
222
 
 
223
 
        if self.readonly:
224
 
            self.assertRaises(TransportNotPossible,
225
 
                    t.mkdir, 'dir_b')
226
 
            os.mkdir('dir_b')
227
 
        else:
228
 
            t.mkdir('dir_b')
229
 
        self.assertEqual(t.has('dir_b'), True)
230
 
        self.assert_(os.path.isdir('dir_b'))
231
 
 
232
 
        if self.readonly:
233
 
            self.assertRaises(TransportNotPossible,
234
 
                    t.mkdir_multi, ['dir_c', 'dir_d'])
235
 
            os.mkdir('dir_c')
236
 
            os.mkdir('dir_d')
237
 
        else:
238
 
            t.mkdir_multi(['dir_c', 'dir_d'])
239
 
 
240
 
        if self.readonly:
241
 
            self.assertRaises(TransportNotPossible,
242
 
                    t.mkdir_multi, iter(['dir_e', 'dir_f']))
243
 
            os.mkdir('dir_e')
244
 
            os.mkdir('dir_f')
245
 
        else:
246
 
            t.mkdir_multi(iter(['dir_e', 'dir_f']))
247
 
        self.assertEqual(list(t.has_multi(
248
 
            ['dir_a', 'dir_b', 'dir_c', 'dir_q',
249
 
             'dir_d', 'dir_e', 'dir_f', 'dir_b'])),
250
 
            [True, True, True, False,
251
 
             True, True, True, True])
252
 
        for d in ['dir_a', 'dir_b', 'dir_c', 'dir_d', 'dir_e', 'dir_f']:
253
 
            self.assert_(os.path.isdir(d))
254
 
 
255
 
        if not self.readonly:
256
 
            self.assertRaises(NoSuchFile, t.mkdir, 'path/doesnt/exist')
257
 
            self.assertRaises(FileExists, t.mkdir, 'dir_a') # Creating a directory again should fail
258
 
 
259
 
        # Make sure the transport recognizes when a
260
 
        # directory is created by other means
261
 
        # Caching Transports will fail, because dir_e was already seen not
262
 
        # to exist. So instead, we will search for a new directory
263
 
        #os.mkdir('dir_e')
264
 
        #if not self.readonly:
265
 
        #    self.assertRaises(FileExists, t.mkdir, 'dir_e')
266
 
 
267
 
        os.mkdir('dir_g')
268
 
        if not self.readonly:
269
 
            self.assertRaises(FileExists, t.mkdir, 'dir_g')
270
 
 
271
 
        # Test get/put in sub-directories
272
 
        if self.readonly:
273
 
            open('dir_a/a', 'wb').write('contents of dir_a/a')
274
 
            open('dir_b/b', 'wb').write('contents of dir_b/b')
275
 
        else:
276
 
            self.assertEqual(
277
 
                t.put_multi([('dir_a/a', 'contents of dir_a/a'),
278
 
                             ('dir_b/b', 'contents of dir_b/b')])
279
 
                          , 2)
280
 
        for f in ('dir_a/a', 'dir_b/b'):
281
 
            self.assertEqual(t.get(f).read(), open(f).read())
282
 
 
283
 
    def test_copy_to(self):
284
 
        import tempfile
285
 
        from bzrlib.transport.local import LocalTransport
286
 
 
287
 
        t = self.get_transport()
288
 
 
289
 
        files = ['a', 'b', 'c', 'd']
290
 
        self.build_tree(files)
291
 
 
292
 
        dtmp = tempfile.mkdtemp(dir=u'.', prefix='test-transport-')
293
 
        dtmp_base = os.path.basename(dtmp)
294
 
        local_t = LocalTransport(dtmp)
295
 
 
296
 
        t.copy_to(files, local_t)
297
 
        for f in files:
298
 
            self.assertEquals(open(f).read(),
299
 
                    open(os.path.join(dtmp_base, f)).read())
300
 
 
301
 
        # Test that copying into a missing directory raises
302
 
        # NoSuchFile
303
 
        os.mkdir('e')
304
 
        open('e/f', 'wb').write('contents of e')
305
 
        self.assertRaises(NoSuchFile, t.copy_to, ['e/f'], local_t)
306
 
 
307
 
        os.mkdir(os.path.join(dtmp_base, 'e'))
308
 
        t.copy_to(['e/f'], local_t)
309
 
 
310
 
        del dtmp, dtmp_base, local_t
311
 
 
312
 
        dtmp = tempfile.mkdtemp(dir=u'.', prefix='test-transport-')
313
 
        dtmp_base = os.path.basename(dtmp)
314
 
        local_t = LocalTransport(dtmp)
315
 
 
316
 
        files = ['a', 'b', 'c', 'd']
317
 
        t.copy_to(iter(files), local_t)
318
 
        for f in files:
319
 
            self.assertEquals(open(f).read(),
320
 
                    open(os.path.join(dtmp_base, f)).read())
321
 
 
322
 
        del dtmp, dtmp_base, local_t
323
 
 
324
 
    def test_append(self):
325
 
        t = self.get_transport()
326
 
 
327
 
        if self.readonly:
328
 
            open('a', 'wb').write('diff\ncontents for\na\n')
329
 
            open('b', 'wb').write('contents\nfor b\n')
330
 
        else:
331
 
            t.put_multi([
332
 
                    ('a', 'diff\ncontents for\na\n'),
333
 
                    ('b', 'contents\nfor b\n')
334
 
                    ])
335
 
 
336
 
        if self.readonly:
337
 
            self.assertRaises(TransportNotPossible,
338
 
                    t.append, 'a', 'add\nsome\nmore\ncontents\n')
339
 
            _append('a', 'add\nsome\nmore\ncontents\n')
340
 
        else:
341
 
            t.append('a', 'add\nsome\nmore\ncontents\n')
342
 
 
343
 
        self.check_file_contents('a', 
344
 
            'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n')
345
 
 
346
 
        if self.readonly:
347
 
            self.assertRaises(TransportNotPossible,
348
 
                    t.append_multi,
349
 
                        [('a', 'and\nthen\nsome\nmore\n'),
350
 
                         ('b', 'some\nmore\nfor\nb\n')])
351
 
            _append('a', 'and\nthen\nsome\nmore\n')
352
 
            _append('b', 'some\nmore\nfor\nb\n')
353
 
        else:
354
 
            t.append_multi([('a', 'and\nthen\nsome\nmore\n'),
355
 
                    ('b', 'some\nmore\nfor\nb\n')])
356
 
        self.check_file_contents('a', 
357
 
            'diff\ncontents for\na\n'
358
 
            'add\nsome\nmore\ncontents\n'
359
 
            'and\nthen\nsome\nmore\n')
360
 
        self.check_file_contents('b', 
361
 
                'contents\nfor b\n'
362
 
                'some\nmore\nfor\nb\n')
363
 
 
364
 
        if self.readonly:
365
 
            _append('a', 'a little bit more\n')
366
 
            _append('b', 'from an iterator\n')
367
 
        else:
368
 
            t.append_multi(iter([('a', 'a little bit more\n'),
369
 
                    ('b', 'from an iterator\n')]))
370
 
        self.check_file_contents('a', 
371
 
            'diff\ncontents for\na\n'
372
 
            'add\nsome\nmore\ncontents\n'
373
 
            'and\nthen\nsome\nmore\n'
374
 
            'a little bit more\n')
375
 
        self.check_file_contents('b', 
376
 
                'contents\nfor b\n'
377
 
                'some\nmore\nfor\nb\n'
378
 
                'from an iterator\n')
379
 
 
380
 
        if self.readonly:
381
 
            _append('c', 'some text\nfor a missing file\n')
382
 
            _append('a', 'some text in a\n')
383
 
            _append('d', 'missing file r\n')
384
 
        else:
385
 
            t.append('c', 'some text\nfor a missing file\n')
386
 
            t.append_multi([('a', 'some text in a\n'),
387
 
                            ('d', 'missing file r\n')])
388
 
        self.check_file_contents('a', 
389
 
            'diff\ncontents for\na\n'
390
 
            'add\nsome\nmore\ncontents\n'
391
 
            'and\nthen\nsome\nmore\n'
392
 
            'a little bit more\n'
393
 
            'some text in a\n')
394
 
        self.check_file_contents('c', 'some text\nfor a missing file\n')
395
 
        self.check_file_contents('d', 'missing file r\n')
396
 
 
397
 
    def test_append_file(self):
398
 
        t = self.get_transport()
399
 
 
400
 
        contents = [
401
 
            ('f1', 'this is a string\nand some more stuff\n'),
402
 
            ('f2', 'here is some text\nand a bit more\n'),
403
 
            ('f3', 'some text for the\nthird file created\n'),
404
 
            ('f4', 'this is a string\nand some more stuff\n'),
405
 
            ('f5', 'here is some text\nand a bit more\n'),
406
 
            ('f6', 'some text for the\nthird file created\n')
407
 
        ]
408
 
        
409
 
        if self.readonly:
410
 
            for f, val in contents:
411
 
                open(f, 'wb').write(val)
412
 
        else:
413
 
            t.put_multi(contents)
414
 
 
415
 
        a1 = StringIO('appending to\none\n')
416
 
        if self.readonly:
417
 
            _append('f1', a1.read())
418
 
        else:
419
 
            t.append('f1', a1)
420
 
 
421
 
        del a1
422
 
 
423
 
        self.check_file_contents('f1', 
424
 
                'this is a string\nand some more stuff\n'
425
 
                'appending to\none\n')
426
 
 
427
 
        a2 = StringIO('adding more\ntext to two\n')
428
 
        a3 = StringIO('some garbage\nto put in three\n')
429
 
 
430
 
        if self.readonly:
431
 
            _append('f2', a2.read())
432
 
            _append('f3', a3.read())
433
 
        else:
434
 
            t.append_multi([('f2', a2), ('f3', a3)])
435
 
 
436
 
        del a2, a3
437
 
 
438
 
        self.check_file_contents('f2',
439
 
                'here is some text\nand a bit more\n'
440
 
                'adding more\ntext to two\n')
441
 
        self.check_file_contents('f3', 
442
 
                'some text for the\nthird file created\n'
443
 
                'some garbage\nto put in three\n')
444
 
 
445
 
        # Test that an actual file object can be used with put
446
 
        a4 = open('f1', 'rb')
447
 
        if self.readonly:
448
 
            _append('f4', a4.read())
449
 
        else:
450
 
            t.append('f4', a4)
451
 
 
452
 
        del a4
453
 
 
454
 
        self.check_file_contents('f4', 
455
 
                'this is a string\nand some more stuff\n'
456
 
                'this is a string\nand some more stuff\n'
457
 
                'appending to\none\n')
458
 
 
459
 
        a5 = open('f2', 'rb')
460
 
        a6 = open('f3', 'rb')
461
 
        if self.readonly:
462
 
            _append('f5', a5.read())
463
 
            _append('f6', a6.read())
464
 
        else:
465
 
            t.append_multi([('f5', a5), ('f6', a6)])
466
 
 
467
 
        del a5, a6
468
 
 
469
 
        self.check_file_contents('f5',
470
 
                'here is some text\nand a bit more\n'
471
 
                'here is some text\nand a bit more\n'
472
 
                'adding more\ntext to two\n')
473
 
        self.check_file_contents('f6',
474
 
                'some text for the\nthird file created\n'
475
 
                'some text for the\nthird file created\n'
476
 
                'some garbage\nto put in three\n')
477
 
 
478
 
        a5 = open('f2', 'rb')
479
 
        a6 = open('f2', 'rb')
480
 
        a7 = open('f3', 'rb')
481
 
        if self.readonly:
482
 
            _append('c', a5.read())
483
 
            _append('a', a6.read())
484
 
            _append('d', a7.read())
485
 
        else:
486
 
            t.append('c', a5)
487
 
            t.append_multi([('a', a6), ('d', a7)])
488
 
        del a5, a6, a7
489
 
        self.check_file_contents('c', open('f2', 'rb').read())
490
 
        self.check_file_contents('d', open('f3', 'rb').read())
491
 
 
492
 
 
493
 
    def test_delete(self):
494
 
        # TODO: Test Transport.delete
495
 
        t = self.get_transport()
496
 
 
497
 
        # Not much to do with a readonly transport
498
 
        if self.readonly:
499
 
            return
500
 
 
501
 
        open('a', 'wb').write('a little bit of text\n')
502
 
        self.failUnless(t.has('a'))
503
 
        self.failUnlessExists('a')
504
 
        t.delete('a')
505
 
        self.failIf(os.path.lexists('a'))
506
 
 
507
 
        self.assertRaises(NoSuchFile, t.delete, 'a')
508
 
 
509
 
        open('a', 'wb').write('a text\n')
510
 
        open('b', 'wb').write('b text\n')
511
 
        open('c', 'wb').write('c text\n')
512
 
        self.assertEqual([True, True, True],
513
 
                list(t.has_multi(['a', 'b', 'c'])))
514
 
        t.delete_multi(['a', 'c'])
515
 
        self.assertEqual([False, True, False],
516
 
                list(t.has_multi(['a', 'b', 'c'])))
517
 
        self.failIf(os.path.lexists('a'))
518
 
        self.failUnlessExists('b')
519
 
        self.failIf(os.path.lexists('c'))
520
 
 
521
 
        self.assertRaises(NoSuchFile,
522
 
                t.delete_multi, ['a', 'b', 'c'])
523
 
 
524
 
        self.assertRaises(NoSuchFile,
525
 
                t.delete_multi, iter(['a', 'b', 'c']))
526
 
 
527
 
        open('a', 'wb').write('another a text\n')
528
 
        open('c', 'wb').write('another c text\n')
529
 
        t.delete_multi(iter(['a', 'b', 'c']))
530
 
 
531
 
        # We should have deleted everything
532
 
        # SftpServer creates control files in the
533
 
        # working directory, so we can just do a
534
 
        # plain "listdir".
535
 
        # self.assertEqual([], os.listdir('.'))
536
 
 
537
 
    def test_move(self):
538
 
        t = self.get_transport()
539
 
 
540
 
        if self.readonly:
541
 
            return
542
 
 
543
 
        # TODO: I would like to use os.listdir() to
544
 
        # make sure there are no extra files, but SftpServer
545
 
        # creates control files in the working directory
546
 
        # perhaps all of this could be done in a subdirectory
547
 
 
548
 
        open('a', 'wb').write('a first file\n')
549
 
        self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
550
 
 
551
 
        t.move('a', 'b')
552
 
        self.failUnlessExists('b')
553
 
        self.failIf(os.path.lexists('a'))
554
 
 
555
 
        self.check_file_contents('b', 'a first file\n')
556
 
        self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
557
 
 
558
 
        # Overwrite a file
559
 
        open('c', 'wb').write('c this file\n')
560
 
        t.move('c', 'b')
561
 
        self.failIf(os.path.lexists('c'))
562
 
        self.check_file_contents('b', 'c this file\n')
563
 
 
564
 
        # TODO: Try to write a test for atomicity
565
 
        # TODO: Test moving into a non-existant subdirectory
566
 
        # TODO: Test Transport.move_multi
567
 
 
568
 
    def test_copy(self):
569
 
        t = self.get_transport()
570
 
 
571
 
        if self.readonly:
572
 
            return
573
 
 
574
 
        open('a', 'wb').write('a file\n')
575
 
        t.copy('a', 'b')
576
 
        self.check_file_contents('b', 'a file\n')
577
 
 
578
 
        self.assertRaises(NoSuchFile, t.copy, 'c', 'd')
579
 
        os.mkdir('c')
580
 
        # What should the assert be if you try to copy a
581
 
        # file over a directory?
582
 
        #self.assertRaises(Something, t.copy, 'a', 'c')
583
 
        open('d', 'wb').write('text in d\n')
584
 
        t.copy('d', 'b')
585
 
        self.check_file_contents('b', 'text in d\n')
586
 
 
587
 
        # TODO: test copy_multi
588
 
 
589
 
    def test_connection_error(self):
590
 
        """ConnectionError is raised when connection is impossible"""
591
 
        if not hasattr(self, "get_bogus_transport"):
592
 
            return
593
 
        t = self.get_bogus_transport()
594
 
        try:
595
 
            t.get('.bzr/branch')
596
 
        except (ConnectionError, NoSuchFile), e:
597
 
            pass
598
 
        except (Exception), e:
599
 
            self.failIf(True, 'Wrong exception thrown: %s' % e)
600
 
        else:
601
 
            self.failIf(True, 'Did not get the expected exception.')
602
 
 
603
 
    def test_stat(self):
604
 
        # TODO: Test stat, just try once, and if it throws, stop testing
605
 
        from stat import S_ISDIR, S_ISREG
606
 
 
607
 
        t = self.get_transport()
608
 
 
609
 
        try:
610
 
            st = t.stat('.')
611
 
        except TransportNotPossible, e:
612
 
            # This transport cannot stat
613
 
            return
614
 
 
615
 
        paths = ['a', 'b/', 'b/c', 'b/d/', 'b/d/e']
616
 
        self.build_tree(paths)
617
 
 
618
 
        local_stats = []
619
 
 
620
 
        for p in paths:
621
 
            st = t.stat(p)
622
 
            local_st = os.stat(p)
623
 
            if p.endswith('/'):
624
 
                self.failUnless(S_ISDIR(st.st_mode))
625
 
            else:
626
 
                self.failUnless(S_ISREG(st.st_mode))
627
 
            self.assertEqual(local_st.st_size, st.st_size)
628
 
            self.assertEqual(local_st.st_mode, st.st_mode)
629
 
            local_stats.append(local_st)
630
 
 
631
 
        remote_stats = list(t.stat_multi(paths))
632
 
        remote_iter_stats = list(t.stat_multi(iter(paths)))
633
 
 
634
 
        for local, remote, remote_iter in \
635
 
            zip(local_stats, remote_stats, remote_iter_stats):
636
 
            self.assertEqual(local.st_mode, remote.st_mode)
637
 
            self.assertEqual(local.st_mode, remote_iter.st_mode)
638
 
 
639
 
            self.assertEqual(local.st_size, remote.st_size)
640
 
            self.assertEqual(local.st_size, remote_iter.st_size)
641
 
            # Should we test UID/GID?
642
 
 
643
 
        self.assertRaises(NoSuchFile, t.stat, 'q')
644
 
        self.assertRaises(NoSuchFile, t.stat, 'b/a')
645
 
 
646
 
        self.assertListRaises(NoSuchFile, t.stat_multi, ['a', 'c', 'd'])
647
 
        self.assertListRaises(NoSuchFile, t.stat_multi, iter(['a', 'c', 'd']))
648
 
 
649
 
    def test_list_dir(self):
650
 
        # TODO: Test list_dir, just try once, and if it throws, stop testing
651
 
        t = self.get_transport()
652
 
        
653
 
        if not t.listable():
654
 
            self.assertRaises(TransportNotPossible, t.list_dir, '.')
655
 
            return
656
 
 
657
 
        def sorted_list(d):
658
 
            l = list(t.list_dir(d))
659
 
            l.sort()
660
 
            return l
661
 
 
662
 
        # SftpServer creates control files in the working directory
663
 
        # so lets move down a directory to be safe
664
 
        os.mkdir('wd')
665
 
        os.chdir('wd')
666
 
        t = t.clone('wd')
667
 
 
668
 
        self.assertEqual([], sorted_list(u'.'))
669
 
        self.build_tree(['a', 'b', 'c/', 'c/d', 'c/e'])
670
 
 
671
 
        self.assertEqual([u'a', u'b', u'c'], sorted_list(u'.'))
672
 
        self.assertEqual([u'd', u'e'], sorted_list(u'c'))
673
 
 
674
 
        os.remove('c/d')
675
 
        os.remove('b')
676
 
        self.assertEqual([u'a', u'c'], sorted_list('.'))
677
 
        self.assertEqual([u'e'], sorted_list(u'c'))
678
 
 
679
 
        self.assertListRaises(NoSuchFile, t.list_dir, 'q')
680
 
        self.assertListRaises(NoSuchFile, t.list_dir, 'c/f')
681
 
        self.assertListRaises(NoSuchFile, t.list_dir, 'a')
682
 
 
683
 
    def test_clone(self):
684
 
        # TODO: Test that clone moves up and down the filesystem
685
 
        t1 = self.get_transport()
686
 
 
687
 
        self.build_tree(['a', 'b/', 'b/c'])
688
 
 
689
 
        self.failUnless(t1.has('a'))
690
 
        self.failUnless(t1.has('b/c'))
691
 
        self.failIf(t1.has('c'))
692
 
 
693
 
        t2 = t1.clone('b')
694
 
        self.failUnless(t2.has('c'))
695
 
        self.failIf(t2.has('a'))
696
 
 
697
 
        t3 = t2.clone('..')
698
 
        self.failUnless(t3.has('a'))
699
 
        self.failIf(t3.has('c'))
700
 
 
701
 
        self.failIf(t1.has('b/d'))
702
 
        self.failIf(t2.has('d'))
703
 
        self.failIf(t3.has('b/d'))
704
 
 
705
 
        if self.readonly:
706
 
            open('b/d', 'wb').write('newfile\n')
707
 
        else:
708
 
            t2.put('d', 'newfile\n')
709
 
 
710
 
        self.failUnless(t1.has('b/d'))
711
 
        self.failUnless(t2.has('d'))
712
 
        self.failUnless(t3.has('b/d'))
713
 
 
714
 
        
715
 
class LocalTransportTest(TestCaseInTempDir, TestTransportMixIn):
716
 
    def get_transport(self):
717
 
        from bzrlib.transport.local import LocalTransport
718
 
        return LocalTransport(u'.')
719
 
 
720
 
 
721
 
class HttpTransportTest(TestCaseWithWebserver, TestTransportMixIn):
722
 
 
723
 
    readonly = True
724
 
 
725
 
    def get_transport(self):
726
 
        from bzrlib.transport.http import HttpTransport
727
 
        url = self.get_remote_url(u'.')
728
 
        return HttpTransport(url)
729
 
 
730
 
    def get_bogus_transport(self):
731
 
        from bzrlib.transport.http import HttpTransport
732
 
        return HttpTransport('http://jasldkjsalkdjalksjdkljasd')
733
 
 
734
 
 
735
 
class TestMemoryTransport(TestCase):
736
 
 
737
 
    def test_get_transport(self):
738
 
        memory.MemoryTransport()
739
 
 
740
 
    def test_clone(self):
741
 
        transport = memory.MemoryTransport()
742
 
        self.failUnless(transport.clone() is transport)
743
 
 
744
 
    def test_abspath(self):
745
 
        transport = memory.MemoryTransport()
746
 
        self.assertEqual("in-memory:relpath", transport.abspath('relpath'))
747
 
 
748
 
    def test_relpath(self):
749
 
        transport = memory.MemoryTransport()
750
 
 
751
 
    def test_append_and_get(self):
752
 
        transport = memory.MemoryTransport()
753
 
        transport.append('path', StringIO('content'))
754
 
        self.assertEqual(transport.get('path').read(), 'content')
755
 
        transport.append('path', StringIO('content'))
756
 
        self.assertEqual(transport.get('path').read(), 'contentcontent')
757
 
 
758
 
    def test_put_and_get(self):
759
 
        transport = memory.MemoryTransport()
760
 
        transport.put('path', StringIO('content'))
761
 
        self.assertEqual(transport.get('path').read(), 'content')
762
 
        transport.put('path', StringIO('content'))
763
 
        self.assertEqual(transport.get('path').read(), 'content')
764
 
 
765
 
    def test_append_without_dir_fails(self):
766
 
        transport = memory.MemoryTransport()
767
 
        self.assertRaises(NoSuchFile,
768
 
                          transport.append, 'dir/path', StringIO('content'))
769
 
 
770
 
    def test_put_without_dir_fails(self):
771
 
        transport = memory.MemoryTransport()
772
 
        self.assertRaises(NoSuchFile,
773
 
                          transport.put, 'dir/path', StringIO('content'))
774
 
 
775
 
    def test_get_missing(self):
776
 
        transport = memory.MemoryTransport()
777
 
        self.assertRaises(NoSuchFile, transport.get, 'foo')
778
 
 
779
 
    def test_has_missing(self):
780
 
        transport = memory.MemoryTransport()
781
 
        self.assertEquals(False, transport.has('foo'))
782
 
 
783
 
    def test_has_present(self):
784
 
        transport = memory.MemoryTransport()
785
 
        transport.append('foo', StringIO('content'))
786
 
        self.assertEquals(True, transport.has('foo'))
787
 
 
788
 
    def test_mkdir(self):
789
 
        transport = memory.MemoryTransport()
790
 
        transport.mkdir('dir')
791
 
        transport.append('dir/path', StringIO('content'))
792
 
        self.assertEqual(transport.get('dir/path').read(), 'content')
793
 
 
794
 
    def test_mkdir_missing_parent(self):
795
 
        transport = memory.MemoryTransport()
796
 
        self.assertRaises(NoSuchFile,
797
 
                          transport.mkdir, 'dir/dir')
798
 
 
799
 
    def test_mkdir_twice(self):
800
 
        transport = memory.MemoryTransport()
801
 
        transport.mkdir('dir')
802
 
        self.assertRaises(FileExists, transport.mkdir, 'dir')
803
 
 
804
 
    def test_parameters(self):
805
 
        transport = memory.MemoryTransport()
806
 
        self.assertEqual(True, transport.listable())
807
 
        self.assertEqual(False, transport.should_cache())
808
 
 
809
 
    def test_iter_files_recursive(self):
810
 
        transport = memory.MemoryTransport()
811
 
        transport.mkdir('dir')
812
 
        transport.put('dir/foo', StringIO('content'))
813
 
        transport.put('dir/bar', StringIO('content'))
814
 
        transport.put('bar', StringIO('content'))
815
 
        paths = set(transport.iter_files_recursive())
816
 
        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
817
 
 
818
 
    def test_stat(self):
819
 
        transport = memory.MemoryTransport()
820
 
        transport.put('foo', StringIO('content'))
821
 
        transport.put('bar', StringIO('phowar'))
822
 
        self.assertEqual(7, transport.stat('foo').st_size)
823
 
        self.assertEqual(6, transport.stat('bar').st_size)
824