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)
102
load_tests = load_tests_apply_scenarios
107
105
class TestContainsWhitespace(tests.TestCase):
109
107
def test_contains_whitespace(self):
110
self.failUnless(osutils.contains_whitespace(u' '))
111
self.failUnless(osutils.contains_whitespace(u'hello there'))
112
self.failUnless(osutils.contains_whitespace(u'hellothere\n'))
113
self.failUnless(osutils.contains_whitespace(u'hello\nthere'))
114
self.failUnless(osutils.contains_whitespace(u'hello\rthere'))
115
self.failUnless(osutils.contains_whitespace(u'hello\tthere'))
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'))
117
115
# \xa0 is "Non-breaking-space" which on some python locales thinks it
118
116
# is whitespace, but we do not.
119
self.failIf(osutils.contains_whitespace(u''))
120
self.failIf(osutils.contains_whitespace(u'hellothere'))
121
self.failIf(osutils.contains_whitespace(u'hello\xa0there'))
117
self.assertFalse(osutils.contains_whitespace(u''))
118
self.assertFalse(osutils.contains_whitespace(u'hellothere'))
119
self.assertFalse(osutils.contains_whitespace(u'hello\xa0there'))
124
122
class TestRename(tests.TestCaseInTempDir):
418
437
self.assertTrue(-eighteen_hours < offset < eighteen_hours)
440
class TestFdatasync(tests.TestCaseInTempDir):
442
def do_fdatasync(self):
443
f = tempfile.NamedTemporaryFile()
444
osutils.fdatasync(f.fileno())
448
def raise_eopnotsupp(*args, **kwargs):
449
raise IOError(errno.EOPNOTSUPP, os.strerror(errno.EOPNOTSUPP))
452
def raise_enotsup(*args, **kwargs):
453
raise IOError(errno.ENOTSUP, os.strerror(errno.ENOTSUP))
455
def test_fdatasync_handles_system_function(self):
456
self.overrideAttr(os, "fdatasync")
459
def test_fdatasync_handles_no_fdatasync_no_fsync(self):
460
self.overrideAttr(os, "fdatasync")
461
self.overrideAttr(os, "fsync")
464
def test_fdatasync_handles_no_EOPNOTSUPP(self):
465
self.overrideAttr(errno, "EOPNOTSUPP")
468
def test_fdatasync_catches_ENOTSUP(self):
469
enotsup = getattr(errno, "ENOTSUP", None)
471
raise tests.TestNotApplicable("No ENOTSUP on this platform")
472
self.overrideAttr(os, "fdatasync", self.raise_enotsup)
475
def test_fdatasync_catches_EOPNOTSUPP(self):
476
enotsup = getattr(errno, "EOPNOTSUPP", None)
478
raise tests.TestNotApplicable("No EOPNOTSUPP on this platform")
479
self.overrideAttr(os, "fdatasync", self.raise_eopnotsupp)
421
483
class TestLinks(tests.TestCaseInTempDir):
423
485
def test_dereference_path(self):
424
self.requireFeature(tests.SymlinkFeature)
486
self.requireFeature(features.SymlinkFeature)
425
487
cwd = osutils.realpath('.')
427
489
bar_path = osutils.pathjoin(cwd, 'bar')
810
872
self.assertEqual(None, osutils.safe_file_id(None))
875
class TestSendAll(tests.TestCase):
877
def test_send_with_disconnected_socket(self):
878
class DisconnectedSocket(object):
879
def __init__(self, err):
881
def send(self, content):
885
# All of these should be treated as ConnectionReset
887
for err_cls in (IOError, socket.error):
888
for errnum in osutils._end_of_stream_errors:
889
errs.append(err_cls(errnum))
891
sock = DisconnectedSocket(err)
892
self.assertRaises(errors.ConnectionReset,
893
osutils.send_all, sock, 'some more content')
895
def test_send_with_no_progress(self):
896
# See https://bugs.launchpad.net/bzr/+bug/1047309
897
# It seems that paramiko can get into a state where it doesn't error,
898
# but it returns 0 bytes sent for requests over and over again.
899
class NoSendingSocket(object):
902
def send(self, bytes):
904
if self.call_count > 100:
905
# Prevent the test suite from hanging
906
raise RuntimeError('too many calls')
908
sock = NoSendingSocket()
909
self.assertRaises(errors.ConnectionReset,
910
osutils.send_all, sock, 'content')
911
self.assertEqual(1, sock.call_count)
914
class TestPosixFuncs(tests.TestCase):
915
"""Test that the posix version of normpath returns an appropriate path
916
when used with 2 leading slashes."""
918
def test_normpath(self):
919
self.assertEqual('/etc/shadow', osutils._posix_normpath('/etc/shadow'))
920
self.assertEqual('/etc/shadow', osutils._posix_normpath('//etc/shadow'))
921
self.assertEqual('/etc/shadow', osutils._posix_normpath('///etc/shadow'))
813
924
class TestWin32Funcs(tests.TestCase):
814
925
"""Test that _win32 versions of os utilities return appropriate paths."""
816
927
def test_abspath(self):
928
self.requireFeature(features.win32_feature)
817
929
self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
818
930
self.assertEqual('C:/foo', osutils._win32_abspath('C:/foo'))
819
931
self.assertEqual('//HOST/path', osutils._win32_abspath(r'\\HOST\path'))
1085
1205
# (It would be ok if it happened earlier but at the moment it
1087
1207
e = self.assertRaises(OSError, list, osutils._walkdirs_utf8("."))
1088
self.assertEquals('./test-unreadable', e.filename)
1089
self.assertEquals(errno.EACCES, e.errno)
1208
self.assertEqual('./test-unreadable', e.filename)
1209
self.assertEqual(errno.EACCES, e.errno)
1090
1210
# Ensure the message contains the file name
1091
1211
self.assertContainsRe(str(e), "\./test-unreadable")
1214
def test_walkdirs_encoding_error(self):
1215
# <https://bugs.launchpad.net/bzr/+bug/488519>
1216
# walkdirs didn't raise a useful message when the filenames
1217
# are not using the filesystem's encoding
1219
# require a bytestring based filesystem
1220
self.requireFeature(features.ByteStringNamedFilesystem)
1231
self.build_tree(tree)
1233
# rename the 1file to a latin-1 filename
1234
os.rename("./1file", "\xe8file")
1235
if "\xe8file" not in os.listdir("."):
1236
self.skip("Lack filesystem that preserves arbitrary bytes")
1238
self._save_platform_info()
1239
win32utils.winver = None # Avoid the win32 detection code
1240
osutils._fs_enc = 'UTF-8'
1242
# this should raise on error
1244
for dirdetail, dirblock in osutils.walkdirs('.'):
1247
self.assertRaises(errors.BadFilenameEncoding, attempt)
1093
1249
def test__walkdirs_utf8(self):
1161
1317
self.requireFeature(UTF8DirReaderFeature)
1162
1318
self._save_platform_info()
1163
1319
win32utils.winver = None # Avoid the win32 detection code
1164
osutils._fs_enc = 'UTF-8'
1165
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1320
osutils._fs_enc = 'utf-8'
1321
self.assertDirReaderIs(
1322
UTF8DirReaderFeature.module.UTF8DirReader)
1167
1324
def test_force_walkdirs_utf8_fs_ascii(self):
1168
1325
self.requireFeature(UTF8DirReaderFeature)
1169
1326
self._save_platform_info()
1170
1327
win32utils.winver = None # Avoid the win32 detection code
1171
osutils._fs_enc = 'US-ASCII'
1172
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1174
def test_force_walkdirs_utf8_fs_ANSI(self):
1175
self.requireFeature(UTF8DirReaderFeature)
1176
self._save_platform_info()
1177
win32utils.winver = None # Avoid the win32 detection code
1178
osutils._fs_enc = 'ANSI_X3.4-1968'
1179
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1328
osutils._fs_enc = 'ascii'
1329
self.assertDirReaderIs(
1330
UTF8DirReaderFeature.module.UTF8DirReader)
1181
1332
def test_force_walkdirs_utf8_fs_latin1(self):
1182
1333
self._save_platform_info()
1183
1334
win32utils.winver = None # Avoid the win32 detection code
1184
osutils._fs_enc = 'latin1'
1335
osutils._fs_enc = 'iso-8859-1'
1185
1336
self.assertDirReaderIs(osutils.UnicodeDirReader)
1187
1338
def test_force_walkdirs_utf8_nt(self):
1679
1830
class TestReCompile(tests.TestCase):
1832
def _deprecated_re_compile_checked(self, *args, **kwargs):
1833
return self.applyDeprecated(symbol_versioning.deprecated_in((2, 2, 0)),
1834
osutils.re_compile_checked, *args, **kwargs)
1681
1836
def test_re_compile_checked(self):
1682
r = osutils.re_compile_checked(r'A*', re.IGNORECASE)
1837
r = self._deprecated_re_compile_checked(r'A*', re.IGNORECASE)
1683
1838
self.assertTrue(r.match('aaaa'))
1684
1839
self.assertTrue(r.match('aAaA'))
1686
1841
def test_re_compile_checked_error(self):
1687
1842
# like https://bugs.launchpad.net/bzr/+bug/251352
1844
# Due to possible test isolation error, re.compile is not lazy at
1845
# this point. We re-install lazy compile.
1846
lazy_regex.install_lazy_compile()
1688
1847
err = self.assertRaises(
1689
1848
errors.BzrCommandError,
1690
osutils.re_compile_checked, '*', re.IGNORECASE, 'test case')
1849
self._deprecated_re_compile_checked, '*', re.IGNORECASE, 'test case')
1691
1850
self.assertEqual(
1692
"Invalid regular expression in test case: '*': "
1693
"nothing to repeat",
1851
'Invalid regular expression in test case: '
1852
'"*" nothing to repeat',
1697
1856
class TestDirReader(tests.TestCaseInTempDir):
1858
scenarios = dir_reader_scenarios()
1699
1860
# Set by load_tests
1700
1861
_dir_reader_class = None
1701
1862
_native_to_unicode = None
1703
1864
def setUp(self):
1704
tests.TestCaseInTempDir.setUp(self)
1865
super(TestDirReader, self).setUp()
1705
1866
self.overrideAttr(osutils,
1706
1867
'_selected_dir_reader', self._dir_reader_class())
1863
2021
self.assertIsInstance(concurrency, int)
1865
2023
def test_local_concurrency_environment_variable(self):
1866
os.environ['BZR_CONCURRENCY'] = '2'
2024
self.overrideEnv('BZR_CONCURRENCY', '2')
1867
2025
self.assertEqual(2, osutils.local_concurrency(use_cache=False))
1868
os.environ['BZR_CONCURRENCY'] = '3'
2026
self.overrideEnv('BZR_CONCURRENCY', '3')
1869
2027
self.assertEqual(3, osutils.local_concurrency(use_cache=False))
1870
os.environ['BZR_CONCURRENCY'] = 'foo'
2028
self.overrideEnv('BZR_CONCURRENCY', 'foo')
1871
2029
self.assertEqual(1, osutils.local_concurrency(use_cache=False))
1873
2031
def test_option_concurrency(self):
1874
os.environ['BZR_CONCURRENCY'] = '1'
2032
self.overrideEnv('BZR_CONCURRENCY', '1')
1875
2033
self.run_bzr('rocks --concurrency 42')
1876
# Command line overrides envrionment variable
1877
self.assertEquals('42', os.environ['BZR_CONCURRENCY'])
1878
self.assertEquals(42, osutils.local_concurrency(use_cache=False))
2034
# Command line overrides environment variable
2035
self.assertEqual('42', os.environ['BZR_CONCURRENCY'])
2036
self.assertEqual(42, osutils.local_concurrency(use_cache=False))
1881
2039
class TestFailedToLoadExtension(tests.TestCase):
2015
2190
osutils.copy_ownership_from_path('test_file')
2017
2192
s = os.stat('..')
2018
self.assertEquals(self.path, 'test_file')
2019
self.assertEquals(self.uid, s.st_uid)
2020
self.assertEquals(self.gid, s.st_gid)
2193
self.assertEqual(self.path, 'test_file')
2194
self.assertEqual(self.uid, s.st_uid)
2195
self.assertEqual(self.gid, s.st_gid)
2198
class TestPathFromEnviron(tests.TestCase):
2200
def test_is_unicode(self):
2201
self.overrideEnv('BZR_TEST_PATH', './anywhere at all/')
2202
path = osutils.path_from_environ('BZR_TEST_PATH')
2203
self.assertIsInstance(path, unicode)
2204
self.assertEqual(u'./anywhere at all/', path)
2206
def test_posix_path_env_ascii(self):
2207
self.overrideEnv('BZR_TEST_PATH', '/tmp')
2208
home = osutils._posix_path_from_environ('BZR_TEST_PATH')
2209
self.assertIsInstance(home, unicode)
2210
self.assertEqual(u'/tmp', home)
2212
def test_posix_path_env_unicode(self):
2213
self.requireFeature(features.ByteStringNamedFilesystem)
2214
self.overrideEnv('BZR_TEST_PATH', '/home/\xa7test')
2215
self.overrideAttr(osutils, "_fs_enc", "iso8859-1")
2216
self.assertEqual(u'/home/\xa7test',
2217
osutils._posix_path_from_environ('BZR_TEST_PATH'))
2218
osutils._fs_enc = "iso8859-5"
2219
self.assertEqual(u'/home/\u0407test',
2220
osutils._posix_path_from_environ('BZR_TEST_PATH'))
2221
osutils._fs_enc = "utf-8"
2222
self.assertRaises(errors.BadFilenameEncoding,
2223
osutils._posix_path_from_environ, 'BZR_TEST_PATH')
2226
class TestGetHomeDir(tests.TestCase):
2228
def test_is_unicode(self):
2229
home = osutils._get_home_dir()
2230
self.assertIsInstance(home, unicode)
2232
def test_posix_homeless(self):
2233
self.overrideEnv('HOME', None)
2234
home = osutils._get_home_dir()
2235
self.assertIsInstance(home, unicode)
2237
def test_posix_home_ascii(self):
2238
self.overrideEnv('HOME', '/home/test')
2239
home = osutils._posix_get_home_dir()
2240
self.assertIsInstance(home, unicode)
2241
self.assertEqual(u'/home/test', home)
2243
def test_posix_home_unicode(self):
2244
self.requireFeature(features.ByteStringNamedFilesystem)
2245
self.overrideEnv('HOME', '/home/\xa7test')
2246
self.overrideAttr(osutils, "_fs_enc", "iso8859-1")
2247
self.assertEqual(u'/home/\xa7test', osutils._posix_get_home_dir())
2248
osutils._fs_enc = "iso8859-5"
2249
self.assertEqual(u'/home/\u0407test', osutils._posix_get_home_dir())
2250
osutils._fs_enc = "utf-8"
2251
self.assertRaises(errors.BadFilenameEncoding,
2252
osutils._posix_get_home_dir)
2255
class TestGetuserUnicode(tests.TestCase):
2257
def test_is_unicode(self):
2258
user = osutils.getuser_unicode()
2259
self.assertIsInstance(user, unicode)
2261
def envvar_to_override(self):
2262
if sys.platform == "win32":
2263
# Disable use of platform calls on windows so envvar is used
2264
self.overrideAttr(win32utils, 'has_ctypes', False)
2265
return 'USERNAME' # only variable used on windows
2266
return 'LOGNAME' # first variable checked by getpass.getuser()
2268
def test_ascii_user(self):
2269
self.overrideEnv(self.envvar_to_override(), 'jrandom')
2270
self.assertEqual(u'jrandom', osutils.getuser_unicode())
2272
def test_unicode_user(self):
2273
ue = osutils.get_user_encoding()
2274
uni_val, env_val = tests.probe_unicode_in_user_encoding()
2276
raise tests.TestSkipped(
2277
'Cannot find a unicode character that works in encoding %s'
2278
% (osutils.get_user_encoding(),))
2279
uni_username = u'jrandom' + uni_val
2280
encoded_username = uni_username.encode(ue)
2281
self.overrideEnv(self.envvar_to_override(), encoded_username)
2282
self.assertEqual(uni_username, osutils.getuser_unicode())
2285
class TestBackupNames(tests.TestCase):
2288
super(TestBackupNames, self).setUp()
2291
def backup_exists(self, name):
2292
return name in self.backups
2294
def available_backup_name(self, name):
2295
backup_name = osutils.available_backup_name(name, self.backup_exists)
2296
self.backups.append(backup_name)
2299
def assertBackupName(self, expected, name):
2300
self.assertEqual(expected, self.available_backup_name(name))
2302
def test_empty(self):
2303
self.assertBackupName('file.~1~', 'file')
2305
def test_existing(self):
2306
self.available_backup_name('file')
2307
self.available_backup_name('file')
2308
self.assertBackupName('file.~3~', 'file')
2309
# Empty slots are found, this is not a strict requirement and may be
2310
# revisited if we test against all implementations.
2311
self.backups.remove('file.~2~')
2312
self.assertBackupName('file.~2~', 'file')
2315
class TestFindExecutableInPath(tests.TestCase):
2317
def test_windows(self):
2318
if sys.platform != 'win32':
2319
raise tests.TestSkipped('test requires win32')
2320
self.assertTrue(osutils.find_executable_on_path('explorer') is not None)
2322
osutils.find_executable_on_path('explorer.exe') is not None)
2324
osutils.find_executable_on_path('EXPLORER.EXE') is not None)
2326
osutils.find_executable_on_path('THIS SHOULD NOT EXIST') is None)
2327
self.assertTrue(osutils.find_executable_on_path('file.txt') is None)
2329
def test_windows_app_path(self):
2330
if sys.platform != 'win32':
2331
raise tests.TestSkipped('test requires win32')
2332
# Override PATH env var so that exe can only be found on App Path
2333
self.overrideEnv('PATH', '')
2334
# Internt Explorer is always registered in the App Path
2335
self.assertTrue(osutils.find_executable_on_path('iexplore') is not None)
2337
def test_other(self):
2338
if sys.platform == 'win32':
2339
raise tests.TestSkipped('test requires non-win32')
2340
self.assertTrue(osutils.find_executable_on_path('sh') is not None)
2342
osutils.find_executable_on_path('THIS SHOULD NOT EXIST') is None)
2345
class TestEnvironmentErrors(tests.TestCase):
2346
"""Test handling of environmental errors"""
2348
def test_is_oserror(self):
2349
self.assertTrue(osutils.is_environment_error(
2350
OSError(errno.EINVAL, "Invalid parameter")))
2352
def test_is_ioerror(self):
2353
self.assertTrue(osutils.is_environment_error(
2354
IOError(errno.EINVAL, "Invalid parameter")))
2356
def test_is_socket_error(self):
2357
self.assertTrue(osutils.is_environment_error(
2358
socket.error(errno.EINVAL, "Invalid parameter")))
2360
def test_is_select_error(self):
2361
self.assertTrue(osutils.is_environment_error(
2362
select.error(errno.EINVAL, "Invalid parameter")))
2364
def test_is_pywintypes_error(self):
2365
self.requireFeature(features.pywintypes)
2367
self.assertTrue(osutils.is_environment_error(
2368
pywintypes.error(errno.EINVAL, "Invalid parameter", "Caller")))