← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

  • Committer: Andrew Bennetts
  • Date: 2009-10-26 06:23:14 UTC
  • mto: This revision was merged to the branch mainline in revision 4775.
  • Revision ID: andrew.bennetts@canonical.com-20091026062314-gc80ohde1kgp0fp8
Make OperationWithCleanups the only public API in bzrlib.cleanup, add test for it, add support for *args and **kwargs for func and for cleanups, use deque.appendleft rather than list.insert(0, ...).

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_osutils.py
1
 
# Copyright (C) 2005-2010 Canonical Ltd
 
1
# Copyright (C) 2005, 2006, 2007, 2008, 2009 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
33
33
    win32utils,
34
34
    )
35
35
from bzrlib.tests import (
36
 
    features,
37
36
    file_utils,
38
37
    test__walkdirs_win32,
39
38
    )
54
53
 
55
54
UTF8DirReaderFeature = _UTF8DirReaderFeature()
56
55
 
57
 
term_ios_feature = tests.ModuleAvailableFeature('termios')
58
 
 
59
56
 
60
57
def _already_unicode(s):
61
58
    return s
62
59
 
63
60
 
 
61
def _fs_enc_to_unicode(s):
 
62
    return s.decode(osutils._fs_enc)
 
63
 
 
64
 
64
65
def _utf8_to_unicode(s):
65
66
    return s.decode('UTF-8')
66
67
 
83
84
                          dict(_dir_reader_class=_readdir_pyx.UTF8DirReader,
84
85
                               _native_to_unicode=_utf8_to_unicode)))
85
86
 
86
 
    if test__walkdirs_win32.win32_readdir_feature.available():
 
87
    if test__walkdirs_win32.Win32ReadDirFeature.available():
87
88
        try:
88
89
            from bzrlib import _walkdirs_win32
 
90
            # TODO: check on windows, it may be that we need to use/add
 
91
            # safe_unicode instead of _fs_enc_to_unicode
89
92
            scenarios.append(
90
93
                ('win32',
91
94
                 dict(_dir_reader_class=_walkdirs_win32.Win32ReadDir,
92
 
                      _native_to_unicode=_already_unicode)))
 
95
                      _native_to_unicode=_fs_enc_to_unicode)))
93
96
        except ImportError:
94
97
            pass
95
98
    return scenarios
123
126
 
124
127
class TestRename(tests.TestCaseInTempDir):
125
128
 
126
 
    def create_file(self, filename, content):
127
 
        f = open(filename, 'wb')
128
 
        try:
129
 
            f.write(content)
130
 
        finally:
131
 
            f.close()
132
 
 
133
 
    def _fancy_rename(self, a, b):
134
 
        osutils.fancy_rename(a, b, rename_func=os.rename,
135
 
                             unlink_func=os.unlink)
136
 
 
137
129
    def test_fancy_rename(self):
138
130
        # This should work everywhere
139
 
        self.create_file('a', 'something in a\n')
140
 
        self._fancy_rename('a', 'b')
 
131
        def rename(a, b):
 
132
            osutils.fancy_rename(a, b,
 
133
                    rename_func=os.rename,
 
134
                    unlink_func=os.unlink)
 
135
 
 
136
        open('a', 'wb').write('something in a\n')
 
137
        rename('a', 'b')
141
138
        self.failIfExists('a')
142
139
        self.failUnlessExists('b')
143
140
        self.check_file_contents('b', 'something in a\n')
144
141
 
145
 
        self.create_file('a', 'new something in a\n')
146
 
        self._fancy_rename('b', 'a')
 
142
        open('a', 'wb').write('new something in a\n')
 
143
        rename('b', 'a')
147
144
 
148
145
        self.check_file_contents('a', 'something in a\n')
149
146
 
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')
157
 
 
158
 
    def test_fancy_rename_fails_if_source_and_target_missing(self):
159
 
        self.assertRaises((IOError, OSError), self._fancy_rename,
160
 
                          'missingsource', 'missingtarget')
161
 
 
162
147
    def test_rename(self):
163
148
        # Rename should be semi-atomic on all platforms
164
 
        self.create_file('a', 'something in a\n')
 
149
        open('a', 'wb').write('something in a\n')
165
150
        osutils.rename('a', 'b')
166
151
        self.failIfExists('a')
167
152
        self.failUnlessExists('b')
168
153
        self.check_file_contents('b', 'something in a\n')
169
154
 
170
 
        self.create_file('a', 'new something in a\n')
 
155
        open('a', 'wb').write('new something in a\n')
171
156
        osutils.rename('b', 'a')
172
157
 
173
158
        self.check_file_contents('a', 'something in a\n')
184
169
        shape = sorted(os.listdir('.'))
185
170
        self.assertEquals(['A', 'B'], shape)
186
171
 
