~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: John Arbash Meinel
  • Author(s): Mark Hammond
  • Date: 2008-09-09 17:02:21 UTC
  • mto: This revision was merged to the branch mainline in revision 3697.
  • Revision ID: john@arbash-meinel.com-20080909170221-svim3jw2mrz0amp3
An updated transparent icon for bzr.

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2005, 2006, 2007 Canonical Ltd
 
2
#
 
3
# This program is free software; you can redistribute it and/or modify
 
4
# it under the terms of the GNU General Public License as published by
 
5
# the Free Software Foundation; either version 2 of the License, or
 
6
# (at your option) any later version.
 
7
#
 
8
# This program is distributed in the hope that it will be useful,
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
11
# GNU General Public License for more details.
 
12
#
 
13
# You should have received a copy of the GNU General Public License
 
14
# along with this program; if not, write to the Free Software
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
16
 
 
17
"""Tests for the osutils wrapper."""
 
18
 
 
19
from cStringIO import StringIO
 
20
import errno
 
21
import os
 
22
import socket
 
23
import stat
 
24
import sys
 
25
import time
 
26
 
 
27
import bzrlib
 
28
from bzrlib import (
 
29
    errors,
 
30
    osutils,
 
31
    tests,
 
32
    win32utils,
 
33
    )
 
34
from bzrlib.errors import BzrBadParameterNotUnicode, InvalidURL
 
35
from bzrlib.osutils import (
 
36
        is_inside_any,
 
37
        is_inside_or_parent_of_any,
 
38
        pathjoin,
 
39
        pumpfile,
 
40
        pump_string_file,
 
41
        )
 
42
from bzrlib.tests import (
 
43
        adapt_tests,
 
44
        Feature,
 
45
        probe_unicode_in_user_encoding,
 
46
        split_suite_by_re,
 
47
        StringIOWrapper,
 
48
        SymlinkFeature,
 
49
        TestCase,
 
50
        TestCaseInTempDir,
 
51
        TestScenarioApplier,
 
52
        TestSkipped,
 
53
        )
 
54
from bzrlib.tests.file_utils import (
 
55
    FakeReadFile,
 
56
    )
 
57
from bzrlib.tests.test__walkdirs_win32 import WalkdirsWin32Feature
 
58
 
 
59
 
 
60
def load_tests(standard_tests, module, loader):
 
61
    """Parameterize readdir tests."""
 
62
    to_adapt, result = split_suite_by_re(standard_tests, "readdir")
 
63
    adapter = TestScenarioApplier()
 
64
    from bzrlib import _readdir_py
 
65
    adapter.scenarios = [('python', {'read_dir': _readdir_py.read_dir})]
 
66
    if ReadDirFeature.available():
 
67
        adapter.scenarios.append(('pyrex',
 
68
            {'read_dir': ReadDirFeature.read_dir}))
 
69
    adapt_tests(to_adapt, adapter, result)
 
70
    return result
 
71
 
 
72
 
 
73
class _ReadDirFeature(Feature):
 
74
 
 
75
    def _probe(self):
 
76
        try:
 
77
            from bzrlib import _readdir_pyx
 
78
            self.read_dir = _readdir_pyx.read_dir
 
79
            return True
 
80
        except ImportError:
 
81
            return False
 
82
 
 
83
    def feature_name(self):
 
84
        return 'bzrlib._readdir_pyx'
 
85
 
 
86
ReadDirFeature = _ReadDirFeature()
 
87
 
 
88
 
 
89
class TestOSUtils(TestCaseInTempDir):
 
90
 
 
91
    def test_contains_whitespace(self):
 
92
        self.failUnless(osutils.contains_whitespace(u' '))
 
93
        self.failUnless(osutils.contains_whitespace(u'hello there'))
 
94
        self.failUnless(osutils.contains_whitespace(u'hellothere\n'))
 
95
        self.failUnless(osutils.contains_whitespace(u'hello\nthere'))
 
96
        self.failUnless(osutils.contains_whitespace(u'hello\rthere'))
 
97
        self.failUnless(osutils.contains_whitespace(u'hello\tthere'))
 
98
 
 
99
        # \xa0 is "Non-breaking-space" which on some python locales thinks it
 
100
        # is whitespace, but we do not.
 
101
        self.failIf(osutils.contains_whitespace(u''))
 
102
        self.failIf(osutils.contains_whitespace(u'hellothere'))
 
103
        self.failIf(osutils.contains_whitespace(u'hello\xa0there'))
 
104
 
 
105
    def test_fancy_rename(self):
 
106
        # This should work everywhere
 
107
        def rename(a, b):
 
108
            osutils.fancy_rename(a, b,
 
109
                    rename_func=os.rename,
 
110
                    unlink_func=os.unlink)
 
111
 
 
112
        open('a', 'wb').write('something in a\n')
 
113
        rename('a', 'b')
 
114
        self.failIfExists('a')
 
115
        self.failUnlessExists('b')
 
116
        self.check_file_contents('b', 'something in a\n')
 
117
 
 
118
        open('a', 'wb').write('new something in a\n')
 
119
        rename('b', 'a')
 
120
 
 
121
        self.check_file_contents('a', 'something in a\n')
 
122
 
 
123
    def test_rename(self):
 
124
        # Rename should be semi-atomic on all platforms
 
125
        open('a', 'wb').write('something in a\n')
 
126
        osutils.rename('a', 'b')
 
127
        self.failIfExists('a')
 
128
        self.failUnlessExists('b')
 
129
        self.check_file_contents('b', 'something in a\n')
 
130
 
 
131
        open('a', 'wb').write('new something in a\n')
 
132
        osutils.rename('b', 'a')
 
133
 
 
134
        self.check_file_contents('a', 'something in a\n')
 
135
 
 
136
    # TODO: test fancy_rename using a MemoryTransport
 
137
 
 
138
    def test_rename_change_case(self):
 
139
        # on Windows we should be able to change filename case by rename
 
140
        self.build_tree(['a', 'b/'])
 
141
        osutils.rename('a', 'A')
 
142
        osutils.rename('b', 'B')
 
143
        # we can't use failUnlessExists on case-insensitive filesystem
 
144
        # so try to check shape of the tree
 
145
        shape = sorted(os.listdir('.'))
 
146
        self.assertEquals(['A', 'B'], shape)
 
147
 
 
148
    def test_01_rand_chars_empty(self):
 
149
        result = osutils.rand_chars(0)
 
150
        self.assertEqual(result, '')
 
151
 
 
152
    def test_02_rand_chars_100(self):
 
153
        result = osutils.rand_chars(100)
 
154
        self.assertEqual(len(result), 100)
 
155
        self.assertEqual(type(result), str)
 
156
        self.assertContainsRe(result, r'^[a-z0-9]{100}$')
 
157
 
 
158
    def test_is_inside(self):
 
159
        is_inside = osutils.is_inside
 
160
        self.assertTrue(is_inside('src', 'src/foo.c'))
 
161
        self.assertFalse(is_inside('src', 'srccontrol'))
 
162
        self.assertTrue(is_inside('src', 'src/a/a/a/foo.c'))
 
163
        self.assertTrue(is_inside('foo.c', 'foo.c'))
 
164
        self.assertFalse(is_inside('foo.c', ''))
 
165
        self.assertTrue(is_inside('', 'foo.c'))
 
166
 
 
167
    def test_is_inside_any(self):
 
168
        SRC_FOO_C = pathjoin('src', 'foo.c')
 
169
        for dirs, fn in [(['src', 'doc'], SRC_FOO_C),
 
170
                         (['src'], SRC_FOO_C),
 
171
                         (['src'], 'src'),
 
172
                         ]:
 
173
            self.assert_(is_inside_any(dirs, fn))
 
174
        for dirs, fn in [(['src'], 'srccontrol'),
 
175
                         (['src'], 'srccontrol/foo')]:
 
176
            self.assertFalse(is_inside_any(dirs, fn))
 
177
 
 
178
    def test_is_inside_or_parent_of_any(self):
 
179
        for dirs, fn in [(['src', 'doc'], 'src/foo.c'),
 
180
                         (['src'], 'src/foo.c'),
 
181
                         (['src/bar.c'], 'src'),
 
182
                         (['src/bar.c', 'bla/foo.c'], 'src'),
 
183
                         (['src'], 'src'),
 
184
                         ]:
 
185
            self.assert_(is_inside_or_parent_of_any(dirs, fn))
 
186
            
 
187
        for dirs, fn in [(['src'], 'srccontrol'),
 
188
                         (['srccontrol/foo.c'], 'src'),
 
189
                         (['src'], 'srccontrol/foo')]:
 
190
            self.assertFalse(is_inside_or_parent_of_any(dirs, fn))
 
191
 
 
192
    def test_rmtree(self):
 
193
        # Check to remove tree with read-only files/dirs
 
194
        os.mkdir('dir')
 
