~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: Jelmer Vernooij
  • Date: 2011-01-19 06:13:58 UTC
  • mfrom: (5619 +trunk)
  • mto: This revision was merged to the branch mainline in revision 5620.
  • Revision ID: jelmer@samba.org-20110119061358-qm58pzvcp2l9cdx4
merge bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2005-2011 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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
16
 
 
17
"""Tests for the osutils wrapper."""
 
18
 
 
19
from cStringIO import StringIO
 
20
import errno
 
21
import os
 
22
import re
 
23
import socket
 
24
import sys
 
25
import time
 
26
 
 
27
from bzrlib import (
 
28
    errors,
 
29
    lazy_regex,
 
30
    osutils,
 
31
    symbol_versioning,
 
32
    tests,
 
33
    trace,
 
34
    win32utils,
 
35
    )
 
36
from bzrlib.tests import (
 
37
    features,
 
38
    file_utils,
 
39
    test__walkdirs_win32,
 
40
    )
 
41
from bzrlib.tests.scenarios import load_tests_apply_scenarios
 
42
 
 
43
 
 
44
class _UTF8DirReaderFeature(tests.Feature):
 
45
 
 
46
    def _probe(self):
 
47
        try:
 
48
            from bzrlib import _readdir_pyx
 
49
            self.reader = _readdir_pyx.UTF8DirReader
 
50
            return True
 
51
        except ImportError:
 
52
            return False
 
53
 
 
54
    def feature_name(self):
 
55
        return 'bzrlib._readdir_pyx'
 
56
 
 
57
UTF8DirReaderFeature = _UTF8DirReaderFeature()
 
58
 
 
59
term_ios_feature = tests.ModuleAvailableFeature('termios')
 
60
 
 
61
 
 
62
def _already_unicode(s):
 
63
    return s
 
64
 
 
65
 
 
66
def _utf8_to_unicode(s):
 
67
    return s.decode('UTF-8')
 
68
 
 
69
 
 
70
def dir_reader_scenarios():
 
71
    # For each dir reader we define:
 
72
 
 
73
    # - native_to_unicode: a function converting the native_abspath as returned
 
74
    #   by DirReader.read_dir to its unicode representation
 
75
 
 
76
    # UnicodeDirReader is the fallback, it should be tested on all platforms.
 
77
    scenarios = [('unicode',
 
78
                  dict(_dir_reader_class=osutils.UnicodeDirReader,
 
79
                       _native_to_unicode=_already_unicode))]
 
80
    # Some DirReaders are platform specific and even there they may not be
 
81
    # available.
 
82
    if UTF8DirReaderFeature.available():
 
83
        from bzrlib import _readdir_pyx
 
84
        scenarios.append(('utf8',
 
85
                          dict(_dir_reader_class=_readdir_pyx.UTF8DirReader,
 
86
                               _native_to_unicode=_utf8_to_unicode)))
 
87
 
 
88
    if test__walkdirs_win32.win32_readdir_feature.available():
 
89
        try:
 
90
            from bzrlib import _walkdirs_win32
 
91
            scenarios.append(
 
92
                ('win32',
 
93
                 dict(_dir_reader_class=_walkdirs_win32.Win32ReadDir,
 
94
                      _native_to_unicode=_already_unicode)))
 
95
        except ImportError:
 
96
            pass
 
97
    return scenarios
 
98
 
 
99
 
 
100
load_tests = load_tests_apply_scenarios
 
101
 
 
102
 
 
103
class TestContainsWhitespace(tests.TestCase):
 
104
 
 
105
    def test_contains_whitespace(self):
 
106
        self.failUnless(osutils.contains_whitespace(u' '))
 
107
        self.failUnless(osutils.contains_whitespace(u'hello there'))
 
108
        self.failUnless(osutils.contains_whitespace(u'hellothere\n'))
 
109
        self.failUnless(osutils.contains_whitespace(u'hello\nthere'))
 
110
        self.failUnless(osutils.contains_whitespace(u'hello\rthere'))
 
111
        self.failUnless(osutils.contains_whitespace(u'hello\tthere'))
 
112
 
 
113
        # \xa0 is "Non-breaking-space" which on some python locales thinks it
 
114
        # is whitespace, but we do not.
 
115
        self.failIf(osutils.contains_whitespace(u''))
 
116
        self.failIf(osutils.contains_whitespace(u'hellothere'))
 
117
        self.failIf(osutils.contains_whitespace(u'hello\xa0there'))
 
118
 
 
119
 
 
120
class TestRename(tests.TestCaseInTempDir):
 
121
 
 
122
    def create_file(self, filename, content):
 
123
        f = open(filename, 'wb')
 
124
        try:
 
125
            f.write(content)
 
126
        finally:
 
127
            f.close()
 
128
 
 
129
    def _fancy_rename(self, a, b):
 
130
        osutils.fancy_rename(a, b, rename_func=os.rename,
 
131
                             unlink_func=os.unlink)
 
132
 
 
133
    def test_fancy_rename(self):
 
134
        # This should work everywhere
 
135
        self.create_file('a', 'something in a\n')
 
136
        self._fancy_rename('a', 'b')
 
137
        self.failIfExists('a')
 
138
        self.failUnlessExists('b')
 
139
        self.check_file_contents('b', 'something in a\n')
 
140
 
 
141
        self.create_file('a', 'new something in a\n')
 
142
        self._fancy_rename('b', 'a')
 
143
 
 
144
        self.check_file_contents('a', 'something in a\n')
 
145
 
 
146
    def test_fancy_rename_fails_source_missing(self):
 
147
        # An exception should be raised, and the target should be left in place
 
148
        self.create_file('target', 'data in target\n')
 
149
        self.assertRaises((IOError, OSError), self._fancy_rename,
 
150
                          'missingsource', 'target')
 
151
        self.failUnlessExists('target')
 
152
        self.check_file_contents('target', 'data in target\n')
 
153
 
 
154
    def test_fancy_rename_fails_if_source_and_target_missing(self):
 
155
        self.assertRaises((IOError, OSError), self._fancy_rename,
 
156
                          'missingsource', 'missingtarget')
 
157
 
 
158
    def test_rename(self):
 
159
        # Rename should be semi-atomic on all platforms
 
160
        self.create_file('a', 'something in a\n')
 
161
        osutils.rename('a', 'b')
 
162
        self.failIfExists('a')
 
163
        self.failUnlessExists('b')
 
164
        self.check_file_contents('b', 'something in a\n')
 
165
 
 
166
        self.create_file('a', 'new something in a\n')
 
167
        osutils.rename('b', 'a')
 
168
 
 
169
        self.check_file_contents('a', 'something in a\n')
 
170
 
 
171
    # TODO: test fancy_rename using a MemoryTransport
 
172
 
 
173
    def test_rename_change_case(self):
 
174
        # on Windows we should be able to change filename case by rename
 
175
        self.build_tree(['a', 'b/'])
 
176
        osutils.rename('a', 'A')
 
177
        osutils.rename('b', 'B')
 
178
        # we can't use failUnlessExists on case-insensitive filesystem
 
179
        # so try to check shape of the tree
 
180
        shape = sorted(os.listdir('.'))
 
181
        self.assertEquals(['A', 'B'], shape)
 
182
 
 
183
 
 
184
class TestRandChars(tests.TestCase):
 
185
 
 
186
    def test_01_rand_chars_empty(self):
 
187
        result = osutils.rand_chars(0)
 
188
        self.assertEqual(result, '')
 
189
 
 
190
    def test_02_rand_chars_100(self):
 
191
        result = osutils.rand_chars(100)
 
192
        self.assertEqual(len(result), 100)
 
193
        self.assertEqual(type(result), str)
 
194
        self.assertContainsRe(result, r'^[a-z0-9]{100}$')
 
195
 
 
196
 
 
197
class TestIsInside(tests.TestCase):
 
198
 
 
199
    def test_is_inside(self):
 
200
        is_inside = osutils.is_inside
 
201
        self.assertTrue(is_inside('src', 'src/foo.c'))
 
202
        self.assertFalse(is_inside('src', 'srccontrol'))
 
203
        self.assertTrue(is_inside('src', 'src/a/a/a/foo.c'))
 
204
        self.assertTrue(is_inside('foo.c', 'foo.c'))
 
205
        self.assertFalse(is_inside('foo.c', ''))
 
206
        self.assertTrue(is_inside('', 'foo.c'))
 
207
 
 
208
    def test_is_inside_any(self):
 
209
        SRC_FOO_C = osutils.pathjoin('src', 'foo.c')
 
210
        for dirs, fn in [(['src', 'doc'], SRC_FOO_C),
 
211
                         (['src'], SRC_FOO_C),
 
212
                         (['src'], 'src'),
 
213
                         ]:
 
214
            self.assert_(osutils.is_inside_any(dirs, fn))
 
215
        for dirs, fn in [(['src'], 'srccontrol'),
 
216
                         (['src'], 'srccontrol/foo')]:
 
217
            self.assertFalse(osutils.is_inside_any(dirs, fn))
 
218
 
 
219
    def test_is_inside_or_parent_of_any(self):
 
220
        for dirs, fn in [(['src', 'doc'], 'src/foo.c'),
 
221
                         (['src'], 'src/foo.c'),
 
222
                         (['src/bar.c'], 'src'),
 
223
                         (['src/bar.c', 'bla/foo.c'], 'src'),
 
224
                         (['src'], 'src'),
 
225
                         ]:
 
226
            self.assert_(osutils.is_inside_or_parent_of_any(dirs, fn))
 
227
 
 
228
        for dirs, fn in [(['src'], 'srccontrol'),
 
229
                         (['srccontrol/foo.c'], 'src'),
 
230
                         (['src'], 'srccontrol/foo')]:
 
231
            self.assertFalse(osutils.is_inside_or_parent_of_any(dirs, fn))
 
232
 
 
233
 
 
234
class TestRmTree(tests.TestCaseInTempDir):
 
235
 
 
236
    def test_rmtree(self):
 
237
        # Check to remove tree with read-only files/dirs
 
238
        os.mkdir('dir')
 
239
        f = file('dir/file', 'w')
 
240
        f.write('spam')
 
241
        f.close()
 
242
        # would like to also try making the directory readonly, but at the
 
243
        # moment python shutil.rmtree doesn't handle that properly - it would
 
244
        # need to chmod the directory before removing things inside it - deferred
 
245
        # for now -- mbp 20060505
 
246
        # osutils.make_readonly('dir')
 
247
        osutils.make_readonly('dir/file')
 
248
 
 
249
        osutils.rmtree('dir')
 
250
 
 
251
        self.failIfExists('dir/file')
 
252
        self.failIfExists('dir')
 
253
 
 
254
 
 
255
class TestDeleteAny(tests.TestCaseInTempDir):
 
256
 
 
257
    def test_delete_any_readonly(self):
 
258
        # from <https://bugs.launchpad.net/bzr/+bug/218206>
 
259
        self.build_tree(['d/', 'f'])
 
260
        osutils.make_readonly('d')
 
261
        osutils.make_readonly('f')
 
262
 
 
263
        osutils.delete_any('f')
 
264
        osutils.delete_any('d')
 
265
 
 
266
 
 
267
class TestKind(tests.TestCaseInTempDir):
 
268
 
 
269
    def test_file_kind(self):
 
270
        self.build_tree(['file', 'dir/'])
 
271
        self.assertEquals('file', osutils.file_kind('file'))
 
272
        self.assertEquals('directory', osutils.file_kind('dir/'))
 
273
        if osutils.has_symlinks():
 
274
            os.symlink('symlink', 'symlink')
 
275
            self.assertEquals('symlink', osutils.file_kind('symlink'))
 
276
 
 
277
        # TODO: jam 20060529 Test a block device
 
278
        try:
 
279
            os.lstat('/dev/null')
 
280
        except OSError, e:
 
281
            if e.errno not in (errno.ENOENT,):
 
282
                raise
 
283
        else:
 
284
            self.assertEquals('chardev', osutils.file_kind('/dev/null'))
 
285
 
 
286
        mkfifo = getattr(os, 'mkfifo', None)
 
287
        if mkfifo:
 
288
            mkfifo('fifo')
 
289
            try:
 
290
                self.assertEquals('fifo', osutils.file_kind('fifo'))
 
291
            finally:
 
292
                os.remove('fifo')
 
293
 
 
294
        AF_UNIX = getattr(socket, 'AF_UNIX', None)
 