187
 
    def test_rename_error(self):
188
 
        # We wrap os.rename to make it give an error including the filenames
189
 
        # https://bugs.launchpad.net/bzr/+bug/491763
190
 
        err = self.assertRaises(OSError, osutils.rename,
191
 
            'nonexistent', 'target')
192
 
        self.assertContainsString(str(err), 'nonexistent')
193
 
 
194
172
 
195
173
class TestRandChars(tests.TestCase):
196
174
 
317
295
        self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))
318
296
        self.assertEqual("@", osutils.kind_marker("symlink"))
319
297
        self.assertEqual("+", osutils.kind_marker("tree-reference"))
320
 
        self.assertEqual("", osutils.kind_marker("fifo"))
321
 
        self.assertEqual("", osutils.kind_marker("socket"))
322
 
        self.assertEqual("", osutils.kind_marker("unknown"))
 
298
        self.assertRaises(errors.BzrError, osutils.kind_marker, "unknown")
323
299
 
324
300
 
325
301
class TestUmask(tests.TestCaseInTempDir):
390
366
        # Instead blackbox.test_locale should check for localized
391
367
        # dates once they do occur in output strings.
392
368
 
393
 
    def test_format_date_with_offset_in_original_timezone(self):
394
 
        self.assertEqual("Thu 1970-01-01 00:00:00 +0000",
395
 
            osutils.format_date_with_offset_in_original_timezone(0))
396
 
        self.assertEqual("Fri 1970-01-02 03:46:40 +0000",
397
 
            osutils.format_date_with_offset_in_original_timezone(100000))
398
 
        self.assertEqual("Fri 1970-01-02 05:46:40 +0200",
399
 
            osutils.format_date_with_offset_in_original_timezone(100000, 7200))
400
 
 
401
369
    def test_local_time_offset(self):
402
370
        """Test that local_time_offset() returns a sane value."""
403
371
        offset = osutils.local_time_offset()
999
967
 
1000
968
    def test_osutils_binding(self):
1001
969
        from bzrlib.tests import test__chunks_to_lines
1002
 
        if test__chunks_to_lines.compiled_chunkstolines_feature.available():
 
970
        if test__chunks_to_lines.CompiledChunksToLinesFeature.available():
1003
971
            from bzrlib._chunks_to_lines_pyx import chunks_to_lines
1004
972
        else:
1005
973
            from bzrlib._chunks_to_lines_py import chunks_to_lines
1145
1113
            dirblock[:] = new_dirblock
1146
1114
 
1147
1115
    def _save_platform_info(self):
1148
 
        self.overrideAttr(win32utils, 'winver')
1149
 
        self.overrideAttr(osutils, '_fs_enc')
1150
 
        self.overrideAttr(osutils, '_selected_dir_reader')
 
1116
        cur_winver = win32utils.winver
 
1117
        cur_fs_enc = osutils._fs_enc
 
1118
        cur_dir_reader = osutils._selected_dir_reader
 
1119
        def restore():
 
1120
            win32utils.winver = cur_winver
 
1121
            osutils._fs_enc = cur_fs_enc
 
1122
            osutils._selected_dir_reader = cur_dir_reader
 
1123
        self.addCleanup(restore)
1151
1124
 
1152
1125
    def assertDirReaderIs(self, expected):
1153
1126
        """Assert the right implementation for _walkdirs_utf8 is chosen."""
1186
1159
 
1187
1160
    def test_force_walkdirs_utf8_nt(self):
1188
1161
        # Disabled because the thunk of the whole walkdirs api is disabled.
1189
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1162
        self.requireFeature(test__walkdirs_win32.Win32ReadDirFeature)
1190
1163
        self._save_platform_info()
1191
1164
        win32utils.winver = 'Windows NT'
1192
1165
        from bzrlib._walkdirs_win32 import Win32ReadDir
1193
1166
        self.assertDirReaderIs(Win32ReadDir)
1194
1167
 
1195
1168
    def test_force_walkdirs_utf8_98(self):
1196
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1169
        self.requireFeature(test__walkdirs_win32.Win32ReadDirFeature)
1197
1170
        self._save_platform_info()
1198
1171
        win32utils.winver = 'Windows 98'
1199
1172
        self.assertDirReaderIs(osutils.UnicodeDirReader)
1350
1323
        self.assertEqual(expected_dirblocks, result)
1351
1324
 
1352
1325
    def test__walkdirs_utf8_win32readdir(self):
1353
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1326
        self.requireFeature(test__walkdirs_win32.Win32ReadDirFeature)
1354
1327
        self.requireFeature(tests.UnicodeFilenameFeature)
1355
1328
        from bzrlib._walkdirs_win32 import Win32ReadDir
1356
1329
        self._save_platform_info()
1407
1380
 
1408
1381
    def test__walkdirs_utf_win32_find_file_stat_file(self):