195
        f = file('dir/file', 'w')
 
196
        f.write('spam')
 
197
        f.close()
 
198
        # would like to also try making the directory readonly, but at the
 
199
        # moment python shutil.rmtree doesn't handle that properly - it would
 
200
        # need to chmod the directory before removing things inside it - deferred
 
201
        # for now -- mbp 20060505
 
202
        # osutils.make_readonly('dir')
 
203
        osutils.make_readonly('dir/file')
 
204
 
 
205
        osutils.rmtree('dir')
 
206
 
 
207
        self.failIfExists('dir/file')
 
208
        self.failIfExists('dir')
 
209
 
 
210
    def test_file_kind(self):
 
211
        self.build_tree(['file', 'dir/'])
 
212
        self.assertEquals('file', osutils.file_kind('file'))
 
213
        self.assertEquals('directory', osutils.file_kind('dir/'))
 
214
        if osutils.has_symlinks():
 
215
            os.symlink('symlink', 'symlink')
 
216
            self.assertEquals('symlink', osutils.file_kind('symlink'))
 
217
        
 
218
        # TODO: jam 20060529 Test a block device
 
219
        try:
 
220
            os.lstat('/dev/null')
 
221
        except OSError, e:
 
222
            if e.errno not in (errno.ENOENT,):
 
223
                raise
 
224
        else:
 
225
            self.assertEquals('chardev', osutils.file_kind('/dev/null'))
 
226
 
 
227
        mkfifo = getattr(os, 'mkfifo', None)
 
228
        if mkfifo:
 
229
            mkfifo('fifo')
 
230
            try:
 
231
                self.assertEquals('fifo', osutils.file_kind('fifo'))
 
232
            finally:
 
233
                os.remove('fifo')
 
234
 
 
235
        AF_UNIX = getattr(socket, 'AF_UNIX', None)
 
236
        if AF_UNIX:
 
237
            s = socket.socket(AF_UNIX)
 
238
            s.bind('socket')
 
239
            try:
 
240
                self.assertEquals('socket', osutils.file_kind('socket'))
 
241
            finally:
 
242
                os.remove('socket')
 
243
 
 
244
    def test_kind_marker(self):
 
245
        self.assertEqual(osutils.kind_marker('file'), '')
 
246
        self.assertEqual(osutils.kind_marker('directory'), '/')
 
247
        self.assertEqual(osutils.kind_marker('symlink'), '@')
 
248
        self.assertEqual(osutils.kind_marker('tree-reference'), '+')
 
249
 
 
250
    def test_get_umask(self):
 
251
        if sys.platform == 'win32':
 
252
            # umask always returns '0', no way to set it
 
253
            self.assertEqual(0, osutils.get_umask())
 
254
            return
 
255
 
 
256
        orig_umask = osutils.get_umask()
 
257
        try:
 
258
            os.umask(0222)
 
259
            self.assertEqual(0222, osutils.get_umask())
 
260
            os.umask(0022)
 
261
            self.assertEqual(0022, osutils.get_umask())
 
262
            os.umask(0002)
 
263
            self.assertEqual(0002, osutils.get_umask())
 
264
            os.umask(0027)
 
265
            self.assertEqual(0027, osutils.get_umask())
 
266
        finally:
 
267
            os.umask(orig_umask)
 
268
 
 
269
    def assertFormatedDelta(self, expected, seconds):
 
270
        """Assert osutils.format_delta formats as expected"""
 
271
        actual = osutils.format_delta(seconds)
 
272
        self.assertEqual(expected, actual)
 
273
 
 
274
    def test_format_delta(self):
 
275
        self.assertFormatedDelta('0 seconds ago', 0)
 
276
        self.assertFormatedDelta('1 second ago', 1)
 
277
        self.assertFormatedDelta('10 seconds ago', 10)
 
278
        self.assertFormatedDelta('59 seconds ago', 59)
 
279
        self.assertFormatedDelta('89 seconds ago', 89)
 
280
        self.assertFormatedDelta('1 minute, 30 seconds ago', 90)
 
281
        self.assertFormatedDelta('3 minutes, 0 seconds ago', 180)
 
282
        self.assertFormatedDelta('3 minutes, 1 second ago', 181)
 
283
        self.assertFormatedDelta('10 minutes, 15 seconds ago', 615)
 
284
        self.assertFormatedDelta('30 minutes, 59 seconds ago', 1859)
 
285
        self.assertFormatedDelta('31 minutes, 0 seconds ago', 1860)
 
286
        self.assertFormatedDelta('60 minutes, 0 seconds ago', 3600)
 
287
        self.assertFormatedDelta('89 minutes, 59 seconds ago', 5399)
 
288
        self.assertFormatedDelta('1 hour, 30 minutes ago', 5400)
 
289
        self.assertFormatedDelta('2 hours, 30 minutes ago', 9017)
 
290
        self.assertFormatedDelta('10 hours, 0 minutes ago', 36000)
 
291
        self.assertFormatedDelta('24 hours, 0 minutes ago', 86400)
 
292
        self.assertFormatedDelta('35 hours, 59 minutes ago', 129599)
 
293
        self.assertFormatedDelta('36 hours, 0 minutes ago', 129600)
 
294
        self.assertFormatedDelta('36 hours, 0 minutes ago', 129601)
 
295
        self.assertFormatedDelta('36 hours, 1 minute ago', 129660)
 
296
        self.assertFormatedDelta('36 hours, 1 minute ago', 129661)
 
297
        self.assertFormatedDelta('84 hours, 10 minutes ago', 303002)
 
298
 
 
299
        # We handle when time steps the wrong direction because computers
 
300
        # don't have synchronized clocks.
 
301
        self.assertFormatedDelta('84 hours, 10 minutes in the future', -303002)
 
302
        self.assertFormatedDelta('1 second in the future', -1)
 
303
        self.assertFormatedDelta('2 seconds in the future', -2)
 
304
 
 
305
    def test_format_date(self):
 
306
        self.assertRaises(errors.UnsupportedTimezoneFormat,
 
307
            osutils.format_date, 0, timezone='foo')
 
308
 
 
309
    def test_dereference_path(self):
 
310
        self.requireFeature(SymlinkFeature)
 
311
        cwd = osutils.realpath('.')
 
312
        os.mkdir('bar')
 
313
        bar_path = osutils.pathjoin(cwd, 'bar')
 
314
        # Using './' to avoid bug #1213894 (first path component not
 
315
        # dereferenced) in Python 2.4.1 and earlier
 
316
        self.assertEqual(bar_path, osutils.realpath('./bar'))
 
317
        os.symlink('bar', 'foo')
 
318
        self.assertEqual(bar_path, osutils.realpath('./foo'))
 
319
        
 
320
        # Does not dereference terminal symlinks
 
321
        foo_path = osutils.pathjoin(cwd, 'foo')
 
322
        self.assertEqual(foo_path, osutils.dereference_path('./foo'))
 
323
 
 
324
        # Dereferences parent symlinks
 
325
        os.mkdir('bar/baz')
 
326
        baz_path = osutils.pathjoin(bar_path, 'baz')
 
327
        self.assertEqual(baz_path, osutils.dereference_path('./foo/baz'))
 
328
 
 
329
        # Dereferences parent symlinks that are the first path element
 
330
        self.assertEqual(baz_path, osutils.dereference_path('foo/baz'))
 
331
 
 
332
        # Dereferences parent symlinks in absolute paths
 
333
        foo_baz_path = osutils.pathjoin(foo_path, 'baz')
 
334
        self.assertEqual(baz_path, osutils.dereference_path(foo_baz_path))
 
335
 
 
336
    def test_changing_access(self):
 
337
        f = file('file', 'w')
 
338
        f.write('monkey')
 
339
        f.close()
 
340
 
 
341
        # Make a file readonly
 
342
        osutils.make_readonly('file')
 
343
        mode = os.lstat('file').st_mode
 
344
        self.assertEqual(mode, mode & 0777555)
 
345
 
 
346
        # Make a file writable
 
347
        osutils.make_writable('file')
 
348
        mode = os.lstat('file').st_mode
 
349
        self.assertEqual(mode, mode | 0200)
 
350
 
 
351
        if osutils.has_symlinks():
 
352
            # should not error when handed a symlink
 
353
            os.symlink('nonexistent', 'dangling')
 
354
            osutils.make_readonly('dangling')
 
355
            osutils.make_writable('dangling')
 
356
 
 
357
    def test_kind_marker(self):
 
358
        self.assertEqual("", osutils.kind_marker("file"))
 
359
        self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
 
360
        self.assertEqual("@", osutils.kind_marker("symlink"))
 
361
        self.assertRaises(errors.BzrError, osutils.kind_marker, "unknown")
 
362
 
 
363
    def test_host_os_dereferences_symlinks(self):
 
364
        osutils.host_os_dereferences_symlinks()
 