295
        if AF_UNIX:
 
296
            s = socket.socket(AF_UNIX)
 
297
            s.bind('socket')
 
298
            try:
 
299
                self.assertEquals('socket', osutils.file_kind('socket'))
 
300
            finally:
 
301
                os.remove('socket')
 
302
 
 
303
    def test_kind_marker(self):
 
304
        self.assertEqual("", osutils.kind_marker("file"))
 
305
        self.assertEqual("/", osutils.kind_marker('directory'))
 
306
        self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
 
307
        self.assertEqual("@", osutils.kind_marker("symlink"))
 
308
        self.assertEqual("+", osutils.kind_marker("tree-reference"))
 
309
        self.assertEqual("", osutils.kind_marker("fifo"))
 
310
        self.assertEqual("", osutils.kind_marker("socket"))
 
311
        self.assertEqual("", osutils.kind_marker("unknown"))
 
312
 
 
313
 
 
314
class TestUmask(tests.TestCaseInTempDir):
 
315
 
 
316
    def test_get_umask(self):
 
317
        if sys.platform == 'win32':
 
318
            # umask always returns '0', no way to set it
 
319
            self.assertEqual(0, osutils.get_umask())
 
320
            return
 
321
 
 
322
        orig_umask = osutils.get_umask()
 
323
        self.addCleanup(os.umask, orig_umask)
 
324
        os.umask(0222)
 
325
        self.assertEqual(0222, osutils.get_umask())
 
326
        os.umask(0022)
 
327
        self.assertEqual(0022, osutils.get_umask())
 
328
        os.umask(0002)
 
329
        self.assertEqual(0002, osutils.get_umask())
 
330
        os.umask(0027)
 
331
        self.assertEqual(0027, osutils.get_umask())
 
332
 
 
333
 
 
334
class TestDateTime(tests.TestCase):
 
335
 
 
336
    def assertFormatedDelta(self, expected, seconds):
 
337
        """Assert osutils.format_delta formats as expected"""
 
338
        actual = osutils.format_delta(seconds)
 
339
        self.assertEqual(expected, actual)
 
340
 
 
341
    def test_format_delta(self):
 
342
        self.assertFormatedDelta('0 seconds ago', 0)
 
343
        self.assertFormatedDelta('1 second ago', 1)
 
344
        self.assertFormatedDelta('10 seconds ago', 10)
 
345
        self.assertFormatedDelta('59 seconds ago', 59)
 
346
        self.assertFormatedDelta('89 seconds ago', 89)
 
347
        self.assertFormatedDelta('1 minute, 30 seconds ago', 90)
 
348
        self.assertFormatedDelta('3 minutes, 0 seconds ago', 180)
 
349
        self.assertFormatedDelta('3 minutes, 1 second ago', 181)
 
350
        self.assertFormatedDelta('10 minutes, 15 seconds ago', 615)
 
351
        self.assertFormatedDelta('30 minutes, 59 seconds ago', 1859)
 
352
        self.assertFormatedDelta('31 minutes, 0 seconds ago', 1860)
 
353
        self.assertFormatedDelta('60 minutes, 0 seconds ago', 3600)
 
354
        self.assertFormatedDelta('89 minutes, 59 seconds ago', 5399)
 
355
        self.assertFormatedDelta('1 hour, 30 minutes ago', 5400)
 
356
        self.assertFormatedDelta('2 hours, 30 minutes ago', 9017)
 
357
        self.assertFormatedDelta('10 hours, 0 minutes ago', 36000)
 
358
        self.assertFormatedDelta('24 hours, 0 minutes ago', 86400)
 
359
        self.assertFormatedDelta('35 hours, 59 minutes ago', 129599)
 
360
        self.assertFormatedDelta('36 hours, 0 minutes ago', 129600)
 
361
        self.assertFormatedDelta('36 hours, 0 minutes ago', 129601)
 
362
        self.assertFormatedDelta('36 hours, 1 minute ago', 129660)
 
363
        self.assertFormatedDelta('36 hours, 1 minute ago', 129661)
 
364
        self.assertFormatedDelta('84 hours, 10 minutes ago', 303002)
 
365
 
 
366
        # We handle when time steps the wrong direction because computers
 
367
        # don't have synchronized clocks.
 
368
        self.assertFormatedDelta('84 hours, 10 minutes in the future', -303002)
 
369
        self.assertFormatedDelta('1 second in the future', -1)
 
370
        self.assertFormatedDelta('2 seconds in the future', -2)
 
371
 
 
372
    def test_format_date(self):
 
373
        self.assertRaises(errors.UnsupportedTimezoneFormat,
 
374
            osutils.format_date, 0, timezone='foo')
 
375
        self.assertIsInstance(osutils.format_date(0), str)
 
376
        self.assertIsInstance(osutils.format_local_date(0), unicode)
 
377
        # Testing for the actual value of the local weekday without
 
378
        # duplicating the code from format_date is difficult.
 
379
        # Instead blackbox.test_locale should check for localized
 
380
        # dates once they do occur in output strings.
 
381
 
 
382
    def test_format_date_with_offset_in_original_timezone(self):
 
383
        self.assertEqual("Thu 1970-01-01 00:00:00 +0000",
 
384
            osutils.format_date_with_offset_in_original_timezone(0))
 
385
        self.assertEqual("Fri 1970-01-02 03:46:40 +0000",
 
386
            osutils.format_date_with_offset_in_original_timezone(100000))
 
387
        self.assertEqual("Fri 1970-01-02 05:46:40 +0200",
 
388
            osutils.format_date_with_offset_in_original_timezone(100000, 7200))
 
389
 
 
390
    def test_local_time_offset(self):
 
391
        """Test that local_time_offset() returns a sane value."""
 
392
        offset = osutils.local_time_offset()
 
393
        self.assertTrue(isinstance(offset, int))
 
394
        # Test that the offset is no more than a eighteen hours in
 
395
        # either direction.
 
396
        # Time zone handling is system specific, so it is difficult to
 
397
        # do more specific tests, but a value outside of this range is
 
398
        # probably wrong.
 
399
        eighteen_hours = 18 * 3600
 
400
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
 
401
 
 
402
    def test_local_time_offset_with_timestamp(self):
 
403
        """Test that local_time_offset() works with a timestamp."""
 
404
        offset = osutils.local_time_offset(1000000000.1234567)
 
405
        self.assertTrue(isinstance(offset, int))
 
406
        eighteen_hours = 18 * 3600
 
407
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
 
408
 
 
409
 
 
410
class TestLinks(tests.TestCaseInTempDir):
 
411
 
 
412
    def test_dereference_path(self):
 
413
        self.requireFeature(tests.SymlinkFeature)
 
414
        cwd = osutils.realpath('.')
 
415
        os.mkdir('bar')
 
416
        bar_path = osutils.pathjoin(cwd, 'bar')
 
417
        # Using './' to avoid bug #1213894 (first path component not
 
418
        # dereferenced) in Python 2.4.1 and earlier
 
419
        self.assertEqual(bar_path, osutils.realpath('./bar'))
 
420
        os.symlink('bar', 'foo')
 
421
        self.assertEqual(bar_path, osutils.realpath('./foo'))
 
422
 
 
423
        # Does not dereference terminal symlinks
 
424
        foo_path = osutils.pathjoin(cwd, 'foo')
 
425
        self.assertEqual(foo_path, osutils.dereference_path('./foo'))
 
426
 
 
427
        # Dereferences parent symlinks
 
428
        os.mkdir('bar/baz')
 
429
        baz_path = osutils.pathjoin(bar_path, 'baz')
 
430
        self.assertEqual(baz_path, osutils.dereference_path('./foo/baz'))
 
431
 
 
432
        # Dereferences parent symlinks that are the first path element
 
433
        self.assertEqual(baz_path, osutils.dereference_path('foo/baz'))
 
434
 
 
435
        # Dereferences parent symlinks in absolute paths
 
436
        foo_baz_path = osutils.pathjoin(foo_path, 'baz')
 
437
        self.assertEqual(baz_path, osutils.dereference_path(foo_baz_path))
 
438
 
 
439
    def test_changing_access(self):
 
440
        f = file('file', 'w')
 
441
        f.write('monkey')
 
442
        f.close()
 
443
 
 
444
        # Make a file readonly
 
445
        osutils.make_readonly('file')
 
446
        mode = os.lstat('file').st_mode
 
447
        self.assertEqual(mode, mode & 0777555)
 
448
 
 
449
        # Make a file writable
 
450
        osutils.make_writable('file')
 
451
        mode = os.lstat('file').st_mode
 
452
        self.assertEqual(mode, mode | 0200)
 
453
 
 
454
        if osutils.has_symlinks():
 
455
            # should not error when handed a symlink
 
456
            os.symlink('nonexistent', 'dangling')
 
457
            osutils.make_readonly('dangling')
 
458
            osutils.make_writable('dangling')
 
459
 
 
460
    def test_host_os_dereferences_symlinks(self):
 
461
        osutils.host_os_dereferences_symlinks()
 
462
 
 
463
 
 
464
class TestCanonicalRelPath(tests.TestCaseInTempDir):
 
465
 
 
466
    _test_needs_features = [tests.CaseInsCasePresFilenameFeature]
 
467
 
 
468
    def test_canonical_relpath_simple(self):
 
469
        f = file('MixedCaseName', 'w')
 
470
        f.close()
 
471
        actual = osutils.canonical_relpath(self.test_base_dir, 'mixedcasename')
 
472
        self.failUnlessEqual('work/MixedCaseName', actual)
 
473
 
 
474
    def test_canonical_relpath_missing_tail(self):
 
475
        os.mkdir('MixedCaseParent')
 
476
        actual = osutils.canonical_relpath(self.test_base_dir,
 
477
                                           'mixedcaseparent/nochild')
 
478
        self.failUnlessEqual('work/MixedCaseParent/nochild', actual)
 
479
 
 
480
 
 
481
class Test_CICPCanonicalRelpath(tests.TestCaseWithTransport):
 
482
 
 
483
    def assertRelpath(self, expected, base, path):
 
484
        actual = osutils._cicp_canonical_relpath(base, path)
 
485
        self.assertEqual(expected, actual)
 
486
 
 
487
    def test_simple(self):
 
488
        self.build_tree(['MixedCaseName'])
 
489
        base = osutils.realpath(self.get_transport('.').local_abspath('.'))
 
490
        self.assertRelpath('MixedCaseName', base, 'mixedcAsename')
 
491
 
 
492
    def test_subdir_missing_tail(self):
 
493
        self.build_tree(['MixedCaseParent/', 'MixedCaseParent/a_child'])
 
494
        base = osutils.realpath(self.get_transport('.').local_abspath('.'))
 
495
        self.assertRelpath('MixedCaseParent/a_child', base,
 
496
                           'MixedCaseParent/a_child')
 
497
        self.assertRelpath('MixedCaseParent/a_child', base,
 
498
                           'MixedCaseParent/A_Child')
 
499
        self.assertRelpath('MixedCaseParent/not_child', base,
 
500
                           'MixedCaseParent/not_child')
 
501
 
 
502
    def test_at_root_slash(self):
 
503
        # We can't test this on Windows, because it has a 'MIN_ABS_PATHLENGTH'
 
504
        # check...
 
505
        if osutils.MIN_ABS_PATHLENGTH > 1:
 
506
            raise tests.TestSkipped('relpath requires %d chars'
 
507
                                    % osutils.MIN_ABS_PATHLENGTH)
 
508
        self.assertRelpath('foo', '/', '/foo')
 
509
 
 
510
    def test_at_root_drive(self):
 
511
        if sys.platform != 'win32':
 
512
            raise tests.TestNotApplicable('we can only test drive-letter relative'
 
513
                                          ' paths on Windows where we have drive'
 
514
                                          ' letters.')
 
515
        # see bug #322807
 
516
        # The specific issue is that when at the root of a drive, 'abspath'
 
517
        # returns "C:/" or just "/". However, the code assumes that abspath
 
518
        # always returns something like "C:/foo" or "/foo" (no trailing slash).
 
519
        self.assertRelpath('foo', 'C:/', 'C:/foo')
 
520
        self.assertRelpath('foo', 'X:/', 'X:/foo')
 
521
        self.assertRelpath('foo', 'X:/', 'X://foo')
 
522
 
 
523
 
 
524
class TestPumpFile(tests.TestCase):
 
525
    """Test pumpfile method."""
 
526
 
 
527
    def setUp(self):
 
