426
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)
429
483
class TestLinks(tests.TestCaseInTempDir):
431
485
def test_dereference_path(self):
432
self.requireFeature(tests.SymlinkFeature)
486
self.requireFeature(features.SymlinkFeature)
433
487
cwd = osutils.realpath('.')
435
489
bar_path = osutils.pathjoin(cwd, 'bar')
818
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'))
821
924
class TestWin32Funcs(tests.TestCase):
822
925
"""Test that _win32 versions of os utilities return appropriate paths."""
840
943
osutils._win32_pathjoin('path/to', 'C:/foo'))
841
944
self.assertEqual('path/to/foo',
842
945
osutils._win32_pathjoin('path/to/', 'foo'))
843
self.assertEqual('/foo',
947
def test_pathjoin_late_bugfix(self):
948
if sys.version_info < (2, 7, 6):
952
self.assertEqual(expected,
844
953
osutils._win32_pathjoin('C:/path/to/', '/foo'))
845
self.assertEqual('/foo',
954
self.assertEqual(expected,
846
955
osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
848
957
def test_normpath(self):
1206
1315
self.requireFeature(UTF8DirReaderFeature)
1207
1316
self._save_platform_info()
1208
1317
win32utils.winver = None # Avoid the win32 detection code
1209
osutils._fs_enc = 'UTF-8'
1210
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1318
osutils._fs_enc = 'utf-8'
1319
self.assertDirReaderIs(
1320
UTF8DirReaderFeature.module.UTF8DirReader)
1212
1322
def test_force_walkdirs_utf8_fs_ascii(self):
1213
1323
self.requireFeature(UTF8DirReaderFeature)
1214
1324
self._save_platform_info()
1215
1325
win32utils.winver = None # Avoid the win32 detection code
1216
osutils._fs_enc = 'US-ASCII'
1217
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1219
def test_force_walkdirs_utf8_fs_ANSI(self):
1220
self.requireFeature(UTF8DirReaderFeature)
1221
self._save_platform_info()
1222
win32utils.winver = None # Avoid the win32 detection code
1223
osutils._fs_enc = 'ANSI_X3.4-1968'
1224
self.assertDirReaderIs(UTF8DirReaderFeature.reader)
1326
osutils._fs_enc = 'ascii'
1327
self.assertDirReaderIs(
1328
UTF8DirReaderFeature.module.UTF8DirReader)
1226
1330
def test_force_walkdirs_utf8_fs_latin1(self):
1227
1331
self._save_platform_info()
1228
1332
win32utils.winver = None # Avoid the win32 detection code
1229
osutils._fs_enc = 'latin1'
1333
osutils._fs_enc = 'iso-8859-1'
1230
1334
self.assertDirReaderIs(osutils.UnicodeDirReader)
1232
1336
def test_force_walkdirs_utf8_nt(self):
1856
1960
return filtered_dirblocks
1858
1962
def test_walk_unicode_tree(self):
1859
self.requireFeature(tests.UnicodeFilenameFeature)
1963
self.requireFeature(features.UnicodeFilenameFeature)
1860
1964
tree, expected_dirblocks = self._get_unicode_tree()
1861
1965
self.build_tree(tree)
1862
1966
result = list(osutils._walkdirs_utf8('.'))
1863
1967
self.assertEqual(expected_dirblocks, self._filter_out(result))
1865
1969
def test_symlink(self):
1866
self.requireFeature(tests.SymlinkFeature)
1867
self.requireFeature(tests.UnicodeFilenameFeature)
1970
self.requireFeature(features.SymlinkFeature)
1971
self.requireFeature(features.UnicodeFilenameFeature)
1868
1972
target = u'target\N{Euro Sign}'
1869
1973
link_name = u'l\N{Euro Sign}nk'
1870
1974
os.symlink(target, link_name)
2089
2193
self.assertEquals(self.gid, s.st_gid)
2196
class TestPathFromEnviron(tests.TestCase):
2198
def test_is_unicode(self):
2199
self.overrideEnv('BZR_TEST_PATH', './anywhere at all/')
2200
path = osutils.path_from_environ('BZR_TEST_PATH')
2201
self.assertIsInstance(path, unicode)
2202
self.assertEqual(u'./anywhere at all/', path)
2204
def test_posix_path_env_ascii(self):
2205
self.overrideEnv('BZR_TEST_PATH', '/tmp')
2206
home = osutils._posix_path_from_environ('BZR_TEST_PATH')
2207
self.assertIsInstance(home, unicode)
2208
self.assertEqual(u'/tmp', home)
2210
def test_posix_path_env_unicode(self):
2211
self.requireFeature(features.ByteStringNamedFilesystem)
2212
self.overrideEnv('BZR_TEST_PATH', '/home/\xa7test')
2213
self.overrideAttr(osutils, "_fs_enc", "iso8859-1")
2214
self.assertEqual(u'/home/\xa7test',
2215
osutils._posix_path_from_environ('BZR_TEST_PATH'))
2216
osutils._fs_enc = "iso8859-5"
2217
self.assertEqual(u'/home/\u0407test',
2218
osutils._posix_path_from_environ('BZR_TEST_PATH'))
2219
osutils._fs_enc = "utf-8"
2220
self.assertRaises(errors.BadFilenameEncoding,
2221
osutils._posix_path_from_environ, 'BZR_TEST_PATH')
2224
class TestGetHomeDir(tests.TestCase):
2226
def test_is_unicode(self):
2227
home = osutils._get_home_dir()
2228
self.assertIsInstance(home, unicode)
2230
def test_posix_homeless(self):
2231
self.overrideEnv('HOME', None)
2232
home = osutils._get_home_dir()
2233
self.assertIsInstance(home, unicode)
2235
def test_posix_home_ascii(self):
2236
self.overrideEnv('HOME', '/home/test')
2237
home = osutils._posix_get_home_dir()
2238
self.assertIsInstance(home, unicode)
2239
self.assertEqual(u'/home/test', home)
2241
def test_posix_home_unicode(self):
2242
self.requireFeature(features.ByteStringNamedFilesystem)
2243
self.overrideEnv('HOME', '/home/\xa7test')
2244
self.overrideAttr(osutils, "_fs_enc", "iso8859-1")
2245
self.assertEqual(u'/home/\xa7test', osutils._posix_get_home_dir())
2246
osutils._fs_enc = "iso8859-5"
2247
self.assertEqual(u'/home/\u0407test', osutils._posix_get_home_dir())
2248
osutils._fs_enc = "utf-8"
2249
self.assertRaises(errors.BadFilenameEncoding,
2250
osutils._posix_get_home_dir)
2092
2253
class TestGetuserUnicode(tests.TestCase):
2255
def test_is_unicode(self):
2256
user = osutils.getuser_unicode()
2257
self.assertIsInstance(user, unicode)
2259
def envvar_to_override(self):
2260
if sys.platform == "win32":
2261
# Disable use of platform calls on windows so envvar is used
2262
self.overrideAttr(win32utils, 'has_ctypes', False)
2263
return 'USERNAME' # only variable used on windows
2264
return 'LOGNAME' # first variable checked by getpass.getuser()
2094
2266
def test_ascii_user(self):
2095
self.overrideEnv('LOGNAME', 'jrandom')
2267
self.overrideEnv(self.envvar_to_override(), 'jrandom')
2096
2268
self.assertEqual(u'jrandom', osutils.getuser_unicode())
2098
2270
def test_unicode_user(self):
2104
2276
% (osutils.get_user_encoding(),))
2105
2277
uni_username = u'jrandom' + uni_val
2106
2278
encoded_username = uni_username.encode(ue)
2107
self.overrideEnv('LOGNAME', encoded_username)
2279
self.overrideEnv(self.envvar_to_override(), encoded_username)
2108
2280
self.assertEqual(uni_username, osutils.getuser_unicode())
2109
self.overrideEnv('LOGNAME', u'jrandom\xb6'.encode(ue))
2110
self.assertEqual(u'jrandom\xb6', osutils.getuser_unicode())
2112
def test_no_username_bug_660174(self):
2113
self.requireFeature(features.win32_feature)
2114
for name in ('LOGNAME', 'USER', 'LNAME', 'USERNAME'):
2115
self.overrideEnv(name, None)
2116
self.assertEqual(u'UNKNOWN', osutils.getuser_unicode())
2119
2283
class TestBackupNames(tests.TestCase):
2166
2338
self.assertTrue(osutils.find_executable_on_path('sh') is not None)
2167
2339
self.assertTrue(
2168
2340
osutils.find_executable_on_path('THIS SHOULD NOT EXIST') is None)
2343
class TestEnvironmentErrors(tests.TestCase):
2344
"""Test handling of environmental errors"""
2346
def test_is_oserror(self):
2347
self.assertTrue(osutils.is_environment_error(
2348
OSError(errno.EINVAL, "Invalid parameter")))
2350
def test_is_ioerror(self):
2351
self.assertTrue(osutils.is_environment_error(
2352
IOError(errno.EINVAL, "Invalid parameter")))
2354
def test_is_socket_error(self):
2355
self.assertTrue(osutils.is_environment_error(
2356
socket.error(errno.EINVAL, "Invalid parameter")))
2358
def test_is_select_error(self):
2359
self.assertTrue(osutils.is_environment_error(
2360
select.error(errno.EINVAL, "Invalid parameter")))
2362
def test_is_pywintypes_error(self):
2363
self.requireFeature(features.pywintypes)
2365
self.assertTrue(osutils.is_environment_error(
2366
pywintypes.error(errno.EINVAL, "Invalid parameter", "Caller")))