← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: mbp at sourcefrog
  • Date: 2006-10-31 01:30:31 UTC
  • mto: This revision was merged to the branch mainline in revision 2114.
  • Revision ID: mbp@sourcefrog.net-20061031013031-da3047f0d9fb1f64
Use regular progress-bar classes, not a special mechanism

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_osutils.py
 
1
# Copyright (C) 2005, 2006 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
"""Tests for the osutils wrapper."""
 
18
 
 
19
import errno
 
20
import os
 
21
import socket
 
22
import stat
 
23
import sys
 
24
 
 
25
import bzrlib
 
26
from bzrlib import (
 
27
    errors,
 
28
    osutils,
 
29
    )
 
30
from bzrlib.errors import BzrBadParameterNotUnicode, InvalidURL
 
31
from bzrlib.tests import (
 
32
        StringIOWrapper,
 
33
        TestCase, 
 
34
        TestCaseInTempDir, 
 
35
        TestSkipped,
 
36
        )
 
37
 
 
38
 
 
39
class TestOSUtils(TestCaseInTempDir):
 
40
 
 
41
    def test_fancy_rename(self):
 
42
        # This should work everywhere
 
43
        def rename(a, b):
 
44
            osutils.fancy_rename(a, b,
 
45
                    rename_func=os.rename,
 
46
                    unlink_func=os.unlink)
 
47
 
 
48
        open('a', 'wb').write('something in a\n')
 
49
        rename('a', 'b')
 
50
        self.failIfExists('a')
 
51
        self.failUnlessExists('b')
 
52
        self.check_file_contents('b', 'something in a\n')
 
53
 
 
54
        open('a', 'wb').write('new something in a\n')
 
55
        rename('b', 'a')
 
56
 
 
57
        self.check_file_contents('a', 'something in a\n')
 
58
 
 
59
    def test_rename(self):
 
60
        # Rename should be semi-atomic on all platforms
 
61
        open('a', 'wb').write('something in a\n')
 
62
        osutils.rename('a', 'b')
 
63
        self.failIfExists('a')
 
64
        self.failUnlessExists('b')
 
65
        self.check_file_contents('b', 'something in a\n')
 
66
 
 
67
        open('a', 'wb').write('new something in a\n')
 
68
        osutils.rename('b', 'a')
 
69
 
 
70
        self.check_file_contents('a', 'something in a\n')
 
71
 
 
72
    # TODO: test fancy_rename using a MemoryTransport
 
73
 
 
74
    def test_01_rand_chars_empty(self):
 
75
        result = osutils.rand_chars(0)
 
76
        self.assertEqual(result, '')
 
77
 
 
78
    def test_02_rand_chars_100(self):
 
79
        result = osutils.rand_chars(100)
 
80
        self.assertEqual(len(result), 100)
 
81
        self.assertEqual(type(result), str)
 
82
        self.assertContainsRe(result, r'^[a-z0-9]{100}$')
 
83
 
 
84
    def test_is_inside(self):
 
85
        is_inside = osutils.is_inside
 
86
        self.assertTrue(is_inside('src', 'src/foo.c'))
 
87
        self.assertFalse(is_inside('src', 'srccontrol'))
 
88
        self.assertTrue(is_inside('src', 'src/a/a/a/foo.c'))
 
89
        self.assertTrue(is_inside('foo.c', 'foo.c'))
 
90
        self.assertFalse(is_inside('foo.c', ''))
 
91
        self.assertTrue(is_inside('', 'foo.c'))
 
92
 
 
93
    def test_rmtree(self):
 
94
        # Check to remove tree with read-only files/dirs
 
95
        os.mkdir('dir')
 
96
        f = file('dir/file', 'w')
 
97
        f.write('spam')
 
98
        f.close()
 
99
        # would like to also try making the directory readonly, but at the
 
100
        # moment python shutil.rmtree doesn't handle that properly - it would
 
101
        # need to chmod the directory before removing things inside it - deferred
 
102
        # for now -- mbp 20060505
 
103
        # osutils.make_readonly('dir')
 
104
        osutils.make_readonly('dir/file')
 
105
 
 
106
        osutils.rmtree('dir')
 