528
        tests.TestCase.setUp(self)
 
529
        # create a test datablock
 
530
        self.block_size = 512
 
531
        pattern = '0123456789ABCDEF'
 
532
        self.test_data = pattern * (3 * self.block_size / len(pattern))
 
533
        self.test_data_len = len(self.test_data)
 
534
 
 
535
    def test_bracket_block_size(self):
 
536
        """Read data in blocks with the requested read size bracketing the
 
537
        block size."""
 
538
        # make sure test data is larger than max read size
 
539
        self.assertTrue(self.test_data_len > self.block_size)
 
540
 
 
541
        from_file = file_utils.FakeReadFile(self.test_data)
 
542
        to_file = StringIO()
 
543
 
 
544
        # read (max / 2) bytes and verify read size wasn't affected
 
545
        num_bytes_to_read = self.block_size / 2
 
546
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
547
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
 
548
        self.assertEqual(from_file.get_read_count(), 1)
 
549
 
 
550
        # read (max) bytes and verify read size wasn't affected
 
551
        num_bytes_to_read = self.block_size
 
552
        from_file.reset_read_count()
 
553
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
554
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
 
555
        self.assertEqual(from_file.get_read_count(), 1)
 
556
 
 
557
        # read (max + 1) bytes and verify read size was limited
 
558
        num_bytes_to_read = self.block_size + 1
 
559
        from_file.reset_read_count()
 
560
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
561
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
 
562
        self.assertEqual(from_file.get_read_count(), 2)
 
563
 
 
564
        # finish reading the rest of the data
 
565
        num_bytes_to_read = self.test_data_len - to_file.tell()
 
566
        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
567
 
 
568
        # report error if the data wasn't equal (we only report the size due
 
569
        # to the length of the data)
 
570
        response_data = to_file.getvalue()
 
571
        if response_data != self.test_data:
 
572
            message = "Data not equal.  Expected %d bytes, received %d."
 
573
            self.fail(message % (len(response_data), self.test_data_len))
 
574
 
 
575
    def test_specified_size(self):
 
576
        """Request a transfer larger than the maximum block size and verify
 
577
        that the maximum read doesn't exceed the block_size."""
 
578
        # make sure test data is larger than max read size
 
579
        self.assertTrue(self.test_data_len > self.block_size)
 
580
 
 
581
        # retrieve data in blocks
 
582
        from_file = file_utils.FakeReadFile(self.test_data)
 
583
        to_file = StringIO()
 
584
        osutils.pumpfile(from_file, to_file, self.test_data_len,
 
585
                         self.block_size)
 
586
 
 
587
        # verify read size was equal to the maximum read size
 
588
        self.assertTrue(from_file.get_max_read_size() > 0)
 
589
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
 
590
        self.assertEqual(from_file.get_read_count(), 3)
 
591
 
 
592
        # report error if the data wasn't equal (we only report the size due
 
593
        # to the length of the data)
 
594
        response_data = to_file.getvalue()
 
595
        if response_data != self.test_data:
 
596
            message = "Data not equal.  Expected %d bytes, received %d."
 
597
            self.fail(message % (len(response_data), self.test_data_len))
 
598
 
 
599
    def test_to_eof(self):
 
600
        """Read to end-of-file and verify that the reads are not larger than
 
601
        the maximum read size."""
 
602
        # make sure test data is larger than max read size
 
603
        self.assertTrue(self.test_data_len > self.block_size)
 
604
 
 
605
        # retrieve data to EOF
 
606
        from_file = file_utils.FakeReadFile(self.test_data)
 
607
        to_file = StringIO()
 
608
        osutils.pumpfile(from_file, to_file, -1, self.block_size)
 
609
 
 
610
        # verify read size was equal to the maximum read size
 
611
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
 
612
        self.assertEqual(from_file.get_read_count(), 4)
 
613
 
 
614
        # report error if the data wasn't equal (we only report the size due
 
615
        # to the length of the data)
 
616
        response_data = to_file.getvalue()
 
617
        if response_data != self.test_data:
 
618
            message = "Data not equal.  Expected %d bytes, received %d."
 
619
            self.fail(message % (len(response_data), self.test_data_len))
 
620
 
 
621
    def test_defaults(self):
 
622
        """Verifies that the default arguments will read to EOF -- this
 
623
        test verifies that any existing usages of pumpfile will not be broken
 
624
        with this new version."""
 
625
        # retrieve data using default (old) pumpfile method
 
626
        from_file = file_utils.FakeReadFile(self.test_data)
 
627
        to_file = StringIO()
 
628
        osutils.pumpfile(from_file, to_file)
 
629
 
 
630
        # report error if the data wasn't equal (we only report the size due
 
631
        # to the length of the data)
 
632
        response_data = to_file.getvalue()
 
633
        if response_data != self.test_data:
 
634
            message = "Data not equal.  Expected %d bytes, received %d."
 
635
            self.fail(message % (len(response_data), self.test_data_len))
 
636
 
 
637
    def test_report_activity(self):
 
638
        activity = []
 
639
        def log_activity(length, direction):
 
640
            activity.append((length, direction))
 
641
        from_file = StringIO(self.test_data)
 
642
        to_file = StringIO()
 
643
        osutils.pumpfile(from_file, to_file, buff_size=500,
 
644
                         report_activity=log_activity, direction='read')
 
645
        self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
 
646
                          (36, 'read')], activity)
 
647
 
 
648
        from_file = StringIO(self.test_data)
 
649
        to_file = StringIO()
 
650
        del activity[:]
 
651
        osutils.pumpfile(from_file, to_file, buff_size=500,
 
652
                         report_activity=log_activity, direction='write')
 
653
        self.assertEqual([(500, 'write'), (500, 'write'), (500, 'write'),
 
654
                          (36, 'write')], activity)
 
655
 
 
656
        # And with a limited amount of data
 
657
        from_file = StringIO(self.test_data)
 
658
        to_file = StringIO()
 
659
        del activity[:]
 
660
        osutils.pumpfile(from_file, to_file, buff_size=500, read_length=1028,
 
661
                         report_activity=log_activity, direction='read')
 
662
        self.assertEqual([(500, 'read'), (500, 'read'), (28, 'read')], activity)
 
663
 
 
664
 
 
665
 
 
666
class TestPumpStringFile(tests.TestCase):
 
667
 
 
668
    def test_empty(self):
 
669
        output = StringIO()
 
670
        osutils.pump_string_file("", output)
 
671
        self.assertEqual("", output.getvalue())
 
672
 
 
673
    def test_more_than_segment_size(self):
 
674
        output = StringIO()
 
675
        osutils.pump_string_file("123456789", output, 2)
 
676
        self.assertEqual("123456789", output.getvalue())
 
677
 
 
678
    def test_segment_size(self):
 
679
        output = StringIO()
 
680
        osutils.pump_string_file("12", output, 2)
 
681
        self.assertEqual("12", output.getvalue())
 
682
 
 
683
    def test_segment_size_multiple(self):
 
684
        output = StringIO()
 
685
        osutils.pump_string_file("1234", output, 2)
 
686
        self.assertEqual("1234", output.getvalue())
 
687
 
 
688
 
 
689
class TestRelpath(tests.TestCase):
 
690
 
 
691
    def test_simple_relpath(self):
 
692
        cwd = osutils.getcwd()
 
693
        subdir = cwd + '/subdir'
 
694
        self.assertEqual('subdir', osutils.relpath(cwd, subdir))
 
695
 
 
696
    def test_deep_relpath(self):
 
697
        cwd = osutils.getcwd()
 
698
        subdir = cwd + '/sub/subsubdir'
 
699
        self.assertEqual('sub/subsubdir', osutils.relpath(cwd, subdir))
 
700
 
 
701
    def test_not_relative(self):
 
702
        self.assertRaises(errors.PathNotChild,
 
703
                          osutils.relpath, 'C:/path', 'H:/path')
 
704
        self.assertRaises(errors.PathNotChild,
 
705
                          osutils.relpath, 'C:/', 'H:/path')
 
706
 
 
707
 
 
708
class TestSafeUnicode(tests.TestCase):
 
709
 
 
710
    def test_from_ascii_string(self):
 
711
        self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
 
712
 
 
713
    def test_from_unicode_string_ascii_contents(self):
 
714
        self.assertEqual(u'bargam', osutils.safe_unicode(u'bargam'))
 
715
 
 
716
    def test_from_unicode_string_unicode_contents(self):
 
717
        self.assertEqual(u'bargam\xae', osutils.safe_unicode(u'bargam\xae'))
 
718
 
 
719
    def test_from_utf8_string(self):
 
720
        self.assertEqual(u'foo\xae', osutils.safe_unicode('foo\xc2\xae'))
 
721
 
 
722
    def test_bad_utf8_string(self):
 
723
        self.assertRaises(errors.BzrBadParameterNotUnicode,
 
724
                          osutils.safe_unicode,
 
725
                          '\xbb\xbb')
 
726
 
 
727
 
 
728
class TestSafeUtf8(tests.TestCase):
 
729
 
 
730
    def test_from_ascii_string(self):
 
731
        f = 'foobar'
 
732
        self.assertEqual('foobar', osutils.safe_utf8(f))
 
733
 
 
734
    def test_from_unicode_string_ascii_contents(self):
 
735
        self.assertEqual('bargam', osutils.safe_utf8(u'bargam'))
 
736
 
 
737
    def test_from_unicode_string_unicode_contents(self):
 
738
        self.assertEqual('bargam\xc2\xae', osutils.safe_utf8(u'bargam\xae'))
 
739
 
 
740
    def test_from_utf8_string(self):
 
741
        self.assertEqual('foo\xc2\xae', osutils.safe_utf8('foo\xc2\xae'))
 
742
 
 
743
    def test_bad_utf8_string(self):
 
744
        self.assertRaises(errors.BzrBadParameterNotUnicode,
 
745
                          osutils.safe_utf8, '\xbb\xbb')
 
746
 
 
747
 
 
748
class TestSafeRevisionId(tests.TestCase):
 
749
 
 
750
    def test_from_ascii_string(self):
 
751
        # this shouldn't give a warning because it's getting an ascii string
 
752
        self.assertEqual('foobar', osutils.safe_revision_id('foobar'))
 
753
 
 
754
    def test_from_unicode_string_ascii_contents(self):
 
755
        self.assertEqual('bargam',
 
756
                         osutils.safe_revision_id(u'bargam', warn=False))
 
757
 
 
758
    def test_from_unicode_deprecated(self):
 
759
        self.assertEqual('bargam',
 
760
            self.callDeprecated([osutils._revision_id_warning],
 
761
                                osutils.safe_revision_id, u'bargam'))
 
762
 
 
763
    def test_from_unicode_string_unicode_contents(self):
 
764
        self.assertEqual('bargam\xc2\xae',
 
765
                         osutils.safe_revision_id(u'bargam\xae', warn=False))
 
766
 
 
767
    def test_from_utf8_string(self):
 
768
        self.assertEqual('foo\xc2\xae',
 
769
                         osutils.safe_revision_id('foo\xc2\xae'))
 
770
 
 
771
    def test_none(self):
 
772
        """Currently, None is a valid revision_id"""
 
773
        self.assertEqual(None, osutils.safe_revision_id(None))
 
774
 
 
775
 
 
776
class TestSafeFileId(tests.TestCase):
 
777
 
 
778
    def test_from_ascii_string(self):
 
779
        self.assertEqual('foobar', osutils.safe_file_id('foobar'))
 
780
 
 
781
    def test_from_unicode_string_ascii_contents(self):
 
782
        self.assertEqual('bargam', osutils.safe_file_id(u'bargam', warn=False))
 
783
 
 
784
    def test_from_unicode_deprecated(self):
 
785
        self.assertEqual('bargam',
 
786
            self.callDeprecated([osutils._file_id_warning],
 
787
                                osutils.safe_file_id, u'bargam'))
 
788
 
 
789
    def test_from_unicode_string_unicode_contents(self):
 
790
        self.assertEqual('bargam\xc2\xae',
 
791
                         osutils.safe_file_id(u'bargam\xae', warn=False))
 
792
 
 
793
    def test_from_utf8_string(self):
 
794
        self.assertEqual('foo\xc2\xae',
 
795
                         osutils.safe_file_id('foo\xc2\xae'))
 
796
 
 
797
    def test_none(self):
 
798
        """Currently, None is a valid revision_id"""
 
799
        self.assertEqual(None, osutils.safe_file_id(None))
 
