~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: Martin Pool
  • Date: 2005-05-09 04:38:31 UTC
  • Revision ID: mbp@sourcefrog.net-20050509043831-d45f7832b7d4d5b0
- better message when refusing to add symlinks (from mpe)

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 TestSplitLines(TestCase):
760
 
 
761
 
    def test_split_unicode(self):
762
 
        self.assertEqual([u'foo\n', u'bar\xae'],
763
 
                         osutils.split_lines(u'foo\nbar\xae'))
764
 
        self.assertEqual([u'foo\n', u'bar\xae\n'],
765
 
                         osutils.split_lines(u'foo\nbar\xae\n'))
766
 
 
767
 
    def test_split_with_carriage_returns(self):
768
 
        self.assertEqual(['foo\rbar\n'],
769
 
                         osutils.split_lines('foo\rbar\n'))
770
 
 
771
 
 
772
 
class TestWalkDirs(TestCaseInTempDir):
773
 
 
774
 
    def test_walkdirs(self):
775
 
        tree = [
776
 
            '.bzr',
777
 
            '0file',
778
 
            '1dir/',
779
 
            '1dir/0file',
780
 
            '1dir/1dir/',
781
 
            '2file'
782
 
            ]
783
 
        self.build_tree(tree)
784
 
        expected_dirblocks = [
785
 
                (('', '.'),
786
 
                 [('0file', '0file', 'file'),
787
 
                  ('1dir', '1dir', 'directory'),
788
 
                  ('2file', '2file', 'file'),
789
 
                 ]
790
 
                ),
791
 
                (('1dir', './1dir'),
792
 
                 [('1dir/0file', '0file', 'file'),
793
 
                  ('1dir/1dir', '1dir', 'directory'),
794
 
                 ]
795
 
                ),
796
 
                (('1dir/1dir', './1dir/1dir'),
797
 
                 [
798
 
                 ]
799
 
                ),
800
 
            ]
801
 
        result = []
802
 
        found_bzrdir = False
803
 
        for dirdetail, dirblock in osutils.walkdirs('.'):
804
 
            if len(dirblock) and dirblock[0][1] == '.bzr':
805
 
                # this tests the filtering of selected paths
806
 
                found_bzrdir = True
807
 
                del dirblock[0]
808
 
            result.append((dirdetail, dirblock))
809
 
 
810
 
        self.assertTrue(found_bzrdir)