107
 
 
108
        self.failIfExists('dir/file')
 
109
        self.failIfExists('dir')
 
110
 
 
111
    def test_file_kind(self):
 
112
        self.build_tree(['file', 'dir/'])
 
113
        self.assertEquals('file', osutils.file_kind('file'))
 
114
        self.assertEquals('directory', osutils.file_kind('dir/'))
 
115
        if osutils.has_symlinks():
 
116
            os.symlink('symlink', 'symlink')
 
117
            self.assertEquals('symlink', osutils.file_kind('symlink'))
 
118
        
 
119
        # TODO: jam 20060529 Test a block device
 
120
        try:
 
121
            os.lstat('/dev/null')
 
122
        except OSError, e:
 
123
            if e.errno not in (errno.ENOENT,):
 
124
                raise
 
125
        else:
 
126
            self.assertEquals('chardev', osutils.file_kind('/dev/null'))
 
127
 
 
128
        mkfifo = getattr(os, 'mkfifo', None)
 
129
        if mkfifo:
 
130
            mkfifo('fifo')
 
131
            try:
 
132
                self.assertEquals('fifo', osutils.file_kind('fifo'))
 
133
            finally:
 
134
                os.remove('fifo')
 
135
 
 
136
        AF_UNIX = getattr(socket, 'AF_UNIX', None)
 
137
        if AF_UNIX:
 
138
            s = socket.socket(AF_UNIX)
 
139
            s.bind('socket')
 
140
            try:
 
141
                self.assertEquals('socket', osutils.file_kind('socket'))
 
142
            finally:
 
143
                os.remove('socket')
 
144
 
 
145
    def test_get_umask(self):
 
146
        if sys.platform == 'win32':
 
147
            # umask always returns '0', no way to set it
 
148
            self.assertEqual(0, osutils.get_umask())
 
149
            return
 
150
 
 
151
        orig_umask = osutils.get_umask()
 
152
        try:
 
153
            os.umask(0222)
 
154
            self.assertEqual(0222, osutils.get_umask())
 
155
            os.umask(0022)
 
156
            self.assertEqual(0022, osutils.get_umask())
 
157
            os.umask(0002)
 
158
            self.assertEqual(0002, osutils.get_umask())
 
159
            os.umask(0027)
 
160
            self.assertEqual(0027, osutils.get_umask())
 
161
        finally:
 
162
            os.umask(orig_umask)
 
163
 
 
164
    def assertFormatedDelta(self, expected, seconds):
 
165
        """Assert osutils.format_delta formats as expected"""
 
166
        actual = osutils.format_delta(seconds)
 
167
        self.assertEqual(expected, actual)
 
168
 
 
169
    def test_format_delta(self):
 
170
        self.assertFormatedDelta('0 seconds ago', 0)
 
171
        self.assertFormatedDelta('1 second ago', 1)
 
172
        self.assertFormatedDelta('10 seconds ago', 10)
 
173
        self.assertFormatedDelta('59 seconds ago', 59)
 
174
        self.assertFormatedDelta('89 seconds ago', 89)
 
175
        self.assertFormatedDelta('1 minute, 30 seconds ago', 90)
 
176
        self.assertFormatedDelta('3 minutes, 0 seconds ago', 180)
 
177
        self.assertFormatedDelta('3 minutes, 1 second ago', 181)
 
178
        self.assertFormatedDelta('10 minutes, 15 seconds ago', 615)
 
179
        self.assertFormatedDelta('30 minutes, 59 seconds ago', 1859)
 
180
        self.assertFormatedDelta('31 minutes, 0 seconds ago', 1860)
 
181
        self.assertFormatedDelta('60 minutes, 0 seconds ago', 3600)
 
182
        self.assertFormatedDelta('89 minutes, 59 seconds ago', 5399)
 
183
        self.assertFormatedDelta('1 hour, 30 minutes ago', 5400)
 
184
        self.assertFormatedDelta('2 hours, 30 minutes ago', 9017)
 
185
        self.assertFormatedDelta('10 hours, 0 minutes ago', 36000)
 
186
        self.assertFormatedDelta('24 hours, 0 minutes ago', 86400)
 
187
        self.assertFormatedDelta('35 hours, 59 minutes ago', 129599)
 
