437
426
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
429
class TestLinks(tests.TestCaseInTempDir):
485
431
def test_dereference_path(self):
872
818
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
821
class TestWin32Funcs(tests.TestCase):
925
822
"""Test that _win32 versions of os utilities return appropriate paths."""
1317
1214
self.requireFeature(UTF8DirReaderFeature)
1318
1215
self._save_platform_info()
1319
1216
win32utils.winver = None # Avoid the win32 detection code
1320
osutils._fs_enc = 'ascii'
1217
osutils._fs_enc = 'US-ASCII'
1218
self.assertDirReaderIs(
1219
UTF8DirReaderFeature.module.UTF8DirReader)
1221
def test_force_walkdirs_utf8_fs_ANSI(self):
1222
self.requireFeature(UTF8DirReaderFeature)
1223
self._save_platform_info()
1224
win32utils.winver = None # Avoid the win32 detection code
1225
osutils._fs_enc = 'ANSI_X3.4-1968'
1321
1226
self.assertDirReaderIs(
1322
1227
UTF8DirReaderFeature.module.UTF8DirReader)
1324
1229
def test_force_walkdirs_utf8_fs_latin1(self):
1325
1230
self._save_platform_info()
1326
1231
win32utils.winver = None # Avoid the win32 detection code
1327
osutils._fs_enc = 'iso-8859-1'
1232
osutils._fs_enc = 'latin1'
1328
1233
self.assertDirReaderIs(osutils.UnicodeDirReader)
1330
1235
def test_force_walkdirs_utf8_nt(self):
2187
2092
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
2095
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
2097
def test_ascii_user(self):
2261
self.overrideEnv(self.envvar_to_override(), 'jrandom')
2098
self.overrideEnv('LOGNAME', 'jrandom')
2262
2099
self.assertEqual(u'jrandom', osutils.getuser_unicode())
2264
2101
def test_unicode_user(self):
2270
2107
% (osutils.get_user_encoding(),))
2271
2108
uni_username = u'jrandom' + uni_val
2272
2109
encoded_username = uni_username.encode(ue)
2273
self.overrideEnv(self.envvar_to_override(), encoded_username)
2110
self.overrideEnv('LOGNAME', encoded_username)
2274
2111
self.assertEqual(uni_username, osutils.getuser_unicode())
2112
self.overrideEnv('LOGNAME', u'jrandom\xb6'.encode(ue))
2113
self.assertEqual(u'jrandom\xb6', osutils.getuser_unicode())
2115
def test_no_username_bug_660174(self):
2116
self.requireFeature(features.win32_feature)
2117
for name in ('LOGNAME', 'USER', 'LNAME', 'USERNAME'):
2118
self.overrideEnv(name, None)
2119
self.assertEqual(u'UNKNOWN', osutils.getuser_unicode())
2277
2122
class TestBackupNames(tests.TestCase):
2332
2169
self.assertTrue(osutils.find_executable_on_path('sh') is not None)
2333
2170
self.assertTrue(
2334
2171
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")))