~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: John Arbash Meinel
  • Date: 2008-08-25 03:43:42 UTC
  • mto: This revision was merged to the branch mainline in revision 3688.
  • Revision ID: john@arbash-meinel.com-20080825034342-owq0858uk1wp2q0l
the new btree index doesn't have 'absent' keys in its _nodes

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