188
        self.assertFormatedDelta('36 hours, 0 minutes ago', 129600)
 
189
        self.assertFormatedDelta('36 hours, 0 minutes ago', 129601)
 
190
        self.assertFormatedDelta('36 hours, 1 minute ago', 129660)
 
191
        self.assertFormatedDelta('36 hours, 1 minute ago', 129661)
 
192
        self.assertFormatedDelta('84 hours, 10 minutes ago', 303002)
 
193
 
 
194
        # We handle when time steps the wrong direction because computers
 
195
        # don't have synchronized clocks.
 
196
        self.assertFormatedDelta('84 hours, 10 minutes in the future', -303002)
 
197
        self.assertFormatedDelta('1 second in the future', -1)
 
198
        self.assertFormatedDelta('2 seconds in the future', -2)
 
199
 
 
200
 
 
201
class TestSafeUnicode(TestCase):
 
202
 
 
203
    def test_from_ascii_string(self):
 
204
        self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
 
205
 
 
206
    def test_from_unicode_string_ascii_contents(self):
 
207
        self.assertEqual(u'bargam', osutils.safe_unicode(u'bargam'))
 
208
 
 
209
    def test_from_unicode_string_unicode_contents(self):
 
210
        self.assertEqual(u'bargam\xae', osutils.safe_unicode(u'bargam\xae'))
 
211
 
 
212
    def test_from_utf8_string(self):
 
213
        self.assertEqual(u'foo\xae', osutils.safe_unicode('foo\xc2\xae'))
 
214
 
 
215
    def test_bad_utf8_string(self):
 
216
        self.assertRaises(BzrBadParameterNotUnicode,
 
217
                          osutils.safe_unicode,
 
218
                          '\xbb\xbb')
 
219
 
 
220
 
 
221
class TestWin32Funcs(TestCase):
 
222
    """Test that the _win32 versions of os utilities return appropriate paths."""
 
223
 
 
224
    def test_abspath(self):
 
225
        self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
 
226
        self.assertEqual('C:/foo', osutils._win32_abspath('C:/foo'))
 
227
 
 
228
    def test_realpath(self):
 
229
        self.assertEqual('C:/foo', osutils._win32_realpath('C:\\foo'))
 
230
        self.assertEqual('C:/foo', osutils._win32_realpath('C:/foo'))
 
231
 
 
232
    def test_pathjoin(self):
 
233
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path', 'to', 'foo'))
 
234
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path\\to', 'C:\\foo'))
 
235
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path/to', 'C:/foo'))
 
236
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path/to/', 'foo'))
 
237
        self.assertEqual('/foo', osutils._win32_pathjoin('C:/path/to/', '/foo'))
 