365
 
 
366
 
 
367
class TestPumpFile(TestCase):
 
368
    """Test pumpfile method."""
 
369
    def setUp(self):
 
370
        # create a test datablock
 
371
        self.block_size = 512
 
372
        pattern = '0123456789ABCDEF'
 
373
        self.test_data = pattern * (3 * self.block_size / len(pattern))
 
374
        self.test_data_len = len(self.test_data)
 
375
 
 
376
    def test_bracket_block_size(self):
 
377
        """Read data in blocks with the requested read size bracketing the
 
378
        block size."""
 
379
        # make sure test data is larger than max read size
 
380
        self.assertTrue(self.test_data_len > self.block_size)
 
381
 
 
382
        from_file = FakeReadFile(self.test_data)
 
383
        to_file = StringIO()
 
384
 
 
385
        # read (max / 2) bytes and verify read size wasn't affected
 
386
        num_bytes_to_read = self.block_size / 2
 
387
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
388
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
 
389
        self.assertEqual(from_file.get_read_count(), 1)
 
390
 
 
391
        # read (max) bytes and verify read size wasn't affected
 
392
        num_bytes_to_read = self.block_size
 
393
        from_file.reset_read_count()
 
394
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
395
        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
 
396
        self.assertEqual(from_file.get_read_count(), 1)
 
397
 
 
398
        # read (max + 1) bytes and verify read size was limited
 
399
        num_bytes_to_read = self.block_size + 1
 
400
        from_file.reset_read_count()
 
401
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
402
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
 
403
        self.assertEqual(from_file.get_read_count(), 2)
 
404
 
 
405
        # finish reading the rest of the data
 
406
        num_bytes_to_read = self.test_data_len - to_file.tell()
 
407
        pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
 
408
 
 
409
        # report error if the data wasn't equal (we only report the size due
 
410
        # to the length of the data)
 
411
        response_data = to_file.getvalue()
 
412
        if response_data != self.test_data:
 
413
            message = "Data not equal.  Expected %d bytes, received %d."
 
414
            self.fail(message % (len(response_data), self.test_data_len))
 
415
 
 
416
    def test_specified_size(self):
 
417
        """Request a transfer larger than the maximum block size and verify
 
418
        that the maximum read doesn't exceed the block_size."""
 
419
        # make sure test data is larger than max read size
 
420
        self.assertTrue(self.test_data_len > self.block_size)
 
421
 
 
422
        # retrieve data in blocks
 
423
        from_file = FakeReadFile(self.test_data)
 
424
        to_file = StringIO()
 
425
        pumpfile(from_file, to_file, self.test_data_len, self.block_size)
 
426
 
 
427
        # verify read size was equal to the maximum read size
 
428
        self.assertTrue(from_file.get_max_read_size() > 0)
 
429
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
 
430
        self.assertEqual(from_file.get_read_count(), 3)
 
431
 
 
432
        # report error if the data wasn't equal (we only report the size due
 
433
        # to the length of the data)
 
434
        response_data = to_file.getvalue()
 
435
        if response_data != self.test_data:
 
436
            message = "Data not equal.  Expected %d bytes, received %d."
 
437
            self.fail(message % (len(response_data), self.test_data_len))
 
438
 
 
439
    def test_to_eof(self):
 
440
        """Read to end-of-file and verify that the reads are not larger than
 
441
        the maximum read size."""
 
442
        # make sure test data is larger than max read size
 
443
        self.assertTrue(self.test_data_len > self.block_size)
 
444
 
 
445
        # retrieve data to EOF
 
446
        from_file = FakeReadFile(self.test_data)
 
447
        to_file = StringIO()
 
448
        pumpfile(from_file, to_file, -1, self.block_size)
 
449
 
 
450
        # verify read size was equal to the maximum read size
 
451
        self.assertEqual(from_file.get_max_read_size(), self.block_size)
 
452
        self.assertEqual(from_file.get_read_count(), 4)
 
453
 
 
454
        # report error if the data wasn't equal (we only report the size due
 
455
        # to the length of the data)
 
456
        response_data = to_file.getvalue()
 
457
        if response_data != self.test_data:
 
458
            message = "Data not equal.  Expected %d bytes, received %d."
 
459
            self.fail(message % (len(response_data), self.test_data_len))
 
460
 
 
461
    def test_defaults(self):
 
462
        """Verifies that the default arguments will read to EOF -- this
 
463
        test verifies that any existing usages of pumpfile will not be broken
 
464
        with this new version."""
 
465
        # retrieve data using default (old) pumpfile method
 
466
        from_file = FakeReadFile(self.test_data)
 
467
        to_file = StringIO()
 
468
        pumpfile(from_file, to_file)
 
469
 
 
470
        # report error if the data wasn't equal (we only report the size due
 
471
        # to the length of the data)
 
472
        response_data = to_file.getvalue()
 
473
        if response_data != self.test_data:
 
474
            message = "Data not equal.  Expected %d bytes, received %d."
 
475
            self.fail(message % (len(response_data), self.test_data_len))
 
476
 
 
477
 
 
478
class TestPumpStringFile(TestCase):
 
479
 
 
480
    def test_empty(self):
 
481
        output = StringIO()
 
482
        pump_string_file("", output)
 
483
        self.assertEqual("", output.getvalue())
 
484
 
 
485
    def test_more_than_segment_size(self):
 
486
        output = StringIO()
 
487
        pump_string_file("123456789", output, 2)
 
488
        self.assertEqual("123456789", output.getvalue())
 
489
 
 
490
    def test_segment_size(self):
 
491
        output = StringIO()
 
492
        pump_string_file("12", output, 2)
 
493
        self.assertEqual("12", output.getvalue())
 
494
 
 
495
    def test_segment_size_multiple(self):
 
496
        output = StringIO()
 
497
        pump_string_file("1234", output, 2)
 
498
        self.assertEqual("1234", output.getvalue())
 
499
 
 
500
 
 
501
class TestSafeUnicode(TestCase):
 
502
 
 
503
    def test_from_ascii_string(self):
 
504
        self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
 
505
 
 
506
    def test_from_unicode_string_ascii_contents(self):
 
507
        self.assertEqual(u'bargam', osutils.safe_unicode(u'bargam'))
 
508
 
 
509
    def test_from_unicode_string_unicode_contents(self):
 
510
        self.assertEqual(u'bargam\xae', osutils.safe_unicode(u'bargam\xae'))
 
511
 
 
512
    def test_from_utf8_string(self):
 
513
        self.assertEqual(u'foo\xae', osutils.safe_unicode('foo\xc2\xae'))
 
514
 
 
515
    def test_bad_utf8_string(self):
 
516
        self.assertRaises(BzrBadParameterNotUnicode,
 
517
                          osutils.safe_unicode,
 
518
                          '\xbb\xbb')
 
519
 
 
520
 
 
521
class TestSafeUtf8(TestCase):
 
522
 
 
523
    def test_from_ascii_string(self):
 
524
        f = 'foobar'
 
525
        self.assertEqual('foobar', osutils.safe_utf8(f))
 
526
 
 
527
    def test_from_unicode_string_ascii_contents(self):
 
528
        self.assertEqual('bargam', osutils.safe_utf8(u'bargam'))
 
529
 
 
530
    def test_from_unicode_string_unicode_contents(self):
 
531
        self.assertEqual('bargam\xc2\xae', osutils.safe_utf8(u'bargam\xae'))
 
532
 
 
533
    def test_from_utf8_string(self):
 
534
        self.assertEqual('foo\xc2\xae', osutils.safe_utf8('foo\xc2\xae'))
 
535
 
 
536
    def test_bad_utf8_string(self):
 
537
        self.assertRaises(BzrBadParameterNotUnicode,
 
538
                          osutils.safe_utf8, '\xbb\xbb')
 
539
 
 
540
 
 
541
class TestSafeRevisionId(TestCase):
 
542
 
 
543
    def test_from_ascii_string(self):
 
544
        # this shouldn't give a warning because it's getting an ascii string
 
545
        self.assertEqual('foobar', osutils.safe_revision_id('foobar'))
 
546
 
 
547
    def test_from_unicode_string_ascii_contents(self):
 
548
        self.assertEqual('bargam',
 
549
                         osutils.safe_revision_id(u'bargam', warn=False))
 
550
 
 
551
    def test_from_unicode_deprecated(self):
 
552
        self.assertEqual('bargam',
 
553
            self.callDeprecated([osutils._revision_id_warning],
 
554
                                osutils.safe_revision_id, u'bargam'))
 
555
 
 
556
    def test_from_unicode_string_unicode_contents(self):
 
557
        self.assertEqual('bargam\xc2\xae',
 
558
                         osutils.safe_revision_id(u'bargam\xae', warn=False))
 
559
 
 
560
    def test_from_utf8_string(self):
 
561
        self.assertEqual('foo\xc2\xae',
 
562
                         osutils.safe_revision_id('foo\xc2\xae'))
 