811
 
        self.assertEqual(expected_dirblocks,
812
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
813
 
        # you can search a subdir only, with a supplied prefix.
814
 
        result = []
815
 
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
816
 
            result.append(dirblock)
817
 
        self.assertEqual(expected_dirblocks[1:],
818
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
819
 
 
820
 
    def test__walkdirs_utf8(self):
821
 
        tree = [
822
 
            '.bzr',
823
 
            '0file',
824
 
            '1dir/',
825
 
            '1dir/0file',
826
 
            '1dir/1dir/',
827
 
            '2file'
828
 
            ]
829
 
        self.build_tree(tree)
830
 
        expected_dirblocks = [
831
 
                (('', '.'),
832
 
                 [('0file', '0file', 'file'),
833
 
                  ('1dir', '1dir', 'directory'),
834
 
                  ('2file', '2file', 'file'),
835
 
                 ]
836
 
                ),
837
 
                (('1dir', './1dir'),
838
 
                 [('1dir/0file', '0file', 'file'),
839
 
                  ('1dir/1dir', '1dir', 'directory'),
840
 
                 ]
841
 
                ),
842
 
                (('1dir/1dir', './1dir/1dir'),
843
 
                 [
844
 
                 ]
845
 
                ),
846
 
            ]
847
 
        result = []
848
 
        found_bzrdir = False
849
 
        for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
850
 
            if len(dirblock) and dirblock[0][1] == '.bzr':
851
 
                # this tests the filtering of selected paths
852
 
                found_bzrdir = True
853
 
                del dirblock[0]
854
 
            result.append((dirdetail, dirblock))
855
 
 
856
 
        self.assertTrue(found_bzrdir)
857
 
        self.assertEqual(expected_dirblocks,
858
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
859
 
        # you can search a subdir only, with a supplied prefix.
860
 
        result = []
861
 
        for dirblock in osutils.walkdirs('./1dir', '1dir'):
862
 
            result.append(dirblock)
863
 
        self.assertEqual(expected_dirblocks[1:],
864
 
            [(dirinfo, [line[0:3] for line in block]) for dirinfo, block in result])
865
 
 
866
 
    def _filter_out_stat(self, result):
867
 
        """Filter out the stat value from the walkdirs result"""
868
 
        for dirdetail, dirblock in result:
869
 
            new_dirblock = []
870
 
            for info in dirblock:
871
 
                # Ignore info[3] which is the stat
872
 
                new_dirblock.append((info[0], info[1], info[2], info[4]))
873
 
            dirblock[:] = new_dirblock
874
 
 
875
 
    def _save_platform_info(self):
876
 
        cur_winver = win32utils.winver
877
 
        cur_fs_enc = osutils._fs_enc
878
 
        cur_dir_reader = osutils._selected_dir_reader
879
 
        def restore():
880
 
            win32utils.winver = cur_winver
881
 
            osutils._fs_enc = cur_fs_enc
882
 
            osutils._selected_dir_reader = cur_dir_reader
883
 
        self.addCleanup(restore)
884
 
 
885
 
    def assertReadFSDirIs(self, expected):
886
 
        """Assert the right implementation for _walkdirs_utf8 is chosen."""
887
 
        # Force it to redetect
888
 
        osutils._selected_dir_reader = None
889
 
        # Nothing to list, but should still trigger the selection logic
890
 
        self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
891
 
        self.assertIsInstance(osutils._selected_dir_reader, expected)
892
 
 
893
 
    def test_force_walkdirs_utf8_fs_utf8(self):
894
 
        self.requireFeature(UTF8DirReaderFeature)
895
 
        self._save_platform_info()
896
 
        win32utils.winver = None # Avoid the win32 detection code
897
 
        osutils._fs_enc = 'UTF-8'
898
 
        self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
899
 
 
900
 
    def test_force_walkdirs_utf8_fs_ascii(self):
901
 
        self.requireFeature(UTF8DirReaderFeature)
902
 
        self._save_platform_info()
903
 
        win32utils.winver = None # Avoid the win32 detection code
904
 
        osutils._fs_enc = 'US-ASCII'
905
 
        self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
906
 
 
907
 
    def test_force_walkdirs_utf8_fs_ANSI(self):
908
 
        self.requireFeature(UTF8DirReaderFeature)
909
 
        self._save_platform_info()
910
 
        win32utils.winver = None # Avoid the win32 detection code
911
 
        osutils._fs_enc = 'ANSI_X3.4-1968'
912
 
        self.assertReadFSDirIs(UTF8DirReaderFeature.reader)
913
 
 
914
 
    def test_force_walkdirs_utf8_fs_latin1(self):
915
 
        self._save_platform_info()
916
 
        win32utils.winver = None # Avoid the win32 detection code
917
 
        osutils._fs_enc = 'latin1'
918
 
        self.assertReadFSDirIs(osutils.UnicodeDirReader)
919
 
 
920
 
    def test_force_walkdirs_utf8_nt(self):
921
 
        # Disabled because the thunk of the whole walkdirs api is disabled.
922
 
        self.requireFeature(Win32ReadDirFeature)
923
 
        self._save_platform_info()
924
 
        win32utils.winver = 'Windows NT'
925
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
926
 
        self.assertReadFSDirIs(Win32ReadDir)
927
 
 
928
 
    def test_force_walkdirs_utf8_98(self):
929
 
        self.requireFeature(Win32ReadDirFeature)
930
 
        self._save_platform_info()
931
 
        win32utils.winver = 'Windows 98'
932
 
        self.assertReadFSDirIs(osutils.UnicodeDirReader)
933
 
 
934
 
    def test_unicode_walkdirs(self):
935
 
        """Walkdirs should always return unicode paths."""
936
 
        name0 = u'0file-\xb6'
937
 
        name1 = u'1dir-\u062c\u0648'
938
 
        name2 = u'2file-\u0633'
939
 
        tree = [
940
 
            name0,
941
 
            name1 + '/',
942
 
            name1 + '/' + name0,
943
 
            name1 + '/' + name1 + '/',
944
 
            name2,
945
 
            ]
946
 
        try:
947
 
            self.build_tree(tree)
948
 
        except UnicodeError:
949
 
            raise TestSkipped('Could not represent Unicode chars'
950
 
                              ' in current encoding.')
951
 
        expected_dirblocks = [
952
 
                ((u'', u'.'),
953
 
                 [(name0, name0, 'file', './' + name0),
954
 
                  (name1, name1, 'directory', './' + name1),
955
 
                  (name2, name2, 'file', './' + name2),
956
 
                 ]
957
 
                ),
958
 
                ((name1, './' + name1),
959
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1
960
 
                                                        + '/' + name0),
961
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1
962
 
                                                            + '/' + name1),
963
 
                 ]
964
 
                ),
965
 
                ((name1 + '/' + name1, './' + name1 + '/' + name1),
966
 
                 [
967
 
                 ]
968
 
                ),
969
 
            ]