238
        self.assertEqual('/foo', osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
 
239
 
 
240
    def test_normpath(self):
 
241
        self.assertEqual('path/to/foo', osutils._win32_normpath(r'path\\from\..\to\.\foo'))
 
242
        self.assertEqual('path/to/foo', osutils._win32_normpath('path//from/../to/./foo'))
 
243
 
 
244
    def test_getcwd(self):
 
245
        cwd = osutils._win32_getcwd()
 
246
        os_cwd = os.getcwdu()
 
247
        self.assertEqual(os_cwd[1:].replace('\\', '/'), cwd[1:])
 
248
        # win32 is inconsistent whether it returns lower or upper case
 
249
        # and even if it was consistent the user might type the other
 
250
        # so we force it to uppercase
 
251
        # running python.exe under cmd.exe return capital C:\\
 
252
        # running win32 python inside a cygwin shell returns lowercase
 
253
        self.assertEqual(os_cwd[0].upper(), cwd[0])
 
254
 
 
255
    def test_fixdrive(self):
 
256
        self.assertEqual('H:/foo', osutils._win32_fixdrive('h:/foo'))
 
257
        self.assertEqual('H:/foo', osutils._win32_fixdrive('H:/foo'))
 
258
        self.assertEqual('C:\\foo', osutils._win32_fixdrive('c:\\foo'))
 
259
 
 
260
 
 
261
class TestWin32FuncsDirs(TestCaseInTempDir):
 
262
    """Test win32 functions that create files."""
 
263
    
 
264
    def test_getcwd(self):
 
265
        # Make sure getcwd can handle unicode filenames
 
266
        try:
 
267
            os.mkdir(u'mu-\xb5')
 
268
        except UnicodeError:
 
269
            raise TestSkipped("Unable to create Unicode filename")
 
270
 
 
271
        os.chdir(u'mu-\xb5')
 
272
        # TODO: jam 20060427 This will probably fail on Mac OSX because
 
273
        #       it will change the normalization of B\xe5gfors
 
274
        #       Consider using a different unicode character, or make
 
275
        #       osutils.getcwd() renormalize the path.
 
276
        self.assertEndsWith(osutils._win32_getcwd(), u'mu-\xb5')
 
277
 
 
278
    def test_mkdtemp(self):
 
279
        tmpdir = osutils._win32_mkdtemp(dir='.')
 
280
        self.assertFalse('\\' in tmpdir)
 
281
 
 
282
    def test_rename(self):
 
283
        a = open('a', 'wb')
 
284
        a.write('foo\n')
 
285
        a.close()
 
286
        b = open('b', 'wb')
 
287
        b.write('baz\n')
 
288
        b.close()
 
289
 
 
290
        osutils._win32_rename('b', 'a')
 
291
        self.failUnlessExists('a')
 
292
        self.failIfExists('b')
 
293
        self.assertFileEqual('baz\n', 'a')
 
294
 
 
295
    def test_rename_missing_file(self):
 
296
        a = open('a', 'wb')
 
297
        a.write('foo\n')
 
298
        a.close()
 
299
 
 
300
        try:
 
301
            osutils._win32_rename('b', 'a')
 
302
        except (IOError, OSError), e:
 
303
            self.assertEqual(errno.ENOENT, e.errno)
 
304
        self.assertFileEqual('foo\n', 'a')
 
305
 
 
306
    def test_rename_missing_dir(self):
 
307
        os.mkdir('a')
 
308
        try:
 
309
            osutils._win32_rename('b', 'a')
 
310
        except (IOError, OSError), e:
 
311
            self.assertEqual(errno.ENOENT, e.errno)
 
312
 
 
313
    def test_rename_current_dir(self):
 
314
        os.mkdir('a')
 
315
        os.chdir('a')
 
316
        # You can't rename the working directory
 
317
        # doing rename non-existant . usually
 
318
        # just raises ENOENT, since non-existant
 
319
        # doesn't exist.
 
320
        try:
 
321
            osutils._win32_rename('b', '.')
 
322
        except (IOError, OSError), e:
 
323
            self.assertEqual(errno.ENOENT, e.errno)
 
324
 
 
325
    def test_splitpath(self):
 
326
        def check(expected, path):
 
327
            self.assertEqual(expected, osutils.splitpath(path))
 
328
 
 
329
        check(['a'], 'a')
 
330
        check(['a', 'b'], 'a/b')
 
331
        check(['a', 'b'], 'a/./b')
 
332
        check(['a', '.b'], 'a/.b')
 
333
        check(['a', '.b'], 'a\\.b')
 
334
 
 
335
        self.assertRaises(errors.BzrError, osutils.splitpath, 'a/../b')
 
336
 
 
337
 
 
338
class TestMacFuncsDirs(TestCaseInTempDir):
 
339
    """Test mac special functions that require directories."""
 
340
 
 
341
    def test_getcwd(self):
 
342
        # On Mac, this will actually create Ba\u030agfors
 
343
        # but chdir will still work, because it accepts both paths
 
344
        try:
 
345
            os.mkdir(u'B\xe5gfors')
 
346
        except UnicodeError:
 
347
            raise TestSkipped("Unable to create Unicode filename")
 
348
 
 
349
        os.chdir(u'B\xe5gfors')
 
350
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
 
351
 
 
352
    def test_getcwd_nonnorm(self):
 
353
        # Test that _mac_getcwd() will normalize this path
 
354
        try:
 
355
            os.mkdir(u'Ba\u030agfors')
 
356
        except UnicodeError:
 
357
            raise TestSkipped("Unable to create Unicode filename")
 
358
 
 
359
        os.chdir(u'Ba\u030agfors')
 
360
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
 
361
 
 
362
 
 
363
class TestSplitLines(TestCase):
 
364
 
 
365
    def test_split_unicode(self):
 
366
        self.assertEqual([u'foo\n', u'bar\xae'],
 
367
                         osutils.split_lines(u'foo\nbar\xae'))
 
368
        self.assertEqual([u'foo\n', u'bar\xae\n'],
 
369
                         osutils.split_lines(u'foo\nbar\xae\n'))
 
370
 
 
371
    def test_split_with_carriage_returns(self):
 
372
        self.assertEqual(['foo\rbar\n'],
 
373
                         osutils.split_lines('foo\rbar\n'))
 
374
 
 
375
 
 
376
class TestWalkDirs(TestCaseInTempDir):
 
377
 
 
378
    def test_walkdirs(self):
 
379
        tree = [
 
380
            '.bzr',
 
381
            '0file',
 
382
            '1dir/',
 
383
            '1dir/0file',
 
384
            '1dir/1dir/',
 
385
            '2file'
 
386
            ]
 
387
        self.build_tree(tree)
 
388
        expected_dirblocks = [
 
389
                (('', '.'),
 
390
                 [('0file', '0file', 'file'),
 
391
                  ('1dir', '1dir', 'directory'),
 
392
                  ('2file', '2file', 'file'),
 
393
                 ]
 
394
                ),
 
395
                (('1dir', './1dir'),
 
396
                 [('1dir/0file', '0file', 'file'),
 
397
                  ('1dir/1dir', '1dir', 'directory'),
 
398
                 ]
 
399
                ),
 
400
                (('1dir/1dir', './1dir/1dir'),
 
401
                 [
 
402
                 ]
 
403
                ),
 
404
            ]
 
405
        result = []
 
406
        found_bzrdir = False
 
407
        for dirdetail, dirblock in osutils.walkdirs('.'):
 
408
            if len(dirblock) and dirblock[0][1] == '.bzr':
 
409
                # this tests the filtering of selected paths
 
410
                found_bzrdir = True
 
411
                del dirblock[0]
 
412
            result.append((dirdetail, dirblock))
 
413
 
 
414
        self.assertTrue(found_bzrdir)
 
415
        self.assertEqual(expected_dirblocks,
 
416
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
417
        # you can search a subdir only, with a supplied prefix.
 
418
        result = []
 
419
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
 
420
            result.append(dirblock)
 
421
        self.assertEqual(expected_dirblocks[1:],
 
422
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
423
 
 
424
    def assertPathCompare(self, path_less, path_greater):
 
425
        """check that path_less and path_greater compare correctly."""
 
426
        self.assertEqual(0, osutils.compare_paths_prefix_order(
 
427
            path_less, path_less))
 
428
        self.assertEqual(0, osutils.compare_paths_prefix_order(
 
429
            path_greater, path_greater))
 
430
        self.assertEqual(-1, osutils.compare_paths_prefix_order(
 
431
            path_less, path_greater))
 
432
        self.assertEqual(1, osutils.compare_paths_prefix_order(
 
433
            path_greater, path_less))
 
434
 
 
435
    def test_compare_paths_prefix_order(self):
 
436
        # root before all else
 
437
        self.assertPathCompare("/", "/a")
 
438
        # alpha within a dir
 
439
        self.assertPathCompare("/a", "/b")
 
440
        self.assertPathCompare("/b", "/z")
 
441
        # high dirs before lower.
 
442
        self.assertPathCompare("/z", "/a/a")
 
443
        # except if the deeper dir should be output first
 
444
        self.assertPathCompare("/a/b/c", "/d/g")
 
445
        # lexical betwen dirs of the same height
 
446
        self.assertPathCompare("/a/z", "/z/z")
 
447
        self.assertPathCompare("/a/c/z", "/a/d/e")
 
448
 
 
449
        # this should also be consistent for no leading / paths
 
450
        # root before all else
 
451
        self.assertPathCompare("", "a")
 
452
        # alpha within a dir
 
453
        self.assertPathCompare("a", "b")
 
454
        self.assertPathCompare("b", "z")
 
455
        # high dirs before lower.
 
456
        self.assertPathCompare("z", "a/a")
 
457
        # except if the deeper dir should be output first
 
458
        self.assertPathCompare("a/b/c", "d/g")
 
459
        # lexical betwen dirs of the same height
 
460
        self.assertPathCompare("a/z", "z/z")
 
461
        self.assertPathCompare("a/c/z", "a/d/e")
 
462
 
 
463
    def test_path_prefix_sorting(self):
 
464
        """Doing a sort on path prefix should match our sample data."""
 
465
        original_paths = [
 
466
            'a',
 
467
            'a/b',
 
468
            'a/b/c',
 
469
            'b',
 
470
            'b/c',
 
471
            'd',
 
472
            'd/e',
 
473
            'd/e/f',
 
474
            'd/f',
 
475
            'd/g',
 
476
            'g',
 
477
            ]
 
478
 
 
479
        dir_sorted_paths = [
 
480
            'a',
 
481
            'b',
 
482
            'd',
 
483
            'g',
 
484
            'a/b',
 
485
            'a/b/c',
 
486
            'b/c',
 
487
            'd/e',
 
488
            'd/f',
 
489
            'd/g',
 
490
            'd/e/f',
 
491
            ]
 
492
 
 
493
        self.assertEqual(
 
494
            dir_sorted_paths,
 
495
            sorted(original_paths, key=osutils.path_prefix_key))
 
496
        # using the comparison routine shoudl work too:
 
497
        self.assertEqual(
 
498
            dir_sorted_paths,
 
499
            sorted(original_paths, cmp=osutils.compare_paths_prefix_order))
 
500
 
 
501
 
 
502
class TestCopyTree(TestCaseInTempDir):
 
503
    
 
504
    def test_copy_basic_tree(self):
 
505
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
 
506
        osutils.copy_tree('source', 'target')
 
507
        self.assertEqual(['a', 'b'], sorted(os.listdir('target')))
 
508
        self.assertEqual(['c'], os.listdir('target/b'))
 
509
 
 
510
    def test_copy_tree_target_exists(self):
 
511
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c',
 
512
                         'target/'])
 
513
        osutils.copy_tree('source', 'target')
 
514
        self.assertEqual(['a', 'b'], sorted(os.listdir('target')))
 
515
        self.assertEqual(['c'], os.listdir('target/b'))
 
516
 
 
517
    def test_copy_tree_symlinks(self):
 
518
        if not osutils.has_symlinks():
 
519
            return
 
520
        self.build_tree(['source/'])
 
521
        os.symlink('a/generic/path', 'source/lnk')
 
522
        osutils.copy_tree('source', 'target')
 
523
        self.assertEqual(['lnk'], os.listdir('target'))
 
524
        self.assertEqual('a/generic/path', os.readlink('target/lnk'))
 
525
 
 
526
    def test_copy_tree_handlers(self):
 
527
        processed_files = []
 
528
        processed_links = []
 
529
        def file_handler(from_path, to_path):
 
530
            processed_files.append(('f', from_path, to_path))
 
531
        def dir_handler(from_path, to_path):
 
532
            processed_files.append(('d', from_path, to_path))
 
533
        def link_handler(from_path, to_path):
 
534
            processed_links.append((from_path, to_path))
 
535
        handlers = {'file':file_handler,
 
536
                    'directory':dir_handler,
 
537
                    'symlink':link_handler,
 
538
                   }
 
539
 
 
540
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
 
541
        if osutils.has_symlinks():
 
542
            os.symlink('a/generic/path', 'source/lnk')
 
543
        osutils.copy_tree('source', 'target', handlers=handlers)
 
544
 
 
545
        self.assertEqual([('d', 'source', 'target'),
 
546
                          ('f', 'source/a', 'target/a'),
 
547
                          ('d', 'source/b', 'target/b'),
 
548
                          ('f', 'source/b/c', 'target/b/c'),
 
549
                         ], processed_files)
 
550
        self.failIfExists('target')
 
551
        if osutils.has_symlinks():
 
552
            self.assertEqual([('source/lnk', 'target/lnk')], processed_links)
 
553
 
 
554
 
 
555
class TestTerminalEncoding(TestCase):
 
556
    """Test the auto-detection of proper terminal encoding."""
 
557
 
 
558
    def setUp(self):
 
559
        self._stdout = sys.stdout
 
560
        self._stderr = sys.stderr
 
561
        self._stdin = sys.stdin
 
562
        self._user_encoding = bzrlib.user_encoding
 
563
 
 
564
        self.addCleanup(self._reset)
 
565
 
 
566
        sys.stdout = StringIOWrapper()
 
567
        sys.stdout.encoding = 'stdout_encoding'
 
568
        sys.stderr = StringIOWrapper()
 
569
        sys.stderr.encoding = 'stderr_encoding'
 
570
        sys.stdin = StringIOWrapper()
 
571
        sys.stdin.encoding = 'stdin_encoding'
 
572
        bzrlib.user_encoding = 'user_encoding'
 
573
 
 
574
    def _reset(self):
 
575
        sys.stdout = self._stdout
 
576
        sys.stderr = self._stderr
 
577
        sys.stdin = self._stdin
 
578
        bzrlib.user_encoding = self._user_encoding
 
579
 
 
580
    def test_get_terminal_encoding(self):
 
581
        # first preference is stdout encoding
 
582
        self.assertEqual('stdout_encoding', osutils.get_terminal_encoding())
 
583
 
 
584
        sys.stdout.encoding = None
 
585
        # if sys.stdout is None, fall back to sys.stdin
 
586
        self.assertEqual('stdin_encoding', osutils.get_terminal_encoding())
 
587
 
 
588
        sys.stdin.encoding = None
 
589
        # and in the worst case, use bzrlib.user_encoding
 
590
        self.assertEqual('user_encoding', osutils.get_terminal_encoding())
 
591
 
 
592
 
 
593
class TestSetUnsetEnv(TestCase):
 
594
    """Test updating the environment"""
 
595
 
 
596
    def setUp(self):
 
597
        super(TestSetUnsetEnv, self).setUp()
 
598
 
 
599
        self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'),
 
600
                         'Environment was not cleaned up properly.'
 
601
                         ' Variable BZR_TEST_ENV_VAR should not exist.')
 