563
 
 
564
    def test_none(self):
 
565
        """Currently, None is a valid revision_id"""
 
566
        self.assertEqual(None, osutils.safe_revision_id(None))
 
567
 
 
568
 
 
569
class TestSafeFileId(TestCase):
 
570
 
 
571
    def test_from_ascii_string(self):
 
572
        self.assertEqual('foobar', osutils.safe_file_id('foobar'))
 
573
 
 
574
    def test_from_unicode_string_ascii_contents(self):
 
575
        self.assertEqual('bargam', osutils.safe_file_id(u'bargam', warn=False))
 
576
 
 
577
    def test_from_unicode_deprecated(self):
 
578
        self.assertEqual('bargam',
 
579
            self.callDeprecated([osutils._file_id_warning],
 
580
                                osutils.safe_file_id, u'bargam'))
 
581
 
 
582
    def test_from_unicode_string_unicode_contents(self):
 
583
        self.assertEqual('bargam\xc2\xae',
 
584
                         osutils.safe_file_id(u'bargam\xae', warn=False))
 
585
 
 
586
    def test_from_utf8_string(self):
 
587
        self.assertEqual('foo\xc2\xae',
 
588
                         osutils.safe_file_id('foo\xc2\xae'))
 
589
 
 
590
    def test_none(self):
 
591
        """Currently, None is a valid revision_id"""
 
592
        self.assertEqual(None, osutils.safe_file_id(None))
 
593
 
 
594
 
 
595
class TestWin32Funcs(TestCase):
 
596
    """Test that the _win32 versions of os utilities return appropriate paths."""
 
597
 
 
598
    def test_abspath(self):
 
599
        self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
 
600
        self.assertEqual('C:/foo', osutils._win32_abspath('C:/foo'))
 
601
        self.assertEqual('//HOST/path', osutils._win32_abspath(r'\\HOST\path'))
 
602
        self.assertEqual('//HOST/path', osutils._win32_abspath('//HOST/path'))
 
603
 
 
604
    def test_realpath(self):
 
605
        self.assertEqual('C:/foo', osutils._win32_realpath('C:\\foo'))
 
606
        self.assertEqual('C:/foo', osutils._win32_realpath('C:/foo'))
 
607
 
 
608
    def test_pathjoin(self):
 
609
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path', 'to', 'foo'))
 
610
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path\\to', 'C:\\foo'))
 
611
        self.assertEqual('C:/foo', osutils._win32_pathjoin('path/to', 'C:/foo'))
 
612
        self.assertEqual('path/to/foo', osutils._win32_pathjoin('path/to/', 'foo'))
 
613
        self.assertEqual('/foo', osutils._win32_pathjoin('C:/path/to/', '/foo'))
 