800
 
 
801
 
 
802
class TestWin32Funcs(tests.TestCase):
 
803
    """Test that _win32 versions of os utilities return appropriate paths."""
 
804
 
 
805
    def test_abspath(self):
 
806
        self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
 
807
        self.assertEqual('C:/foo', osutils._win32_abspath('C:/foo'))
 
808
        self.assertEqual('//HOST/path', osutils._win32_abspath(r'\\HOST\path'))
 
809
        self.assertEqual('//HOST/path', osutils._win32_abspath('//HOST/path'))
 
810
 
 
811
    def test_realpath(self):
 
812
        self.assertEqual('C:/foo', osutils._win32_realpath('C:\\foo'))
 
813
        self.assertEqual('C:/foo', osutils._win32_realpath('C:/foo'))
 
814
 
 
815
    def test_pathjoin(self):
 
816
        self.assertEqual('path/to/foo',
 
817
                         osutils._win32_pathjoin('path', 'to', 'foo'))
 
818
        self.assertEqual('C:/foo',
 
819
                         osutils._win32_pathjoin('path\\to', 'C:\\foo'))
 
820
        self.assertEqual('C:/foo',
 
821
                         osutils._win32_pathjoin('path/to', 'C:/foo'))
 
822
        self.assertEqual('path/to/foo',
 
823
                         osutils._win32_pathjoin('path/to/', 'foo'))
 
824
        self.assertEqual('/foo',
 
825
                         osutils._win32_pathjoin('C:/path/to/', '/foo'))
 
826
        self.assertEqual('/foo',
 
827
                         osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
 
828
 
 
829
    def test_normpath(self):
 
830
        self.assertEqual('path/to/foo',
 
831
                         osutils._win32_normpath(r'path\\from\..\to\.\foo'))
 
832
        self.assertEqual('path/to/foo',
 
833
                         osutils._win32_normpath('path//from/../to/./foo'))
 
834
 
 
835
    def test_getcwd(self):
 
836
        cwd = osutils._win32_getcwd()
 
837
        os_cwd = os.getcwdu()
 
838
        self.assertEqual(os_cwd[1:].replace('\\', '/'), cwd[1:])
 
839
        # win32 is inconsistent whether it returns lower or upper case
 
840
        # and even if it was consistent the user might type the other
 
841
        # so we force it to uppercase
 
842
        # running python.exe under cmd.exe return capital C:\\
 
843
        # running win32 python inside a cygwin shell returns lowercase
 
844
        self.assertEqual(os_cwd[0].upper(), cwd[0])
 
845
 
 
846
    def test_fixdrive(self):
 
847
        self.assertEqual('H:/foo', osutils._win32_fixdrive('h:/foo'))
 
848
        self.assertEqual('H:/foo', osutils._win32_fixdrive('H:/foo'))
 
849
        self.assertEqual('C:\\foo', osutils._win32_fixdrive('c:\\foo'))
 
850
 
 
851
    def test_win98_abspath(self):
 
852
        # absolute path
 
853
        self.assertEqual('C:/foo', osutils._win98_abspath('C:\\foo'))
 
854
        self.assertEqual('C:/foo', osutils._win98_abspath('C:/foo'))
 
855
        # UNC path
 
856
        self.assertEqual('//HOST/path', osutils._win98_abspath(r'\\HOST\path'))
 
857
        self.assertEqual('//HOST/path', osutils._win98_abspath('//HOST/path'))
 
858
        # relative path
 
859
        cwd = osutils.getcwd().rstrip('/')
 
860
        drive = osutils.ntpath.splitdrive(cwd)[0]
 
861
        self.assertEqual(cwd+'/path', osutils._win98_abspath('path'))
 
862
        self.assertEqual(drive+'/path', osutils._win98_abspath('/path'))
 
863
        # unicode path
 
864
        u = u'\u1234'
 
865
        self.assertEqual(cwd+'/'+u, osutils._win98_abspath(u))
 
866
 
 
867
 
 
868
class TestWin32FuncsDirs(tests.TestCaseInTempDir):
 
869
    """Test win32 functions that create files."""
 
870
 
 
871
    def test_getcwd(self):
 
872
        self.requireFeature(tests.UnicodeFilenameFeature)
 
873
        os.mkdir(u'mu-\xb5')
 
874
        os.chdir(u'mu-\xb5')
 
875
        # TODO: jam 20060427 This will probably fail on Mac OSX because
 
876
        #       it will change the normalization of B\xe5gfors
 
877
        #       Consider using a different unicode character, or make
 
878
        #       osutils.getcwd() renormalize the path.
 
879
        self.assertEndsWith(osutils._win32_getcwd(), u'mu-\xb5')
 
880
 
 
881
    def test_minimum_path_selection(self):
 
882
        self.assertEqual(set(),
 
883
            osutils.minimum_path_selection([]))
 
884
        self.assertEqual(set(['a']),
 
885
            osutils.minimum_path_selection(['a']))
 
886
        self.assertEqual(set(['a', 'b']),
 
887
            osutils.minimum_path_selection(['a', 'b']))
 
888
        self.assertEqual(set(['a/', 'b']),
 
889
            osutils.minimum_path_selection(['a/', 'b']))
 
890
        self.assertEqual(set(['a/', 'b']),
 
891
            osutils.minimum_path_selection(['a/c', 'a/', 'b']))
 
892
        self.assertEqual(set(['a-b', 'a', 'a0b']),
 
893
            osutils.minimum_path_selection(['a-b', 'a/b', 'a0b', 'a']))
 
894
 
 
895
    def test_mkdtemp(self):
 
896
        tmpdir = osutils._win32_mkdtemp(dir='.')
 
897
        self.assertFalse('\\' in tmpdir)
 
898
 
 
899
    def test_rename(self):
 
900
        a = open('a', 'wb')
 
901
        a.write('foo\n')
 
902
        a.close()
 
903
        b = open('b', 'wb')
 
904
        b.write('baz\n')
 
905
        b.close()
 
906
 
 
907
        osutils._win32_rename('b', 'a')
 
908
        self.failUnlessExists('a')
 
909
        self.failIfExists('b')
 
910
        self.assertFileEqual('baz\n', 'a')
 
911
 
 
912
    def test_rename_missing_file(self):
 
913
        a = open('a', 'wb')
 
914
        a.write('foo\n')
 
915
        a.close()
 
916
 
 
917
        try:
 
918
            osutils._win32_rename('b', 'a')
 
919
        except (IOError, OSError), e:
 
920
            self.assertEqual(errno.ENOENT, e.errno)
 
921
        self.assertFileEqual('foo\n', 'a')
 
922
 
 
923
    def test_rename_missing_dir(self):
 
924
        os.mkdir('a')
 
925
        try:
 
926
            osutils._win32_rename('b', 'a')
 
927
        except (IOError, OSError), e:
 
928
            self.assertEqual(errno.ENOENT, e.errno)
 
929
 
 
930
    def test_rename_current_dir(self):
 
931
        os.mkdir('a')
 
932
        os.chdir('a')
 
933
        # You can't rename the working directory
 
934
        # doing rename non-existant . usually
 
935
        # just raises ENOENT, since non-existant
 
936
        # doesn't exist.
 
937
        try:
 
938
            osutils._win32_rename('b', '.')
 
939
        except (IOError, OSError), e:
 
940
            self.assertEqual(errno.ENOENT, e.errno)
 
941
 
 
942
    def test_splitpath(self):
 
943
        def check(expected, path):
 
944
            self.assertEqual(expected, osutils.splitpath(path))
 
945
 
 
946
        check(['a'], 'a')
 
947
        check(['a', 'b'], 'a/b')
 
948
        check(['a', 'b'], 'a/./b')
 
949
        check(['a', '.b'], 'a/.b')
 
950
        check(['a', '.b'], 'a\\.b')
 
951
 
 
952
        self.assertRaises(errors.BzrError, osutils.splitpath, 'a/../b')
 
953
 
 
954
 
 
955
class TestParentDirectories(tests.TestCaseInTempDir):
 
956
    """Test osutils.parent_directories()"""
 
957
 
 
958
    def test_parent_directories(self):
 
959
        self.assertEqual([], osutils.parent_directories('a'))
 
960
        self.assertEqual(['a'], osutils.parent_directories('a/b'))
 
961
        self.assertEqual(['a/b', 'a'], osutils.parent_directories('a/b/c'))
 
962
 
 
963
 
 
964
class TestMacFuncsDirs(tests.TestCaseInTempDir):
 
965
    """Test mac special functions that require directories."""
 
966
 
 
967
    def test_getcwd(self):
 
968
        self.requireFeature(tests.UnicodeFilenameFeature)
 
969
        os.mkdir(u'B\xe5gfors')
 
970
        os.chdir(u'B\xe5gfors')
 
971
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
 
972
 
 
973
    def test_getcwd_nonnorm(self):
 
974
        self.requireFeature(tests.UnicodeFilenameFeature)
 
975
        # Test that _mac_getcwd() will normalize this path
 
976
        os.mkdir(u'Ba\u030agfors')
 
977
        os.chdir(u'Ba\u030agfors')
 
978
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
 
979
 
 
980
 
 
981
class TestChunksToLines(tests.TestCase):
 
982
 
 
983
    def test_smoketest(self):
 
984
        self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
 
985
                         osutils.chunks_to_lines(['foo\nbar', '\nbaz\n']))
 
986
        self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
 
987
                         osutils.chunks_to_lines(['foo\n', 'bar\n', 'baz\n']))
 
988
 
 
989
    def test_osutils_binding(self):
 
990
        from bzrlib.tests import test__chunks_to_lines
 
991
        if test__chunks_to_lines.compiled_chunkstolines_feature.available():
 
992
            from bzrlib._chunks_to_lines_pyx import chunks_to_lines
 
993
        else:
 
994
            from bzrlib._chunks_to_lines_py import chunks_to_lines
 
995
        self.assertIs(chunks_to_lines, osutils.chunks_to_lines)
 
996
 
 
997
 
 
998
class TestSplitLines(tests.TestCase):
 
999
 
 
1000
    def test_split_unicode(self):
 
1001
        self.assertEqual([u'foo\n', u'bar\xae'],
 
1002
                         osutils.split_lines(u'foo\nbar\xae'))
 
1003
        self.assertEqual([u'foo\n', u'bar\xae\n'],
 
1004
                         osutils.split_lines(u'foo\nbar\xae\n'))
 
1005
 
 
1006
    def test_split_with_carriage_returns(self):
 
1007
        self.assertEqual(['foo\rbar\n'],
 
1008
                         osutils.split_lines('foo\rbar\n'))
 
1009
 
 
1010
 
 
1011
class TestWalkDirs(tests.TestCaseInTempDir):
 
1012
 
 
1013
    def assertExpectedBlocks(self, expected, result):
 
1014
        self.assertEqual(expected,
 
1015
                         [(dirinfo, [line[0:3] for line in block])
 
1016
                          for dirinfo, block in result])
 
1017
 
 
1018
    def test_walkdirs(self):
 
1019
        tree = [
 
1020
            '.bzr',
 
1021
            '0file',
 
1022
            '1dir/',
 
1023
            '1dir/0file',
 
1024
            '1dir/1dir/',
 
1025
            '2file'
 
1026
            ]
 
1027
        self.build_tree(tree)
 
1028
        expected_dirblocks = [
 
1029
                (('', '.'),
 
1030
                 [('0file', '0file', 'file'),
 
1031
                  ('1dir', '1dir', 'directory'),
 
1032
                  ('2file', '2file', 'file'),
 
1033
                 ]
 
1034
                ),
 
1035
                (('1dir', './1dir'),
 
1036
                 [('1dir/0file', '0file', 'file'),
 
1037
                  ('1dir/1dir', '1dir', 'directory'),
 
1038
                 ]
 
1039
                ),
 
1040
                (('1dir/1dir', './1dir/1dir'),
 
1041
                 [
 
1042
                 ]
 
1043
                ),
 
1044
            ]
 
1045
        result = []
 
1046
        found_bzrdir = False
 
1047
        for dirdetail, dirblock in osutils.walkdirs('.'):
 
1048
            if len(dirblock) and dirblock[0][1] == '.bzr':
 
1049
                # this tests the filtering of selected paths
 
1050
                found_bzrdir = True
 
1051
                del dirblock[0]
 
1052
            result.append((dirdetail, dirblock))
 
