/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: Robert Collins
  • Date: 2010-06-21 01:30:45 UTC
  • mfrom: (5309 +trunk)
  • mto: This revision was merged to the branch mainline in revision 5310.
  • Revision ID: robertc@robertcollins.net-20100621013045-s59nfjps3rkcn53j
Merge trunk to fix NEWS sections.

Show diffs side-by-side

added added

removed removed

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