614
        self.assertEqual('/foo', osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
 
615
 
 
616
    def test_normpath(self):
 
617
        self.assertEqual('path/to/foo', osutils._win32_normpath(r'path\\from\..\to\.\foo'))
 
618
        self.assertEqual('path/to/foo', osutils._win32_normpath('path//from/../to/./foo'))
 
619
 
 
620
    def test_getcwd(self):
 
621
        cwd = osutils._win32_getcwd()
 
622
        os_cwd = os.getcwdu()
 
623
        self.assertEqual(os_cwd[1:].replace('\\', '/'), cwd[1:])
 
624
        # win32 is inconsistent whether it returns lower or upper case
 
625
        # and even if it was consistent the user might type the other
 
626
        # so we force it to uppercase
 
627
        # running python.exe under cmd.exe return capital C:\\
 
628
        # running win32 python inside a cygwin shell returns lowercase
 
629
        self.assertEqual(os_cwd[0].upper(), cwd[0])
 
630
 
 
631
    def test_fixdrive(self):
 
632
        self.assertEqual('H:/foo', osutils._win32_fixdrive('h:/foo'))
 
633
        self.assertEqual('H:/foo', osutils._win32_fixdrive('H:/foo'))
 
634
        self.assertEqual('C:\\foo', osutils._win32_fixdrive('c:\\foo'))
 
635
 
 
636
    def test_win98_abspath(self):
 
637
        # absolute path
 
638
        self.assertEqual('C:/foo', osutils._win98_abspath('C:\\foo'))
 
639
        self.assertEqual('C:/foo', osutils._win98_abspath('C:/foo'))
 
640
        # UNC path
 
641
        self.assertEqual('//HOST/path', osutils._win98_abspath(r'\\HOST\path'))
 
642
        self.assertEqual('//HOST/path', osutils._win98_abspath('//HOST/path'))
 
643
        # relative path
 
644
        cwd = osutils.getcwd().rstrip('/')
 
645
        drive = osutils._nt_splitdrive(cwd)[0]
 
646
        self.assertEqual(cwd+'/path', osutils._win98_abspath('path'))
 
647
        self.assertEqual(drive+'/path', osutils._win98_abspath('/path'))
 
648
        # unicode path
 
649
        u = u'\u1234'
 
650
        self.assertEqual(cwd+'/'+u, osutils._win98_abspath(u))
 
651
 
 
652
 
 
653
class TestWin32FuncsDirs(TestCaseInTempDir):
 
654
    """Test win32 functions that create files."""
 
655
    
 
656
    def test_getcwd(self):
 
657
        if win32utils.winver == 'Windows 98':
 
658
            raise TestSkipped('Windows 98 cannot handle unicode filenames')
 
659
        # Make sure getcwd can handle unicode filenames
 
660
        try:
 
661
            os.mkdir(u'mu-\xb5')
 
662
        except UnicodeError:
 
663
            raise TestSkipped("Unable to create Unicode filename")
 
664
 
 
665
        os.chdir(u'mu-\xb5')
 
666
        # TODO: jam 20060427 This will probably fail on Mac OSX because
 
667
        #       it will change the normalization of B\xe5gfors
 
668
        #       Consider using a different unicode character, or make
 
669
        #       osutils.getcwd() renormalize the path.
 
670
        self.assertEndsWith(osutils._win32_getcwd(), u'mu-\xb5')
 
671
 
 
672
    def test_minimum_path_selection(self):
 
673
        self.assertEqual(set(),
 
674
            osutils.minimum_path_selection([]))
 
675
        self.assertEqual(set(['a', 'b']),
 
676
            osutils.minimum_path_selection(['a', 'b']))
 
677
        self.assertEqual(set(['a/', 'b']),
 
678
            osutils.minimum_path_selection(['a/', 'b']))
 
679
        self.assertEqual(set(['a/', 'b']),
 
680
            osutils.minimum_path_selection(['a/c', 'a/', 'b']))
 
681
 
 
682
    def test_mkdtemp(self):
 
683
        tmpdir = osutils._win32_mkdtemp(dir='.')
 
684
        self.assertFalse('\\' in tmpdir)
 
685
 
 
686
    def test_rename(self):
 
687
        a = open('a', 'wb')
 
688
        a.write('foo\n')
 
689
        a.close()
 
690
        b = open('b', 'wb')
 
691
        b.write('baz\n')
 
692
        b.close()
 
693
 
 
694
        osutils._win32_rename('b', 'a')
 
695
        self.failUnlessExists('a')
 
696
        self.failIfExists('b')
 
697
        self.assertFileEqual('baz\n', 'a')
 
698
 
 
699
    def test_rename_missing_file(self):
 
700
        a = open('a', 'wb')
 
701
        a.write('foo\n')
 
702
        a.close()
 
703
 
 
704
        try:
 
705
            osutils._win32_rename('b', 'a')
 
706
        except (IOError, OSError), e:
 
707
            self.assertEqual(errno.ENOENT, e.errno)
 
708
        self.assertFileEqual('foo\n', 'a')
 
709
 
 
710
    def test_rename_missing_dir(self):
 
711
        os.mkdir('a')
 
712
        try:
 
713
            osutils._win32_rename('b', 'a')
 
714
        except (IOError, OSError), e:
 
715
            self.assertEqual(errno.ENOENT, e.errno)
 
716
 
 
717
    def test_rename_current_dir(self):
 
718
        os.mkdir('a')
 
719
        os.chdir('a')
 
720
        # You can't rename the working directory
 
721
        # doing rename non-existant . usually
 
722
        # just raises ENOENT, since non-existant
 
723
        # doesn't exist.
 
724
        try:
 
725
            osutils._win32_rename('b', '.')
 
726
        except (IOError, OSError), e:
 
727
            self.assertEqual(errno.ENOENT, e.errno)
 
728
 
 
729
    def test_splitpath(self):
 
730
        def check(expected, path):
 
731
            self.assertEqual(expected, osutils.splitpath(path))
 
732
 
 
733
        check(['a'], 'a')
 
734
        check(['a', 'b'], 'a/b')
 
735
        check(['a', 'b'], 'a/./b')
 
736
        check(['a', '.b'], 'a/.b')
 
737
        check(['a', '.b'], 'a\\.b')
 
738
 
 
739
        self.assertRaises(errors.BzrError, osutils.splitpath, 'a/../b')
 
740
 
 
741
 
 
742
class TestMacFuncsDirs(TestCaseInTempDir):
 
743
    """Test mac special functions that require directories."""
 
744
 
 
745
    def test_getcwd(self):
 
746
        # On Mac, this will actually create Ba\u030agfors
 
747
        # but chdir will still work, because it accepts both paths
 
748
        try:
 
749
            os.mkdir(u'B\xe5gfors')
 
750
        except UnicodeError:
 
751
            raise TestSkipped("Unable to create Unicode filename")
 
752
 
 
753
        os.chdir(u'B\xe5gfors')
 
754
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
 
755
 
 
756
    def test_getcwd_nonnorm(self):
 
757
        # Test that _mac_getcwd() will normalize this path
 
758
        try:
 
759
            os.mkdir(u'Ba\u030agfors')
 
760
        except UnicodeError:
 
761
            raise TestSkipped("Unable to create Unicode filename")
 
762
 
 
763
        os.chdir(u'Ba\u030agfors')
 
764
        self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
 
765
 
 
766
 
 
767
class TestSplitLines(TestCase):
 
768
 
 
769
    def test_split_unicode(self):
 
770
        self.assertEqual([u'foo\n', u'bar\xae'],
 
771
                         osutils.split_lines(u'foo\nbar\xae'))
 
772
        self.assertEqual([u'foo\n', u'bar\xae\n'],
 
773
                         osutils.split_lines(u'foo\nbar\xae\n'))
 
774
 
 
775
    def test_split_with_carriage_returns(self):
 
776
        self.assertEqual(['foo\rbar\n'],
 
777
                         osutils.split_lines('foo\rbar\n'))
 
778
 
 
779
 
 
780
class TestWalkDirs(TestCaseInTempDir):
 
781
 
 
782
    def test_readdir(self):
 
783
        tree = [
 
784
            '.bzr/',
 
785
            '0file',
 
786
            '1dir/',
 
787
            '1dir/0file',
 
788
            '1dir/1dir/',
 
789
            '2file'
 
790
            ]
 
791
        self.build_tree(tree)
 
792
        expected_names = ['.bzr', '0file', '1dir', '2file']
 
793
        # read_dir returns pairs, which form a table with either None in all
 
794
        # the first columns, or a sort key to get best on-disk-read order, 
 
795
        # and the disk path name in utf-8 encoding in the second column.
 
796
        read_result = self.read_dir('.')
 
797
        # The second column is always the names, and every name except "." and
 
798
        # ".." should be present.
 
799
        names = sorted([row[1] for row in read_result])
 
800
        self.assertEqual(expected_names, names)
 
801
        expected_sort_key = None
 
802
        if read_result[0][0] is None:
 
803
            # No sort key returned - all keys must None
 
804
            operator = self.assertEqual
 
805
        else:
 
806
            # A sort key in the first row implies sort keys in the other rows.
 
807
            operator = self.assertNotEqual
 
808
        for row in read_result:
 
809
            operator(None, row[0])
 
810
 
 
811
    def test_compiled_extension_exists(self):
 
812
        self.requireFeature(ReadDirFeature)
 
813
        
 
814
    def test_walkdirs(self):
 
815
        tree = [
 
816
            '.bzr',
 
817
            '0file',
 
818
            '1dir/',
 
819
            '1dir/0file',
 
820
            '1dir/1dir/',
 
821
            '2file'
 
822
            ]
 
823
        self.build_tree(tree)
 
824
        expected_dirblocks = [
 
825
                (('', '.'),
 
826
                 [('0file', '0file', 'file'),
 
827
                  ('1dir', '1dir', 'directory'),
 
828
                  ('2file', '2file', 'file'),
 
829
                 ]
 
830
                ),
 
831
                (('1dir', './1dir'),
 
832
                 [('1dir/0file', '0file', 'file'),
 
833
                  ('1dir/1dir', '1dir', 'directory'),
 
834
                 ]
 
835
                ),
 
836
                (('1dir/1dir', './1dir/1dir'),
 
837
                 [
 
838
                 ]
 
839
                ),
 
840
            ]
 
841
        result = []
 
842
        found_bzrdir = False
 
843
        for dirdetail, dirblock in osutils.walkdirs('.'):
 
844
            if len(dirblock) and dirblock[0][1] == '.bzr':
 
845
                # this tests the filtering of selected paths
 
846
                found_bzrdir = True
 
847
                del dirblock[0]
 
848
            result.append((dirdetail, dirblock))
 
849
 
 
850
        self.assertTrue(found_bzrdir)
 
851
        self.assertEqual(expected_dirblocks,
 
852
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
853
        # you can search a subdir only, with a supplied prefix.
 
854
        result = []
 
855
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
 
856
            result.append(dirblock)
 
857
        self.assertEqual(expected_dirblocks[1:],
 
858
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
859
 
 
860
    def test__walkdirs_utf8(self):
 
861
        tree = [
 
862
            '.bzr',
 
863
            '0file',
 
864
            '1dir/',
 
865
            '1dir/0file',
 
866
            '1dir/1dir/',
 
867
            '2file'
 
868
            ]
 
869
        self.build_tree(tree)
 
870
        expected_dirblocks = [
 
871
                (('', '.'),
 
872
                 [('0file', '0file', 'file'),
 
873
                  ('1dir', '1dir', 'directory'),
 
874
                  ('2file', '2file', 'file'),
 
875
                 ]
 
876
                ),
 
877
                (('1dir', './1dir'),
 
878
                 [('1dir/0file', '0file', 'file'),
 
879
                  ('1dir/1dir', '1dir', 'directory'),
 
880
                 ]
 
881
                ),
 
882
                (('1dir/1dir', './1dir/1dir'),
 
883
                 [
 
884
                 ]
 
885
                ),
 
886
            ]
 
887
        result = []
 
888
        found_bzrdir = False
 
889
        for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
 
890
            if len(dirblock) and dirblock[0][1] == '.bzr':
 
891
                # this tests the filtering of selected paths
 
892
                found_bzrdir = True
 
893
                del dirblock[0]
 
894
            result.append((dirdetail, dirblock))
 
895
 
 
896
        self.assertTrue(found_bzrdir)
 
897
        self.assertEqual(expected_dirblocks,
 
898
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
899
        # you can search a subdir only, with a supplied prefix.
 
900
        result = []
 
901
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
 
902
            result.append(dirblock)
 
903
        self.assertEqual(expected_dirblocks[1:],
 
904
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
 
905
 
 
906
    def _filter_out_stat(self, result):
 
907
        """Filter out the stat value from the walkdirs result"""
 
908
        for dirdetail, dirblock in result:
 
909
            new_dirblock = []
 
910
            for info in dirblock:
 
911
                # Ignore info[3] which is the stat
 
912
                new_dirblock.append((info[0], info[1], info[2], info[4]))
 
913
            dirblock[:] = new_dirblock
 
914
 
 
915
    def test__walkdirs_utf8_selection(self):
 
916
        # Just trigger the function once, to make sure it has selected a real
 
917
        # implementation.
 
918
        list(osutils._walkdirs_utf8('.'))
 
919
        if WalkdirsWin32Feature.available():
 
920
            # If the compiled form is available, make sure it is used
 
921
            from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
 
922
            self.assertIs(_walkdirs_utf8_win32_find_file,
 
923
                          osutils._real_walkdirs_utf8)
 
924
        elif sys.platform == 'win32':
 
925
            self.assertIs(osutils._walkdirs_unicode_to_utf8,
 
926
                          osutils._real_walkdirs_utf8)
 
927
        elif osutils._fs_enc.upper() in ('UTF-8', 'ASCII', 'ANSI_X3.4-1968'): # ascii
 
928
            self.assertIs(osutils._walkdirs_fs_utf8,
 
929
                          osutils._real_walkdirs_utf8)
 
930
        else:
 
931
            self.assertIs(osutils._walkdirs_unicode_to_utf8,
 
932
                          osutils._real_walkdirs_utf8)
 
933
 
 
934
    def _save_platform_info(self):
 
935
        cur_winver = win32utils.winver
 
936
        cur_fs_enc = osutils._fs_enc
 
937
        cur_real_walkdirs_utf8 = osutils._real_walkdirs_utf8
 
938
        def restore():
 
939
            win32utils.winver = cur_winver
 
940
            osutils._fs_enc = cur_fs_enc
 
941
            osutils._real_walkdirs_utf8 = cur_real_walkdirs_utf8
 
942
        self.addCleanup(restore)
 
943
 
 
944
    def assertWalkdirsUtf8Is(self, expected):
 
945
        """Assert the right implementation for _walkdirs_utf8 is chosen."""
 
946
        # Force it to redetect
 
947
        osutils._real_walkdirs_utf8 = None
 
948
        # Nothing to list, but should still trigger the selection logic
 
949
        self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
 
950
        self.assertIs(expected, osutils._real_walkdirs_utf8)
 
951
 
 
952
    def test_force_walkdirs_utf8_fs_utf8(self):
 
953
        self._save_platform_info()
 
954
        win32utils.winver = None # Avoid the win32 detection code
 
955
        osutils._fs_enc = 'UTF-8'
 
956
        self.assertWalkdirsUtf8Is(osutils._walkdirs_fs_utf8)
 
957
 
 
958
    def test_force_walkdirs_utf8_fs_ascii(self):
 
959
        self._save_platform_info()
 
960
        win32utils.winver = None # Avoid the win32 detection code
 
961
        osutils._fs_enc = 'US-ASCII'
 
962
        self.assertWalkdirsUtf8Is(osutils._walkdirs_fs_utf8)
 
963
 
 
964
    def test_force_walkdirs_utf8_fs_ANSI(self):
 
965
        self._save_platform_info()
 
966
        win32utils.winver = None # Avoid the win32 detection code
 
967
        osutils._fs_enc = 'ANSI_X3.4-1968'
 
968
        self.assertWalkdirsUtf8Is(osutils._walkdirs_fs_utf8)
 
969
 
 
970
    def test_force_walkdirs_utf8_fs_latin1(self):
 
971
        self._save_platform_info()
 
972
        win32utils.winver = None # Avoid the win32 detection code
 
973
        osutils._fs_enc = 'latin1'
 
974
        self.assertWalkdirsUtf8Is(osutils._walkdirs_unicode_to_utf8)
 
975
 
 
976
    def test_force_walkdirs_utf8_nt(self):
 
977
        self.requireFeature(WalkdirsWin32Feature)
 
978
        self._save_platform_info()
 
979
        win32utils.winver = 'Windows NT'
 
980
        from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
 
981
        self.assertWalkdirsUtf8Is(_walkdirs_utf8_win32_find_file)
 
982
 
 
983
    def test_force_walkdirs_utf8_nt(self):
 
984
        self.requireFeature(WalkdirsWin32Feature)
 
985
        self._save_platform_info()
 
986
        win32utils.winver = 'Windows 98'
 
987
        self.assertWalkdirsUtf8Is(osutils._walkdirs_unicode_to_utf8)
 
988
 
 
989
    def test_unicode_walkdirs(self):
 
990
        """Walkdirs should always return unicode paths."""
 
991
        name0 = u'0file-\xb6'
 
992
        name1 = u'1dir-\u062c\u0648'
 
993
        name2 = u'2file-\u0633'
 
994
        tree = [
 
995
            name0,
 
996
            name1 + '/',
 
997
            name1 + '/' + name0,
 
998
            name1 + '/' + name1 + '/',
 
999
            name2,
 
1000
            ]
 
1001
        try:
 
1002
            self.build_tree(tree)
 
1003
        except UnicodeError:
 
1004
            raise TestSkipped('Could not represent Unicode chars'
 
1005
                              ' in current encoding.')
 
1006
        expected_dirblocks = [
 
1007
                ((u'', u'.'),
 
1008
                 [(name0, name0, 'file', './' + name0),
 
1009
                  (name1, name1, 'directory', './' + name1),
 
1010
                  (name2, name2, 'file', './' + name2),
 
1011
                 ]
 
1012
                ),
 
1013
                ((name1, './' + name1),
 
1014
                 [(name1 + '/' + name0, name0, 'file', './' + name1
 
1015
                                                        + '/' + name0),
 
1016
                  (name1 + '/' + name1, name1, 'directory', './' + name1
 
1017
                                                            + '/' + name1),
 
1018
                 ]
 
1019
                ),
 
1020
                ((name1 + '/' + name1, './' + name1 + '/' + name1),
 
1021
                 [
 
1022
                 ]
 
1023
                ),
 
1024
            ]
 
1025
        result = list(osutils.walkdirs('.'))
 
1026
        self._filter_out_stat(result)
 
1027
        self.assertEqual(expected_dirblocks, result)
 
1028
        result = list(osutils.walkdirs(u'./'+name1, name1))
 
1029
        self._filter_out_stat(result)
 
1030
        self.assertEqual(expected_dirblocks[1:], result)
 
1031
 
 
1032
    def test_unicode__walkdirs_utf8(self):
 
1033
        """Walkdirs_utf8 should always return utf8 paths.
 
1034
 
 
1035
        The abspath portion might be in unicode or utf-8
 
1036
        """
 
1037
        name0 = u'0file-\xb6'
 
1038
        name1 = u'1dir-\u062c\u0648'
 
1039
        name2 = u'2file-\u0633'
 
1040
        tree = [
 
1041
            name0,
 
1042
            name1 + '/',
 
1043
            name1 + '/' + name0,
 
1044
            name1 + '/' + name1 + '/',
 
1045
            name2,
 
1046
            ]
 
1047
        try:
 
1048
            self.build_tree(tree)
 
1049
        except UnicodeError:
 
1050
            raise TestSkipped('Could not represent Unicode chars'
 
1051
                              ' in current encoding.')
 
1052
        name0 = name0.encode('utf8')
 
1053
        name1 = name1.encode('utf8')
 
1054
        name2 = name2.encode('utf8')
 
1055
 
 
1056
        expected_dirblocks = [
 
1057
                (('', '.'),
 
1058
                 [(name0, name0, 'file', './' + name0),
 
1059
                  (name1, name1, 'directory', './' + name1),
 
1060
                  (name2, name2, 'file', './' + name2),
 
1061
                 ]
 
1062
                ),
 
1063
                ((name1, './' + name1),
 
1064
                 [(name1 + '/' + name0, name0, 'file', './' + name1
 
1065
                                                        + '/' + name0),
 
1066
                  (name1 + '/' + name1, name1, 'directory', './' + name1
 
1067
                                                            + '/' + name1),
 
1068
                 ]
 
1069
                ),
 
1070
                ((name1 + '/' + name1, './' + name1 + '/' + name1),
 
1071
                 [
 
1072
                 ]
 
1073
                ),
 
1074
            ]
 
1075
        result = []
 
1076
        # For ease in testing, if walkdirs_utf8 returns Unicode, assert that
 
1077
        # all abspaths are Unicode, and encode them back into utf8.
 
1078
        for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
 
1079
            self.assertIsInstance(dirdetail[0], str)
 
1080
            if isinstance(dirdetail[1], unicode):
 
1081
                dirdetail = (dirdetail[0], dirdetail[1].encode('utf8'))
 
1082
                dirblock = [list(info) for info in dirblock]
 
1083
                for info in dirblock:
 
1084
                    self.assertIsInstance(info[4], unicode)
 
1085
                    info[4] = info[4].encode('utf8')
 
1086
            new_dirblock = []
 
1087
            for info in dirblock:
 
1088
                self.assertIsInstance(info[0], str)
 
1089
                self.assertIsInstance(info[1], str)
 
1090
                self.assertIsInstance(info[4], str)
 
1091
                # Remove the stat information
 
1092
                new_dirblock.append((info[0], info[1], info[2], info[4]))
 
1093
            result.append((dirdetail, new_dirblock))
 
1094
        self.assertEqual(expected_dirblocks, result)
 
1095
 
 
1096
    def test_unicode__walkdirs_unicode_to_utf8(self):
 
1097
        """walkdirs_unicode_to_utf8 should be a safe fallback everywhere
 
1098
 
 
1099
        The abspath portion should be in unicode
 
1100
        """
 
1101
        name0u = u'0file-\xb6'
 
1102
        name1u = u'1dir-\u062c\u0648'
 
1103
        name2u = u'2file-\u0633'
 
1104
        tree = [
 
1105
            name0u,
 
1106
            name1u + '/',
 
1107
            name1u + '/' + name0u,
 
1108
            name1u + '/' + name1u + '/',
 
1109
            name2u,
 
1110
            ]
 
1111
        try:
 
1112
            self.build_tree(tree)
 
1113
        except UnicodeError:
 
1114
            raise TestSkipped('Could not represent Unicode chars'
 
1115
                              ' in current encoding.')
 
1116
        name0 = name0u.encode('utf8')
 
1117
        name1 = name1u.encode('utf8')
 
1118
        name2 = name2u.encode('utf8')
 
1119
 
 
1120
        # All of the abspaths should be in unicode, all of the relative paths
 
1121
        # should be in utf8
 
1122
        expected_dirblocks = [
 
1123
                (('', '.'),
 
1124
                 [(name0, name0, 'file', './' + name0u),
 
1125
                  (name1, name1, 'directory', './' + name1u),
 
1126
                  (name2, name2, 'file', './' + name2u),
 
1127
                 ]
 
1128
                ),
 
1129
                ((name1, './' + name1u),
 
1130
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
1131
                                                        + '/' + name0u),
 
1132
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
1133
                                                            + '/' + name1u),
 
1134
                 ]
 
1135
                ),
 
1136
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
1137
                 [
 
1138
                 ]
 
1139
                ),
 
1140
            ]
 
1141
        result = list(osutils._walkdirs_unicode_to_utf8('.'))
 
1142
        self._filter_out_stat(result)
 
1143
        self.assertEqual(expected_dirblocks, result)
 
1144
 
 
1145
    def test__walkdirs_utf_win32_find_file(self):
 
1146
        self.requireFeature(WalkdirsWin32Feature)
 
1147
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1148
        from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
 
1149
        name0u = u'0file-\xb6'
 
1150
        name1u = u'1dir-\u062c\u0648'
 
1151
        name2u = u'2file-\u0633'
 
1152
        tree = [
 
1153
            name0u,
 
1154
            name1u + '/',
 
1155
            name1u + '/' + name0u,
 
1156
            name1u + '/' + name1u + '/',
 
1157
            name2u,
 
1158
            ]
 
1159
        self.build_tree(tree)
 
1160
        name0 = name0u.encode('utf8')
 
1161
        name1 = name1u.encode('utf8')
 
1162
        name2 = name2u.encode('utf8')
 
1163
 
 
1164
        # All of the abspaths should be in unicode, all of the relative paths
 
1165
        # should be in utf8
 
1166
        expected_dirblocks = [
 
1167
                (('', '.'),
 
1168
                 [(name0, name0, 'file', './' + name0u),
 
1169
                  (name1, name1, 'directory', './' + name1u),
 
1170
                  (name2, name2, 'file', './' + name2u),
 
1171
                 ]
 
1172
                ),
 
1173
                ((name1, './' + name1u),
 
1174
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
 
1175
                                                        + '/' + name0u),
 
1176
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
 
1177
                                                            + '/' + name1u),
 
1178
                 ]
 
