13
13
# You should have received a copy of the GNU General Public License
14
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
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
17
"""Tests for the osutils wrapper."""
19
from cStringIO import StringIO
29
26
from bzrlib import (
31
from bzrlib.errors import BzrBadParameterNotUnicode, InvalidURL
32
from bzrlib.osutils import (
34
is_inside_or_parent_of_any,
38
37
from bzrlib.tests import (
43
from bzrlib.tests.scenarios import load_tests_apply_scenarios
46
class _UTF8DirReaderFeature(features.Feature):
50
from bzrlib import _readdir_pyx
51
self.reader = _readdir_pyx.UTF8DirReader
56
def feature_name(self):
57
return 'bzrlib._readdir_pyx'
59
UTF8DirReaderFeature = features.ModuleAvailableFeature('bzrlib._readdir_pyx')
61
term_ios_feature = features.ModuleAvailableFeature('termios')
64
def _already_unicode(s):
68
def _utf8_to_unicode(s):
69
return s.decode('UTF-8')
72
def dir_reader_scenarios():
73
# For each dir reader we define:
75
# - native_to_unicode: a function converting the native_abspath as returned
76
# by DirReader.read_dir to its unicode representation
78
# UnicodeDirReader is the fallback, it should be tested on all platforms.
79
scenarios = [('unicode',
80
dict(_dir_reader_class=osutils.UnicodeDirReader,
81
_native_to_unicode=_already_unicode))]
82
# Some DirReaders are platform specific and even there they may not be
84
if UTF8DirReaderFeature.available():
85
from bzrlib import _readdir_pyx
86
scenarios.append(('utf8',
87
dict(_dir_reader_class=_readdir_pyx.UTF8DirReader,
88
_native_to_unicode=_utf8_to_unicode)))
90
if test__walkdirs_win32.win32_readdir_feature.available():
92
from bzrlib import _walkdirs_win32
95
dict(_dir_reader_class=_walkdirs_win32.Win32ReadDir,
96
_native_to_unicode=_already_unicode)))
102
load_tests = load_tests_apply_scenarios
105
class TestContainsWhitespace(tests.TestCase):
38
probe_unicode_in_user_encoding,
47
class TestOSUtils(TestCaseInTempDir):
107
49
def test_contains_whitespace(self):
108
self.assertTrue(osutils.contains_whitespace(u' '))
109
self.assertTrue(osutils.contains_whitespace(u'hello there'))
110
self.assertTrue(osutils.contains_whitespace(u'hellothere\n'))
111
self.assertTrue(osutils.contains_whitespace(u'hello\nthere'))
112
self.assertTrue(osutils.contains_whitespace(u'hello\rthere'))
113
self.assertTrue(osutils.contains_whitespace(u'hello\tthere'))
50
self.failUnless(osutils.contains_whitespace(u' '))
51
self.failUnless(osutils.contains_whitespace(u'hello there'))
52
self.failUnless(osutils.contains_whitespace(u'hellothere\n'))
53
self.failUnless(osutils.contains_whitespace(u'hello\nthere'))
54
self.failUnless(osutils.contains_whitespace(u'hello\rthere'))
55
self.failUnless(osutils.contains_whitespace(u'hello\tthere'))
115
57
# \xa0 is "Non-breaking-space" which on some python locales thinks it
116
58
# is whitespace, but we do not.
117
self.assertFalse(osutils.contains_whitespace(u''))
118
self.assertFalse(osutils.contains_whitespace(u'hellothere'))
119
self.assertFalse(osutils.contains_whitespace(u'hello\xa0there'))
122
class TestRename(tests.TestCaseInTempDir):
124
def create_file(self, filename, content):
125
f = open(filename, 'wb')
131
def _fancy_rename(self, a, b):
132
osutils.fancy_rename(a, b, rename_func=os.rename,
133
unlink_func=os.unlink)
59
self.failIf(osutils.contains_whitespace(u''))
60
self.failIf(osutils.contains_whitespace(u'hellothere'))
61
self.failIf(osutils.contains_whitespace(u'hello\xa0there'))
135
63
def test_fancy_rename(self):
136
64
# This should work everywhere
137
self.create_file('a', 'something in a\n')
138
self._fancy_rename('a', 'b')
139
self.assertPathDoesNotExist('a')
140
self.assertPathExists('b')
66
osutils.fancy_rename(a, b,
67
rename_func=os.rename,
68
unlink_func=os.unlink)
70
open('a', 'wb').write('something in a\n')
72
self.failIfExists('a')
73
self.failUnlessExists('b')
141
74
self.check_file_contents('b', 'something in a\n')
143
self.create_file('a', 'new something in a\n')
144
self._fancy_rename('b', 'a')
76
open('a', 'wb').write('new something in a\n')
146
79
self.check_file_contents('a', 'something in a\n')
148
def test_fancy_rename_fails_source_missing(self):
149
# An exception should be raised, and the target should be left in place
150
self.create_file('target', 'data in target\n')
151
self.assertRaises((IOError, OSError), self._fancy_rename,
152
'missingsource', 'target')
153
self.assertPathExists('target')
154
self.check_file_contents('target', 'data in target\n')
156
def test_fancy_rename_fails_if_source_and_target_missing(self):
157
self.assertRaises((IOError, OSError), self._fancy_rename,
158
'missingsource', 'missingtarget')
160
81
def test_rename(self):
161
82
# Rename should be semi-atomic on all platforms
162
self.create_file('a', 'something in a\n')
83
open('a', 'wb').write('something in a\n')
163
84
osutils.rename('a', 'b')
164
self.assertPathDoesNotExist('a')
165
self.assertPathExists('b')
85
self.failIfExists('a')
86
self.failUnlessExists('b')
166
87
self.check_file_contents('b', 'something in a\n')
168
self.create_file('a', 'new something in a\n')
89
open('a', 'wb').write('new something in a\n')
169
90
osutils.rename('b', 'a')
171
92
self.check_file_contents('a', 'something in a\n')
530
312
osutils.make_readonly('dangling')
531
313
osutils.make_writable('dangling')
533
def test_host_os_dereferences_symlinks(self):
534
osutils.host_os_dereferences_symlinks()
537
class TestCanonicalRelPath(tests.TestCaseInTempDir):
539
_test_needs_features = [features.CaseInsCasePresFilenameFeature]
541
def test_canonical_relpath_simple(self):
542
f = file('MixedCaseName', 'w')
544
actual = osutils.canonical_relpath(self.test_base_dir, 'mixedcasename')
545
self.assertEqual('work/MixedCaseName', actual)
547
def test_canonical_relpath_missing_tail(self):
548
os.mkdir('MixedCaseParent')
549
actual = osutils.canonical_relpath(self.test_base_dir,
550
'mixedcaseparent/nochild')
551
self.assertEqual('work/MixedCaseParent/nochild', actual)
554
class Test_CICPCanonicalRelpath(tests.TestCaseWithTransport):
556
def assertRelpath(self, expected, base, path):
557
actual = osutils._cicp_canonical_relpath(base, path)
558
self.assertEqual(expected, actual)
560
def test_simple(self):
561
self.build_tree(['MixedCaseName'])
562
base = osutils.realpath(self.get_transport('.').local_abspath('.'))
563
self.assertRelpath('MixedCaseName', base, 'mixedcAsename')
565
def test_subdir_missing_tail(self):
566
self.build_tree(['MixedCaseParent/', 'MixedCaseParent/a_child'])
567
base = osutils.realpath(self.get_transport('.').local_abspath('.'))
568
self.assertRelpath('MixedCaseParent/a_child', base,
569
'MixedCaseParent/a_child')
570
self.assertRelpath('MixedCaseParent/a_child', base,
571
'MixedCaseParent/A_Child')
572
self.assertRelpath('MixedCaseParent/not_child', base,
573
'MixedCaseParent/not_child')
575
def test_at_root_slash(self):
576
# We can't test this on Windows, because it has a 'MIN_ABS_PATHLENGTH'
578
if osutils.MIN_ABS_PATHLENGTH > 1:
579
raise tests.TestSkipped('relpath requires %d chars'
580
% osutils.MIN_ABS_PATHLENGTH)
581
self.assertRelpath('foo', '/', '/foo')
583
def test_at_root_drive(self):
584
if sys.platform != 'win32':
585
raise tests.TestNotApplicable('we can only test drive-letter relative'
586
' paths on Windows where we have drive'
589
# The specific issue is that when at the root of a drive, 'abspath'
590
# returns "C:/" or just "/". However, the code assumes that abspath
591
# always returns something like "C:/foo" or "/foo" (no trailing slash).
592
self.assertRelpath('foo', 'C:/', 'C:/foo')
593
self.assertRelpath('foo', 'X:/', 'X:/foo')
594
self.assertRelpath('foo', 'X:/', 'X://foo')
597
class TestPumpFile(tests.TestCase):
598
"""Test pumpfile method."""
601
super(TestPumpFile, self).setUp()
602
# create a test datablock
603
self.block_size = 512
604
pattern = '0123456789ABCDEF'
605
self.test_data = pattern * (3 * self.block_size / len(pattern))
606
self.test_data_len = len(self.test_data)
608
def test_bracket_block_size(self):
609
"""Read data in blocks with the requested read size bracketing the
611
# make sure test data is larger than max read size
612
self.assertTrue(self.test_data_len > self.block_size)
614
from_file = file_utils.FakeReadFile(self.test_data)
617
# read (max / 2) bytes and verify read size wasn't affected
618
num_bytes_to_read = self.block_size / 2
619
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
620
self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
621
self.assertEqual(from_file.get_read_count(), 1)
623
# read (max) bytes and verify read size wasn't affected
624
num_bytes_to_read = self.block_size
625
from_file.reset_read_count()
626
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
627
self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
628
self.assertEqual(from_file.get_read_count(), 1)
630
# read (max + 1) bytes and verify read size was limited
631
num_bytes_to_read = self.block_size + 1
632
from_file.reset_read_count()
633
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
634
self.assertEqual(from_file.get_max_read_size(), self.block_size)
635
self.assertEqual(from_file.get_read_count(), 2)
637
# finish reading the rest of the data
638
num_bytes_to_read = self.test_data_len - to_file.tell()
639
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
641
# report error if the data wasn't equal (we only report the size due
642
# to the length of the data)
643
response_data = to_file.getvalue()
644
if response_data != self.test_data:
645
message = "Data not equal. Expected %d bytes, received %d."
646
self.fail(message % (len(response_data), self.test_data_len))
648
def test_specified_size(self):
649
"""Request a transfer larger than the maximum block size and verify
650
that the maximum read doesn't exceed the block_size."""
651
# make sure test data is larger than max read size
652
self.assertTrue(self.test_data_len > self.block_size)
654
# retrieve data in blocks
655
from_file = file_utils.FakeReadFile(self.test_data)
657
osutils.pumpfile(from_file, to_file, self.test_data_len,
660
# verify read size was equal to the maximum read size
661
self.assertTrue(from_file.get_max_read_size() > 0)
662
self.assertEqual(from_file.get_max_read_size(), self.block_size)
663
self.assertEqual(from_file.get_read_count(), 3)
665
# report error if the data wasn't equal (we only report the size due
666
# to the length of the data)
667
response_data = to_file.getvalue()
668
if response_data != self.test_data:
669
message = "Data not equal. Expected %d bytes, received %d."
670
self.fail(message % (len(response_data), self.test_data_len))
672
def test_to_eof(self):
673
"""Read to end-of-file and verify that the reads are not larger than
674
the maximum read size."""
675
# make sure test data is larger than max read size
676
self.assertTrue(self.test_data_len > self.block_size)
678
# retrieve data to EOF
679
from_file = file_utils.FakeReadFile(self.test_data)
681
osutils.pumpfile(from_file, to_file, -1, self.block_size)
683
# verify read size was equal to the maximum read size
684
self.assertEqual(from_file.get_max_read_size(), self.block_size)
685
self.assertEqual(from_file.get_read_count(), 4)
687
# report error if the data wasn't equal (we only report the size due
688
# to the length of the data)
689
response_data = to_file.getvalue()
690
if response_data != self.test_data:
691
message = "Data not equal. Expected %d bytes, received %d."
692
self.fail(message % (len(response_data), self.test_data_len))
694
def test_defaults(self):
695
"""Verifies that the default arguments will read to EOF -- this
696
test verifies that any existing usages of pumpfile will not be broken
697
with this new version."""
698
# retrieve data using default (old) pumpfile method
699
from_file = file_utils.FakeReadFile(self.test_data)
701
osutils.pumpfile(from_file, to_file)
703
# report error if the data wasn't equal (we only report the size due
704
# to the length of the data)
705
response_data = to_file.getvalue()
706
if response_data != self.test_data:
707
message = "Data not equal. Expected %d bytes, received %d."
708
self.fail(message % (len(response_data), self.test_data_len))
710
def test_report_activity(self):
712
def log_activity(length, direction):
713
activity.append((length, direction))
714
from_file = StringIO(self.test_data)
716
osutils.pumpfile(from_file, to_file, buff_size=500,
717
report_activity=log_activity, direction='read')
718
self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
719
(36, 'read')], activity)
721
from_file = StringIO(self.test_data)
724
osutils.pumpfile(from_file, to_file, buff_size=500,
725
report_activity=log_activity, direction='write')
726
self.assertEqual([(500, 'write'), (500, 'write'), (500, 'write'),
727
(36, 'write')], activity)
729
# And with a limited amount of data
730
from_file = StringIO(self.test_data)
733
osutils.pumpfile(from_file, to_file, buff_size=500, read_length=1028,
734
report_activity=log_activity, direction='read')
735
self.assertEqual([(500, 'read'), (500, 'read'), (28, 'read')], activity)
739
class TestPumpStringFile(tests.TestCase):
741
def test_empty(self):
743
osutils.pump_string_file("", output)
744
self.assertEqual("", output.getvalue())
746
def test_more_than_segment_size(self):
748
osutils.pump_string_file("123456789", output, 2)
749
self.assertEqual("123456789", output.getvalue())
751
def test_segment_size(self):
753
osutils.pump_string_file("12", output, 2)
754
self.assertEqual("12", output.getvalue())
756
def test_segment_size_multiple(self):
758
osutils.pump_string_file("1234", output, 2)
759
self.assertEqual("1234", output.getvalue())
762
class TestRelpath(tests.TestCase):
764
def test_simple_relpath(self):
765
cwd = osutils.getcwd()
766
subdir = cwd + '/subdir'
767
self.assertEqual('subdir', osutils.relpath(cwd, subdir))
769
def test_deep_relpath(self):
770
cwd = osutils.getcwd()
771
subdir = cwd + '/sub/subsubdir'
772
self.assertEqual('sub/subsubdir', osutils.relpath(cwd, subdir))
774
def test_not_relative(self):
775
self.assertRaises(errors.PathNotChild,
776
osutils.relpath, 'C:/path', 'H:/path')
777
self.assertRaises(errors.PathNotChild,
778
osutils.relpath, 'C:/', 'H:/path')
781
class TestSafeUnicode(tests.TestCase):
315
def test_kind_marker(self):
316
self.assertEqual("", osutils.kind_marker("file"))
317
self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
318
self.assertEqual("@", osutils.kind_marker("symlink"))
319
self.assertRaises(errors.BzrError, osutils.kind_marker, "unknown")
322
class TestSafeUnicode(TestCase):
783
324
def test_from_ascii_string(self):
784
325
self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
1442
808
result.append((dirdetail, new_dirblock))
1443
809
self.assertEqual(expected_dirblocks, result)
1445
def test__walkdirs_utf8_with_unicode_fs(self):
1446
"""UnicodeDirReader should be a safe fallback everywhere
811
def test_unicode__walkdirs_unicode_to_utf8(self):
812
"""walkdirs_unicode_to_utf8 should be a safe fallback everywhere
1448
814
The abspath portion should be in unicode
1450
self.requireFeature(features.UnicodeFilenameFeature)
1451
# Use the unicode reader. TODO: split into driver-and-driven unit
1453
self._save_platform_info()
1454
osutils._selected_dir_reader = osutils.UnicodeDirReader()
1455
name0u = u'0file-\xb6'
1456
name1u = u'1dir-\u062c\u0648'
1457
name2u = u'2file-\u0633'
1461
name1u + '/' + name0u,
1462
name1u + '/' + name1u + '/',
1465
self.build_tree(tree)
1466
name0 = name0u.encode('utf8')
1467
name1 = name1u.encode('utf8')
1468
name2 = name2u.encode('utf8')
1470
# All of the abspaths should be in unicode, all of the relative paths
1472
expected_dirblocks = [
1474
[(name0, name0, 'file', './' + name0u),
1475
(name1, name1, 'directory', './' + name1u),
1476
(name2, name2, 'file', './' + name2u),
1479
((name1, './' + name1u),
1480
[(name1 + '/' + name0, name0, 'file', './' + name1u
1482
(name1 + '/' + name1, name1, 'directory', './' + name1u
1486
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1491
result = list(osutils._walkdirs_utf8('.'))
1492
self._filter_out_stat(result)
1493
self.assertEqual(expected_dirblocks, result)
1495
def test__walkdirs_utf8_win32readdir(self):
1496
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1497
self.requireFeature(features.UnicodeFilenameFeature)
1498
from bzrlib._walkdirs_win32 import Win32ReadDir
1499
self._save_platform_info()
1500
osutils._selected_dir_reader = Win32ReadDir()
1501
name0u = u'0file-\xb6'
1502
name1u = u'1dir-\u062c\u0648'
1503
name2u = u'2file-\u0633'
1507
name1u + '/' + name0u,
1508
name1u + '/' + name1u + '/',
1511
self.build_tree(tree)
1512
name0 = name0u.encode('utf8')
1513
name1 = name1u.encode('utf8')
1514
name2 = name2u.encode('utf8')
1516
# All of the abspaths should be in unicode, all of the relative paths
1518
expected_dirblocks = [
1520
[(name0, name0, 'file', './' + name0u),
1521
(name1, name1, 'directory', './' + name1u),
1522
(name2, name2, 'file', './' + name2u),
1525
((name1, './' + name1u),
1526
[(name1 + '/' + name0, name0, 'file', './' + name1u
1528
(name1 + '/' + name1, name1, 'directory', './' + name1u
1532
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1537
result = list(osutils._walkdirs_utf8(u'.'))
1538
self._filter_out_stat(result)
1539
self.assertEqual(expected_dirblocks, result)
1541
def assertStatIsCorrect(self, path, win32stat):
1542
os_stat = os.stat(path)
1543
self.assertEqual(os_stat.st_size, win32stat.st_size)
1544
self.assertAlmostEqual(os_stat.st_mtime, win32stat.st_mtime, places=4)
1545
self.assertAlmostEqual(os_stat.st_ctime, win32stat.st_ctime, places=4)
1546
self.assertAlmostEqual(os_stat.st_atime, win32stat.st_atime, places=4)
1547
self.assertEqual(os_stat.st_dev, win32stat.st_dev)
1548
self.assertEqual(os_stat.st_ino, win32stat.st_ino)
1549
self.assertEqual(os_stat.st_mode, win32stat.st_mode)
1551
def test__walkdirs_utf_win32_find_file_stat_file(self):
1552
"""make sure our Stat values are valid"""
1553
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1554
self.requireFeature(features.UnicodeFilenameFeature)
1555
from bzrlib._walkdirs_win32 import Win32ReadDir
1556
name0u = u'0file-\xb6'
1557
name0 = name0u.encode('utf8')
1558
self.build_tree([name0u])
1559
# I hate to sleep() here, but I'm trying to make the ctime different
1562
f = open(name0u, 'ab')
816
name0u = u'0file-\xb6'
817
name1u = u'1dir-\u062c\u0648'
818
name2u = u'2file-\u0633'
822
name1u + '/' + name0u,
823
name1u + '/' + name1u + '/',
1564
f.write('just a small update')
1568
result = Win32ReadDir().read_dir('', u'.')
1570
self.assertEqual((name0, name0, 'file'), entry[:3])
1571
self.assertEqual(u'./' + name0u, entry[4])
1572
self.assertStatIsCorrect(entry[4], entry[3])
1573
self.assertNotEqual(entry[3].st_mtime, entry[3].st_ctime)
1575
def test__walkdirs_utf_win32_find_file_stat_directory(self):
1576
"""make sure our Stat values are valid"""
1577
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1578
self.requireFeature(features.UnicodeFilenameFeature)
1579
from bzrlib._walkdirs_win32 import Win32ReadDir
1580
name0u = u'0dir-\u062c\u0648'
827
self.build_tree(tree)
829
raise TestSkipped('Could not represent Unicode chars'
830
' in current encoding.')
1581
831
name0 = name0u.encode('utf8')
1582
self.build_tree([name0u + '/'])
832
name1 = name1u.encode('utf8')
833
name2 = name2u.encode('utf8')
1584
result = Win32ReadDir().read_dir('', u'.')
1586
self.assertEqual((name0, name0, 'directory'), entry[:3])
1587
self.assertEqual(u'./' + name0u, entry[4])
1588
self.assertStatIsCorrect(entry[4], entry[3])
835
# All of the abspaths should be in unicode, all of the relative paths
837
expected_dirblocks = [
839
[(name0, name0, 'file', './' + name0u),
840
(name1, name1, 'directory', './' + name1u),
841
(name2, name2, 'file', './' + name2u),
844
((name1, './' + name1u),
845
[(name1 + '/' + name0, name0, 'file', './' + name1u
847
(name1 + '/' + name1, name1, 'directory', './' + name1u
851
((name1 + '/' + name1, './' + name1u + '/' + name1u),
856
result = list(osutils._walkdirs_unicode_to_utf8('.'))
857
self._filter_out_stat(result)
858
self.assertEqual(expected_dirblocks, result)
1590
860
def assertPathCompare(self, path_less, path_greater):
1591
861
"""check that path_less and path_greater compare correctly."""
1764
1038
old = osutils.set_or_unset_env('BZR_TEST_ENV_VAR', None)
1765
1039
self.assertEqual('foo', old)
1766
1040
self.assertEqual(None, os.environ.get('BZR_TEST_ENV_VAR'))
1767
self.assertFalse('BZR_TEST_ENV_VAR' in os.environ)
1770
class TestSizeShaFile(tests.TestCaseInTempDir):
1772
def test_sha_empty(self):
1773
self.build_tree_contents([('foo', '')])
1774
expected_sha = osutils.sha_string('')
1776
self.addCleanup(f.close)
1777
size, sha = osutils.size_sha_file(f)
1778
self.assertEqual(0, size)
1779
self.assertEqual(expected_sha, sha)
1781
def test_sha_mixed_endings(self):
1782
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1783
self.build_tree_contents([('foo', text)])
1784
expected_sha = osutils.sha_string(text)
1785
f = open('foo', 'rb')
1786
self.addCleanup(f.close)
1787
size, sha = osutils.size_sha_file(f)
1788
self.assertEqual(38, size)
1789
self.assertEqual(expected_sha, sha)
1792
class TestShaFileByName(tests.TestCaseInTempDir):
1794
def test_sha_empty(self):
1795
self.build_tree_contents([('foo', '')])
1796
expected_sha = osutils.sha_string('')
1797
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1799
def test_sha_mixed_endings(self):
1800
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1801
self.build_tree_contents([('foo', text)])
1802
expected_sha = osutils.sha_string(text)
1803
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1806
class TestResourceLoading(tests.TestCaseInTempDir):
1041
self.failIf('BZR_TEST_ENV_VAR' in os.environ)
1044
class TestLocalTimeOffset(TestCase):
1046
def test_local_time_offset(self):
1047
"""Test that local_time_offset() returns a sane value."""
1048
offset = osutils.local_time_offset()
1049
self.assertTrue(isinstance(offset, int))
1050
# Test that the offset is no more than a eighteen hours in
1052
# Time zone handling is system specific, so it is difficult to
1053
# do more specific tests, but a value outside of this range is
1055
eighteen_hours = 18 * 3600
1056
self.assertTrue(-eighteen_hours < offset < eighteen_hours)
1058
def test_local_time_offset_with_timestamp(self):
1059
"""Test that local_time_offset() works with a timestamp."""
1060
offset = osutils.local_time_offset(1000000000.1234567)
1061
self.assertTrue(isinstance(offset, int))
1062
eighteen_hours = 18 * 3600
1063
self.assertTrue(-eighteen_hours < offset < eighteen_hours)
1066
class TestShaFileByName(TestCaseInTempDir):
1068
def test_sha_empty(self):
1069
self.build_tree_contents([('foo', '')])
1070
expected_sha = osutils.sha_string('')
1071
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1073
def test_sha_mixed_endings(self):
1074
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1075
self.build_tree_contents([('foo', text)])
1076
expected_sha = osutils.sha_string(text)
1077
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1081
r'''# Copyright (C) 2005, 2006 Canonical Ltd
1083
# This program is free software; you can redistribute it and/or modify
1084
# it under the terms of the GNU General Public License as published by
1085
# the Free Software Foundation; either version 2 of the License, or
1086
# (at your option) any later version.
1088
# This program is distributed in the hope that it will be useful,
1089
# but WITHOUT ANY WARRANTY; without even the implied warranty of
1090
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
1091
# GNU General Public License for more details.
1093
# You should have received a copy of the GNU General Public License
1094
# along with this program; if not, write to the Free Software
1095
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
1098
# NOTE: If update these, please also update the help for global-options in
1099
# bzrlib/help_topics/__init__.py
1102
"""Set of flags that enable different debug behaviour.
1104
These are set with eg ``-Dlock`` on the bzr command line.
1108
* auth - show authentication sections used
1109
* error - show stack traces for all top level exceptions
1110
* evil - capture call sites that do expensive or badly-scaling operations.
1111
* fetch - trace history copying between repositories
1112
* hashcache - log every time a working file is read to determine its hash
1113
* hooks - trace hook execution
1114
* hpss - trace smart protocol requests and responses
1115
* http - trace http connections, requests and responses
1116
* index - trace major index operations
1117
* knit - trace knit operations
1118
* lock - trace when lockdir locks are taken or released
1119
* merge - emit information for debugging merges
1120
* pack - emit information about pack operations
1121
* selftest_debug - do not disable all debug flags when running selftest
1127
class TestResourceLoading(TestCaseInTempDir):
1808
1129
def test_resource_string(self):
1809
1130
# test resource in bzrlib
1810
1131
text = osutils.resource_string('bzrlib', 'debug.py')
1811
self.assertContainsRe(text, "debug_flags = set()")
1132
self.assertEquals(_debug_text, text)
1812
1133
# test resource under bzrlib
1813
1134
text = osutils.resource_string('bzrlib.ui', 'text.py')
1814
1135
self.assertContainsRe(text, "class TextUIFactory")
1818
1139
# test unknown resource
1819
1140
self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')
1822
class TestReCompile(tests.TestCase):
1824
def _deprecated_re_compile_checked(self, *args, **kwargs):
1825
return self.applyDeprecated(symbol_versioning.deprecated_in((2, 2, 0)),
1826
osutils.re_compile_checked, *args, **kwargs)
1828
def test_re_compile_checked(self):
1829
r = self._deprecated_re_compile_checked(r'A*', re.IGNORECASE)
1830
self.assertTrue(r.match('aaaa'))
1831
self.assertTrue(r.match('aAaA'))
1833
def test_re_compile_checked_error(self):
1834
# like https://bugs.launchpad.net/bzr/+bug/251352
1836
# Due to possible test isolation error, re.compile is not lazy at
1837
# this point. We re-install lazy compile.
1838
lazy_regex.install_lazy_compile()
1839
err = self.assertRaises(
1840
errors.BzrCommandError,
1841
self._deprecated_re_compile_checked, '*', re.IGNORECASE, 'test case')
1843
'Invalid regular expression in test case: '
1844
'"*" nothing to repeat',
1848
class TestDirReader(tests.TestCaseInTempDir):
1850
scenarios = dir_reader_scenarios()
1853
_dir_reader_class = None
1854
_native_to_unicode = None
1857
super(TestDirReader, self).setUp()
1858
self.overrideAttr(osutils,
1859
'_selected_dir_reader', self._dir_reader_class())
1861
def _get_ascii_tree(self):
1869
expected_dirblocks = [
1871
[('0file', '0file', 'file'),
1872
('1dir', '1dir', 'directory'),
1873
('2file', '2file', 'file'),
1876
(('1dir', './1dir'),
1877
[('1dir/0file', '0file', 'file'),
1878
('1dir/1dir', '1dir', 'directory'),
1881
(('1dir/1dir', './1dir/1dir'),
1886
return tree, expected_dirblocks
1888
def test_walk_cur_dir(self):
1889
tree, expected_dirblocks = self._get_ascii_tree()
1890
self.build_tree(tree)
1891
result = list(osutils._walkdirs_utf8('.'))
1892
# Filter out stat and abspath
1893
self.assertEqual(expected_dirblocks,
1894
[(dirinfo, [line[0:3] for line in block])
1895
for dirinfo, block in result])
1897
def test_walk_sub_dir(self):
1898
tree, expected_dirblocks = self._get_ascii_tree()
1899
self.build_tree(tree)
1900
# you can search a subdir only, with a supplied prefix.
1901
result = list(osutils._walkdirs_utf8('./1dir', '1dir'))
1902
# Filter out stat and abspath
1903
self.assertEqual(expected_dirblocks[1:],
1904
[(dirinfo, [line[0:3] for line in block])
1905
for dirinfo, block in result])
1907
def _get_unicode_tree(self):
1908
name0u = u'0file-\xb6'
1909
name1u = u'1dir-\u062c\u0648'
1910
name2u = u'2file-\u0633'
1914
name1u + '/' + name0u,
1915
name1u + '/' + name1u + '/',
1918
name0 = name0u.encode('UTF-8')
1919
name1 = name1u.encode('UTF-8')
1920
name2 = name2u.encode('UTF-8')
1921
expected_dirblocks = [
1923
[(name0, name0, 'file', './' + name0u),
1924
(name1, name1, 'directory', './' + name1u),
1925
(name2, name2, 'file', './' + name2u),
1928
((name1, './' + name1u),
1929
[(name1 + '/' + name0, name0, 'file', './' + name1u
1931
(name1 + '/' + name1, name1, 'directory', './' + name1u
1935
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1940
return tree, expected_dirblocks
1942
def _filter_out(self, raw_dirblocks):
1943
"""Filter out a walkdirs_utf8 result.
1945
stat field is removed, all native paths are converted to unicode
1947
filtered_dirblocks = []
1948
for dirinfo, block in raw_dirblocks:
1949
dirinfo = (dirinfo[0], self._native_to_unicode(dirinfo[1]))
1952
details.append(line[0:3] + (self._native_to_unicode(line[4]), ))
1953
filtered_dirblocks.append((dirinfo, details))
1954
return filtered_dirblocks
1956
def test_walk_unicode_tree(self):
1957
self.requireFeature(features.UnicodeFilenameFeature)
1958
tree, expected_dirblocks = self._get_unicode_tree()
1959
self.build_tree(tree)
1960
result = list(osutils._walkdirs_utf8('.'))
1961
self.assertEqual(expected_dirblocks, self._filter_out(result))
1963
def test_symlink(self):
1964
self.requireFeature(features.SymlinkFeature)
1965
self.requireFeature(features.UnicodeFilenameFeature)
1966
target = u'target\N{Euro Sign}'
1967
link_name = u'l\N{Euro Sign}nk'
1968
os.symlink(target, link_name)
1969
target_utf8 = target.encode('UTF-8')
1970
link_name_utf8 = link_name.encode('UTF-8')
1971
expected_dirblocks = [
1973
[(link_name_utf8, link_name_utf8,
1974
'symlink', './' + link_name),],
1976
result = list(osutils._walkdirs_utf8('.'))
1977
self.assertEqual(expected_dirblocks, self._filter_out(result))
1980
class TestReadLink(tests.TestCaseInTempDir):
1981
"""Exposes os.readlink() problems and the osutils solution.
1983
The only guarantee offered by os.readlink(), starting with 2.6, is that a
1984
unicode string will be returned if a unicode string is passed.
1986
But prior python versions failed to properly encode the passed unicode
1989
_test_needs_features = [features.SymlinkFeature, features.UnicodeFilenameFeature]
1992
super(tests.TestCaseInTempDir, self).setUp()
1993
self.link = u'l\N{Euro Sign}ink'
1994
self.target = u'targe\N{Euro Sign}t'
1995
os.symlink(self.target, self.link)
1997
def test_os_readlink_link_encoding(self):
1998
self.assertEquals(self.target, os.readlink(self.link))
2000
def test_os_readlink_link_decoding(self):
2001
self.assertEquals(self.target.encode(osutils._fs_enc),
2002
os.readlink(self.link.encode(osutils._fs_enc)))
2005
class TestConcurrency(tests.TestCase):
2008
super(TestConcurrency, self).setUp()
2009
self.overrideAttr(osutils, '_cached_local_concurrency')
2011
def test_local_concurrency(self):
2012
concurrency = osutils.local_concurrency()
2013
self.assertIsInstance(concurrency, int)
2015
def test_local_concurrency_environment_variable(self):
2016
self.overrideEnv('BZR_CONCURRENCY', '2')
2017
self.assertEqual(2, osutils.local_concurrency(use_cache=False))
2018
self.overrideEnv('BZR_CONCURRENCY', '3')
2019
self.assertEqual(3, osutils.local_concurrency(use_cache=False))
2020
self.overrideEnv('BZR_CONCURRENCY', 'foo')
2021
self.assertEqual(1, osutils.local_concurrency(use_cache=False))
2023
def test_option_concurrency(self):
2024
self.overrideEnv('BZR_CONCURRENCY', '1')
2025
self.run_bzr('rocks --concurrency 42')
2026
# Command line overrides environment variable
2027
self.assertEquals('42', os.environ['BZR_CONCURRENCY'])
2028
self.assertEquals(42, osutils.local_concurrency(use_cache=False))
2031
class TestFailedToLoadExtension(tests.TestCase):
2033
def _try_loading(self):
2035
import bzrlib._fictional_extension_py
2036
except ImportError, e:
2037
osutils.failed_to_load_extension(e)
2041
super(TestFailedToLoadExtension, self).setUp()
2042
self.overrideAttr(osutils, '_extension_load_failures', [])
2044
def test_failure_to_load(self):
2046
self.assertLength(1, osutils._extension_load_failures)
2047
self.assertEquals(osutils._extension_load_failures[0],
2048
"No module named _fictional_extension_py")
2050
def test_report_extension_load_failures_no_warning(self):
2051
self.assertTrue(self._try_loading())
2052
warnings, result = self.callCatchWarnings(osutils.report_extension_load_failures)
2053
# it used to give a Python warning; it no longer does
2054
self.assertLength(0, warnings)
2056
def test_report_extension_load_failures_message(self):
2058
trace.push_log_file(log)
2059
self.assertTrue(self._try_loading())
2060
osutils.report_extension_load_failures()
2061
self.assertContainsRe(
2063
r"bzr: warning: some compiled extensions could not be loaded; "
2064
"see <https://answers\.launchpad\.net/bzr/\+faq/703>\n"
2068
class TestTerminalWidth(tests.TestCase):
2071
super(TestTerminalWidth, self).setUp()
2072
self._orig_terminal_size_state = osutils._terminal_size_state
2073
self._orig_first_terminal_size = osutils._first_terminal_size
2074
self.addCleanup(self.restore_osutils_globals)
2075
osutils._terminal_size_state = 'no_data'
2076
osutils._first_terminal_size = None
2078
def restore_osutils_globals(self):
2079
osutils._terminal_size_state = self._orig_terminal_size_state
2080
osutils._first_terminal_size = self._orig_first_terminal_size
2082
def replace_stdout(self, new):
2083
self.overrideAttr(sys, 'stdout', new)
2085
def replace__terminal_size(self, new):
2086
self.overrideAttr(osutils, '_terminal_size', new)
2088
def set_fake_tty(self):
2090
class I_am_a_tty(object):
2094
self.replace_stdout(I_am_a_tty())
2096
def test_default_values(self):
2097
self.assertEqual(80, osutils.default_terminal_width)
2099
def test_defaults_to_BZR_COLUMNS(self):
2100
# BZR_COLUMNS is set by the test framework
2101
self.assertNotEqual('12', os.environ['BZR_COLUMNS'])
2102
self.overrideEnv('BZR_COLUMNS', '12')
2103
self.assertEqual(12, osutils.terminal_width())
2105
def test_BZR_COLUMNS_0_no_limit(self):
2106
self.overrideEnv('BZR_COLUMNS', '0')
2107
self.assertEqual(None, osutils.terminal_width())
2109
def test_falls_back_to_COLUMNS(self):
2110
self.overrideEnv('BZR_COLUMNS', None)
2111
self.assertNotEqual('42', os.environ['COLUMNS'])
2113
self.overrideEnv('COLUMNS', '42')
2114
self.assertEqual(42, osutils.terminal_width())
2116
def test_tty_default_without_columns(self):
2117
self.overrideEnv('BZR_COLUMNS', None)
2118
self.overrideEnv('COLUMNS', None)
2120
def terminal_size(w, h):
2124
# We need to override the osutils definition as it depends on the
2125
# running environment that we can't control (PQM running without a
2126
# controlling terminal is one example).
2127
self.replace__terminal_size(terminal_size)
2128
self.assertEqual(42, osutils.terminal_width())
2130
def test_non_tty_default_without_columns(self):
2131
self.overrideEnv('BZR_COLUMNS', None)
2132
self.overrideEnv('COLUMNS', None)
2133
self.replace_stdout(None)
2134
self.assertEqual(None, osutils.terminal_width())
2136
def test_no_TIOCGWINSZ(self):
2137
self.requireFeature(term_ios_feature)
2138
termios = term_ios_feature.module
2139
# bug 63539 is about a termios without TIOCGWINSZ attribute
2141
orig = termios.TIOCGWINSZ
2142
except AttributeError:
2143
# We won't remove TIOCGWINSZ, because it doesn't exist anyway :)
2146
self.overrideAttr(termios, 'TIOCGWINSZ')
2147
del termios.TIOCGWINSZ
2148
self.overrideEnv('BZR_COLUMNS', None)
2149
self.overrideEnv('COLUMNS', None)
2150
# Whatever the result is, if we don't raise an exception, it's ok.
2151
osutils.terminal_width()
2154
class TestCreationOps(tests.TestCaseInTempDir):
2155
_test_needs_features = [features.chown_feature]
2158
super(TestCreationOps, self).setUp()
2159
self.overrideAttr(os, 'chown', self._dummy_chown)
2161
# params set by call to _dummy_chown
2162
self.path = self.uid = self.gid = None
2164
def _dummy_chown(self, path, uid, gid):
2165
self.path, self.uid, self.gid = path, uid, gid
2167
def test_copy_ownership_from_path(self):
2168
"""copy_ownership_from_path test with specified src."""
2170
f = open('test_file', 'wt')
2171
osutils.copy_ownership_from_path('test_file', ownsrc)
2174
self.assertEquals(self.path, 'test_file')
2175
self.assertEquals(self.uid, s.st_uid)
2176
self.assertEquals(self.gid, s.st_gid)
2178
def test_copy_ownership_nonesrc(self):
2179
"""copy_ownership_from_path test with src=None."""
2180
f = open('test_file', 'wt')
2181
# should use parent dir for permissions
2182
osutils.copy_ownership_from_path('test_file')
2185
self.assertEquals(self.path, 'test_file')
2186
self.assertEquals(self.uid, s.st_uid)
2187
self.assertEquals(self.gid, s.st_gid)
2190
class TestPathFromEnviron(tests.TestCase):
2192
def test_is_unicode(self):
2193
self.overrideEnv('BZR_TEST_PATH', './anywhere at all/')
2194
path = osutils.path_from_environ('BZR_TEST_PATH')
2195
self.assertIsInstance(path, unicode)
2196
self.assertEqual(u'./anywhere at all/', path)
2198
def test_posix_path_env_ascii(self):
2199
self.overrideEnv('BZR_TEST_PATH', '/tmp')
2200
home = osutils._posix_path_from_environ('BZR_TEST_PATH')
2201
self.assertIsInstance(home, unicode)
2202
self.assertEqual(u'/tmp', home)
2204
def test_posix_path_env_unicode(self):
2205
self.requireFeature(features.ByteStringNamedFilesystem)
2206
self.overrideEnv('BZR_TEST_PATH', '/home/\xa7test')
2207
self.overrideAttr(osutils, "_fs_enc", "iso8859-1")
2208
self.assertEqual(u'/home/\xa7test',
2209
osutils._posix_path_from_environ('BZR_TEST_PATH'))
2210
osutils._fs_enc = "iso8859-5"
2211
self.assertEqual(u'/home/\u0407test',
2212
osutils._posix_path_from_environ('BZR_TEST_PATH'))
2213
osutils._fs_enc = "utf-8"
2214
self.assertRaises(errors.BadFilenameEncoding,
2215
osutils._posix_path_from_environ, 'BZR_TEST_PATH')
2218
class TestGetHomeDir(tests.TestCase):
2220
def test_is_unicode(self):
2221
home = osutils._get_home_dir()
2222
self.assertIsInstance(home, unicode)
2224
def test_posix_homeless(self):
2225
self.overrideEnv('HOME', None)
2226
home = osutils._get_home_dir()
2227
self.assertIsInstance(home, unicode)
2229
def test_posix_home_ascii(self):
2230
self.overrideEnv('HOME', '/home/test')
2231
home = osutils._posix_get_home_dir()
2232
self.assertIsInstance(home, unicode)
2233
self.assertEqual(u'/home/test', home)
2235
def test_posix_home_unicode(self):
2236
self.requireFeature(features.ByteStringNamedFilesystem)
2237
self.overrideEnv('HOME', '/home/\xa7test')
2238
self.overrideAttr(osutils, "_fs_enc", "iso8859-1")
2239
self.assertEqual(u'/home/\xa7test', osutils._posix_get_home_dir())
2240
osutils._fs_enc = "iso8859-5"
2241
self.assertEqual(u'/home/\u0407test', osutils._posix_get_home_dir())
2242
osutils._fs_enc = "utf-8"
2243
self.assertRaises(errors.BadFilenameEncoding,
2244
osutils._posix_get_home_dir)
2247
class TestGetuserUnicode(tests.TestCase):
2249
def test_is_unicode(self):
2250
user = osutils.getuser_unicode()
2251
self.assertIsInstance(user, unicode)
2253
def envvar_to_override(self):
2254
if sys.platform == "win32":
2255
# Disable use of platform calls on windows so envvar is used
2256
self.overrideAttr(win32utils, 'has_ctypes', False)
2257
return 'USERNAME' # only variable used on windows
2258
return 'LOGNAME' # first variable checked by getpass.getuser()
2260
def test_ascii_user(self):
2261
self.overrideEnv(self.envvar_to_override(), 'jrandom')
2262
self.assertEqual(u'jrandom', osutils.getuser_unicode())
2264
def test_unicode_user(self):
2265
ue = osutils.get_user_encoding()
2266
uni_val, env_val = tests.probe_unicode_in_user_encoding()
2268
raise tests.TestSkipped(
2269
'Cannot find a unicode character that works in encoding %s'
2270
% (osutils.get_user_encoding(),))
2271
uni_username = u'jrandom' + uni_val
2272
encoded_username = uni_username.encode(ue)
2273
self.overrideEnv(self.envvar_to_override(), encoded_username)
2274
self.assertEqual(uni_username, osutils.getuser_unicode())
2277
class TestBackupNames(tests.TestCase):
2280
super(TestBackupNames, self).setUp()
2283
def backup_exists(self, name):
2284
return name in self.backups
2286
def available_backup_name(self, name):
2287
backup_name = osutils.available_backup_name(name, self.backup_exists)
2288
self.backups.append(backup_name)
2291
def assertBackupName(self, expected, name):
2292
self.assertEqual(expected, self.available_backup_name(name))
2294
def test_empty(self):
2295
self.assertBackupName('file.~1~', 'file')
2297
def test_existing(self):
2298
self.available_backup_name('file')
2299
self.available_backup_name('file')
2300
self.assertBackupName('file.~3~', 'file')
2301
# Empty slots are found, this is not a strict requirement and may be
2302
# revisited if we test against all implementations.
2303
self.backups.remove('file.~2~')
2304
self.assertBackupName('file.~2~', 'file')
2307
class TestFindExecutableInPath(tests.TestCase):
2309
def test_windows(self):
2310
if sys.platform != 'win32':
2311
raise tests.TestSkipped('test requires win32')
2312
self.assertTrue(osutils.find_executable_on_path('explorer') is not None)
2314
osutils.find_executable_on_path('explorer.exe') is not None)
2316
osutils.find_executable_on_path('EXPLORER.EXE') is not None)
2318
osutils.find_executable_on_path('THIS SHOULD NOT EXIST') is None)
2319
self.assertTrue(osutils.find_executable_on_path('file.txt') is None)
2321
def test_windows_app_path(self):
2322
if sys.platform != 'win32':
2323
raise tests.TestSkipped('test requires win32')
2324
# Override PATH env var so that exe can only be found on App Path
2325
self.overrideEnv('PATH', '')
2326
# Internt Explorer is always registered in the App Path
2327
self.assertTrue(osutils.find_executable_on_path('iexplore') is not None)
2329
def test_other(self):
2330
if sys.platform == 'win32':
2331
raise tests.TestSkipped('test requires non-win32')
2332
self.assertTrue(osutils.find_executable_on_path('sh') is not None)
2334
osutils.find_executable_on_path('THIS SHOULD NOT EXIST') is None)
2337
class TestEnvironmentErrors(tests.TestCase):
2338
"""Test handling of environmental errors"""
2340
def test_is_oserror(self):
2341
self.assertTrue(osutils.is_environment_error(
2342
OSError(errno.EINVAL, "Invalid parameter")))
2344
def test_is_ioerror(self):
2345
self.assertTrue(osutils.is_environment_error(
2346
IOError(errno.EINVAL, "Invalid parameter")))
2348
def test_is_socket_error(self):
2349
self.assertTrue(osutils.is_environment_error(
2350
socket.error(errno.EINVAL, "Invalid parameter")))
2352
def test_is_select_error(self):
2353
self.assertTrue(osutils.is_environment_error(
2354
select.error(errno.EINVAL, "Invalid parameter")))
2356
def test_is_pywintypes_error(self):
2357
self.requireFeature(features.pywintypes)
2359
self.assertTrue(osutils.is_environment_error(
2360
pywintypes.error(errno.EINVAL, "Invalid parameter", "Caller")))
added
removed
bzrlib/tests/test_osutils.py