970
 
        result = list(osutils.walkdirs('.'))
971
 
        self._filter_out_stat(result)
972
 
        self.assertEqual(expected_dirblocks, result)
973
 
        result = list(osutils.walkdirs(u'./'+name1, name1))
974
 
        self._filter_out_stat(result)
975
 
        self.assertEqual(expected_dirblocks[1:], result)
976
 
 
977
 
    def test_unicode__walkdirs_utf8(self):
978
 
        """Walkdirs_utf8 should always return utf8 paths.
979
 
 
980
 
        The abspath portion might be in unicode or utf-8
981
 
        """
982
 
        name0 = u'0file-\xb6'
983
 
        name1 = u'1dir-\u062c\u0648'
984
 
        name2 = u'2file-\u0633'
985
 
        tree = [
986
 
            name0,
987
 
            name1 + '/',
988
 
            name1 + '/' + name0,
989
 
            name1 + '/' + name1 + '/',
990
 
            name2,
991
 
            ]
992
 
        try:
993
 
            self.build_tree(tree)
994
 
        except UnicodeError:
995
 
            raise TestSkipped('Could not represent Unicode chars'
996
 
                              ' in current encoding.')
997
 
        name0 = name0.encode('utf8')
998
 
        name1 = name1.encode('utf8')
999
 
        name2 = name2.encode('utf8')
1000
 
 
1001
 
        expected_dirblocks = [
1002
 
                (('', '.'),
1003
 
                 [(name0, name0, 'file', './' + name0),
1004
 
                  (name1, name1, 'directory', './' + name1),
1005
 
                  (name2, name2, 'file', './' + name2),
1006
 
                 ]
1007
 
                ),
1008
 
                ((name1, './' + name1),
1009
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1
1010
 
                                                        + '/' + name0),
1011
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1
1012
 
                                                            + '/' + name1),
1013
 
                 ]
1014
 
                ),
1015
 
                ((name1 + '/' + name1, './' + name1 + '/' + name1),
1016
 
                 [
1017
 
                 ]
1018
 
                ),
1019
 
            ]
1020
 
        result = []
1021
 
        # For ease in testing, if walkdirs_utf8 returns Unicode, assert that
1022
 
        # all abspaths are Unicode, and encode them back into utf8.
1023
 
        for dirdetail, dirblock in osutils._walkdirs_utf8('.'):
1024
 
            self.assertIsInstance(dirdetail[0], str)
1025
 
            if isinstance(dirdetail[1], unicode):
1026
 
                dirdetail = (dirdetail[0], dirdetail[1].encode('utf8'))
1027
 
                dirblock = [list(info) for info in dirblock]
1028
 
                for info in dirblock:
1029
 
                    self.assertIsInstance(info[4], unicode)
1030
 
                    info[4] = info[4].encode('utf8')
1031
 
            new_dirblock = []
1032
 
            for info in dirblock:
1033
 
                self.assertIsInstance(info[0], str)
