~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

MergeĀ fromĀ martin

Show diffs side-by-side

added added

removed removed

Lines of Context:
16
16
 
17
17
 
18
18
import os
 
19
import sys
 
20
import stat
19
21
from cStringIO import StringIO
20
22
 
21
23
from bzrlib.errors import (NoSuchFile, FileExists,
23
25
from bzrlib.tests import TestCase, TestCaseInTempDir
24
26
from bzrlib.tests.HTTPTestUtil import TestCaseWithWebserver
25
27
from bzrlib.transport import memory, urlescape
 
28
from bzrlib.osutils import pathjoin
26
29
 
27
30
 
28
31
def _append(fn, txt):
33
36
    f.close()
34
37
    del f
35
38
 
 
39
 
 
40
if sys.platform != 'win32':
 
41
    def check_mode(test, path, mode):
 
42
        """Check that a particular path has the correct mode."""
 
43
        actual_mode = stat.S_IMODE(os.stat(path).st_mode)
 
44
        test.assertEqual(mode, actual_mode,
 
45
            'mode of %r incorrect (%o != %o)' % (path, mode, actual_mode))
 
46
else:
 
47
    def check_mode(test, path, mode):
 
48
        """On win32 chmod doesn't have any effect, 
 
49
        so don't actually check anything
 
50
        """
 
51
        return
 
52
 
 
53
 
36
54
class TestTransport(TestCase):
37
55
    """Test the non transport-concrete class functionality."""
38
56
 
95
113
        self.assertEqual(open('a', 'rb').read(), t.get('a').read())
96
114
        content_f = t.get_multi(files)
97
115
        for path,f in zip(files, content_f):
98
 
            self.assertEqual(open(path).read(), f.read())
 
116
            self.assertEqual(f.read(), open(path, 'rb').read())
99
117
 
100
118
        content_f = t.get_multi(iter(files))
101
119
        for path,f in zip(files, content_f):
102
 
            self.assertEqual(f.read(), open(path).read())
 
120
            self.assertEqual(f.read(), open(path, 'rb').read())
103
121
 
104
122
        self.assertRaises(NoSuchFile, t.get, 'c')
105
123
        self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
108
126
    def test_put(self):
109
127
        t = self.get_transport()
110
128
 
 
129
        # TODO: jam 20051215 No need to do anything if the test is readonly
 
130
        #                    origininally it was thought that it would give
 
131
        #                    more of a workout to readonly tests. By now the
 
132
        #                    suite is probably thorough enough without testing
 
133
        #                    readonly protocols in write sections
 
134
        #                    The only thing that needs to be tested is that the
 
135
        #                    right error is raised
 
136
 
111
137
        if self.readonly:
112
138
            self.assertRaises(TransportNotPossible,
113
139
                    t.put, 'a', 'some text for a\n')
158
184
            self.assertRaises(NoSuchFile,
159
185
                    t.put, 'path/doesnt/exist/c', 'contents')
160
186
 
 
187
        if not self.readonly:
 
188
            t.put('mode644', 'test text\n', mode=0644)
 
189
            check_mode(self, 'mode644', 0644)
 
190
 
 
191
            t.put('mode666', 'test text\n', mode=0666)
 
192
            check_mode(self, 'mode666', 0666)
 
193
 
 
194
            t.put('mode600', 'test text\n', mode=0600)
 
195
            check_mode(self, 'mode600', 0600)
 
196
 
 
197
            # Yes, you can put a file such that it becomes readonly
 
198
            t.put('mode400', 'test text\n', mode=0400)
 
199
            check_mode(self, 'mode400', 0400)
 
200
 
 
201
            t.put_multi([('mmode644', 'text\n')], mode=0644)
 
202
            check_mode(self, 'mmode644', 0644)
 
203
 
 
204
        # TODO: jam 20051215 test put_multi with a mode. I didn't bother because
 
205
        #                    it seems most people don't like the _multi functions
 
206
 
161
207
    def test_put_file(self):
162
208
        t = self.get_transport()
163
209
 
212
258
        self.check_file_contents('f5', 'here is some text\nand a bit more\n')
213
259
        self.check_file_contents('f6', 'some text for the\nthird file created\n')
214
260
 
 
261
        if not self.readonly:
 
262
            sio = StringIO('test text\n')
 
263
            t.put('mode644', sio, mode=0644)
 
264
            check_mode(self, 'mode644', 0644)
 
265
 
 
266
            a = open('mode644', 'rb')
 
267
            t.put('mode666', a, mode=0666)
 
268
            check_mode(self, 'mode666', 0666)
 
269
 
 
270
            a = open('mode644', 'rb')
 
271
            t.put('mode600', a, mode=0600)
 
272
            check_mode(self, 'mode600', 0600)
 
273
 
 
274
            # Yes, you can put a file such that it becomes readonly
 
275
            a = open('mode644', 'rb')
 
276
            t.put('mode400', a, mode=0400)
 
277
            check_mode(self, 'mode400', 0400)
 
278
 
215
279
    def test_mkdir(self):
216
280
        t = self.get_transport()
217
281
 
278
342
                             ('dir_b/b', 'contents of dir_b/b')])
