/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: John Arbash Meinel
  • Date: 2008-09-02 19:15:57 UTC
  • mto: This revision was merged to the branch mainline in revision 3680.
  • Revision ID: john@arbash-meinel.com-20080902191557-s88j283bmnnnlvhr
PQM's pyrex needs a Py_ssize_t typedef.

Show diffs side-by-side

added added

removed removed

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