1034
 
                self.assertIsInstance(info[1], str)
1035
 
                self.assertIsInstance(info[4], str)
1036
 
                # Remove the stat information
1037
 
                new_dirblock.append((info[0], info[1], info[2], info[4]))
1038
 
            result.append((dirdetail, new_dirblock))
1039
 
        self.assertEqual(expected_dirblocks, result)
1040
 
 
1041
 
    def test__walkdirs_utf8_with_unicode_fs(self):
1042
 
        """UnicodeDirReader should be a safe fallback everywhere
1043
 
 
1044
 
        The abspath portion should be in unicode
1045
 
        """
1046
 
        # Use the unicode reader. TODO: split into driver-and-driven unit
1047
 
        # tests.
1048
 
        self._save_platform_info()
1049
 
        osutils._selected_dir_reader = osutils.UnicodeDirReader()
1050
 
        name0u = u'0file-\xb6'
1051
 
        name1u = u'1dir-\u062c\u0648'
1052
 
        name2u = u'2file-\u0633'
1053
 
        tree = [
1054
 
            name0u,
1055
 
            name1u + '/',
1056
 
            name1u + '/' + name0u,
1057
 
            name1u + '/' + name1u + '/',
1058
 
            name2u,
1059
 
            ]
1060
 
        try:
1061
 
            self.build_tree(tree)
1062
 
        except UnicodeError:
1063
 
            raise TestSkipped('Could not represent Unicode chars'
1064
 
                              ' in current encoding.')
1065
 
        name0 = name0u.encode('utf8')
1066
 
        name1 = name1u.encode('utf8')
1067
 
        name2 = name2u.encode('utf8')
1068
 
 
1069
 
        # All of the abspaths should be in unicode, all of the relative paths
1070
 
        # should be in utf8
1071
 
        expected_dirblocks = [
1072
 
                (('', '.'),
1073
 
                 [(name0, name0, 'file', './' + name0u),
1074
 
                  (name1, name1, 'directory', './' + name1u),
1075
 
                  (name2, name2, 'file', './' + name2u),
1076
 
                 ]
1077
 
                ),
1078
 
                ((name1, './' + name1u),
1079
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
1080
 
                                                        + '/' + name0u),
1081
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
1082
 
                                                            + '/' + name1u),
1083
 
                 ]
1084
 
                ),
1085
 
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
1086
 
                 [
1087
 
                 ]
1088
 
                ),
1089
 
            ]
1090
 
        result = list(osutils._walkdirs_utf8('.'))
1091
 
        self._filter_out_stat(result)
1092
 
        self.assertEqual(expected_dirblocks, result)
1093
 
 
1094
 
    def test__walkdirs_utf8_win32readdir(self):
1095
 
        self.requireFeature(Win32ReadDirFeature)
1096
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1097
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1098
 
        self._save_platform_info()
1099
 
        osutils._selected_dir_reader = Win32ReadDir()
1100
 
        name0u = u'0file-\xb6'
1101
 
        name1u = u'1dir-\u062c\u0648'
1102
 
        name2u = u'2file-\u0633'
1103
 
        tree = [
1104
 
            name0u,
1105
 
            name1u + '/',
1106
 
            name1u + '/' + name0u,
1107
 
            name1u + '/' + name1u + '/',
1108
 
            name2u,
1109
 
            ]
1110
 
        self.build_tree(tree)
1111
 
        name0 = name0u.encode('utf8')
1112
 
        name1 = name1u.encode('utf8')
1113
 
        name2 = name2u.encode('utf8')
1114
 
 
1115
 
        # All of the abspaths should be in unicode, all of the relative paths
1116
 
        # should be in utf8
1117
 
        expected_dirblocks = [
1118
 
                (('', '.'),
1119
 
                 [(name0, name0, 'file', './' + name0u),
1120
 
                  (name1, name1, 'directory', './' + name1u),
1121
 
                  (name2, name2, 'file', './' + name2u),
1122
 
                 ]
1123
 
                ),
1124
 
                ((name1, './' + name1u),
1125
 
                 [(name1 + '/' + name0, name0, 'file', './' + name1u
1126
 
                                                        + '/' + name0u),
1127
 
                  (name1 + '/' + name1, name1, 'directory', './' + name1u
1128
 
                                                            + '/' + name1u),
1129
 
                 ]
1130
 
                ),
1131
 
                ((name1 + '/' + name1, './' + name1u + '/' + name1u),
1132
 
                 [
1133
 
                 ]
1134
 
                ),
1135
 
            ]