1409
1382
        """make sure our Stat values are valid"""
1410
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1383
        self.requireFeature(test__walkdirs_win32.Win32ReadDirFeature)
1411
1384
        self.requireFeature(tests.UnicodeFilenameFeature)
1412
1385
        from bzrlib._walkdirs_win32 import Win32ReadDir
1413
1386
        name0u = u'0file-\xb6'
1431
1404
 
1432
1405
    def test__walkdirs_utf_win32_find_file_stat_directory(self):
1433
1406
        """make sure our Stat values are valid"""
1434
 
        self.requireFeature(test__walkdirs_win32.win32_readdir_feature)
 
1407
        self.requireFeature(test__walkdirs_win32.Win32ReadDirFeature)
1435
1408
        self.requireFeature(tests.UnicodeFilenameFeature)
1436
1409
        from bzrlib._walkdirs_win32 import Win32ReadDir
1437
1410
        name0u = u'0dir-\u062c\u0648'
1586
1559
        def cleanup():
1587
1560
            if 'BZR_TEST_ENV_VAR' in os.environ:
1588
1561
                del os.environ['BZR_TEST_ENV_VAR']
 
1562
 
1589
1563
        self.addCleanup(cleanup)
1590
1564
 
1591
1565
    def test_set(self):
1639
1613
        text = 'test\r\nwith\nall\rpossible line endings\r\n'
1640
1614
        self.build_tree_contents([('foo', text)])
1641
1615
        expected_sha = osutils.sha_string(text)
1642
 
        f = open('foo', 'rb')
 
1616
        f = open('foo')
1643
1617
        self.addCleanup(f.close)
1644
1618
        size, sha = osutils.size_sha_file(f)
1645
1619
        self.assertEqual(38, size)
1702
1676
 
1703
1677
    def setUp(self):
1704
1678
        tests.TestCaseInTempDir.setUp(self)
1705
 
        self.overrideAttr(osutils,
1706
 
                          '_selected_dir_reader', self._dir_reader_class())
 
1679
 
 
1680
        # Save platform specific info and reset it
 
1681
        cur_dir_reader = osutils._selected_dir_reader
 
1682
 
 
1683
        def restore():
 
1684
            osutils._selected_dir_reader = cur_dir_reader
 
1685
        self.addCleanup(restore)
 
1686
 
 
1687
        osutils._selected_dir_reader = self._dir_reader_class()
1707
1688
 
1708
1689
    def _get_ascii_tree(self):
1709
1690
        tree = [
1854
1835
 
1855
1836
class TestConcurrency(tests.TestCase):
1856
1837
 
1857
 
    def setUp(self):
1858
 
        super(TestConcurrency, self).setUp()
1859
 
        self.overrideAttr(osutils, '_cached_local_concurrency')
1860
 
 
1861
1838
    def test_local_concurrency(self):
1862
1839
        concurrency = osutils.local_concurrency()
1863
1840
        self.assertIsInstance(concurrency, int)
1864
1841
 
1865
 
    def test_local_concurrency_environment_variable(self):
1866
 
        os.environ['BZR_CONCURRENCY'] = '2'
1867
 
        self.assertEqual(2, osutils.local_concurrency(use_cache=False))
1868
 
        os.environ['BZR_CONCURRENCY'] = '3'
1869
 
        self.assertEqual(3, osutils.local_concurrency(use_cache=False))
1870
 
        os.environ['BZR_CONCURRENCY'] = 'foo'
1871
 
        self.assertEqual(1, osutils.local_concurrency(use_cache=False))
1872
 
 
1873
 
    def test_option_concurrency(self):
1874
 
        os.environ['BZR_CONCURRENCY'] = '1'
1875
 
        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))
1879
 
 
1880
1842
 
1881
1843
class TestFailedToLoadExtension(tests.TestCase):
1882
1844
 
1889
1851
 
1890
1852
    def setUp(self):
1891
1853
        super(TestFailedToLoadExtension, self).setUp()
1892
 
        self.overrideAttr(osutils, '_extension_load_failures', [])
 
1854
        self.saved_failures = osutils._extension_load_failures[:]
 
1855
        del osutils._extension_load_failures[:]
 
1856
        self.addCleanup(self.restore_failures)
 
1857
 
 
1858
    def restore_failures(self):
 
1859
        osutils._extension_load_failures = self.saved_failures
1893
1860
 
1894
1861
    def test_failure_to_load(self):
1895
1862
        self._try_loading()
1913
1880
            r"bzr: warning: some compiled extensions could not be loaded; "
1914
1881
            "see <https://answers\.launchpad\.net/bzr/\+faq/703>\n"
1915
1882
            )
1916
 
 
1917
 
 
1918
 
class TestTerminalWidth(tests.TestCase):
1919
 
 
1920
 
    def replace_stdout(self, new):