1053
 
 
1054
        self.assertTrue(found_bzrdir)
 
1055
        self.assertExpectedBlocks(expected_dirblocks, result)
 
1056
        # you can search a subdir only, with a supplied prefix.
 
1057
        result = []
 
1058
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
 
1059
            result.append(dirblock)
 
1060
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
 
1061
 
 
1062
    def test_walkdirs_os_error(self):
 
1063
        # <https://bugs.launchpad.net/bzr/+bug/338653>
 
1064
        # Pyrex readdir didn't raise useful messages if it had an error
 
1065
        # reading the directory
 
1066
        if sys.platform == 'win32':
 
1067
            raise tests.TestNotApplicable(
 
1068
                "readdir IOError not tested on win32")
 
1069
        self.requireFeature(features.not_running_as_root)
 
1070
        os.mkdir("test-unreadable")
 
1071
        os.chmod("test-unreadable", 0000)
 
1072
        # must chmod it back so that it can be removed
 
1073
        self.addCleanup(os.chmod, "test-unreadable", 0700)
 
1074
        # The error is not raised until the generator is actually evaluated.
 
1075
        # (It would be ok if it happened earlier but at the moment it
 
1076
        # doesn't.)
 
1077
        e = self.assertRaises(OSError, list, osutils._walkdirs_utf8("."))
 
1078
        self.assertEquals('./test-unreadable', e.filename)
 
1079
        self.assertEquals(errno.EACCES, e.errno)
 
1080
        # Ensure the message contains the file name
 
1081
        self.assertContainsRe(str(e), "\./test-unreadable")
 
1082
 
 
1083
 
 
1084
    def test_walkdirs_encoding_error(self):
 
1085
        # <https://bugs.launchpad.net/bzr/+bug/488519>
 
1086
        # walkdirs didn't raise a useful message when the filenames
 
1087
        # are not using the filesystem's encoding
 
1088
 
 
1089
        # require a bytestring based filesystem
 
1090
        self.requireFeature(tests.ByteStringNamedFilesystem)
 
1091
 
 
1092
        tree = [
 
1093
            '.bzr',
 
1094
            '0file',
 
1095
            '1dir/',
 
1096
            '1dir/0file',
 
1097
            '1dir/1dir/',
 
1098
            '1file'
 
1099
            ]
 
1100
 
 
1101
        self.build_tree(tree)
 
1102
 
 
1103
        # rename the 1file to a latin-1 filename
 
1104
        os.rename("./1file", "\xe8file")
 
1105
        if "\xe8file" not in os.listdir("."):
 
1106
            self.skip("Lack filesystem that preserves arbitrary bytes")
 
1107
 
 
1108
        self._save_platform_info()
 
1109
        win32utils.winver = None # Avoid the win32 detection code
 
1110
        osutils._fs_enc = 'UTF-8'
 
1111
 
 
1112
        # this should raise on error
 
1113
        def attempt():
 
1114
            for dirdetail, dirblock in osutils.walkdirs('.'):
 
1115
                pass
 
1116
 
 
1117
        self.assertRaises(errors.BadFilenameEncoding, attempt)
 
1118
 
 
1119
    def test__walkdirs_utf8(self):
 
1120
        tree = [
 
1121
            '.bzr',
 
1122
            '0file',
 
1123
            '1dir/',
 
1124
            '1dir/0file',
 
1125
            '1dir/1dir/',
 
1126
            '2file'
 
1127
            ]
 
1128
        self.build_tree(tree)
 
1129
        expected_dirblocks = [
 
1130
                (('', '.'),
 
1131
                 [('0file', '0file', 'file'),
 
1132
                  ('1dir', '1dir', 'directory'),
 
1133
                  ('2file', '2file', 'file'),
 
1134
                 ]
 
1135
                ),
 
1136
                (('1dir', './1dir'),
 
1137
                 [('1dir/0file', '0file', 'file'),
 
1138
                  ('1dir/1dir', '1dir', 'directory'),
 
1139
                 ]
 
1140
                ),
 
1141
                (('1dir/1dir', './1dir/1dir'),
 
1142
                 [
 
1143
                 ]
 
1144
                ),
 
1145
            ]
 
1146
        result = []
 
1147
        found_bzrdir = False
 
1148
        for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
 
1149
            if len(dirblock) and dirblock[0][1] == '.bzr':
 
1150
                # this tests the filtering of selected paths
 
1151
                found_bzrdir = True
 
1152
                del dirblock[0]
 
1153
            result.append((dirdetail, dirblock))
 
1154
 
 
1155
        self.assertTrue(found_bzrdir)
 
1156
        self.assertExpectedBlocks(expected_dirblocks, result)
 
1157
 
 
1158
        # you can search a subdir only, with a supplied prefix.
 
1159
        result = []
 
1160
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
 
1161
            result.append(dirblock)
 
1162
        self.assertExpectedBlocks(expected_dirblocks[1:], result)
 
1163
 
 
1164
    def _filter_out_stat(self, result):
 
1165
        """Filter out the stat value from the walkdirs result"""
 
1166
        for dirdetail, dirblock in result:
 
1167
            new_dirblock = []
 
1168
            for info in dirblock:
 
1169
                # Ignore info[3] which is the stat
 
1170
                new_dirblock.append((info[0], info[1], info[2], info[4]))
 
1171
            dirblock[:] = new_dirblock
 
1172
 
 
1173
    def _save_platform_info(self):
 
1174
        self.overrideAttr(win32utils, 'winver')
 
1175
        self.overrideAttr(osutils, '_fs_enc')
 
1176
        self.overrideAttr(osutils, '_selected_dir_reader')
 
1177
 
 
1178
    def assertDirReaderIs(self, expected):
 
1179
        """Assert the right implementation for _walkdirs_utf8 is chosen."""
 
1180
        # Force it to redetect
 
1181
        osutils._selected_dir_reader = None
 
1182
        # Nothing to list, but should still trigger the selection logic
 
1183
        self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
 
1184
        self.assertIsInstance(osutils._selected_dir_reader, expected)
 
1185
 
 
1186
    def test_force_walkdirs_utf8_fs_utf8(self):
 
1187
        self.requireFeature(UTF8DirReaderFeature)
 
1188
        self._save_platform_info()
 
1189
        win32utils.winver = None # Avoid the win32 detection code
 
1190
        osutils._fs_enc = 'UTF-8'
 
1191
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
 
1192
 
 
1193
    def test_force_walkdirs_utf8_fs_ascii(self):
 
1194
        self.requireFeature(UTF8DirReaderFeature)
 
1195
        self._save_platform_info()
 
1196
        win32utils.winver = None # Avoid the win32 detection code
 
1197
        osutils._fs_enc = 'US-ASCII'
 
1198
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
 
1199
 
 
1200
    def test_force_walkdirs_utf8_fs_ANSI(self):
 
1201
        self.requireFeature(UTF8DirReaderFeature)
 
1202
        self._save_platform_info()
 
1203
        win32utils.winver = None # Avoid the win32 detection code
 
1204
        osutils._fs_enc = 'ANSI_X3.4-1968'
 
1205
        self.assertDirReaderIs(UTF8DirReaderFeature.reader)
 
1206
 
 
1207
    def test_force_walkdirs_utf8_fs_latin1(self):
 
1208
        self._save_platform_info()
 
1209
        win32utils.winver = None # Avoid the win32 detection code
 
1210
        osutils._fs_enc = 'latin1'
 
1211
        self.assertDirReaderIs(osutils.UnicodeDirReader)
 
1212
 
 
1213
    def test_force_walkdirs_utf8_nt(self):
 
1214
        # Disabled because the thunk of the whole walkdirs api is disabled.
 
1215
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1216
        self._save_platform_info()
 
1217
        win32utils.winver = 'Windows NT'
 
1218
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1219
        self.assertDirReaderIs(Win32ReadDir)
 
1220
 
 
1221
    def test_force_walkdirs_utf8_98(self):
 
1222
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1223
        self._save_platform_info()
 
1224
        win32utils.winver = 'Windows 98'
 
1225
        self.assertDirReaderIs(osutils.UnicodeDirReader)
 
1226
 
 
1227
    def test_unicode_walkdirs(self):
 
1228
        """Walkdirs should always return unicode paths."""
 
1229
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1230
        name0 = u'0file-\xb6'
 
1231
        name1 = u'1dir-\u062c\u0648'
 
1232
        name2 = u'2file-\u0633'
 
1233
        tree = [
 
1234
            name0,
 
1235
            name1 + '/',
 
1236
            name1 + '/' + name0,
 
1237
            name1 + '/' + name1 + '/',
 
1238
            name2,
 
1239
            ]
 
1240
        self.build_tree(tree)
 
1241
        expected_dirblocks = [
 
1242
                ((u'', u'.'),
 
1243
                 [(name0, name0, 'file', './' + name0),
 
1244
                  (name1, name1, 'directory', './' + name1),
 
1245
                  (name2, name2, 'file', './' + name2),
 
1246
                 ]
 
1247
                ),
 
1248
                ((name1, './' + name1),
 
1249
                 [(name1 + '/' + name0, name0, 'file', './' + name1
 
1250
                                                        + '/' + name0),
 
1251
                  (name1 + '/' + name1, name1, 'directory', './' + name1
 
1252
                                                            + '/' + name1),
 
1253
                 ]
 
1254
                ),
 
1255
                ((name1 + '/' + name1, './' + name1 + '/' + name1),
 
1256
                 [
 
1257
                 ]
 
1258
                ),
 
1259
            ]
 
1260
        result = list(osutils.walkdirs('.'))
 
1261
        self._filter_out_stat(result)
 
1262
        self.assertEqual(expected_dirblocks, result)
 
1263
        result = list(osutils.walkdirs(u'./'+name1, name1))
 
1264
        self._filter_out_stat(result)
 
1265
        self.assertEqual(expected_dirblocks[1:], result)
 
1266
 
 
1267
    def test_unicode__walkdirs_utf8(self):
 
1268
        """Walkdirs_utf8 should always return utf8 paths.
 
1269
 
 
1270
        The abspath portion might be in unicode or utf-8
 
1271
        """
 
1272
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1273
        name0 = u'0file-\xb6'
 
1274
        name1 = u'1dir-\u062c\u0648'
 
1275
        name2 = u'2file-\u0633'
 
1276
        tree = [
 
1277
            name0,
 
1278
            name1 + '/',
 
1279
            name1 + '/' + name0,
 
1280
            name1 + '/' + name1 + '/',
 
1281
            name2,
 
1282
            ]
 
1283
        self.build_tree(tree)
 
1284
        name0 = name0.encode('utf8')
 
1285
        name1 = name1.encode('utf8')
 
1286
        name2 = name2.encode('utf8')
 
1287
 
 
1288
        expected_dirblocks = [
 
1289
                (('', '.'),
 
1290
                 [(name0, name0, 'file', './' + name0),
 
1291
                  (name1, name1, 'directory', './' + name1),
 
1292
                  (name2, name2, 'file', './' + name2),
 
1293
                 ]
 
1294
                ),
 
1295
                ((name1, './' + name1),
 
1296
                 [(name1 + '/' + name0, name0, 'file', './' + name1
 
1297
                                                        + '/' + name0),
 
1298
                  (name1 + '/' + name1, name1, 'directory', './' + name1
 
1299
                                                            + '/' + name1),
 
1300
                 ]
 
1301
                ),
 
1302
                ((name1 + '/' + name1, './' + name1 + '/' + name1),
 
1303
                 [
 
1304
                 ]
 
1305
                ),
 
1306
            ]
 
1307
        result = []
 
1308
        # For ease in testing, if walkdirs_utf8 returns Unicode, assert that
 
1309
        # all abspaths are Unicode, and encode them back into utf8.
 
1310
        for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
 
1311
            self.assertIsInstance(dirdetail[0], str)
 
1312
            if isinstance(dirdetail[1], unicode):
 
1313
                dirdetail = (dirdetail[0], dirdetail[1].encode('utf8'))
 
1314
                dirblock = [list(info) for info in dirblock]
 
1315
                for info in dirblock:
 
1316
                    self.assertIsInstance(info[4], unicode)
 
1317
                    info[4] = info[4].encode('utf8')
 
1318
            new_dirblock = []
 
1319
            for info in dirblock:
 
1320
                self.assertIsInstance(info[0], str)
 
1321
                self.assertIsInstance(info[1], str)
 
1322
                self.assertIsInstance(info[4], str)
 
1323
                # Remove the stat information
 