1136
 
        result = list(osutils._walkdirs_utf8(u'.'))
1137
 
        self._filter_out_stat(result)
1138
 
        self.assertEqual(expected_dirblocks, result)
1139
 
 
1140
 
    def assertStatIsCorrect(self, path, win32stat):
1141
 
        os_stat = os.stat(path)
1142
 
        self.assertEqual(os_stat.st_size, win32stat.st_size)
1143
 
        self.assertAlmostEqual(os_stat.st_mtime, win32stat.st_mtime, places=4)
1144
 
        self.assertAlmostEqual(os_stat.st_ctime, win32stat.st_ctime, places=4)
1145
 
        self.assertAlmostEqual(os_stat.st_atime, win32stat.st_atime, places=4)
1146
 
        self.assertEqual(os_stat.st_dev, win32stat.st_dev)
1147
 
        self.assertEqual(os_stat.st_ino, win32stat.st_ino)
1148
 
        self.assertEqual(os_stat.st_mode, win32stat.st_mode)
1149
 
 
1150
 
    def test__walkdirs_utf_win32_find_file_stat_file(self):
1151
 
        """make sure our Stat values are valid"""
1152
 
        self.requireFeature(Win32ReadDirFeature)
1153
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1154
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1155
 
        name0u = u'0file-\xb6'
1156
 
        name0 = name0u.encode('utf8')
1157
 
        self.build_tree([name0u])
1158
 
        # I hate to sleep() here, but I'm trying to make the ctime different
1159
 
        # from the mtime
1160
 
        time.sleep(2)
1161
 
        f = open(name0u, 'ab')
1162
 
        try:
1163
 
            f.write('just a small update')
1164
 
        finally:
1165
 
            f.close()
1166
 
 
1167
 
        result = Win32ReadDir().read_dir('', u'.')
1168
 
        entry = result[0]
1169
 
        self.assertEqual((name0, name0, 'file'), entry[:3])
1170
 
        self.assertEqual(u'./' + name0u, entry[4])
1171
 
        self.assertStatIsCorrect(entry[4], entry[3])
1172
 
        self.assertNotEqual(entry[3].st_mtime, entry[3].st_ctime)
1173
 
 
1174
 
    def test__walkdirs_utf_win32_find_file_stat_directory(self):
1175
 
        """make sure our Stat values are valid"""
1176
 
        self.requireFeature(Win32ReadDirFeature)
1177
 
        self.requireFeature(tests.UnicodeFilenameFeature)
1178
 
        from bzrlib._walkdirs_win32 import Win32ReadDir
1179
 
        name0u = u'0dir-\u062c\u0648'
1180
 
        name0 = name0u.encode('utf8')
1181
 
        self.build_tree([name0u + '/'])
1182
 
 
1183
 
        result = Win32ReadDir().read_dir('', u'.')
1184
 
        entry = result[0]
1185
 
        self.assertEqual((name0, name0, 'directory'), entry[:3])
1186
 
        self.assertEqual(u'./' + name0u, entry[4])
1187
 
        self.assertStatIsCorrect(entry[4], entry[3])
1188
 
 
1189
 
    def assertPathCompare(self, path_less, path_greater):
1190
 
        """check that path_less and path_greater compare correctly."""
1191
 
        self.assertEqual(0, osutils.compare_paths_prefix_order(
1192
 
            path_less, path_less))
1193
 
        self.assertEqual(0, osutils.compare_paths_prefix_order(
1194
 
            path_greater, path_greater))
1195
 
        self.assertEqual(-1, osutils.compare_paths_prefix_order(
1196
 
            path_less, path_greater))
1197
 
        self.assertEqual(1, osutils.compare_paths_prefix_order(
1198
 
            path_greater, path_less))
