/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: John Arbash Meinel
  • Date: 2008-12-11 02:18:59 UTC
  • mto: This revision was merged to the branch mainline in revision 3895.
  • Revision ID: john@arbash-meinel.com-20081211021859-3ds8cwdqiq387t83
A Pyrex extension is about 5x faster than the fastest python code I could write.

Seems worth having after all.

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