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., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
17
"""Tests for the osutils wrapper."""
19
from cStringIO import StringIO
26
28
from bzrlib import (
31
from bzrlib.errors import BzrBadParameterNotUnicode, InvalidURL
32
from bzrlib.osutils import (
34
is_inside_or_parent_of_any,
37
35
from bzrlib.tests import (
38
probe_unicode_in_user_encoding,
47
class TestOSUtils(TestCaseInTempDir):
42
class _UTF8DirReaderFeature(tests.Feature):
46
from bzrlib import _readdir_pyx
47
self.reader = _readdir_pyx.UTF8DirReader
52
def feature_name(self):
53
return 'bzrlib._readdir_pyx'
55
UTF8DirReaderFeature = _UTF8DirReaderFeature()
57
term_ios_feature = tests.ModuleAvailableFeature('termios')
60
def _already_unicode(s):
64
def _utf8_to_unicode(s):
65
return s.decode('UTF-8')
68
def dir_reader_scenarios():
69
# For each dir reader we define:
71
# - native_to_unicode: a function converting the native_abspath as returned
72
# by DirReader.read_dir to its unicode representation
74
# UnicodeDirReader is the fallback, it should be tested on all platforms.
75
scenarios = [('unicode',
76
dict(_dir_reader_class=osutils.UnicodeDirReader,
77
_native_to_unicode=_already_unicode))]
78
# Some DirReaders are platform specific and even there they may not be
80
if UTF8DirReaderFeature.available():
81
from bzrlib import _readdir_pyx
82
scenarios.append(('utf8',
83
dict(_dir_reader_class=_readdir_pyx.UTF8DirReader,
84
_native_to_unicode=_utf8_to_unicode)))
86
if test__walkdirs_win32.win32_readdir_feature.available():
88
from bzrlib import _walkdirs_win32
91
dict(_dir_reader_class=_walkdirs_win32.Win32ReadDir,
92
_native_to_unicode=_already_unicode)))
98
def load_tests(basic_tests, module, loader):
99
suite = loader.suiteClass()
100
dir_reader_tests, remaining_tests = tests.split_suite_by_condition(
101
basic_tests, tests.condition_isinstance(TestDirReader))
102
tests.multiply_tests(dir_reader_tests, dir_reader_scenarios(), suite)
103
suite.addTest(remaining_tests)
107
class TestContainsWhitespace(tests.TestCase):
49
109
def test_contains_whitespace(self):
50
110
self.failUnless(osutils.contains_whitespace(u' '))
60
120
self.failIf(osutils.contains_whitespace(u'hellothere'))
61
121
self.failIf(osutils.contains_whitespace(u'hello\xa0there'))
124
class TestRename(tests.TestCaseInTempDir):
126
def create_file(self, filename, content):
127
f = open(filename, 'wb')
133
def _fancy_rename(self, a, b):
134
osutils.fancy_rename(a, b, rename_func=os.rename,
135
unlink_func=os.unlink)
63
137
def test_fancy_rename(self):
64
138
# This should work everywhere
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')
139
self.create_file('a', 'something in a\n')
140
self._fancy_rename('a', 'b')
72
141
self.failIfExists('a')
73
142
self.failUnlessExists('b')
74
143
self.check_file_contents('b', 'something in a\n')
76
open('a', 'wb').write('new something in a\n')
145
self.create_file('a', 'new something in a\n')
146
self._fancy_rename('b', 'a')
79
148
self.check_file_contents('a', 'something in a\n')
150
def test_fancy_rename_fails_source_missing(self):
151
# An exception should be raised, and the target should be left in place
152
self.create_file('target', 'data in target\n')
153
self.assertRaises((IOError, OSError), self._fancy_rename,
154
'missingsource', 'target')
155
self.failUnlessExists('target')
156
self.check_file_contents('target', 'data in target\n')
158
def test_fancy_rename_fails_if_source_and_target_missing(self):
159
self.assertRaises((IOError, OSError), self._fancy_rename,
160
'missingsource', 'missingtarget')
81
162
def test_rename(self):
82
163
# Rename should be semi-atomic on all platforms
83
open('a', 'wb').write('something in a\n')
164
self.create_file('a', 'something in a\n')
84
165
osutils.rename('a', 'b')
85
166
self.failIfExists('a')
86
167
self.failUnlessExists('b')
87
168
self.check_file_contents('b', 'something in a\n')
89
open('a', 'wb').write('new something in a\n')
170
self.create_file('a', 'new something in a\n')
90
171
osutils.rename('b', 'a')
92
173
self.check_file_contents('a', 'something in a\n')
312
461
osutils.make_readonly('dangling')
313
462
osutils.make_writable('dangling')
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):
464
def test_host_os_dereferences_symlinks(self):
465
osutils.host_os_dereferences_symlinks()
468
class TestCanonicalRelPath(tests.TestCaseInTempDir):
470
_test_needs_features = [tests.CaseInsCasePresFilenameFeature]
472
def test_canonical_relpath_simple(self):
473
f = file('MixedCaseName', 'w')
475
actual = osutils.canonical_relpath(self.test_base_dir, 'mixedcasename')
476
self.failUnlessEqual('work/MixedCaseName', actual)
478
def test_canonical_relpath_missing_tail(self):
479
os.mkdir('MixedCaseParent')
480
actual = osutils.canonical_relpath(self.test_base_dir,
481
'mixedcaseparent/nochild')
482
self.failUnlessEqual('work/MixedCaseParent/nochild', actual)
485
class Test_CICPCanonicalRelpath(tests.TestCaseWithTransport):
487
def assertRelpath(self, expected, base, path):
488
actual = osutils._cicp_canonical_relpath(base, path)
489
self.assertEqual(expected, actual)
491
def test_simple(self):
492
self.build_tree(['MixedCaseName'])
493
base = osutils.realpath(self.get_transport('.').local_abspath('.'))
494
self.assertRelpath('MixedCaseName', base, 'mixedcAsename')
496
def test_subdir_missing_tail(self):
497
self.build_tree(['MixedCaseParent/', 'MixedCaseParent/a_child'])
498
base = osutils.realpath(self.get_transport('.').local_abspath('.'))
499
self.assertRelpath('MixedCaseParent/a_child', base,
500
'MixedCaseParent/a_child')
501
self.assertRelpath('MixedCaseParent/a_child', base,
502
'MixedCaseParent/A_Child')
503
self.assertRelpath('MixedCaseParent/not_child', base,
504
'MixedCaseParent/not_child')
506
def test_at_root_slash(self):
507
# We can't test this on Windows, because it has a 'MIN_ABS_PATHLENGTH'
509
if osutils.MIN_ABS_PATHLENGTH > 1:
510
raise tests.TestSkipped('relpath requires %d chars'
511
% osutils.MIN_ABS_PATHLENGTH)
512
self.assertRelpath('foo', '/', '/foo')
514
def test_at_root_drive(self):
515
if sys.platform != 'win32':
516
raise tests.TestNotApplicable('we can only test drive-letter relative'
517
' paths on Windows where we have drive'
520
# The specific issue is that when at the root of a drive, 'abspath'
521
# returns "C:/" or just "/". However, the code assumes that abspath
522
# always returns something like "C:/foo" or "/foo" (no trailing slash).
523
self.assertRelpath('foo', 'C:/', 'C:/foo')
524
self.assertRelpath('foo', 'X:/', 'X:/foo')
525
self.assertRelpath('foo', 'X:/', 'X://foo')
528
class TestPumpFile(tests.TestCase):
529
"""Test pumpfile method."""
532
tests.TestCase.setUp(self)
533
# create a test datablock
534
self.block_size = 512
535
pattern = '0123456789ABCDEF'
536
self.test_data = pattern * (3 * self.block_size / len(pattern))
537
self.test_data_len = len(self.test_data)
539
def test_bracket_block_size(self):
540
"""Read data in blocks with the requested read size bracketing the
542
# make sure test data is larger than max read size
543
self.assertTrue(self.test_data_len > self.block_size)
545
from_file = file_utils.FakeReadFile(self.test_data)
548
# read (max / 2) bytes and verify read size wasn't affected
549
num_bytes_to_read = self.block_size / 2
550
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
551
self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
552
self.assertEqual(from_file.get_read_count(), 1)
554
# read (max) bytes and verify read size wasn't affected
555
num_bytes_to_read = self.block_size
556
from_file.reset_read_count()
557
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
558
self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
559
self.assertEqual(from_file.get_read_count(), 1)
561
# read (max + 1) bytes and verify read size was limited
562
num_bytes_to_read = self.block_size + 1
563
from_file.reset_read_count()
564
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
565
self.assertEqual(from_file.get_max_read_size(), self.block_size)
566
self.assertEqual(from_file.get_read_count(), 2)
568
# finish reading the rest of the data
569
num_bytes_to_read = self.test_data_len - to_file.tell()
570
osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
572
# report error if the data wasn't equal (we only report the size due
573
# to the length of the data)
574
response_data = to_file.getvalue()
575
if response_data != self.test_data:
576
message = "Data not equal. Expected %d bytes, received %d."
577
self.fail(message % (len(response_data), self.test_data_len))
579
def test_specified_size(self):
580
"""Request a transfer larger than the maximum block size and verify
581
that the maximum read doesn't exceed the block_size."""
582
# make sure test data is larger than max read size
583
self.assertTrue(self.test_data_len > self.block_size)
585
# retrieve data in blocks
586
from_file = file_utils.FakeReadFile(self.test_data)
588
osutils.pumpfile(from_file, to_file, self.test_data_len,
591
# verify read size was equal to the maximum read size
592
self.assertTrue(from_file.get_max_read_size() > 0)
593
self.assertEqual(from_file.get_max_read_size(), self.block_size)
594
self.assertEqual(from_file.get_read_count(), 3)
596
# report error if the data wasn't equal (we only report the size due
597
# to the length of the data)
598
response_data = to_file.getvalue()
599
if response_data != self.test_data:
600
message = "Data not equal. Expected %d bytes, received %d."
601
self.fail(message % (len(response_data), self.test_data_len))
603
def test_to_eof(self):
604
"""Read to end-of-file and verify that the reads are not larger than
605
the maximum read size."""
606
# make sure test data is larger than max read size
607
self.assertTrue(self.test_data_len > self.block_size)
609
# retrieve data to EOF
610
from_file = file_utils.FakeReadFile(self.test_data)
612
osutils.pumpfile(from_file, to_file, -1, self.block_size)
614
# verify read size was equal to the maximum read size
615
self.assertEqual(from_file.get_max_read_size(), self.block_size)
616
self.assertEqual(from_file.get_read_count(), 4)
618
# report error if the data wasn't equal (we only report the size due
619
# to the length of the data)
620
response_data = to_file.getvalue()
621
if response_data != self.test_data:
622
message = "Data not equal. Expected %d bytes, received %d."
623
self.fail(message % (len(response_data), self.test_data_len))
625
def test_defaults(self):
626
"""Verifies that the default arguments will read to EOF -- this
627
test verifies that any existing usages of pumpfile will not be broken
628
with this new version."""
629
# retrieve data using default (old) pumpfile method
630
from_file = file_utils.FakeReadFile(self.test_data)
632
osutils.pumpfile(from_file, to_file)
634
# report error if the data wasn't equal (we only report the size due
635
# to the length of the data)
636
response_data = to_file.getvalue()
637
if response_data != self.test_data:
638
message = "Data not equal. Expected %d bytes, received %d."
639
self.fail(message % (len(response_data), self.test_data_len))
641
def test_report_activity(self):
643
def log_activity(length, direction):
644
activity.append((length, direction))
645
from_file = StringIO(self.test_data)
647
osutils.pumpfile(from_file, to_file, buff_size=500,
648
report_activity=log_activity, direction='read')
649
self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),
650
(36, 'read')], activity)
652
from_file = StringIO(self.test_data)
655
osutils.pumpfile(from_file, to_file, buff_size=500,
656
report_activity=log_activity, direction='write')
657
self.assertEqual([(500, 'write'), (500, 'write'), (500, 'write'),
658
(36, 'write')], activity)
660
# And with a limited amount of data
661
from_file = StringIO(self.test_data)
664
osutils.pumpfile(from_file, to_file, buff_size=500, read_length=1028,
665
report_activity=log_activity, direction='read')
666
self.assertEqual([(500, 'read'), (500, 'read'), (28, 'read')], activity)
670
class TestPumpStringFile(tests.TestCase):
672
def test_empty(self):
674
osutils.pump_string_file("", output)
675
self.assertEqual("", output.getvalue())
677
def test_more_than_segment_size(self):
679
osutils.pump_string_file("123456789", output, 2)
680
self.assertEqual("123456789", output.getvalue())
682
def test_segment_size(self):
684
osutils.pump_string_file("12", output, 2)
685
self.assertEqual("12", output.getvalue())
687
def test_segment_size_multiple(self):
689
osutils.pump_string_file("1234", output, 2)
690
self.assertEqual("1234", output.getvalue())
693
class TestRelpath(tests.TestCase):
695
def test_simple_relpath(self):
696
cwd = osutils.getcwd()
697
subdir = cwd + '/subdir'
698
self.assertEqual('subdir', osutils.relpath(cwd, subdir))
700
def test_deep_relpath(self):
701
cwd = osutils.getcwd()
702
subdir = cwd + '/sub/subsubdir'
703
self.assertEqual('sub/subsubdir', osutils.relpath(cwd, subdir))
705
def test_not_relative(self):
706
self.assertRaises(errors.PathNotChild,
707
osutils.relpath, 'C:/path', 'H:/path')
708
self.assertRaises(errors.PathNotChild,
709
osutils.relpath, 'C:/', 'H:/path')
712
class TestSafeUnicode(tests.TestCase):
324
714
def test_from_ascii_string(self):
325
715
self.assertEqual(u'foobar', osutils.safe_unicode('foobar'))
560
956
self.assertRaises(errors.BzrError, osutils.splitpath, 'a/../b')
563
class TestMacFuncsDirs(TestCaseInTempDir):
959
class TestParentDirectories(tests.TestCaseInTempDir):
960
"""Test osutils.parent_directories()"""
962
def test_parent_directories(self):
963
self.assertEqual([], osutils.parent_directories('a'))
964
self.assertEqual(['a'], osutils.parent_directories('a/b'))
965
self.assertEqual(['a/b', 'a'], osutils.parent_directories('a/b/c'))
968
class TestMacFuncsDirs(tests.TestCaseInTempDir):
564
969
"""Test mac special functions that require directories."""
566
971
def test_getcwd(self):
567
# On Mac, this will actually create Ba\u030agfors
568
# but chdir will still work, because it accepts both paths
570
os.mkdir(u'B\xe5gfors')
572
raise TestSkipped("Unable to create Unicode filename")
972
self.requireFeature(tests.UnicodeFilenameFeature)
973
os.mkdir(u'B\xe5gfors')
574
974
os.chdir(u'B\xe5gfors')
575
975
self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
577
977
def test_getcwd_nonnorm(self):
978
self.requireFeature(tests.UnicodeFilenameFeature)
578
979
# Test that _mac_getcwd() will normalize this path
580
os.mkdir(u'Ba\u030agfors')
582
raise TestSkipped("Unable to create Unicode filename")
980
os.mkdir(u'Ba\u030agfors')
584
981
os.chdir(u'Ba\u030agfors')
585
982
self.assertEndsWith(osutils._mac_getcwd(), u'B\xe5gfors')
588
class TestSplitLines(TestCase):
985
class TestChunksToLines(tests.TestCase):
987
def test_smoketest(self):
988
self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
989
osutils.chunks_to_lines(['foo\nbar', '\nbaz\n']))
990
self.assertEqual(['foo\n', 'bar\n', 'baz\n'],
991
osutils.chunks_to_lines(['foo\n', 'bar\n', 'baz\n']))
993
def test_osutils_binding(self):
994
from bzrlib.tests import test__chunks_to_lines
995
if test__chunks_to_lines.compiled_chunkstolines_feature.available():
996
from bzrlib._chunks_to_lines_pyx import chunks_to_lines
998
from bzrlib._chunks_to_lines_py import chunks_to_lines
999
self.assertIs(chunks_to_lines, osutils.chunks_to_lines)
1002
class TestSplitLines(tests.TestCase):
590
1004
def test_split_unicode(self):
591
1005
self.assertEqual([u'foo\n', u'bar\xae'],
701
1137
new_dirblock.append((info[0], info[1], info[2], info[4]))
702
1138
dirblock[:] = new_dirblock
1140
def _save_platform_info(self):
1141
self.overrideAttr(win32utils, 'winver')
1142
self.overrideAttr(osutils, '_fs_enc')
1143
self.overrideAttr(osutils, '_selected_dir_reader')
1145
def assertDirReaderIs(self, expected):
1146
"""Assert the right implementation for _walkdirs_utf8 is chosen."""
1147
# Force it to redetect
1148
osutils._selected_dir_reader = None
1149
# Nothing to list, but should still trigger the selection logic
1150
self.assertEqual([(('', '.'), [])], list(osutils._walkdirs_utf8('.')))
1151
self.assertIsInstance(osutils._selected_dir_reader, expected)
1153
def test_force_walkdirs_utf8_fs_utf8(self):
1154
self.requireFeature(UTF8DirReaderFeature)
1155
self._save_platform_info()
1156
win32utils.winver = None # Avoid the win32 detection code
1157
osutils._fs_enc = 'UTF-8'
1158
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1160
def test_force_walkdirs_utf8_fs_ascii(self):
1161
self.requireFeature(UTF8DirReaderFeature)
1162
self._save_platform_info()
1163
win32utils.winver = None # Avoid the win32 detection code
1164
osutils._fs_enc = 'US-ASCII'
1165
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1167
def test_force_walkdirs_utf8_fs_ANSI(self):
1168
self.requireFeature(UTF8DirReaderFeature)
1169
self._save_platform_info()
1170
win32utils.winver = None # Avoid the win32 detection code
1171
osutils._fs_enc = 'ANSI_X3.4-1968'
1172
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1174
def test_force_walkdirs_utf8_fs_latin1(self):
1175
self._save_platform_info()
1176
win32utils.winver = None # Avoid the win32 detection code
1177
osutils._fs_enc = 'latin1'
1178
self.assertDirReaderIs(osutils.UnicodeDirReader)
1180
def test_force_walkdirs_utf8_nt(self):
1181
# Disabled because the thunk of the whole walkdirs api is disabled.
1182
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1183
self._save_platform_info()
1184
win32utils.winver = 'Windows NT'
1185
from bzrlib._walkdirs_win32 import Win32ReadDir
1186
self.assertDirReaderIs(Win32ReadDir)
1188
def test_force_walkdirs_utf8_98(self):
1189
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1190
self._save_platform_info()
1191
win32utils.winver = 'Windows 98'
1192
self.assertDirReaderIs(osutils.UnicodeDirReader)
704
1194
def test_unicode_walkdirs(self):
705
1195
"""Walkdirs should always return unicode paths."""
1196
self.requireFeature(tests.UnicodeFilenameFeature)
706
1197
name0 = u'0file-\xb6'
707
1198
name1 = u'1dir-\u062c\u0648'
708
1199
name2 = u'2file-\u0633'
808
1292
result.append((dirdetail, new_dirblock))
809
1293
self.assertEqual(expected_dirblocks, result)
811
def test_unicode__walkdirs_unicode_to_utf8(self):
812
"""walkdirs_unicode_to_utf8 should be a safe fallback everywhere
1295
def test__walkdirs_utf8_with_unicode_fs(self):
1296
"""UnicodeDirReader should be a safe fallback everywhere
814
1298
The abspath portion should be in unicode
816
name0u = u'0file-\xb6'
817
name1u = u'1dir-\u062c\u0648'
818
name2u = u'2file-\u0633'
822
name1u + '/' + name0u,
823
name1u + '/' + name1u + '/',
1300
self.requireFeature(tests.UnicodeFilenameFeature)
1301
# Use the unicode reader. TODO: split into driver-and-driven unit
1303
self._save_platform_info()
1304
osutils._selected_dir_reader = osutils.UnicodeDirReader()
1305
name0u = u'0file-\xb6'
1306
name1u = u'1dir-\u062c\u0648'
1307
name2u = u'2file-\u0633'
1311
name1u + '/' + name0u,
1312
name1u + '/' + name1u + '/',
1315
self.build_tree(tree)
1316
name0 = name0u.encode('utf8')
1317
name1 = name1u.encode('utf8')
1318
name2 = name2u.encode('utf8')
1320
# All of the abspaths should be in unicode, all of the relative paths
1322
expected_dirblocks = [
1324
[(name0, name0, 'file', './' + name0u),
1325
(name1, name1, 'directory', './' + name1u),
1326
(name2, name2, 'file', './' + name2u),
1329
((name1, './' + name1u),
1330
[(name1 + '/' + name0, name0, 'file', './' + name1u
1332
(name1 + '/' + name1, name1, 'directory', './' + name1u
1336
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1341
result = list(osutils._walkdirs_utf8('.'))
1342
self._filter_out_stat(result)
1343
self.assertEqual(expected_dirblocks, result)
1345
def test__walkdirs_utf8_win32readdir(self):
1346
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1347
self.requireFeature(tests.UnicodeFilenameFeature)
1348
from bzrlib._walkdirs_win32 import Win32ReadDir
1349
self._save_platform_info()
1350
osutils._selected_dir_reader = Win32ReadDir()
1351
name0u = u'0file-\xb6'
1352
name1u = u'1dir-\u062c\u0648'
1353
name2u = u'2file-\u0633'
1357
name1u + '/' + name0u,
1358
name1u + '/' + name1u + '/',
1361
self.build_tree(tree)
1362
name0 = name0u.encode('utf8')
1363
name1 = name1u.encode('utf8')
1364
name2 = name2u.encode('utf8')
1366
# All of the abspaths should be in unicode, all of the relative paths
1368
expected_dirblocks = [
1370
[(name0, name0, 'file', './' + name0u),
1371
(name1, name1, 'directory', './' + name1u),
1372
(name2, name2, 'file', './' + name2u),
1375
((name1, './' + name1u),
1376
[(name1 + '/' + name0, name0, 'file', './' + name1u
1378
(name1 + '/' + name1, name1, 'directory', './' + name1u
1382
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1387
result = list(osutils._walkdirs_utf8(u'.'))
1388
self._filter_out_stat(result)
1389
self.assertEqual(expected_dirblocks, result)
1391
def assertStatIsCorrect(self, path, win32stat):
1392
os_stat = os.stat(path)
1393
self.assertEqual(os_stat.st_size, win32stat.st_size)
1394
self.assertAlmostEqual(os_stat.st_mtime, win32stat.st_mtime, places=4)
1395
self.assertAlmostEqual(os_stat.st_ctime, win32stat.st_ctime, places=4)
1396
self.assertAlmostEqual(os_stat.st_atime, win32stat.st_atime, places=4)
1397
self.assertEqual(os_stat.st_dev, win32stat.st_dev)
1398
self.assertEqual(os_stat.st_ino, win32stat.st_ino)
1399
self.assertEqual(os_stat.st_mode, win32stat.st_mode)
1401
def test__walkdirs_utf_win32_find_file_stat_file(self):
1402
"""make sure our Stat values are valid"""
1403
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1404
self.requireFeature(tests.UnicodeFilenameFeature)
1405
from bzrlib._walkdirs_win32 import Win32ReadDir
1406
name0u = u'0file-\xb6'
1407
name0 = name0u.encode('utf8')
1408
self.build_tree([name0u])
1409
# I hate to sleep() here, but I'm trying to make the ctime different
1412
f = open(name0u, 'ab')
827
self.build_tree(tree)
829
raise TestSkipped('Could not represent Unicode chars'
830
' in current encoding.')
1414
f.write('just a small update')
1418
result = Win32ReadDir().read_dir('', u'.')
1420
self.assertEqual((name0, name0, 'file'), entry[:3])
1421
self.assertEqual(u'./' + name0u, entry[4])
1422
self.assertStatIsCorrect(entry[4], entry[3])
1423
self.assertNotEqual(entry[3].st_mtime, entry[3].st_ctime)
1425
def test__walkdirs_utf_win32_find_file_stat_directory(self):
1426
"""make sure our Stat values are valid"""
1427
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
1428
self.requireFeature(tests.UnicodeFilenameFeature)
1429
from bzrlib._walkdirs_win32 import Win32ReadDir
1430
name0u = u'0dir-\u062c\u0648'
831
1431
name0 = name0u.encode('utf8')
832
name1 = name1u.encode('utf8')
833
name2 = name2u.encode('utf8')
1432
self.build_tree([name0u + '/'])
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)
1434
result = Win32ReadDir().read_dir('', u'.')
1436
self.assertEqual((name0, name0, 'directory'), entry[:3])
1437
self.assertEqual(u'./' + name0u, entry[4])
1438
self.assertStatIsCorrect(entry[4], entry[3])
860
1440
def assertPathCompare(self, path_less, path_greater):
861
1441
"""check that path_less and path_greater compare correctly."""
1041
1617
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
* graph - trace graph traversal information
1113
* hashcache - log every time a working file is read to determine its hash
1114
* hooks - trace hook execution
1115
* hpss - trace smart protocol requests and responses
1116
* http - trace http connections, requests and responses
1117
* index - trace major index operations
1118
* knit - trace knit operations
1119
* lock - trace when lockdir locks are taken or released
1120
* merge - emit information for debugging merges
1121
* pack - emit information about pack operations
1127
class TestResourceLoading(TestCaseInTempDir):
1620
class TestSizeShaFile(tests.TestCaseInTempDir):
1622
def test_sha_empty(self):
1623
self.build_tree_contents([('foo', '')])
1624
expected_sha = osutils.sha_string('')
1626
self.addCleanup(f.close)
1627
size, sha = osutils.size_sha_file(f)
1628
self.assertEqual(0, size)
1629
self.assertEqual(expected_sha, sha)
1631
def test_sha_mixed_endings(self):
1632
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1633
self.build_tree_contents([('foo', text)])
1634
expected_sha = osutils.sha_string(text)
1635
f = open('foo', 'rb')
1636
self.addCleanup(f.close)
1637
size, sha = osutils.size_sha_file(f)
1638
self.assertEqual(38, size)
1639
self.assertEqual(expected_sha, sha)
1642
class TestShaFileByName(tests.TestCaseInTempDir):
1644
def test_sha_empty(self):
1645
self.build_tree_contents([('foo', '')])
1646
expected_sha = osutils.sha_string('')
1647
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1649
def test_sha_mixed_endings(self):
1650
text = 'test\r\nwith\nall\rpossible line endings\r\n'
1651
self.build_tree_contents([('foo', text)])
1652
expected_sha = osutils.sha_string(text)
1653
self.assertEqual(expected_sha, osutils.sha_file_by_name('foo'))
1656
class TestResourceLoading(tests.TestCaseInTempDir):
1129
1658
def test_resource_string(self):
1130
1659
# test resource in bzrlib
1131
1660
text = osutils.resource_string('bzrlib', 'debug.py')
1132
self.assertEquals(_debug_text, text)
1661
self.assertContainsRe(text, "debug_flags = set()")
1133
1662
# test resource under bzrlib
1134
1663
text = osutils.resource_string('bzrlib.ui', 'text.py')
1135
1664
self.assertContainsRe(text, "class TextUIFactory")
1139
1668
# test unknown resource
1140
1669
self.assertRaises(IOError, osutils.resource_string, 'bzrlib', 'yyy.xx')
1672
class TestReCompile(tests.TestCase):
1674
def test_re_compile_checked(self):
1675
r = osutils.re_compile_checked(r'A*', re.IGNORECASE)
1676
self.assertTrue(r.match('aaaa'))
1677
self.assertTrue(r.match('aAaA'))
1679
def test_re_compile_checked_error(self):
1680
# like https://bugs.launchpad.net/bzr/+bug/251352
1681
err = self.assertRaises(
1682
errors.BzrCommandError,
1683
osutils.re_compile_checked, '*', re.IGNORECASE, 'test case')
1685
"Invalid regular expression in test case: '*': "
1686
"nothing to repeat",
1690
class TestDirReader(tests.TestCaseInTempDir):
1693
_dir_reader_class = None
1694
_native_to_unicode = None
1697
tests.TestCaseInTempDir.setUp(self)
1698
self.overrideAttr(osutils,
1699
'_selected_dir_reader', self._dir_reader_class())
1701
def _get_ascii_tree(self):
1709
expected_dirblocks = [
1711
[('0file', '0file', 'file'),
1712
('1dir', '1dir', 'directory'),
1713
('2file', '2file', 'file'),
1716
(('1dir', './1dir'),
1717
[('1dir/0file', '0file', 'file'),
1718
('1dir/1dir', '1dir', 'directory'),
1721
(('1dir/1dir', './1dir/1dir'),
1726
return tree, expected_dirblocks
1728
def test_walk_cur_dir(self):
1729
tree, expected_dirblocks = self._get_ascii_tree()
1730
self.build_tree(tree)
1731
result = list(osutils._walkdirs_utf8('.'))
1732
# Filter out stat and abspath
1733
self.assertEqual(expected_dirblocks,
1734
[(dirinfo, [line[0:3] for line in block])
1735
for dirinfo, block in result])
1737
def test_walk_sub_dir(self):
1738
tree, expected_dirblocks = self._get_ascii_tree()
1739
self.build_tree(tree)
1740
# you can search a subdir only, with a supplied prefix.
1741
result = list(osutils._walkdirs_utf8('./1dir', '1dir'))
1742
# Filter out stat and abspath
1743
self.assertEqual(expected_dirblocks[1:],
1744
[(dirinfo, [line[0:3] for line in block])
1745
for dirinfo, block in result])
1747
def _get_unicode_tree(self):
1748
name0u = u'0file-\xb6'
1749
name1u = u'1dir-\u062c\u0648'
1750
name2u = u'2file-\u0633'
1754
name1u + '/' + name0u,
1755
name1u + '/' + name1u + '/',
1758
name0 = name0u.encode('UTF-8')
1759
name1 = name1u.encode('UTF-8')
1760
name2 = name2u.encode('UTF-8')
1761
expected_dirblocks = [
1763
[(name0, name0, 'file', './' + name0u),
1764
(name1, name1, 'directory', './' + name1u),
1765
(name2, name2, 'file', './' + name2u),
1768
((name1, './' + name1u),
1769
[(name1 + '/' + name0, name0, 'file', './' + name1u
1771
(name1 + '/' + name1, name1, 'directory', './' + name1u
1775
((name1 + '/' + name1, './' + name1u + '/' + name1u),
1780
return tree, expected_dirblocks
1782
def _filter_out(self, raw_dirblocks):
1783
"""Filter out a walkdirs_utf8 result.
1785
stat field is removed, all native paths are converted to unicode
1787
filtered_dirblocks = []
1788
for dirinfo, block in raw_dirblocks:
1789
dirinfo = (dirinfo[0], self._native_to_unicode(dirinfo[1]))
1792
details.append(line[0:3] + (self._native_to_unicode(line[4]), ))
1793
filtered_dirblocks.append((dirinfo, details))
1794
return filtered_dirblocks
1796
def test_walk_unicode_tree(self):
1797
self.requireFeature(tests.UnicodeFilenameFeature)
1798
tree, expected_dirblocks = self._get_unicode_tree()
1799
self.build_tree(tree)
1800
result = list(osutils._walkdirs_utf8('.'))
1801
self.assertEqual(expected_dirblocks, self._filter_out(result))
1803
def test_symlink(self):
1804
self.requireFeature(tests.SymlinkFeature)
1805
self.requireFeature(tests.UnicodeFilenameFeature)
1806
target = u'target\N{Euro Sign}'
1807
link_name = u'l\N{Euro Sign}nk'
1808
os.symlink(target, link_name)
1809
target_utf8 = target.encode('UTF-8')
1810
link_name_utf8 = link_name.encode('UTF-8')
1811
expected_dirblocks = [
1813
[(link_name_utf8, link_name_utf8,
1814
'symlink', './' + link_name),],
1816
result = list(osutils._walkdirs_utf8('.'))
1817
self.assertEqual(expected_dirblocks, self._filter_out(result))
1820
class TestReadLink(tests.TestCaseInTempDir):
1821
"""Exposes os.readlink() problems and the osutils solution.
1823
The only guarantee offered by os.readlink(), starting with 2.6, is that a
1824
unicode string will be returned if a unicode string is passed.
1826
But prior python versions failed to properly encode the passed unicode
1829
_test_needs_features = [tests.SymlinkFeature, tests.UnicodeFilenameFeature]
1832
super(tests.TestCaseInTempDir, self).setUp()
1833
self.link = u'l\N{Euro Sign}ink'
1834
self.target = u'targe\N{Euro Sign}t'
1835
os.symlink(self.target, self.link)
1837
def test_os_readlink_link_encoding(self):
1838
if sys.version_info < (2, 6):
1839
self.assertRaises(UnicodeEncodeError, os.readlink, self.link)
1841
self.assertEquals(self.target, os.readlink(self.link))
1843
def test_os_readlink_link_decoding(self):
1844
self.assertEquals(self.target.encode(osutils._fs_enc),
1845
os.readlink(self.link.encode(osutils._fs_enc)))
1848
class TestConcurrency(tests.TestCase):
1851
super(TestConcurrency, self).setUp()
1852
self.overrideAttr(osutils, '_cached_local_concurrency')
1854
def test_local_concurrency(self):
1855
concurrency = osutils.local_concurrency()
1856
self.assertIsInstance(concurrency, int)
1858
def test_local_concurrency_environment_variable(self):
1859
os.environ['BZR_CONCURRENCY'] = '2'
1860
self.assertEqual(2, osutils.local_concurrency(use_cache=False))
1861
os.environ['BZR_CONCURRENCY'] = '3'
1862
self.assertEqual(3, osutils.local_concurrency(use_cache=False))
1863
os.environ['BZR_CONCURRENCY'] = 'foo'
1864
self.assertEqual(1, osutils.local_concurrency(use_cache=False))
1866
def test_option_concurrency(self):
1867
os.environ['BZR_CONCURRENCY'] = '1'
1868
self.run_bzr('rocks --concurrency 42')
1869
# Command line overrides envrionment variable
1870
self.assertEquals('42', os.environ['BZR_CONCURRENCY'])
1871
self.assertEquals(42, osutils.local_concurrency(use_cache=False))
1874
class TestFailedToLoadExtension(tests.TestCase):
1876
def _try_loading(self):
1878
import bzrlib._fictional_extension_py
1879
except ImportError, e:
1880
osutils.failed_to_load_extension(e)
1884
super(TestFailedToLoadExtension, self).setUp()
1885
self.overrideAttr(osutils, '_extension_load_failures', [])
1887
def test_failure_to_load(self):
1889
self.assertLength(1, osutils._extension_load_failures)
1890
self.assertEquals(osutils._extension_load_failures[0],
1891
"No module named _fictional_extension_py")
1893
def test_report_extension_load_failures_no_warning(self):
1894
self.assertTrue(self._try_loading())
1895
warnings, result = self.callCatchWarnings(osutils.report_extension_load_failures)
1896
# it used to give a Python warning; it no longer does
1897
self.assertLength(0, warnings)
1899
def test_report_extension_load_failures_message(self):
1901
trace.push_log_file(log)
1902
self.assertTrue(self._try_loading())
1903
osutils.report_extension_load_failures()
1904
self.assertContainsRe(
1906
r"bzr: warning: some compiled extensions could not be loaded; "
1907
"see <https://answers\.launchpad\.net/bzr/\+faq/703>\n"
1911
class TestTerminalWidth(tests.TestCase):
1913
def replace_stdout(self, new):
1914
self.overrideAttr(sys, 'stdout', new)
1916
def replace__terminal_size(self, new):
1917
self.overrideAttr(osutils, '_terminal_size', new)
1919
def set_fake_tty(self):
1921
class I_am_a_tty(object):
1925
self.replace_stdout(I_am_a_tty())
1927
def test_default_values(self):
1928
self.assertEqual(80, osutils.default_terminal_width)
1930
def test_defaults_to_BZR_COLUMNS(self):
1931
# BZR_COLUMNS is set by the test framework
1932
self.assertNotEqual('12', os.environ['BZR_COLUMNS'])
1933
os.environ['BZR_COLUMNS'] = '12'
1934
self.assertEqual(12, osutils.terminal_width())
1936
def test_falls_back_to_COLUMNS(self):
1937
del os.environ['BZR_COLUMNS']
1938
self.assertNotEqual('42', os.environ['COLUMNS'])
1940
os.environ['COLUMNS'] = '42'
1941
self.assertEqual(42, osutils.terminal_width())
1943
def test_tty_default_without_columns(self):
1944
del os.environ['BZR_COLUMNS']
1945
del os.environ['COLUMNS']
1947
def terminal_size(w, h):
1951
# We need to override the osutils definition as it depends on the
1952
# running environment that we can't control (PQM running without a
1953
# controlling terminal is one example).
1954
self.replace__terminal_size(terminal_size)
1955
self.assertEqual(42, osutils.terminal_width())
1957
def test_non_tty_default_without_columns(self):
1958
del os.environ['BZR_COLUMNS']
1959
del os.environ['COLUMNS']
1960
self.replace_stdout(None)
1961
self.assertEqual(None, osutils.terminal_width())
1963
def test_no_TIOCGWINSZ(self):
1964
self.requireFeature(term_ios_feature)
1965
termios = term_ios_feature.module
1966
# bug 63539 is about a termios without TIOCGWINSZ attribute
1968
orig = termios.TIOCGWINSZ
1969
except AttributeError:
1970
# We won't remove TIOCGWINSZ, because it doesn't exist anyway :)
1973
self.overrideAttr(termios, 'TIOCGWINSZ')
1974
del termios.TIOCGWINSZ
1975
del os.environ['BZR_COLUMNS']
1976
del os.environ['COLUMNS']
1977
# Whatever the result is, if we don't raise an exception, it's ok.
1978
osutils.terminal_width()
1980
class TestCreationOps(tests.TestCaseInTempDir):
1981
_test_needs_features = [features.chown_feature]
1984
tests.TestCaseInTempDir.setUp(self)
1985
self.overrideAttr(os, 'chown', self._dummy_chown)
1987
# params set by call to _dummy_chown
1988
self.path = self.uid = self.gid = None
1990
def _dummy_chown(self, path, uid, gid):
1991
self.path, self.uid, self.gid = path, uid, gid
1993
def test_mkdir_with_ownership_chown(self):
1994
"""Ensure that osutils.mkdir_with_ownership chowns correctly with ownership_src.
1997
osutils.mkdir_with_ownership('foo', ownsrc)
2000
self.assertEquals(self.path, 'foo')
2001
self.assertEquals(self.uid, s.st_uid)
2002
self.assertEquals(self.gid, s.st_gid)
2004
def test_open_with_ownership_chown(self):
2005
"""Ensure that osutils.open_with_ownership chowns correctly with ownership_src.
2008
f = osutils.open_with_ownership('foo', 'w', ownership_src=ownsrc)
2010
# do a test write and close
2015
self.assertEquals(self.path, 'foo')
2016
self.assertEquals(self.uid, s.st_uid)
2017
self.assertEquals(self.gid, s.st_gid)
added
removed
bzrlib/tests/test_osutils.py