602
        def cleanup():
 
603
            if 'BZR_TEST_ENV_VAR' in os.environ:
 
604
                del os.environ['BZR_TEST_ENV_VAR']
 
605
 
 
606
        self.addCleanup(cleanup)
 
607
 
 
608
    def test_set(self):
 
609
        """Test that we can set an env variable"""
 
610
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
 
611
        self.assertEqual(None, old)
 
612
        self.assertEqual('foo', os.environ.get('BZR_TEST_ENV_VAR'))
 
613
 
 
614
    def test_double_set(self):
 
615
        """Test that we get the old value out"""
 
616
        osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
 
617
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'bar')
 
618
        self.assertEqual('foo', old)
 
619
        self.assertEqual('bar', os.environ.get('BZR_TEST_ENV_VAR'))
 
620
 
 
621
    def test_unicode(self):
 
622
        """Environment can only contain plain strings
 
623
        
 
624
        So Unicode strings must be encoded.
 
625
        """
 
626
        # Try a few different characters, to see if we can get
 
627
        # one that will be valid in the user_encoding
 
628
        possible_vals = [u'm\xb5', u'\xe1', u'\u0410']
 
629
        for uni_val in possible_vals:
 
630
            try:
 
631
                env_val = uni_val.encode(bzrlib.user_encoding)
 
632
            except UnicodeEncodeError:
 
633
                # Try a different character
 
634
                pass
 
635
            else:
 
636
                break
 
637
        else:
 
638
            raise TestSkipped('Cannot find a unicode character that works in'
 
639
                              ' encoding %s' % (bzrlib.user_encoding,))
 
640
 
 
641
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', uni_val)
 
642
        self.assertEqual(env_val, os.environ.get('BZR_TEST_ENV_VAR'))
 
643
 
 
644
    def test_unset(self):
 
645
        """Test that passing None will remove the env var"""
 
646
        osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
 
647
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', None)
 
648
        self.assertEqual('foo', old)
 
649
        self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'))
 
650
        self.failIf('BZR_TEST_ENV_VAR' in os.environ)
 
651