279
343
                          , 2)
280
344
        for f in ('dir_a/a', 'dir_b/b'):
281
 
            self.assertEqual(t.get(f).read(), open(f).read())
 
345
            self.assertEqual(t.get(f).read(), open(f, 'rb').read())
 
346
 
 
347
        if not self.readonly:
 
348
            # Test mkdir with a mode
 
349
            t.mkdir('dmode755', mode=0755)
 
350
            check_mode(self, 'dmode755', 0755)
 
351
 
 
352
            t.mkdir('dmode555', mode=0555)
 
353
            check_mode(self, 'dmode555', 0555)
 
354
 
 
355
            t.mkdir('dmode777', mode=0777)
 
356
            check_mode(self, 'dmode777', 0777)
 
357
 
 
358
            t.mkdir('dmode700', mode=0700)
 
359
            check_mode(self, 'dmode700', 0700)
 
360
 
 
361
            # TODO: jam 20051215 test mkdir_multi with a mode
 
362
            t.mkdir_multi(['mdmode755'], mode=0755)
 
363
            check_mode(self, 'mdmode755', 0755)
 
364
 
282
365
 
283
366
    def test_copy_to(self):
284
367
        import tempfile
289
372
        files = ['a', 'b', 'c', 'd']
290
373
        self.build_tree(files)
291
374
 
292
 
        dtmp = tempfile.mkdtemp(dir=u'.', prefix='test-transport-')
293
 
        dtmp_base = os.path.basename(dtmp)
294
 
        local_t = LocalTransport(dtmp)
 
375
        def get_temp_local():
 
376
            dtmp = tempfile.mkdtemp(dir=u'.', prefix='test-transport-')
 
377
            dtmp_base = os.path.basename(dtmp)
 
378
            return dtmp_base, LocalTransport(dtmp)
 
379
        dtmp_base, local_t = get_temp_local()
295
380
 
296
381
        t.copy_to(files, local_t)
297
382
        for f in files:
298
 
            self.assertEquals(open(f).read(),
299
 
                    open(os.path.join(dtmp_base, f)).read())
 
383
            self.assertEquals(open(f, 'rb').read(),
 
384
                    open(pathjoin(dtmp_base, f), 'rb').read())
300
385
 
301
386
        # Test that copying into a missing directory raises
302
387
        # NoSuchFile
304
389
        open('e/f', 'wb').write('contents of e')
305
390
        self.assertRaises(NoSuchFile, t.copy_to, ['e/f'], local_t)
306
391
 
307
 
        os.mkdir(os.path.join(dtmp_base, 'e'))
 
392
        os.mkdir(pathjoin(dtmp_base, 'e'))
308
393
        t.copy_to(['e/f'], local_t)
309
394
 
310
 
        del dtmp, dtmp_base, local_t
 
395
        del dtmp_base, local_t
311
396
 
312
 
        dtmp = tempfile.mkdtemp(dir=u'.', prefix='test-transport-')
313
 
        dtmp_base = os.path.basename(dtmp)
314
 
        local_t = LocalTransport(dtmp)
 
397
        dtmp_base, local_t = get_temp_local()
315
398
 
316
399
        files = ['a', 'b', 'c', 'd']
317
400
        t.copy_to(iter(files), local_t)
318
401
        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
 
402
            self.assertEquals(open(f, 'rb').read(),
 
403
                    open(pathjoin(dtmp_base, f), 'rb').read())
 
404
 
 
405
        del dtmp_base, local_t
 
406
 
 
407
        for mode in (0666, 0644, 0600, 0400):
 
408
            dtmp_base, local_t = get_temp_local()
 
409
            t.copy_to(files, local_t, mode=mode)
 
410
            for f in files:
 
411
                check_mode(self, os.path.join(dtmp_base, f), mode)
323
412
 
324
413
    def test_append(self):
325
414
        t = self.get_transport()