~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: Martin Pool
  • Date: 2009-03-03 01:45:32 UTC
  • mto: (4070.4.5 gnu-changelog)
  • mto: This revision was merged to the branch mainline in revision 4081.
  • Revision ID: mbp@sourcefrog.net-20090303014532-d223fxy97cb1og8f
Recommend setting timestamp in BranchBuilder

Show diffs side-by-side

added added

removed removed

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