1199
 
 
1200
 
    def test_compare_paths_prefix_order(self):
1201
 
        # root before all else
1202
 
        self.assertPathCompare("/", "/a")
1203
 
        # alpha within a dir
1204
 
        self.assertPathCompare("/a", "/b")
1205
 
        self.assertPathCompare("/b", "/z")
1206
 
        # high dirs before lower.
1207
 
        self.assertPathCompare("/z", "/a/a")
1208
 
        # except if the deeper dir should be output first
1209
 
        self.assertPathCompare("/a/b/c", "/d/g")
1210
 
        # lexical betwen dirs of the same height
1211
 
        self.assertPathCompare("/a/z", "/z/z")
1212
 
        self.assertPathCompare("/a/c/z", "/a/d/e")
1213
 
 
1214
 
        # this should also be consistent for no leading / paths
1215
 
        # root before all else
1216
 
        self.assertPathCompare("", "a")
1217
 
        # alpha within a dir
1218
 
        self.assertPathCompare("a", "b")
1219
 
        self.assertPathCompare("b", "z")
1220
 
        # high dirs before lower.
1221
 
        self.assertPathCompare("z", "a/a")
1222
 
        # except if the deeper dir should be output first
1223
 
        self.assertPathCompare("a/b/c", "d/g")
1224
 
        # lexical betwen dirs of the same height
1225
 
        self.assertPathCompare("a/z", "z/z")
1226
 
        self.assertPathCompare("a/c/z", "a/d/e")
1227
 
 
1228
 
    def test_path_prefix_sorting(self):
1229
 
        """Doing a sort on path prefix should match our sample data."""
1230
 
        original_paths = [
1231
 
            'a',
1232
 
            'a/b',
1233
 
            'a/b/c',
1234
 
            'b',
1235
 
            'b/c',
1236
 
            'd',
1237
 
            'd/e',
1238
 
            'd/e/f',
1239
 
            'd/f',
1240
 
            'd/g',
1241
 
            'g',
1242
 
            ]
1243
 
 
1244
 
        dir_sorted_paths = [
1245
 
            'a',
1246
 
            'b',
1247
 
            'd',
1248
 
            'g',
1249
 
            'a/b',
1250
 
            'a/b/c',
1251
 
            'b/c',
1252
 
            'd/e',
1253
 
            'd/f',
1254
 
            'd/g',
1255
 
            'd/e/f',
1256
 
            ]
1257
 
 
1258
 
        self.assertEqual(
1259
 
            dir_sorted_paths,
1260
 
            sorted(original_paths, key=osutils.path_prefix_key))
1261
 
        # using the comparison routine shoudl work too:
1262
 
        self.assertEqual(
1263
 
            dir_sorted_paths,
1264
 
            sorted(original_paths, cmp=osutils.compare_paths_prefix_order))
1265
 
 
1266
 
 
1267
 
class TestCopyTree(TestCaseInTempDir):
1268
 
    
1269
 
    def test_copy_basic_tree(self):
1270
 
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
1271
 
        osutils.copy_tree('source', 'target')
1272
 
        self.assertEqual(['a', 'b'], sorted(os.listdir('target')))
1273
 
        self.assertEqual(['c'], os.listdir('target/b'))
1274
 
 
1275
 
    def test_copy_tree_target_exists(self):
1276
 
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c',
1277
 
                         'target/'])
1278
 
        osutils.copy_tree('source', 'target')
1279
 
        self.assertEqual(['a', 'b'], sorted(os.listdir('target')))
1280
 
        self.assertEqual(['c'], os.listdir('target/b'))
1281
 
 
1282
 
    def test_copy_tree_symlinks(self):
1283
 
        self.requireFeature(SymlinkFeature)
1284
 
        self.build_tree(['source/'])
1285
 
        os.symlink('a/generic/path', 'source/lnk')
1286
 
        osutils.copy_tree('source', 'target')
1287
 
        self.assertEqual(['lnk'], os.listdir('target'))
1288
 
        self.assertEqual('a/generic/path', os.readlink('target/lnk'))
