~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: John Arbash Meinel
  • Date: 2008-08-18 22:34:21 UTC
  • mto: (3606.5.6 1.6)
  • mto: This revision was merged to the branch mainline in revision 3641.
  • Revision ID: john@arbash-meinel.com-20080818223421-todjny24vj4faj4t
Add tests for the fetching behavior.

The proper parameter passed is 'unordered' add an assert for it, and
fix callers that were passing 'unsorted' instead.
Add tests that we make the right get_record_stream call based
on the value of _fetch_uses_deltas.
Fix the fetch request for signatures.

Show diffs side-by-side

added added

removed removed

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