1324
                new_dirblock.append((info[0], info[1], info[2], info[4]))
 
1325
            result.append((dirdetail, new_dirblock))
 
1326
        self.assertEqual(expected_dirblocks, result)
 
1327
 
 
1328
    def test__walkdirs_utf8_with_unicode_fs(self):
 
1329
        """UnicodeDirReader should be a safe fallback everywhere
 
1330
 
 
1331
        The abspath portion should be in unicode
 
1332
        """
 
1333
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1334
        # Use the unicode reader. TODO: split into driver-and-driven unit
 
1335
        # tests.
 
1336
        self._save_platform_info()
 
1337
        osutils._selected_dir_reader = osutils.UnicodeDirReader()
 
1338
        name0u = u'0file-\xb6'
 
1339
        name1u = u'1dir-\u062c\u0648'
 
1340
        name2u = u'2file-\u0633'
 
1341
        tree = [
 
1342
            name0u,
 
1343
            name1u + '/',
 
1344
            name1u + '/' + name0u,
 
1345
            name1u + '/' + name1u + '/',
 
1346
            name2u,
 
1347
            ]
 
1348
        self.build_tree(tree)
 
1349
        name0 = name0u.encode('utf8')
 
1350
        name1 = name1u.encode('utf8')
 
1351
        name2 = name2u.encode('utf8')
 
1352
 
 
1353
        # All of the abspaths should be in unicode, all of the relative paths
 
1354
        # should be in utf8
 
1355
        expected_dirblocks = [
 
1356
                (('', '.'),
 
1357
                 [(name0, name0, 'file', './' + name0u),
 
1358
                  (name1, name1, 'directory', './' + name1u),
 
1359
                  (name2, name2, 'file', './' + name2u),
 
1360
                 ]
 
1361
                ),
 
1362
                ((name1, './' + name1u),
 
1363
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
1364
                                                        + '/' + name0u),
 
1365
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
1366
                                                            + '/' + name1u),
 
1367
                 ]
 
1368
                ),
 
1369
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
1370
                 [
 
1371
                 ]
 
1372
                ),
 
1373
            ]
 
1374
        result = list(osutils._walkdirs_utf8('.'))
 
1375
        self._filter_out_stat(result)
 
1376
        self.assertEqual(expected_dirblocks, result)
 
1377
 
 
1378
    def test__walkdirs_utf8_win32readdir(self):
 
1379
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1380
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1381
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1382
        self._save_platform_info()
 
1383
        osutils._selected_dir_reader = Win32ReadDir()
 
1384
        name0u = u'0file-\xb6'
 
1385
        name1u = u'1dir-\u062c\u0648'
 
1386
        name2u = u'2file-\u0633'
 
1387
        tree = [
 
1388
            name0u,
 
1389
            name1u + '/',
 
1390
            name1u + '/' + name0u,
 
1391
            name1u + '/' + name1u + '/',
 
1392
            name2u,
 
1393
            ]
 
1394
        self.build_tree(tree)
 
1395
        name0 = name0u.encode('utf8')
 
1396
        name1 = name1u.encode('utf8')
 
1397
        name2 = name2u.encode('utf8')
 
1398
 
 
1399
        # All of the abspaths should be in unicode, all of the relative paths
 
1400
        # should be in utf8
 
1401
        expected_dirblocks = [
 
1402
                (('', '.'),
 
1403
                 [(name0, name0, 'file', './' + name0u),
 
1404
                  (name1, name1, 'directory', './' + name1u),
 
1405
                  (name2, name2, 'file', './' + name2u),
 
1406
                 ]
 
1407
                ),
 
1408
                ((name1, './' + name1u),
 
1409
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
1410
                                                        + '/' + name0u),
 
1411
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
1412
                                                            + '/' + name1u),
 
1413
                 ]
 
1414
                ),
 
1415
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
1416
                 [
 
1417
                 ]
 
1418
                ),
 
1419
            ]
 
1420
        result = list(osutils._walkdirs_utf8(u'.'))
 
1421
        self._filter_out_stat(result)
 
1422
        self.assertEqual(expected_dirblocks, result)
 
1423
 
 
1424
    def assertStatIsCorrect(self, path, win32stat):
 
1425
        os_stat = os.stat(path)
 
1426
        self.assertEqual(os_stat.st_size, win32stat.st_size)
 
1427
        self.assertAlmostEqual(os_stat.st_mtime, win32stat.st_mtime, places=4)
 
1428
        self.assertAlmostEqual(os_stat.st_ctime, win32stat.st_ctime, places=4)
 
1429
        self.assertAlmostEqual(os_stat.st_atime, win32stat.st_atime, places=4)
 
1430
        self.assertEqual(os_stat.st_dev, win32stat.st_dev)
 
1431
        self.assertEqual(os_stat.st_ino, win32stat.st_ino)
 
1432
        self.assertEqual(os_stat.st_mode, win32stat.st_mode)
 
1433
 
 
1434
    def test__walkdirs_utf_win32_find_file_stat_file(self):
 
1435
        """make sure our Stat values are valid"""
 
1436
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1437
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1438
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1439
        name0u = u'0file-\xb6'
 
1440
        name0 = name0u.encode('utf8')
 
1441
        self.build_tree([name0u])
 
1442
        # I hate to sleep() here, but I'm trying to make the ctime different
 
1443
        # from the mtime
 
1444
        time.sleep(2)
 
1445
        f = open(name0u, 'ab')
 
1446
        try:
 
1447
            f.write('just a small update')
 
1448
        finally:
 
1449
            f.close()
 
1450
 
 
1451
        result = Win32ReadDir().read_dir('', u'.')
 
1452
        entry = result[0]
 
1453
        self.assertEqual((name0, name0, 'file'), entry[:3])
 
1454
        self.assertEqual(u'./' + name0u, entry[4])
 
1455
        self.assertStatIsCorrect(entry[4], entry[3])
 
1456
        self.assertNotEqual(entry[3].st_mtime, entry[3].st_ctime)
 
1457
 
 
1458
    def test__walkdirs_utf_win32_find_file_stat_directory(self):
 
1459
        """make sure our Stat values are valid"""
 
1460
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1461
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1462
        from bzrlib._walkdirs_win32 import Win32ReadDir
 
1463
        name0u = u'0dir-\u062c\u0648'
 
1464
        name0 = name0u.encode('utf8')
 
1465
        self.build_tree([name0u + '/'])
 
1466
 
 
1467
        result = Win32ReadDir().read_dir('', u'.')
 
1468
        entry = result[0]
 
1469
        self.assertEqual((name0, name0, 'directory'), entry[:3])
 
1470
        self.assertEqual(u'./' + name0u, entry[4])
 
1471
        self.assertStatIsCorrect(entry[4], entry[3])
 
1472
 
 
1473
    def assertPathCompare(self, path_less, path_greater):
 
1474
        """check that path_less and path_greater compare correctly."""
 
1475
        self.assertEqual(0, osutils.compare_paths_prefix_order(
 
1476
            path_less, path_less))
 
1477
        self.assertEqual(0, osutils.compare_paths_prefix_order(
 
1478
            path_greater, path_greater))
 
1479
        self.assertEqual(-1, osutils.compare_paths_prefix_order(
 
1480
            path_less, path_greater))
 
1481
        self.assertEqual(1, osutils.compare_paths_prefix_order(
 
1482
            path_greater, path_less))
 
1483
 
 
1484
    def test_compare_paths_prefix_order(self):
 
1485
        # root before all else
 
1486
        self.assertPathCompare("/", "/a")
 
1487
        # alpha within a dir
 
1488
        self.assertPathCompare("/a", "/b")
 
1489
        self.assertPathCompare("/b", "/z")
 
1490
        # high dirs before lower.
 
1491
        self.assertPathCompare("/z", "/a/a")
 
1492
        # except if the deeper dir should be output first
 
1493
        self.assertPathCompare("/a/b/c", "/d/g")
 
1494
        # lexical betwen dirs of the same height
 
1495
        self.assertPathCompare("/a/z", "/z/z")
 
1496
        self.assertPathCompare("/a/c/z", "/a/d/e")
 
1497
 
 
1498
        # this should also be consistent for no leading / paths
 
1499
        # root before all else
 
1500
        self.assertPathCompare("", "a")
 
1501
        # alpha within a dir
 
1502
        self.assertPathCompare("a", "b")
 
1503
        self.assertPathCompare("b", "z")
 
1504
        # high dirs before lower.
 
1505
        self.assertPathCompare("z", "a/a")
 
1506
        # except if the deeper dir should be output first
 
1507
        self.assertPathCompare("a/b/c", "d/g")
 
1508
        # lexical betwen dirs of the same height
 
1509
        self.assertPathCompare("a/z", "z/z")
 
1510
        self.assertPathCompare("a/c/z", "a/d/e")
 
1511
 
 
1512
    def test_path_prefix_sorting(self):
 
1513
        """Doing a sort on path prefix should match our sample data."""
 
1514
        original_paths = [
 
1515
            'a',
 
1516
            'a/b',
 
1517
            'a/b/c',
 
1518
            'b',
 
1519
            'b/c',
 
1520
            'd',
 
1521
            'd/e',
 
1522
            'd/e/f',
 
1523
            'd/f',
 
1524
            'd/g',
 
1525
            'g',
 
1526
            ]
 
1527
 
 
1528
        dir_sorted_paths = [
 
1529
            'a',
 
1530
            'b',
 
1531
            'd',
 
1532
            'g',
 
1533
            'a/b',
 
1534
            'a/b/c',
 
1535
            'b/c',
 
1536
            'd/e',
 
1537
            'd/f',
 
1538
            'd/g',
 
1539
            'd/e/f',
 
1540
            ]
 
1541
 
 
1542
        self.assertEqual(
 
1543
            dir_sorted_paths,
 
1544
            sorted(original_paths, key=osutils.path_prefix_key))
 
1545
        # using the comparison routine shoudl work too:
 
1546
        self.assertEqual(
 
1547
            dir_sorted_paths,
 
1548
            sorted(original_paths, cmp=osutils.compare_paths_prefix_order))
 
1549
 
 
1550
 
 
1551
class TestCopyTree(tests.TestCaseInTempDir):
 
1552
 
 
1553
    def test_copy_basic_tree(self):
 
1554
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
 
1555
        osutils.copy_tree('source', 'target')
 
1556
        self.assertEqual(['a', 'b'], sorted(os.listdir('target')))
 
1557
        self.assertEqual(['c'], os.listdir('target/b'))
 
1558
 
 
1559
    def test_copy_tree_target_exists(self):
 
1560
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c',
 
1561
                         'target/'])
 
1562
        osutils.copy_tree('source', 'target')
 
1563
        self.assertEqual(['a', 'b'], sorted(os.listdir('target')))
 
1564
        self.assertEqual(['c'], os.listdir('target/b'))
 
1565
 
 
1566
    def test_copy_tree_symlinks(self):
 
1567
        self.requireFeature(tests.SymlinkFeature)
 
1568
        self.build_tree(['source/'])
 
1569
        os.symlink('a/generic/path', 'source/lnk')
 
1570
        osutils.copy_tree('source', 'target')
 
1571
        self.assertEqual(['lnk'], os.listdir('target'))
 
1572
        self.assertEqual('a/generic/path', os.readlink('target/lnk'))
 
1573
 
 
1574
    def test_copy_tree_handlers(self):
 
1575
        processed_files = []
 
1576
        processed_links = []
 
1577
        def file_handler(from_path, to_path):
 
1578
            processed_files.append(('f', from_path, to_path))
 
1579
        def dir_handler(from_path, to_path):
 
1580
            processed_files.append(('d', from_path, to_path))
 
1581
        def link_handler(from_path, to_path):
 
1582
            processed_links.append((from_path, to_path))
 
1583
        handlers = {'file':file_handler,
 
1584
                    'directory':dir_handler,
 
1585
                    'symlink':link_handler,
 
1586
                   }
 
1587
 
 
1588
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
 
1589
        if osutils.has_symlinks():
 
1590
            os.symlink('a/generic/path', 'source/lnk')
 
1591
        osutils.copy_tree('source', 'target', handlers=handlers)
 
1592
 
 
1593
        self.assertEqual([('d', 'source', 'target'),
 
1594
                          ('f', 'source/a', 'target/a'),
 
1595
                          ('d', 'source/b', 'target/b'),
 
1596
                          ('f', 'source/b/c', 'target/b/c'),
 
1597
                         ], processed_files)
 
