/brz/remove-bazaar

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

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2009-03-30 05:50:28 UTC
  • mfrom: (4183.5.9 commit-uses-ric)
  • Revision ID: pqm@pqm.ubuntu.com-20090330055028-lhmncpzf7ebkd2yc
(robertc) Teach commit to use record_iter_changes for some commits.
        (Robert Collins)

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