1289
 
 
1290
 
    def test_copy_tree_handlers(self):
1291
 
        processed_files = []
1292
 
        processed_links = []
1293
 
        def file_handler(from_path, to_path):
1294
 
            processed_files.append(('f', from_path, to_path))
1295
 
        def dir_handler(from_path, to_path):
1296
 
            processed_files.append(('d', from_path, to_path))
1297
 
        def link_handler(from_path, to_path):
1298
 
            processed_links.append((from_path, to_path))
1299
 
        handlers = {'file':file_handler,
1300
 
                    'directory':dir_handler,
1301
 
                    'symlink':link_handler,
1302
 
                   }
1303
 
 
1304
 
        self.build_tree(['source/', 'source/a', 'source/b/', 'source/b/c'])
1305
 
        if osutils.has_symlinks():
1306
 
            os.symlink('a/generic/path', 'source/lnk')
1307
 
        osutils.copy_tree('source', 'target', handlers=handlers)
1308
 
 
1309
 
        self.assertEqual([('d', 'source', 'target'),
1310
 
                          ('f', 'source/a', 'target/a'),
1311
 
                          ('d', 'source/b', 'target/b'),
1312
 
                          ('f', 'source/b/c', 'target/b/c'),
1313
 
                         ], processed_files)
1314
 
        self.failIfExists('target')
1315
 
        if osutils.has_symlinks():
1316
 
            self.assertEqual([('source/lnk', 'target/lnk')], processed_links)
1317
 
 
1318
 
 
1319
 
#class TestTerminalEncoding has been moved to test_osutils_encodings.py
1320
 
# [bialix] 2006/12/26
1321
 
 
1322
 
 
1323
 
class TestSetUnsetEnv(TestCase):
1324
 
    """Test updating the environment"""
1325
 
 
1326
 
    def setUp(self):
1327
 
        super(TestSetUnsetEnv, self).setUp()
1328
 
 
1329
 
        self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'),
1330
 
                         'Environment was not cleaned up properly.'
1331
 
                         ' Variable BZR_TEST_ENV_VAR should not exist.')
1332
 
        def cleanup():
1333
 
            if 'BZR_TEST_ENV_VAR' in os.environ:
1334
 
                del os.environ['BZR_TEST_ENV_VAR']
1335
 
 
1336
 
        self.addCleanup(cleanup)
1337
 
 
1338
 
    def test_set(self):
1339
 
        """Test that we can set an env variable"""
1340
 
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
1341
 
        self.assertEqual(None, old)
1342
 
        self.assertEqual('foo', os.environ.get('BZR_TEST_ENV_VAR'))
1343
 
 
1344
 
    def test_double_set(self):
1345
 
        """Test that we get the old value out"""
1346
 
        osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
1347
 
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'bar')
1348
 
        self.assertEqual('foo', old)
1349
 
        self.assertEqual('bar', os.environ.get('BZR_TEST_ENV_VAR'))
1350
 
 
1351
 
    def test_unicode(self):
1352
 
        """Environment can only contain plain strings
1353
 
        
1354
 
        So Unicode strings must be encoded.
1355
 
        """
1356
 
        uni_val, env_val = probe_unicode_in_user_encoding()
1357
 
        if uni_val is None:
1358
 
            raise TestSkipped('Cannot find a unicode character that works in'
1359
 
                              ' encoding %s' % (osutils.get_user_encoding(),))
1360
 
 
1361
 
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', uni_val)
1362
 
        self.assertEqual(env_val, os.environ.get('BZR_TEST_ENV_VAR'))
1363
 
 
1364
 
    def test_unset(self):
1365
 
        """Test that passing None will remove the env var"""
1366
 
        osutils.set_or_unset_env('BZR_TEST_ENV_VAR', 'foo')
1367
 
        old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', None)
1368
 
        self.assertEqual('foo', old)
1369
 
        self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'))
1370
 
        self.failIf('BZR_TEST_ENV_VAR' in os.environ)
1371
 
 
1372
 
 
1373
 
class TestLocalTimeOffset(TestCase):
1374
 
 
1375
 
    def test_local_time_offset(self):