1179
                ),
 
1180
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
 
1181
                 [
 
1182
                 ]
 
1183
                ),
 
1184
            ]
 
1185
        result = list(_walkdirs_utf8_win32_find_file(u'.'))
 
1186
        self._filter_out_stat(result)
 
1187
        self.assertEqual(expected_dirblocks, result)
 
1188
 
 
1189
    def assertStatIsCorrect(self, path, win32stat):
 
1190
        os_stat = os.stat(path)
 
1191
        self.assertEqual(os_stat.st_size, win32stat.st_size)
 
1192
        self.assertAlmostEqual(os_stat.st_mtime, win32stat.st_mtime, places=4)
 
1193
        self.assertAlmostEqual(os_stat.st_ctime, win32stat.st_ctime, places=4)
 
1194
        self.assertAlmostEqual(os_stat.st_atime, win32stat.st_atime, places=4)
 
1195
        self.assertEqual(os_stat.st_dev, win32stat.st_dev)
 
1196
        self.assertEqual(os_stat.st_ino, win32stat.st_ino)
 
1197
        self.assertEqual(os_stat.st_mode, win32stat.st_mode)
 
1198
 
 
1199
    def test__walkdirs_utf_win32_find_file_stat_file(self):
 
1200
        """make sure our Stat values are valid"""
 