1598
        self.failIfExists('target')
 
1599
        if osutils.has_symlinks():
 
1600
            self.assertEqual([('source/lnk', 'target/lnk')], processed_links)
 
1601
 
 
1602
 
 
1603
class TestSetUnsetEnv(tests.TestCase):
 
1604
    """Test updating the environment"""
 
1605
 
 
1606
    def setUp(self):
 
1607
        super(TestSetUnsetEnv, self).setUp()
 
1608
 
 
1609
        self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'),
 
1610
                         'Environment was not cleaned up properly.'
 
1611
                         ' Variable BZR_TEST_ENV_VAR should not exist.')
 
1612
        def cleanup():
 
1613
            if 'BZR_TEST_ENV_VAR' in os.environ:
 
1614
                del os.environ['BZR_TEST_ENV_VAR']
 
1615
        self.addCleanup(cleanup)
 
1616
 
 
1617
    def test_set(self):
 
1618
        """Test that we can set an env variable"""
 
1619
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
 
1620
        self.assertEqual(None, old)
 
1621
        self.assertEqual('foo', os.environ.get('BZR_TEST_ENV_VAR'))
 
1622
 
 
1623
    def test_double_set(self):
 
1624
        """Test that we get the old value out"""
 
1625
        osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
 
1626
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'bar')
 
1627
        self.assertEqual('foo', old)
 
1628
        self.assertEqual('bar', os.environ.get('BZR_TEST_ENV_VAR'))
 
1629
 
 
1630
    def test_unicode(self):
 
1631
        """Environment can only contain plain strings
 
1632
 
 
1633
        So Unicode strings must be encoded.
 
1634
        """
 
1635
        uni_val, env_val = tests.probe_unicode_in_user_encoding()
 
1636
        if uni_val is None:
 
1637
            raise tests.TestSkipped(
 
1638
                'Cannot find a unicode character that works in encoding %s'
 
1639
                % (osutils.get_user_encoding(),))
 
1640
 
 
1641
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', uni_val)
 
1642
        self.assertEqual(env_val, os.environ.get('BZR_TEST_ENV_VAR'))
 
1643
 
 
1644
    def test_unset(self):
 
1645
        """Test that passing None will remove the env var"""
 
1646
        osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
 
1647
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', None)
 
1648
        self.assertEqual('foo', old)
 
1649
        self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'))
 
1650
        self.failIf('BZR_TEST_ENV_VAR' in os.environ)
 
1651
 
 
1652
 
 
1653
class TestSizeShaFile(tests.TestCaseInTempDir):
 
1654
 
 
1655
    def test_sha_empty(self):
 
1656
        self.build_tree_contents([('foo', '')])
 
1657
        expected_sha = osutils.sha_string('')
 
1658
        f = open('foo')
 
1659
        self.addCleanup(f.close)
 
1660
        size, sha = osutils.size_sha_file(f)
 
1661
        self.assertEqual(0, size)
 
1662
        self.assertEqual(expected_sha, sha)
 
1663
 
 
1664
    def test_sha_mixed_endings(self):
 
1665
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
 
1666
        self.build_tree_contents([('foo', text)])
 
1667
        expected_sha = osutils.sha_string(text)
 
1668
        f = open('foo', 'rb')
 
1669
        self.addCleanup(f.close)
 
1670
        size, sha = osutils.size_sha_file(f)
 
1671
        self.assertEqual(38, size)
 
1672
        self.assertEqual(expected_sha, sha)
 
1673
 
 
1674
 
 
1675
class TestShaFileByName(tests.TestCaseInTempDir):
 
1676
 
 
1677
    def test_sha_empty(self):
 
1678
        self.build_tree_contents([('foo', '')])
 
1679
        expected_sha = osutils.sha_string('')
 
1680
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
 
1681
 
 
1682
    def test_sha_mixed_endings(self):
 
1683
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
 
1684
        self.build_tree_contents([('foo', text)])
 
1685
        expected_sha = osutils.sha_string(text)
 
1686
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
 
1687
 
 
1688
 
 
1689
class TestResourceLoading(tests.TestCaseInTempDir):
 
1690
 
 
1691
    def test_resource_string(self):
 
1692
        # test resource in bzrlib
 
1693
        text = osutils.resource_string('bzrlib', 'debug.py')
 
1694
        self.assertContainsRe(text, "debug_flags = set()")
 
1695
        # test resource under bzrlib
 
1696
        text = osutils.resource_string('bzrlib.ui', 'text.py')
 
1697
        self.assertContainsRe(text, "class TextUIFactory")
 
1698
        # test unsupported package
 
1699
        self.assertRaises(errors.BzrError, osutils.resource_string, 'zzzz',
 
1700
            'yyy.xx')
 
1701
        # test unknown resource
 
1702
        self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')
 
1703
 
 
1704
 
 
1705
class TestReCompile(tests.TestCase):
 
1706
 
 
1707
    def _deprecated_re_compile_checked(self, *args, **kwargs):
 
1708
        return self.applyDeprecated(symbol_versioning.deprecated_in((2, 2, 0)),
 
1709
            osutils.re_compile_checked, *args, **kwargs)
 
1710
 
 
1711
    def test_re_compile_checked(self):
 
1712
        r = self._deprecated_re_compile_checked(r'A*', re.IGNORECASE)
 
1713
        self.assertTrue(r.match('aaaa'))
 
1714
        self.assertTrue(r.match('aAaA'))
 
1715
 
 
1716
    def test_re_compile_checked_error(self):
 
1717
        # like https://bugs.launchpad.net/bzr/+bug/251352
 
1718
 
 
1719
        # Due to possible test isolation error, re.compile is not lazy at
 
1720
        # this point. We re-install lazy compile.
 
1721
        lazy_regex.install_lazy_compile()
 
1722
        err = self.assertRaises(
 
1723
            errors.BzrCommandError,
 
1724
            self._deprecated_re_compile_checked, '*', re.IGNORECASE, 'test case')
 
1725
        self.assertEqual(
 
1726
            'Invalid regular expression in test case: '
 
1727
            '"*" nothing to repeat',
 
1728
            str(err))
 
1729
 
 
1730
 
 
1731
class TestDirReader(tests.TestCaseInTempDir):
 
1732
 
 
1733
    scenarios = dir_reader_scenarios()
 
1734
 
 
1735
    # Set by load_tests
 
1736
    _dir_reader_class = None
 
1737
    _native_to_unicode = None
 
1738
 
 
1739
    def setUp(self):
 
1740
        tests.TestCaseInTempDir.setUp(self)
 
1741
        self.overrideAttr(osutils,
 
1742
                          '_selected_dir_reader', self._dir_reader_class())
 
1743
 
 
1744
    def _get_ascii_tree(self):
 
1745
        tree = [
 
1746
            '0file',
 
1747
            '1dir/',
 
1748
            '1dir/0file',
 
1749
            '1dir/1dir/',
 
1750
            '2file'
 
1751
            ]
 
1752
        expected_dirblocks = [
 
1753
                (('', '.'),
 
1754
                 [('0file', '0file', 'file'),
 
1755
                  ('1dir', '1dir', 'directory'),
 
1756
                  ('2file', '2file', 'file'),
 
1757
                 ]
 
1758
                ),
 
1759
                (('1dir', './1dir'),
 
1760
                 [('1dir/0file', '0file', 'file'),
 
1761
                  ('1dir/1dir', '1dir', 'directory'),
 
1762
                 ]
 
1763
                ),
 
1764
                (('1dir/1dir', './1dir/1dir'),
 
1765
                 [
 
1766
                 ]
 
1767
                ),
 
1768
            ]
 
1769
        return tree, expected_dirblocks
 
1770
 
 
1771
    def test_walk_cur_dir(self):
 
1772
        tree, expected_dirblocks = self._get_ascii_tree()
 
1773
        self.build_tree(tree)
 
1774
        result = list(osutils._walkdirs_utf8('.'))
 
1775
        # Filter out stat and abspath
 
1776
        self.assertEqual(expected_dirblocks,
 
1777
                         [(dirinfo, [line[0:3] for line in block])
 
1778
                          for dirinfo, block in result])
 
1779
 
 
1780
    def test_walk_sub_dir(self):
 
1781
        tree, expected_dirblocks = self._get_ascii_tree()
 
1782
        self.build_tree(tree)
 
1783
        # you can search a subdir only, with a supplied prefix.
 
1784
        result = list(osutils._walkdirs_utf8('./1dir', '1dir'))
 
1785
        # Filter out stat and abspath
 
1786
        self.assertEqual(expected_dirblocks[1:],
 
1787
                         [(dirinfo, [line[0:3] for line in block])
 
1788
                          for dirinfo, block in result])
 
1789
 
 
1790
    def _get_unicode_tree(self):
 
1791
        name0u = u'0file-\xb6'
 
1792
        name1u = u'1dir-\u062c\u0648'
 
1793
        name2u = u'2file-\u0633'
 
1794
        tree = [
 
1795
            name0u,
 
1796
            name1u + '/',
 
1797
            name1u + '/' + name0u,
 
1798
            name1u + '/' + name1u + '/',
 
1799
            name2u,
 
1800
            ]
 
1801
        name0 = name0u.encode('UTF-8')
 
1802
        name1 = name1u.encode('UTF-8')
 
1803
        name2 = name2u.encode('UTF-8')
 
1804
        expected_dirblocks = [
 
1805
                (('', '.'),
 
1806
                 [(name0, name0, 'file', './' + name0u),
 
1807
                  (name1, name1, 'directory', './' + name1u),
 
1808
                  (name2, name2, 'file', './' + name2u),
 
1809
                 ]
 
1810
                ),
 
1811
                ((name1, './' + name1u),
 
1812
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
1813
                                                        + '/' + name0u),
 
1814
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
1815
                                                            + '/' + name1u),
 
1816
                 ]
 
1817
                ),
 
1818
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
1819
                 [
 
1820
                 ]
 
1821
                ),
 
1822
            ]
 
1823
        return tree, expected_dirblocks
 
1824
 
 
1825
    def _filter_out(self, raw_dirblocks):
 
1826
        """Filter out a walkdirs_utf8 result.
 
1827
 
 
1828
        stat field is removed, all native paths are converted to unicode
 
1829
        """
 
1830
        filtered_dirblocks = []
 
1831
        for dirinfo, block in raw_dirblocks:
 
1832
            dirinfo = (dirinfo[0], self._native_to_unicode(dirinfo[1]))
 
1833
            details = []
 
1834
            for line in block:
 
1835
                details.append(line[0:3] + (self._native_to_unicode(line[4]), ))
 
1836
            filtered_dirblocks.append((dirinfo, details))
 
1837
        return filtered_dirblocks
 
1838
 
 
1839
    def test_walk_unicode_tree(self):
 
1840
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1841
        tree, expected_dirblocks = self._get_unicode_tree()
 
1842
        self.build_tree(tree)
 
1843
        result = list(osutils._walkdirs_utf8('.'))
 
1844
        self.assertEqual(expected_dirblocks, self._filter_out(result))
 
1845
 
 
1846
    def test_symlink(self):
 
1847
        self.requireFeature(tests.SymlinkFeature)
 
1848
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1849
        target = u'target\N{Euro Sign}'
 
1850
        link_name = u'l\N{Euro Sign}nk'
 
1851
        os.symlink(target, link_name)
 
1852
        target_utf8 = target.encode('UTF-8')
 
1853
        link_name_utf8 = link_name.encode('UTF-8')
 
1854
        expected_dirblocks = [
 
1855
                (('', '.'),
 
1856
                 [(link_name_utf8, link_name_utf8,
 
1857
                   'symlink', './' + link_name),],
 
1858
                 )]
 
1859
        result = list(osutils._walkdirs_utf8('.'))
 
1860
        self.assertEqual(expected_dirblocks, self._filter_out(result))
 
1861
 
 
1862
 
 
1863
class TestReadLink(tests.TestCaseInTempDir):
 
1864
    """Exposes os.readlink() problems and the osutils solution.
 
1865
 
 
1866
    The only guarantee offered by os.readlink(), starting with 2.6, is that a
 
1867
    unicode string will be returned if a unicode string is passed.
 
1868
 
 
1869
    But prior python versions failed to properly encode the passed unicode
 
1870
    string.
 
1871
    """
 
1872
    _test_needs_features = [tests.SymlinkFeature, tests.UnicodeFilenameFeature]
 