1376
 
        """Test that local_time_offset() returns a sane value."""
1377
 
        offset = osutils.local_time_offset()
1378
 
        self.assertTrue(isinstance(offset, int))
1379
 
        # Test that the offset is no more than a eighteen hours in
1380
 
        # either direction.
1381
 
        # Time zone handling is system specific, so it is difficult to
1382
 
        # do more specific tests, but a value outside of this range is
1383
 
        # probably wrong.
1384
 
        eighteen_hours = 18 * 3600
1385
 
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
1386
 
 
1387
 
    def test_local_time_offset_with_timestamp(self):
1388
 
        """Test that local_time_offset() works with a timestamp."""
1389
 
        offset = osutils.local_time_offset(1000000000.1234567)
1390
 
        self.assertTrue(isinstance(offset, int))
1391
 
        eighteen_hours = 18 * 3600
1392
 
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
1393
 
 
1394
 
 
1395
 
class TestShaFileByName(TestCaseInTempDir):
1396
 
 
1397
 
    def test_sha_empty(self):
1398
 
        self.build_tree_contents([('foo', '')])
1399
 
        expected_sha = osutils.sha_string('')
1400
 
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1401
 
 
1402
 
    def test_sha_mixed_endings(self):
1403
 
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
1404
 
        self.build_tree_contents([('foo', text)])
1405
 
        expected_sha = osutils.sha_string(text)
1406
 
        self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1407
 
 
1408
 
 
1409
 
_debug_text = \
1410
 
r'''# Copyright (C) 2005, 2006 Canonical Ltd
1411
 
#
1412
 
# This program is free software; you can redistribute it and/or modify
1413
 
# it under the terms of the GNU General Public License as published by
1414
 
# the Free Software Foundation; either version 2 of the License, or
1415
 
# (at your option) any later version.
1416
 
#
1417
 
# This program is distributed in the hope that it will be useful,
1418
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
1419
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
1420
 
# GNU General Public License for more details.
1421
 
#
1422
 
# You should have received a copy of the GNU General Public License
1423
 
# along with this program; if not, write to the Free Software
1424
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
1425
 
 
1426
 
 
1427
 
# NOTE: If update these, please also update the help for global-options in
1428
 
#       bzrlib/help_topics/__init__.py
1429
 
 
1430
 
debug_flags = set()
1431
 
"""Set of flags that enable different debug behaviour.
1432
 
 
1433
 
These are set with eg ``-Dlock`` on the bzr command line.
1434
 
 
1435
 
Options include:
1436
 
 
1437
 
 * auth - show authentication sections used
1438
 
 * error - show stack traces for all top level exceptions
1439
 
 * evil - capture call sites that do expensive or badly-scaling operations.
1440
 
 * fetch - trace history copying between repositories
1441
 
 * graph - trace graph traversal information
1442
 
 * hashcache - log every time a working file is read to determine its hash
1443
 
 * hooks - trace hook execution
1444
 
 * hpss - trace smart protocol requests and responses
1445
 
 * http - trace http connections, requests and responses
1446
 
 * index - trace major index operations
1447
 
 * knit - trace knit operations
1448
 
 * lock - trace when lockdir locks are taken or released
1449
 
 * merge - emit information for debugging merges
1450
 
 * pack - emit information about pack operations
1451
 
 
1452
 
"""
1453
 
'''
1454
 
 
1455
 
 
1456
 
class TestResourceLoading(TestCaseInTempDir):
1457
 
 
1458
 
    def test_resource_string(self):
1459
 
        # test resource in bzrlib
1460
 
        text = osutils.resource_string('bzrlib', 'debug.py')
1461
 
        self.assertEquals(_debug_text, text)
1462
 
        # test resource under bzrlib
1463
 
        text = osutils.resource_string('bzrlib.ui', 'text.py')
1464
 
        self.assertContainsRe(text, "class TextUIFactory")
1465
 
        # test unsupported package
1466
 
        self.assertRaises(errors.BzrError, osutils.resource_string, 'zzzz',
1467
 
            'yyy.xx')
1468
 
        # test unknown resource
1469
 
        self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')