/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: Canonical.com Patch Queue Manager
  • Date: 2008-10-01 06:49:10 UTC
  • mfrom: (3731.2.7 hpss-call-count)
  • Revision ID: pqm@pqm.ubuntu.com-20081001064910-typ6d7diwd53z6iv
Extend -Dhpss flag to report HPSS call count on stderr. (Andrew
        Bennetts)

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