1921
 
        self.overrideAttr(sys, 'stdout', new)
1922
 
 
1923
 
    def replace__terminal_size(self, new):
1924
 
        self.overrideAttr(osutils, '_terminal_size', new)
1925
 
 
1926
 
    def set_fake_tty(self):
1927
 
 
1928
 
        class I_am_a_tty(object):
1929
 
            def isatty(self):
1930
 
                return True
1931
 
 
1932
 
        self.replace_stdout(I_am_a_tty())
1933
 
 
1934
 
    def test_default_values(self):
1935
 
        self.assertEqual(80, osutils.default_terminal_width)
1936
 
 
1937
 
    def test_defaults_to_BZR_COLUMNS(self):
1938
 
        # BZR_COLUMNS is set by the test framework
1939
 
        self.assertNotEqual('12', os.environ['BZR_COLUMNS'])
1940
 
        os.environ['BZR_COLUMNS'] = '12'
1941
 
        self.assertEqual(12, osutils.terminal_width())
1942
 
 
1943
 
    def test_falls_back_to_COLUMNS(self):
1944
 
        del os.environ['BZR_COLUMNS']
1945
 
        self.assertNotEqual('42', os.environ['COLUMNS'])
1946
 
        self.set_fake_tty()
1947
 
        os.environ['COLUMNS'] = '42'
1948
 
        self.assertEqual(42, osutils.terminal_width())
1949
 
 
1950
 
    def test_tty_default_without_columns(self):
1951
 
        del os.environ['BZR_COLUMNS']
1952
 
        del os.environ['COLUMNS']
1953
 
 
1954
 
        def terminal_size(w, h):
1955
 
            return 42, 42
1956
 
 
1957
 
        self.set_fake_tty()
1958
 
        # We need to override the osutils definition as it depends on the
1959
 
        # running environment that we can't control (PQM running without a
1960
 
        # controlling terminal is one example).
1961
 
        self.replace__terminal_size(terminal_size)
1962
 
        self.assertEqual(42, osutils.terminal_width())
1963
 
 
1964
 
    def test_non_tty_default_without_columns(self):
1965
 
        del os.environ['BZR_COLUMNS']
1966
 
        del os.environ['COLUMNS']
1967
 
        self.replace_stdout(None)
1968
 
        self.assertEqual(None, osutils.terminal_width())
1969
 
 
1970
 
    def test_no_TIOCGWINSZ(self):
1971
 
        self.requireFeature(term_ios_feature)
1972
 
        termios = term_ios_feature.module
1973
 
        # bug 63539 is about a termios without TIOCGWINSZ attribute
1974
 
        try:
1975
 
            orig = termios.TIOCGWINSZ
1976
 
        except AttributeError:
1977
 
            # We won't remove TIOCGWINSZ, because it doesn't exist anyway :)
1978
 
            pass
1979
 
        else:
1980
 
            self.overrideAttr(termios, 'TIOCGWINSZ')
1981
 
            del termios.TIOCGWINSZ
1982
 
        del os.environ['BZR_COLUMNS']
1983
 
        del os.environ['COLUMNS']
1984
 
        # Whatever the result is, if we don't raise an exception, it's ok.
1985
 
        osutils.terminal_width()
1986
 
 
1987
 
class TestCreationOps(tests.TestCaseInTempDir):
1988
 
    _test_needs_features = [features.chown_feature]
1989
 
 
1990
 
    def setUp(self):
1991
 
        tests.TestCaseInTempDir.setUp(self)
1992
 
        self.overrideAttr(os, 'chown', self._dummy_chown)
1993
 
 
1994
 
        # params set by call to _dummy_chown
1995
 
        self.path = self.uid = self.gid = None
1996
 
 
1997
 
    def _dummy_chown(self, path, uid, gid):
1998
 
        self.path, self.uid, self.gid = path, uid, gid
1999
 
 
2000
 
    def test_copy_ownership_from_path(self):
2001
 
        """copy_ownership_from_path test with specified src."""
2002
 
        ownsrc = '/'
2003
 
        f = open('test_file', 'wt')
2004
 
        osutils.copy_ownership_from_path('test_file', ownsrc)
2005
 
 
2006
 
        s = os.stat(ownsrc)
2007
 
        self.assertEquals(self.path, 'test_file')
2008
 
        self.assertEquals(self.uid, s.st_uid)
2009
 
        self.assertEquals(self.gid, s.st_gid)
2010
 
 
2011
 
    def test_copy_ownership_nonesrc(self):
2012
 
        """copy_ownership_from_path test with src=None."""
2013
 
        f = open('test_file', 'wt')
2014
 
        # should use parent dir for permissions
2015
 
        osutils.copy_ownership_from_path('test_file')
2016
 
 
2017
 
        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)