1201
        self.requireFeature(WalkdirsWin32Feature)
 
1202
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1203
        from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
 
1204
        name0u = u'0file-\xb6'
 
1205
        name0 = name0u.encode('utf8')
 
1206
        self.build_tree([name0u])
 
1207
        # I hate to sleep() here, but I'm trying to make the ctime different
 
1208
        # from the mtime
 
1209
        time.sleep(2)
 
1210
        f = open(name0u, 'ab')
 
1211
        try:
 
1212
            f.write('just a small update')
 
1213
        finally:
 
1214
            f.close()
 
1215
 
 
1216
        result = list(_walkdirs_utf8_win32_find_file(u'.'))
 
1217
        entry = result[0][1][0]
 
1218
        self.assertEqual((name0, name0, 'file'), entry[:3])
 
1219
        self.assertEqual(u'./' + name0u, entry[4])
 
1220
        self.assertStatIsCorrect(entry[4], entry[3])
 
1221
        self.assertNotEqual(entry[3].st_mtime, entry[3].st_ctime)
 
1222
 
 
1223
    def test__walkdirs_utf_win32_find_file_stat_directory(self):
 
1224
        """make sure our Stat values are valid"""
 
1225
        self.requireFeature(WalkdirsWin32Feature)
 
1226
        self.requireFeature(tests.UnicodeFilenameFeature)
 
1227
        from bzrlib._walkdirs_win32 import _walkdirs_utf8_win32_find_file
 
1228
        name0u = u'0dir-\u062c\u0648'
 
1229
        name0 = name0u.encode('utf8')
 
1230
        self.build_tree([name0u + '/'])
 
1231
 
 
1232
        result = list(_walkdirs_utf8_win32_find_file(u'.'))
 
1233
        entry = result[0][1][0]
 
1234
        self.assertEqual((name0, name0, 'directory'), entry[:3])
 
1235
        self.assertEqual(u'./' + name0u, entry[4])
 
1236
        self.assertStatIsCorrect(entry[4], entry[3])
 
1237
 
 
1238
    def assertPathCompare(self, path_less, path_greater):
 
1239
        """check that path_less and path_greater compare correctly."""
 
1240
        self.assertEqual(0, osutils.compare_paths_prefix_order(
 
1241
            path_less, path_less))
 
1242
        self.assertEqual(0, osutils.compare_paths_prefix_order(
 
1243
            path_greater, path_greater))
 
1244
        self.assertEqual(-1, osutils.compare_paths_prefix_order(
 
1245
            path_less, path_greater))
 
1246
        self.assertEqual(1, osutils.compare_paths_prefix_order(
 
1247
            path_greater, path_less))
 
1248
 
 
1249
    def test_compare_paths_prefix_order(self):
 
1250
        # root before all else
 
1251
        self.assertPathCompare("/", "/a")
 
1252
        # alpha within a dir
 
1253
        self.assertPathCompare("/a", "/b")
 
1254
        self.assertPathCompare("/b", "/z")
 
1255
        # high dirs before lower.
 
1256
        self.assertPathCompare("/z", "/a/a")
 
1257
        # except if the deeper dir should be output first
 
1258
        self.assertPathCompare("/a/b/c", "/d/g")
 
1259
        # lexical betwen dirs of the same height
 
1260
        self.assertPathCompare("/a/z", "/z/z")
 
1261
        self.assertPathCompare("/a/c/z", "/a/d/e")
 
1262
 
 
1263
        # this should also be consistent for no leading / paths
 
1264
        # root before all else
 
1265
        self.assertPathCompare("", "a")
 
1266
        # alpha within a dir
 
1267
        self.assertPathCompare("a", "b")
 
1268
        self.assertPathCompare("b", "z")
 
1269
        # high dirs before lower.
 
1270
        self.assertPathCompare("z", "a/a")
 
1271
        # except if the deeper dir should be output first
 
1272
        self.assertPathCompare("a/b/c", "d/g")
 
1273
        # lexical betwen dirs of the same height
 
1274
        self.assertPathCompare("a/z", "z/z")
 
1275
        self.assertPathCompare("a/c/z", "a/d/e")
 
1276
 
 
1277
    def test_path_prefix_sorting(self):
 
1278
        """Doing a sort on path prefix should match our sample data."""
 
1279
        original_paths = [
 
1280
            'a',
 
1281
            'a/b',
 
1282
            'a/b/c',
 
1283
            'b',
 
1284
            'b/c',
 
1285
            'd',
 
1286
            'd/e',
 
1287
            'd/e/f',
 
1288
            'd/f',
 
1289
            'd/g',
 
1290
            'g',
 
1291
            ]
 
1292
 
 
1293
        dir_sorted_paths = [
 
1294
            'a',
 
1295
            'b',
 
1296
            'd',
 
1297
            'g',
 
1298
            'a/b',
 
1299
            'a/b/c',
 
1300
            'b/c',
 
1301
            'd/e',
 
1302
            'd/f',
 
1303
            'd/g',
 
1304
            'd/e/f',
 
1305
            ]
 
1306
 
 
1307
        self.assertEqual(
 
1308
            dir_sorted_paths,
 
1309
            sorted(original_paths, key=osutils.path_prefix_key))
 
1310
        # using the comparison routine shoudl work too:
 
1311
        self.assertEqual(
 
1312
            dir_sorted_paths,
 
1313
            sorted(original_paths, cmp=osutils.compare_paths_prefix_order))
 
1314
 
 
1315
 
 
1316
class TestCopyTree(TestCaseInTempDir):
 
1317
    
 
1318
    def test_copy_basic_tree(self):
 
1319
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
 
1320
        osutils.copy_tree('source', 'target')
 
