102
load_tests = load_tests_apply_scenarios
99
def load_tests(basic_tests, module, loader):
100
suite = loader.suiteClass()
101
dir_reader_tests, remaining_tests = tests.split_suite_by_condition(
102
basic_tests, tests.condition_isinstance(TestDirReader))
103
tests.multiply_tests(dir_reader_tests, dir_reader_scenarios(), suite)
104
suite.addTest(remaining_tests)
105
108
class TestContainsWhitespace(tests.TestCase):
107
110
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'))
111
self.failUnless(osutils.contains_whitespace(u' '))
112
self.failUnless(osutils.contains_whitespace(u'hello there'))
113
self.failUnless(osutils.contains_whitespace(u'hellothere\n'))
114
self.failUnless(osutils.contains_whitespace(u'hello\nthere'))
115
self.failUnless(osutils.contains_whitespace(u'hello\rthere'))
116
self.failUnless(osutils.contains_whitespace(u'hello\tthere'))
115
118
# \xa0 is "Non-breaking-space" which on some python locales thinks it
116
119
# 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'))
120
self.failIf(osutils.contains_whitespace(u''))
121
self.failIf(osutils.contains_whitespace(u'hellothere'))
122
self.failIf(osutils.contains_whitespace(u'hello\xa0there'))
122
125
class TestRename(tests.TestCaseInTempDir):
180
183
# we can't use failUnlessExists on case-insensitive filesystem
181
184
# so try to check shape of the tree
182
185
shape = sorted(os.listdir('.'))
183
self.assertEqual(['A', 'B'], shape)
185
def test_rename_exception(self):
187
osutils.rename('nonexistent_path', 'different_nonexistent_path')
189
self.assertEqual(e.old_filename, 'nonexistent_path')
190
self.assertEqual(e.new_filename, 'different_nonexistent_path')
191
self.assertTrue('nonexistent_path' in e.strerror)
192
self.assertTrue('different_nonexistent_path' in e.strerror)
186
self.assertEquals(['A', 'B'], shape)
195
189
class TestRandChars(tests.TestCase):
242
236
self.assertFalse(osutils.is_inside_or_parent_of_any(dirs, fn))
245
class TestLstat(tests.TestCaseInTempDir):
247
def test_lstat_matches_fstat(self):
248
# On Windows, lstat and fstat don't always agree, primarily in the
249
# 'st_ino' and 'st_dev' fields. So we force them to be '0' in our
250
# custom implementation.
251
if sys.platform == 'win32':
252
# We only have special lstat/fstat if we have the extension.
253
# Without it, we may end up re-reading content when we don't have
254
# to, but otherwise it doesn't effect correctness.
255
self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
256
f = open('test-file.txt', 'wb')
257
self.addCleanup(f.close)
258
f.write('some content\n')
260
self.assertEqualStat(osutils.fstat(f.fileno()),
261
osutils.lstat('test-file.txt'))
264
239
class TestRmTree(tests.TestCaseInTempDir):
266
241
def test_rmtree(self):
299
274
def test_file_kind(self):
300
275
self.build_tree(['file', 'dir/'])
301
self.assertEqual('file', osutils.file_kind('file'))
302
self.assertEqual('directory', osutils.file_kind('dir/'))
276
self.assertEquals('file', osutils.file_kind('file'))
277
self.assertEquals('directory', osutils.file_kind('dir/'))
303
278
if osutils.has_symlinks():
304
279
os.symlink('symlink', 'symlink')
305
self.assertEqual('symlink', osutils.file_kind('symlink'))
280
self.assertEquals('symlink', osutils.file_kind('symlink'))
307
282
# TODO: jam 20060529 Test a block device
437
412
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)
483
415
class TestLinks(tests.TestCaseInTempDir):
485
417
def test_dereference_path(self):
486
self.requireFeature(features.SymlinkFeature)
418
self.requireFeature(tests.SymlinkFeature)
487
419
cwd = osutils.realpath('.')
489
421
bar_path = osutils.pathjoin(cwd, 'bar')
537
469
class TestCanonicalRelPath(tests.TestCaseInTempDir):
539
_test_needs_features = [features.CaseInsCasePresFilenameFeature]
471
_test_needs_features = [tests.CaseInsCasePresFilenameFeature]
541
473
def test_canonical_relpath_simple(self):
542
474
f = file('MixedCaseName', 'w')
544
476
actual = osutils.canonical_relpath(self.test_base_dir, 'mixedcasename')
545
self.assertEqual('work/MixedCaseName', actual)
477
self.failUnlessEqual('work/MixedCaseName', actual)
547
479
def test_canonical_relpath_missing_tail(self):
548
480
os.mkdir('MixedCaseParent')
549
481
actual = osutils.canonical_relpath(self.test_base_dir,
550
482
'mixedcaseparent/nochild')
551
self.assertEqual('work/MixedCaseParent/nochild', actual)
483
self.failUnlessEqual('work/MixedCaseParent/nochild', actual)
554
486
class Test_CICPCanonicalRelpath(tests.TestCaseWithTransport):
872
804
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'))
924
807
class TestWin32Funcs(tests.TestCase):
925
808
"""Test that _win32 versions of os utilities return appropriate paths."""
927
810
def test_abspath(self):
928
self.requireFeature(features.win32_feature)
929
811
self.assertEqual('C:/foo', osutils._win32_abspath('C:\\foo'))
930
812
self.assertEqual('C:/foo', osutils._win32_abspath('C:/foo'))
931
813
self.assertEqual('//HOST/path', osutils._win32_abspath(r'\\HOST\path'))
1317
1189
self.requireFeature(UTF8DirReaderFeature)
1318
1190
self._save_platform_info()
1319
1191
win32utils.winver = None # Avoid the win32 detection code
1320
osutils._fs_enc = 'utf-8'
1321
self.assertDirReaderIs(
1322
UTF8DirReaderFeature.module.UTF8DirReader)
1192
osutils._fs_enc = 'UTF-8'
1193
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1324
1195
def test_force_walkdirs_utf8_fs_ascii(self):
1325
1196
self.requireFeature(UTF8DirReaderFeature)
1326
1197
self._save_platform_info()
1327
1198
win32utils.winver = None # Avoid the win32 detection code
1328
osutils._fs_enc = 'ascii'
1329
self.assertDirReaderIs(
1330
UTF8DirReaderFeature.module.UTF8DirReader)
1199
osutils._fs_enc = 'US-ASCII'
1200
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1202
def test_force_walkdirs_utf8_fs_ANSI(self):
1203
self.requireFeature(UTF8DirReaderFeature)
1204
self._save_platform_info()
1205
win32utils.winver = None # Avoid the win32 detection code
1206
osutils._fs_enc = 'ANSI_X3.4-1968'
1207
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1332
1209
def test_force_walkdirs_utf8_fs_latin1(self):
1333
1210
self._save_platform_info()
1334
1211
win32utils.winver = None # Avoid the win32 detection code
1335
osutils._fs_enc = 'iso-8859-1'
1212
osutils._fs_enc = 'latin1'
1336
1213
self.assertDirReaderIs(osutils.UnicodeDirReader)
1338
1215
def test_force_walkdirs_utf8_nt(self):
1962
1837
return filtered_dirblocks
1964
1839
def test_walk_unicode_tree(self):
1965
self.requireFeature(features.UnicodeFilenameFeature)
1840
self.requireFeature(tests.UnicodeFilenameFeature)
1966
1841
tree, expected_dirblocks = self._get_unicode_tree()
1967
1842
self.build_tree(tree)
1968
1843
result = list(osutils._walkdirs_utf8('.'))
1969
1844
self.assertEqual(expected_dirblocks, self._filter_out(result))
1971
1846
def test_symlink(self):
1972
self.requireFeature(features.SymlinkFeature)
1973
self.requireFeature(features.UnicodeFilenameFeature)
1847
self.requireFeature(tests.SymlinkFeature)
1848
self.requireFeature(tests.UnicodeFilenameFeature)
1974
1849
target = u'target\N{Euro Sign}'
1975
1850
link_name = u'l\N{Euro Sign}nk'
1976
1851
os.symlink(target, link_name)
2003
1878
os.symlink(self.target, self.link)
2005
1880
def test_os_readlink_link_encoding(self):
2006
self.assertEqual(self.target, os.readlink(self.link))
1881
if sys.version_info < (2, 6):
1882
self.assertRaises(UnicodeEncodeError, os.readlink, self.link)
1884
self.assertEquals(self.target, os.readlink(self.link))
2008
1886
def test_os_readlink_link_decoding(self):
2009
self.assertEqual(self.target.encode(osutils._fs_enc),
1887
self.assertEquals(self.target.encode(osutils._fs_enc),
2010
1888
os.readlink(self.link.encode(osutils._fs_enc)))
2021
1899
self.assertIsInstance(concurrency, int)
2023
1901
def test_local_concurrency_environment_variable(self):
2024
self.overrideEnv('BZR_CONCURRENCY', '2')
1902
os.environ['BZR_CONCURRENCY'] = '2'
2025
1903
self.assertEqual(2, osutils.local_concurrency(use_cache=False))
2026
self.overrideEnv('BZR_CONCURRENCY', '3')
1904
os.environ['BZR_CONCURRENCY'] = '3'
2027
1905
self.assertEqual(3, osutils.local_concurrency(use_cache=False))
2028
self.overrideEnv('BZR_CONCURRENCY', 'foo')
1906
os.environ['BZR_CONCURRENCY'] = 'foo'
2029
1907
self.assertEqual(1, osutils.local_concurrency(use_cache=False))
2031
1909
def test_option_concurrency(self):
2032
self.overrideEnv('BZR_CONCURRENCY', '1')
1910
os.environ['BZR_CONCURRENCY'] = '1'
2033
1911
self.run_bzr('rocks --concurrency 42')
2034
# Command line overrides environment variable
2035
self.assertEqual('42', os.environ['BZR_CONCURRENCY'])
2036
self.assertEqual(42, osutils.local_concurrency(use_cache=False))
1912
# Command line overrides envrionment variable
1913
self.assertEquals('42', os.environ['BZR_CONCURRENCY'])
1914
self.assertEquals(42, osutils.local_concurrency(use_cache=False))
2039
1917
class TestFailedToLoadExtension(tests.TestCase):
2107
1985
def test_defaults_to_BZR_COLUMNS(self):
2108
1986
# BZR_COLUMNS is set by the test framework
2109
1987
self.assertNotEqual('12', os.environ['BZR_COLUMNS'])
2110
self.overrideEnv('BZR_COLUMNS', '12')
1988
os.environ['BZR_COLUMNS'] = '12'
2111
1989
self.assertEqual(12, osutils.terminal_width())
2113
def test_BZR_COLUMNS_0_no_limit(self):
2114
self.overrideEnv('BZR_COLUMNS', '0')
2115
self.assertEqual(None, osutils.terminal_width())
2117
1991
def test_falls_back_to_COLUMNS(self):
2118
self.overrideEnv('BZR_COLUMNS', None)
1992
del os.environ['BZR_COLUMNS']
2119
1993
self.assertNotEqual('42', os.environ['COLUMNS'])
2120
1994
self.set_fake_tty()
2121
self.overrideEnv('COLUMNS', '42')
1995
os.environ['COLUMNS'] = '42'
2122
1996
self.assertEqual(42, osutils.terminal_width())
2124
1998
def test_tty_default_without_columns(self):
2125
self.overrideEnv('BZR_COLUMNS', None)
2126
self.overrideEnv('COLUMNS', None)
1999
del os.environ['BZR_COLUMNS']
2000
del os.environ['COLUMNS']
2128
2002
def terminal_size(w, h):
2154
2028
self.overrideAttr(termios, 'TIOCGWINSZ')
2155
2029
del termios.TIOCGWINSZ
2156
self.overrideEnv('BZR_COLUMNS', None)
2157
self.overrideEnv('COLUMNS', None)
2030
del os.environ['BZR_COLUMNS']
2031
del os.environ['COLUMNS']
2158
2032
# Whatever the result is, if we don't raise an exception, it's ok.
2159
2033
osutils.terminal_width()
2162
2035
class TestCreationOps(tests.TestCaseInTempDir):
2163
2036
_test_needs_features = [features.chown_feature]
2165
2038
def setUp(self):
2166
super(TestCreationOps, self).setUp()
2039
tests.TestCaseInTempDir.setUp(self)
2167
2040
self.overrideAttr(os, 'chown', self._dummy_chown)
2169
2042
# params set by call to _dummy_chown
2190
2063
osutils.copy_ownership_from_path('test_file')
2192
2065
s = os.stat('..')
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)
2066
self.assertEquals(self.path, 'test_file')
2067
self.assertEquals(self.uid, s.st_uid)
2068
self.assertEquals(self.gid, s.st_gid)
2255
2070
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
2072
def test_ascii_user(self):
2269
self.overrideEnv(self.envvar_to_override(), 'jrandom')
2073
osutils.set_or_unset_env('LOGNAME', 'jrandom')
2270
2074
self.assertEqual(u'jrandom', osutils.getuser_unicode())
2272
2076
def test_unicode_user(self):
2278
2082
% (osutils.get_user_encoding(),))
2279
2083
uni_username = u'jrandom' + uni_val
2280
2084
encoded_username = uni_username.encode(ue)
2281
self.overrideEnv(self.envvar_to_override(), encoded_username)
2085
osutils.set_or_unset_env('LOGNAME', encoded_username)
2282
2086
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")))