1873
 
 
1874
    def setUp(self):
 
1875
        super(tests.TestCaseInTempDir, self).setUp()
 
1876
        self.link = u'l\N{Euro Sign}ink'
 
1877
        self.target = u'targe\N{Euro Sign}t'
 
1878
        os.symlink(self.target, self.link)
 
1879
 
 
1880
    def test_os_readlink_link_encoding(self):
 
1881
        if sys.version_info < (2, 6):
 
1882
            self.assertRaises(UnicodeEncodeError, os.readlink, self.link)
 
1883
        else:
 
1884
            self.assertEquals(self.target,  os.readlink(self.link))
 
1885
 
 
1886
    def test_os_readlink_link_decoding(self):
 
1887
        self.assertEquals(self.target.encode(osutils._fs_enc),
 
1888
                          os.readlink(self.link.encode(osutils._fs_enc)))
 
1889
 
 
1890
 
 
1891
class TestConcurrency(tests.TestCase):
 
1892
 
 
1893
    def setUp(self):
 
1894
        super(TestConcurrency, self).setUp()
 
1895
        self.overrideAttr(osutils, '_cached_local_concurrency')
 
1896
 
 
1897
    def test_local_concurrency(self):
 
1898
        concurrency = osutils.local_concurrency()
 
1899
        self.assertIsInstance(concurrency, int)
 
1900
 
 
1901
    def test_local_concurrency_environment_variable(self):
 
1902
        self.overrideEnv('BZR_CONCURRENCY', '2')
 
1903
        self.assertEqual(2, osutils.local_concurrency(use_cache=False))
 
1904
        self.overrideEnv('BZR_CONCURRENCY', '3')
 
1905
        self.assertEqual(3, osutils.local_concurrency(use_cache=False))
 
1906
        self.overrideEnv('BZR_CONCURRENCY', 'foo')
 
1907
        self.assertEqual(1, osutils.local_concurrency(use_cache=False))
 
1908
 
 
1909
    def test_option_concurrency(self):
 
1910
        self.overrideEnv('BZR_CONCURRENCY', '1')
 
1911
        self.run_bzr('rocks --concurrency 42')
 
1912
        # Command line overrides environment variable
 
1913
        self.assertEquals('42', os.environ['BZR_CONCURRENCY'])
 
1914
        self.assertEquals(42, osutils.local_concurrency(use_cache=False))
 
1915
 
 
1916
 
 
1917
class TestFailedToLoadExtension(tests.TestCase):
 
1918
 
 
1919
    def _try_loading(self):
 
1920
        try:
 
1921
            import bzrlib._fictional_extension_py
 
1922
        except ImportError, e:
 
1923
            osutils.failed_to_load_extension(e)
 
1924
            return True
 
1925
 
 
1926
    def setUp(self):
 
1927
        super(TestFailedToLoadExtension, self).setUp()
 
1928
        self.overrideAttr(osutils, '_extension_load_failures', [])
 
1929
 
 
1930
    def test_failure_to_load(self):
 
1931
        self._try_loading()
 
1932
        self.assertLength(1, osutils._extension_load_failures)
 
1933
        self.assertEquals(osutils._extension_load_failures[0],
 
1934
            "No module named _fictional_extension_py")
 
1935
 
 
1936
    def test_report_extension_load_failures_no_warning(self):
 
1937
        self.assertTrue(self._try_loading())
 
1938
        warnings, result = self.callCatchWarnings(osutils.report_extension_load_failures)
 
1939
        # it used to give a Python warning; it no longer does
 
1940
        self.assertLength(0, warnings)
 
1941
 
 
1942
    def test_report_extension_load_failures_message(self):
 
1943
        log = StringIO()
 
1944
        trace.push_log_file(log)
 
1945
        self.assertTrue(self._try_loading())
 
1946
        osutils.report_extension_load_failures()
 
1947
        self.assertContainsRe(
 
1948
            log.getvalue(),
 
1949
            r"bzr: warning: some compiled extensions could not be loaded; "
 
1950
            "see <https://answers\.launchpad\.net/bzr/\+faq/703>\n"
 
1951
            )
 
1952
 
 
1953
 
 
1954
class TestTerminalWidth(tests.TestCase):
 
1955
 
 
1956
    def setUp(self):
 
1957
        tests.TestCase.setUp(self)
 
1958
        self._orig_terminal_size_state = osutils._terminal_size_state
 
1959
        self._orig_first_terminal_size = osutils._first_terminal_size
 
1960
        self.addCleanup(self.restore_osutils_globals)
 
1961
        osutils._terminal_size_state = 'no_data'
 
1962
        osutils._first_terminal_size = None
 
1963
 
 
1964
    def restore_osutils_globals(self):
 
1965
        osutils._terminal_size_state = self._orig_terminal_size_state
 
1966
        osutils._first_terminal_size = self._orig_first_terminal_size
 
1967
 
 
1968
    def replace_stdout(self, new):
 
1969
        self.overrideAttr(sys, 'stdout', new)
 
1970
 
 
1971
    def replace__terminal_size(self, new):
 
1972
        self.overrideAttr(osutils, '_terminal_size', new)
 
1973
 
 
1974
    def set_fake_tty(self):
 
1975
 
 
1976
        class I_am_a_tty(object):
 
1977
            def isatty(self):
 
1978
                return True
 
1979
 
 
1980
        self.replace_stdout(I_am_a_tty())
 
1981
 
 
1982
    def test_default_values(self):
 
1983
        self.assertEqual(80, osutils.default_terminal_width)
 
1984
 
 
1985
    def test_defaults_to_BZR_COLUMNS(self):
 
1986
        # BZR_COLUMNS is set by the test framework
 
1987
        self.assertNotEqual('12', os.environ['BZR_COLUMNS'])
 
1988
        self.overrideEnv('BZR_COLUMNS', '12')
 
1989
        self.assertEqual(12, osutils.terminal_width())
 
1990
 
 
1991
    def test_BZR_COLUMNS_0_no_limit(self):
 
1992
        self.overrideEnv('BZR_COLUMNS', '0')
 
1993
        self.assertEqual(None, osutils.terminal_width())
 
1994
 
 
1995
    def test_falls_back_to_COLUMNS(self):
 
1996
        self.overrideEnv('BZR_COLUMNS', None)
 
1997
        self.assertNotEqual('42', os.environ['COLUMNS'])
 
1998
        self.set_fake_tty()
 
1999
        self.overrideEnv('COLUMNS', '42')
 
2000
        self.assertEqual(42, osutils.terminal_width())
 
2001
 
 
2002
    def test_tty_default_without_columns(self):
 
2003
        self.overrideEnv('BZR_COLUMNS', None)
 
2004
        self.overrideEnv('COLUMNS', None)
 
2005
 
 
2006
        def terminal_size(w, h):
 
2007
            return 42, 42
 
2008
 
 
2009
        self.set_fake_tty()
 
2010
        # We need to override the osutils definition as it depends on the
 
2011
        # running environment that we can't control (PQM running without a
 
2012
        # controlling terminal is one example).
 
2013
        self.replace__terminal_size(terminal_size)
 
2014
        self.assertEqual(42, osutils.terminal_width())
 
2015
 
 
2016
    def test_non_tty_default_without_columns(self):
 
2017
        self.overrideEnv('BZR_COLUMNS', None)
 
2018
        self.overrideEnv('COLUMNS', None)
 
2019
        self.replace_stdout(None)
 
2020
        self.assertEqual(None, osutils.terminal_width())
 
2021
 
 
2022
    def test_no_TIOCGWINSZ(self):
 
2023
        self.requireFeature(term_ios_feature)
 
2024
        termios = term_ios_feature.module
 
2025
        # bug 63539 is about a termios without TIOCGWINSZ attribute
 
2026
        try:
 
2027
            orig = termios.TIOCGWINSZ
 
2028
        except AttributeError:
 
2029
            # We won't remove TIOCGWINSZ, because it doesn't exist anyway :)
 
2030
            pass
 
2031
        else:
 
2032
            self.overrideAttr(termios, 'TIOCGWINSZ')
 
2033
            del termios.TIOCGWINSZ
 
2034
        self.overrideEnv('BZR_COLUMNS', None)
 
2035
        self.overrideEnv('COLUMNS', None)
 
2036
        # Whatever the result is, if we don't raise an exception, it's ok.
 
2037
        osutils.terminal_width()
 
2038
 
 
2039
class TestCreationOps(tests.TestCaseInTempDir):
 
2040
    _test_needs_features = [features.chown_feature]
 
2041
 
 
2042
    def setUp(self):
 
2043
        tests.TestCaseInTempDir.setUp(self)
 
2044
        self.overrideAttr(os, 'chown', self._dummy_chown)
 
2045
 
 
2046
        # params set by call to _dummy_chown
 
2047
        self.path = self.uid = self.gid = None
 
2048
 
 
2049
    def _dummy_chown(self, path, uid, gid):
 
2050
        self.path, self.uid, self.gid = path, uid, gid
 
2051
 
 
2052
    def test_copy_ownership_from_path(self):
 
2053
        """copy_ownership_from_path test with specified src."""
 
2054
        ownsrc = '/'
 
2055
        f = open('test_file', 'wt')
 
2056
        osutils.copy_ownership_from_path('test_file', ownsrc)
 
2057
 
 
2058
        s = os.stat(ownsrc)
 
2059
        self.assertEquals(self.path, 'test_file')
 
2060
        self.assertEquals(self.uid, s.st_uid)
 
2061
        self.assertEquals(self.gid, s.st_gid)
 
2062
 
 
2063
    def test_copy_ownership_nonesrc(self):
 
2064
        """copy_ownership_from_path test with src=None."""
 
2065
        f = open('test_file', 'wt')
 
2066
        # should use parent dir for permissions
 
2067
        osutils.copy_ownership_from_path('test_file')
 
2068
 
 
2069
        s = os.stat('..')
 
2070
        self.assertEquals(self.path, 'test_file')
 
2071
        self.assertEquals(self.uid, s.st_uid)
 
2072
        self.assertEquals(self.gid, s.st_gid)
 
2073
 
 
2074
class TestGetuserUnicode(tests.TestCase):
 
2075
 
 
2076
    def test_ascii_user(self):
 
2077
        self.overrideEnv('LOGNAME', 'jrandom')
 
2078
        self.assertEqual(u'jrandom', osutils.getuser_unicode())
 
2079
 
 
2080
    def test_unicode_user(self):
 
2081
        ue = osutils.get_user_encoding()
 
2082
        uni_val, env_val = tests.probe_unicode_in_user_encoding()
 
2083
        if uni_val is None:
 
2084
            raise tests.TestSkipped(
 
2085
                'Cannot find a unicode character that works in encoding %s'
 
2086
                % (osutils.get_user_encoding(),))
 
2087
        uni_username = u'jrandom' + uni_val
 
2088
        encoded_username = uni_username.encode(ue)
 
2089
        self.overrideEnv('LOGNAME', encoded_username)
 
2090
        self.assertEqual(uni_username, osutils.getuser_unicode())
 
2091
        self.overrideEnv('LOGNAME', u'jrandom\xb6'.encode(ue))
 
2092
        self.assertEqual(u'jrandom\xb6', osutils.getuser_unicode())
 
2093
 
 
2094
class TestBackupNames(tests.TestCase):
 
2095
 
 
2096
    def setUp(self):
 
2097
        super(TestBackupNames, self).setUp()
 
2098
        self.backups = []
 
2099
 
 
2100
    def backup_exists(self, name):
 
2101
        return name in self.backups
 
2102
 
 
2103
    def available_backup_name(self, name):
 
2104
        backup_name = osutils.available_backup_name(name, self.backup_exists)
 
2105
        self.backups.append(backup_name)
 
2106
        return backup_name
 
2107
 
 
2108
    def assertBackupName(self, expected, name):
 
2109
        self.assertEqual(expected, self.available_backup_name(name))
 
2110
 
 
2111
    def test_empty(self):
 
2112
        self.assertBackupName('file.~1~', 'file')
 
2113
 
 
2114
    def test_existing(self):
 
2115
        self.available_backup_name('file')
 
2116
        self.available_backup_name('file')
 
2117
        self.assertBackupName('file.~3~', 'file')
 
2118
        # Empty slots are found, this is not a strict requirement and may be
 
2119
        # revisited if we test against all implementations.
 
2120
        self.backups.remove('file.~2~')
 
2121
        self.assertBackupName('file.~2~', 'file')