1321
        self.assertEqual(['a', 'b'], sorted(os.listdir('target')))
 
1322
        self.assertEqual(['c'], os.listdir('target/b'))
 
1323
 
 
1324
    def test_copy_tree_target_exists(self):
 
1325
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c',
 
1326
                         'target/'])
 
1327
        osutils.copy_tree('source', 'target')
 
1328
        self.assertEqual(['a', 'b'], sorted(os.listdir('target')))
 
1329
        self.assertEqual(['c'], os.listdir('target/b'))
 
1330
 
 
1331
    def test_copy_tree_symlinks(self):
 
1332
        self.requireFeature(SymlinkFeature)
 
1333
        self.build_tree(['source/'])
 
1334
        os.symlink('a/generic/path', 'source/lnk')
 
1335
        osutils.copy_tree('source', 'target')
 
1336
        self.assertEqual(['lnk'], os.listdir('target'))
 
1337
        self.assertEqual('a/generic/path', os.readlink('target/lnk'))
 
1338
 
 
1339
    def test_copy_tree_handlers(self):
 
1340
        processed_files = []
 
1341
        processed_links = []
 
1342
        def file_handler(from_path, to_path):
 
1343
            processed_files.append(('f', from_path, to_path))
 
1344
        def dir_handler(from_path, to_path):
 
1345
            processed_files.append(('d', from_path, to_path))
 
1346
        def link_handler(from_path, to_path):
 
1347
            processed_links.append((from_path, to_path))
 
1348
        handlers = {'file':file_handler,
 
1349
                    'directory':dir_handler,
 
1350
                    'symlink':link_handler,
 
1351
                   }
 
1352
 
 
1353
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
 
1354
        if osutils.has_symlinks():
 
1355
            os.symlink('a/generic/path', 'source/lnk')
 
1356
        osutils.copy_tree('source', 'target', handlers=handlers)
 
1357
 
 
1358
        self.assertEqual([('d', 'source', 'target'),
 
1359
                          ('f', 'source/a', 'target/a'),
 
1360
                          ('d', 'source/b', 'target/b'),
 
1361
                          ('f', 'source/b/c', 'target/b/c'),
 
1362
                         ], processed_files)
 
1363
        self.failIfExists('target')
 
1364
        if osutils.has_symlinks():
 
1365
            self.assertEqual([('source/lnk', 'target/lnk')], processed_links)
 
1366
 
 
1367
 
 
1368
#class TestTerminalEncoding has been moved to test_osutils_encodings.py
 
1369
# [bialix] 2006/12/26
 
1370
 
 
1371
 
 
1372
class TestSetUnsetEnv(TestCase):
 
1373
    """Test updating the environment"""
 
1374
 
 
1375
    def setUp(self):
 
1376
        super(TestSetUnsetEnv, self).setUp()
 
1377
 
 
1378
        self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'),
 
1379
                         'Environment was not cleaned up properly.'
 
1380
                         ' Variable BZR_TEST_ENV_VAR should not exist.')
 
1381
        def cleanup():
 
1382
            if 'BZR_TEST_ENV_VAR' in os.environ:
 
1383
                del os.environ['BZR_TEST_ENV_VAR']
 
1384
 
 
1385
        self.addCleanup(cleanup)
 
1386
 
 
1387
    def test_set(self):
 
1388
        """Test that we can set an env variable"""
 
1389
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
 
1390
        self.assertEqual(None, old)
 
1391
        self.assertEqual('foo', os.environ.get('BZR_TEST_ENV_VAR'))
 
1392
 
 
1393
    def test_double_set(self):
 
1394
        """Test that we get the old value out"""
 
1395
        osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
 
1396
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'bar')
 
1397
        self.assertEqual('foo', old)
 
1398
        self.assertEqual('bar', os.environ.get('BZR_TEST_ENV_VAR'))
 
1399
 
 
1400
    def test_unicode(self):
 
1401
        """Environment can only contain plain strings
 
1402
        
 
1403
        So Unicode strings must be encoded.
 
1404
        """
 
1405
        uni_val, env_val = probe_unicode_in_user_encoding()
 
1406
        if uni_val is None:
 
1407
            raise TestSkipped('Cannot find a unicode character that works in'
 
1408
                              ' encoding %s' % (bzrlib.user_encoding,))
 
1409
 
 
1410
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', uni_val)
 
1411
        self.assertEqual(env_val, os.environ.get('BZR_TEST_ENV_VAR'))
 
1412
 
 
1413
    def test_unset(self):
 
1414
        """Test that passing None will remove the env var"""
 
1415
        osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
 
1416
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', None)
 
1417
        self.assertEqual('foo', old)
 
1418
        self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'))
 
1419
        self.failIf('BZR_TEST_ENV_VAR' in os.environ)
 
1420
 
 
1421
 
 
1422
class TestLocalTimeOffset(TestCase):
 
1423
 
 
1424
    def test_local_time_offset(self):
 
1425
        """Test that local_time_offset() returns a sane value."""
 
1426
        offset = osutils.local_time_offset()
 
1427
        self.assertTrue(isinstance(offset, int))
 
1428
        # Test that the offset is no more than a eighteen hours in
 
1429
        # either direction.
 
1430
        # Time zone handling is system specific, so it is difficult to
 
1431
        # do more specific tests, but a value outside of this range is
 
1432
        # probably wrong.
 
1433
        eighteen_hours = 18 * 3600
 
1434
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
 
1435
 
 
1436
    def test_local_time_offset_with_timestamp(self):
 
1437
        """Test that local_time_offset() works with a timestamp."""
 
1438
        offset = osutils.local_time_offset(1000000000.1234567)
 
1439
        self.assertTrue(isinstance(offset, int))
 
1440
        eighteen_hours = 18 * 3600
 
1441
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
 
1442
 
 
1443
 
 
1444
class TestShaFileByName(TestCaseInTempDir):
 
1445
 
 
1446
    def test_sha_empty(self):
 
1447
        self.build_tree_contents([('foo', '')])
 
1448
        expected_sha = osutils.sha_string('')
 
1449
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
 
1450
 
 
1451
    def test_sha_mixed_endings(self):
 
1452
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
 
1453
        self.build_tree_contents([('foo', text)])
 
1454
        expected_sha = osutils.sha_string(text)
 
1455
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
 
1456
 
 
1457
 
 
1458
_debug_text = \
 
1459
r'''# Copyright (C) 2005, 2006 Canonical Ltd
 
1460
#
 
1461
# This program is free software; you can redistribute it and/or modify
 
1462
# it under the terms of the GNU General Public License as published by
 
1463
# the Free Software Foundation; either version 2 of the License, or
 
1464
# (at your option) any later version.
 
1465
#
 
1466
# This program is distributed in the hope that it will be useful,
 
1467
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
1468
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
1469
# GNU General Public License for more details.
 
1470
#
 
1471
# You should have received a copy of the GNU General Public License
 
1472
# along with this program; if not, write to the Free Software
 
1473
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
1474
 
 
1475
 
 
1476
# NOTE: If update these, please also update the help for global-options in
 
1477
#       bzrlib/help_topics/__init__.py
 
1478
 
 
1479
debug_flags = set()
 
1480
"""Set of flags that enable different debug behaviour.
 
1481
 
 
1482
These are set with eg ``-Dlock`` on the bzr command line.
 
1483
 
 
1484
Options include:
 
1485
 
 
1486
 * auth - show authentication sections used
 
1487
 * error - show stack traces for all top level exceptions
 
1488
 * evil - capture call sites that do expensive or badly-scaling operations.
 
1489
 * fetch - trace history copying between repositories
 
1490
 * graph - trace graph traversal information
 
1491
 * hashcache - log every time a working file is read to determine its hash
 
1492
 * hooks - trace hook execution
 
1493
 * hpss - trace smart protocol requests and responses
 
1494
 * http - trace http connections, requests and responses
 
1495
 * index - trace major index operations
 
1496
 * knit - trace knit operations
 
1497
 * lock - trace when lockdir locks are taken or released
 
1498
 * merge - emit information for debugging merges
 
1499
 * pack - emit information about pack operations
 
1500
 
 
1501
"""
 
1502
'''
 
1503
 
 
1504
 
 
1505
class TestResourceLoading(TestCaseInTempDir):
 
1506
 
 
1507
    def test_resource_string(self):
 
1508
        # test resource in bzrlib
 
1509
        text = osutils.resource_string('bzrlib', 'debug.py')
 
1510
        self.assertEquals(_debug_text, text)
 
1511
        # test resource under bzrlib
 
1512
        text = osutils.resource_string('bzrlib.ui', 'text.py')
 
1513
        self.assertContainsRe(text, "class TextUIFactory")
 
1514
        # test unsupported package
 
1515
        self.assertRaises(errors.BzrError, osutils.resource_string, 'zzzz',
 
1516
            'yyy.xx')
 
1517
        # test unknown resource
 
1518
        self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')