~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2005-2010 Canonical Ltd

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

"""Tests for the osutils wrapper."""

from cStringIO import StringIO

import errno

import os

import re

import socket

import stat

import sys

import time

from bzrlib import (

    errors,

    lazy_regex,

    osutils,

    symbol_versioning,

    tests,

    trace,

    win32utils,

    )

from bzrlib.tests import (

    features,

    file_utils,

    test__walkdirs_win32,

    )

class _UTF8DirReaderFeature(tests.Feature):

    def _probe(self):

        try:

            from bzrlib import _readdir_pyx

            self.reader = _readdir_pyx.UTF8DirReader

            return True

        except ImportError:

            return False

    def feature_name(self):

        return 'bzrlib._readdir_pyx'

UTF8DirReaderFeature = _UTF8DirReaderFeature()

term_ios_feature = tests.ModuleAvailableFeature('termios')

def _already_unicode(s):

    return s

def _utf8_to_unicode(s):

    return s.decode('UTF-8')

def dir_reader_scenarios():

    # For each dir reader we define:

    # - native_to_unicode: a function converting the native_abspath as returned

    #   by DirReader.read_dir to its unicode representation

    # UnicodeDirReader is the fallback, it should be tested on all platforms.

    scenarios = [('unicode',

                  dict(_dir_reader_class=osutils.UnicodeDirReader,

                       _native_to_unicode=_already_unicode))]

    # Some DirReaders are platform specific and even there they may not be

    # available.

    if UTF8DirReaderFeature.available():

        from bzrlib import _readdir_pyx

        scenarios.append(('utf8',

                          dict(_dir_reader_class=_readdir_pyx.UTF8DirReader,

                               _native_to_unicode=_utf8_to_unicode)))

    if test__walkdirs_win32.win32_readdir_feature.available():

        try:

            from bzrlib import _walkdirs_win32

            scenarios.append(

                ('win32',

                 dict(_dir_reader_class=_walkdirs_win32.Win32ReadDir,

                      _native_to_unicode=_already_unicode)))

        except ImportError:

            pass

    return scenarios

def load_tests(basic_tests, module, loader):

    suite = loader.suiteClass()

    dir_reader_tests, remaining_tests = tests.split_suite_by_condition(

        basic_tests, tests.condition_isinstance(TestDirReader))

    tests.multiply_tests(dir_reader_tests, dir_reader_scenarios(), suite)

    suite.addTest(remaining_tests)

    return suite

class TestContainsWhitespace(tests.TestCase):

    def test_contains_whitespace(self):

        self.failUnless(osutils.contains_whitespace(u' '))

        self.failUnless(osutils.contains_whitespace(u'hello there'))

        self.failUnless(osutils.contains_whitespace(u'hellothere\n'))

        self.failUnless(osutils.contains_whitespace(u'hello\nthere'))

        self.failUnless(osutils.contains_whitespace(u'hello\rthere'))

        self.failUnless(osutils.contains_whitespace(u'hello\tthere'))

        # \xa0 is "Non-breaking-space" which on some python locales thinks it

        # is whitespace, but we do not.

        self.failIf(osutils.contains_whitespace(u''))

        self.failIf(osutils.contains_whitespace(u'hellothere'))

        self.failIf(osutils.contains_whitespace(u'hello\xa0there'))

class TestRename(tests.TestCaseInTempDir):

    def create_file(self, filename, content):

        f = open(filename, 'wb')

        try:

            f.write(content)

        finally:

            f.close()

    def _fancy_rename(self, a, b):

        osutils.fancy_rename(a, b, rename_func=os.rename,

                             unlink_func=os.unlink)

    def test_fancy_rename(self):

        # This should work everywhere

        self.create_file('a', 'something in a\n')

        self._fancy_rename('a', 'b')

        self.failIfExists('a')

        self.failUnlessExists('b')

        self.check_file_contents('b', 'something in a\n')

        self.create_file('a', 'new something in a\n')

        self._fancy_rename('b', 'a')

        self.check_file_contents('a', 'something in a\n')

    def test_fancy_rename_fails_source_missing(self):

        # An exception should be raised, and the target should be left in place

        self.create_file('target', 'data in target\n')

        self.assertRaises((IOError, OSError), self._fancy_rename,

                          'missingsource', 'target')

        self.failUnlessExists('target')

        self.check_file_contents('target', 'data in target\n')

    def test_fancy_rename_fails_if_source_and_target_missing(self):

        self.assertRaises((IOError, OSError), self._fancy_rename,

                          'missingsource', 'missingtarget')

    def test_rename(self):

        # Rename should be semi-atomic on all platforms

        self.create_file('a', 'something in a\n')

        osutils.rename('a', 'b')

        self.failIfExists('a')

        self.failUnlessExists('b')

        self.check_file_contents('b', 'something in a\n')

        self.create_file('a', 'new something in a\n')

        osutils.rename('b', 'a')

        self.check_file_contents('a', 'something in a\n')

    # TODO: test fancy_rename using a MemoryTransport

    def test_rename_change_case(self):

        # on Windows we should be able to change filename case by rename

        self.build_tree(['a', 'b/'])

        osutils.rename('a', 'A')

        osutils.rename('b', 'B')

        # we can't use failUnlessExists on case-insensitive filesystem

        # so try to check shape of the tree

        shape = sorted(os.listdir('.'))

        self.assertEquals(['A', 'B'], shape)

class TestRandChars(tests.TestCase):

    def test_01_rand_chars_empty(self):

        result = osutils.rand_chars(0)

        self.assertEqual(result, '')

    def test_02_rand_chars_100(self):

        result = osutils.rand_chars(100)

        self.assertEqual(len(result), 100)

        self.assertEqual(type(result), str)

        self.assertContainsRe(result, r'^[a-z0-9]{100}$')

class TestIsInside(tests.TestCase):

    def test_is_inside(self):

        is_inside = osutils.is_inside

        self.assertTrue(is_inside('src', 'src/foo.c'))

        self.assertFalse(is_inside('src', 'srccontrol'))

        self.assertTrue(is_inside('src', 'src/a/a/a/foo.c'))

        self.assertTrue(is_inside('foo.c', 'foo.c'))

        self.assertFalse(is_inside('foo.c', ''))

        self.assertTrue(is_inside('', 'foo.c'))

    def test_is_inside_any(self):

        SRC_FOO_C = osutils.pathjoin('src', 'foo.c')

        for dirs, fn in [(['src', 'doc'], SRC_FOO_C),

                         (['src'], SRC_FOO_C),

                         (['src'], 'src'),

                         ]:

            self.assert_(osutils.is_inside_any(dirs, fn))

        for dirs, fn in [(['src'], 'srccontrol'),

                         (['src'], 'srccontrol/foo')]:

            self.assertFalse(osutils.is_inside_any(dirs, fn))

    def test_is_inside_or_parent_of_any(self):

        for dirs, fn in [(['src', 'doc'], 'src/foo.c'),

                         (['src'], 'src/foo.c'),

                         (['src/bar.c'], 'src'),

                         (['src/bar.c', 'bla/foo.c'], 'src'),

                         (['src'], 'src'),

                         ]:

            self.assert_(osutils.is_inside_or_parent_of_any(dirs, fn))

        for dirs, fn in [(['src'], 'srccontrol'),

                         (['srccontrol/foo.c'], 'src'),

                         (['src'], 'srccontrol/foo')]:

            self.assertFalse(osutils.is_inside_or_parent_of_any(dirs, fn))

class TestRmTree(tests.TestCaseInTempDir):

    def test_rmtree(self):

        # Check to remove tree with read-only files/dirs

        os.mkdir('dir')

        f = file('dir/file', 'w')

        f.write('spam')

        f.close()

        # would like to also try making the directory readonly, but at the

        # moment python shutil.rmtree doesn't handle that properly - it would

        # need to chmod the directory before removing things inside it - deferred

        # for now -- mbp 20060505

        # osutils.make_readonly('dir')

        osutils.make_readonly('dir/file')

        osutils.rmtree('dir')

        self.failIfExists('dir/file')

        self.failIfExists('dir')

class TestDeleteAny(tests.TestCaseInTempDir):

    def test_delete_any_readonly(self):

        # from <https://bugs.launchpad.net/bzr/+bug/218206>

        self.build_tree(['d/', 'f'])

        osutils.make_readonly('d')

        osutils.make_readonly('f')

        osutils.delete_any('f')

        osutils.delete_any('d')

class TestKind(tests.TestCaseInTempDir):

    def test_file_kind(self):

        self.build_tree(['file', 'dir/'])

        self.assertEquals('file', osutils.file_kind('file'))

        self.assertEquals('directory', osutils.file_kind('dir/'))

        if osutils.has_symlinks():

            os.symlink('symlink', 'symlink')

            self.assertEquals('symlink', osutils.file_kind('symlink'))

        # TODO: jam 20060529 Test a block device

        try:

            os.lstat('/dev/null')

        except OSError, e:

            if e.errno not in (errno.ENOENT,):

                raise

        else:

            self.assertEquals('chardev', osutils.file_kind('/dev/null'))

        mkfifo = getattr(os, 'mkfifo', None)

        if mkfifo:

            mkfifo('fifo')

            try:

                self.assertEquals('fifo', osutils.file_kind('fifo'))

            finally:

                os.remove('fifo')

        AF_UNIX = getattr(socket, 'AF_UNIX', None)

        if AF_UNIX:

            s = socket.socket(AF_UNIX)

            s.bind('socket')

            try:

                self.assertEquals('socket', osutils.file_kind('socket'))

            finally:

                os.remove('socket')

    def test_kind_marker(self):

        self.assertEqual("", osutils.kind_marker("file"))

        self.assertEqual("/", osutils.kind_marker('directory'))

        self.assertEqual("/", osutils.kind_marker(osutils._directory_kind))

        self.assertEqual("@", osutils.kind_marker("symlink"))

        self.assertEqual("+", osutils.kind_marker("tree-reference"))

        self.assertEqual("", osutils.kind_marker("fifo"))

        self.assertEqual("", osutils.kind_marker("socket"))

        self.assertEqual("", osutils.kind_marker("unknown"))

class TestUmask(tests.TestCaseInTempDir):

    def test_get_umask(self):

        if sys.platform == 'win32':

            # umask always returns '0', no way to set it

            self.assertEqual(0, osutils.get_umask())

            return

        orig_umask = osutils.get_umask()

        self.addCleanup(os.umask, orig_umask)

        os.umask(0222)

        self.assertEqual(0222, osutils.get_umask())

        os.umask(0022)

        self.assertEqual(0022, osutils.get_umask())

        os.umask(0002)

        self.assertEqual(0002, osutils.get_umask())

        os.umask(0027)

        self.assertEqual(0027, osutils.get_umask())

class TestDateTime(tests.TestCase):

    def assertFormatedDelta(self, expected, seconds):

        """Assert osutils.format_delta formats as expected"""

        actual = osutils.format_delta(seconds)

        self.assertEqual(expected, actual)

    def test_format_delta(self):

        self.assertFormatedDelta('0 seconds ago', 0)

        self.assertFormatedDelta('1 second ago', 1)

        self.assertFormatedDelta('10 seconds ago', 10)

        self.assertFormatedDelta('59 seconds ago', 59)

        self.assertFormatedDelta('89 seconds ago', 89)

        self.assertFormatedDelta('1 minute, 30 seconds ago', 90)

        self.assertFormatedDelta('3 minutes, 0 seconds ago', 180)

        self.assertFormatedDelta('3 minutes, 1 second ago', 181)

        self.assertFormatedDelta('10 minutes, 15 seconds ago', 615)

        self.assertFormatedDelta('30 minutes, 59 seconds ago', 1859)

        self.assertFormatedDelta('31 minutes, 0 seconds ago', 1860)

        self.assertFormatedDelta('60 minutes, 0 seconds ago', 3600)

        self.assertFormatedDelta('89 minutes, 59 seconds ago', 5399)

        self.assertFormatedDelta('1 hour, 30 minutes ago', 5400)

        self.assertFormatedDelta('2 hours, 30 minutes ago', 9017)

        self.assertFormatedDelta('10 hours, 0 minutes ago', 36000)

        self.assertFormatedDelta('24 hours, 0 minutes ago', 86400)

        self.assertFormatedDelta('35 hours, 59 minutes ago', 129599)

        self.assertFormatedDelta('36 hours, 0 minutes ago', 129600)

        self.assertFormatedDelta('36 hours, 0 minutes ago', 129601)

        self.assertFormatedDelta('36 hours, 1 minute ago', 129660)

        self.assertFormatedDelta('36 hours, 1 minute ago', 129661)

        self.assertFormatedDelta('84 hours, 10 minutes ago', 303002)

        # We handle when time steps the wrong direction because computers

        # don't have synchronized clocks.

        self.assertFormatedDelta('84 hours, 10 minutes in the future', -303002)

        self.assertFormatedDelta('1 second in the future', -1)

        self.assertFormatedDelta('2 seconds in the future', -2)

    def test_format_date(self):

        self.assertRaises(errors.UnsupportedTimezoneFormat,

            osutils.format_date, 0, timezone='foo')

        self.assertIsInstance(osutils.format_date(0), str)

        self.assertIsInstance(osutils.format_local_date(0), unicode)

        # Testing for the actual value of the local weekday without

        # duplicating the code from format_date is difficult.

        # Instead blackbox.test_locale should check for localized

        # dates once they do occur in output strings.

    def test_format_date_with_offset_in_original_timezone(self):

        self.assertEqual("Thu 1970-01-01 00:00:00 +0000",

            osutils.format_date_with_offset_in_original_timezone(0))

        self.assertEqual("Fri 1970-01-02 03:46:40 +0000",

            osutils.format_date_with_offset_in_original_timezone(100000))

        self.assertEqual("Fri 1970-01-02 05:46:40 +0200",

            osutils.format_date_with_offset_in_original_timezone(100000, 7200))

    def test_local_time_offset(self):

        """Test that local_time_offset() returns a sane value."""

        offset = osutils.local_time_offset()

        self.assertTrue(isinstance(offset, int))

        # Test that the offset is no more than a eighteen hours in

        # either direction.

        # Time zone handling is system specific, so it is difficult to

        # do more specific tests, but a value outside of this range is

        # probably wrong.

        eighteen_hours = 18 * 3600

        self.assertTrue(-eighteen_hours < offset < eighteen_hours)

    def test_local_time_offset_with_timestamp(self):

        """Test that local_time_offset() works with a timestamp."""

        offset = osutils.local_time_offset(1000000000.1234567)

        self.assertTrue(isinstance(offset, int))

        eighteen_hours = 18 * 3600

        self.assertTrue(-eighteen_hours < offset < eighteen_hours)

class TestLinks(tests.TestCaseInTempDir):

    def test_dereference_path(self):

        self.requireFeature(tests.SymlinkFeature)

        cwd = osutils.realpath('.')

        os.mkdir('bar')

        bar_path = osutils.pathjoin(cwd, 'bar')

        # Using './' to avoid bug #1213894 (first path component not

        # dereferenced) in Python 2.4.1 and earlier

        self.assertEqual(bar_path, osutils.realpath('./bar'))

        os.symlink('bar', 'foo')

        self.assertEqual(bar_path, osutils.realpath('./foo'))

        # Does not dereference terminal symlinks

        foo_path = osutils.pathjoin(cwd, 'foo')

        self.assertEqual(foo_path, osutils.dereference_path('./foo'))

        # Dereferences parent symlinks

        os.mkdir('bar/baz')

        baz_path = osutils.pathjoin(bar_path, 'baz')

        self.assertEqual(baz_path, osutils.dereference_path('./foo/baz'))

        # Dereferences parent symlinks that are the first path element

        self.assertEqual(baz_path, osutils.dereference_path('foo/baz'))

        # Dereferences parent symlinks in absolute paths

        foo_baz_path = osutils.pathjoin(foo_path, 'baz')

        self.assertEqual(baz_path, osutils.dereference_path(foo_baz_path))

    def test_changing_access(self):

        f = file('file', 'w')

        f.write('monkey')

        f.close()

        # Make a file readonly

        osutils.make_readonly('file')

        mode = os.lstat('file').st_mode

        self.assertEqual(mode, mode & 0777555)

        # Make a file writable

        osutils.make_writable('file')

        mode = os.lstat('file').st_mode

        self.assertEqual(mode, mode | 0200)

        if osutils.has_symlinks():

            # should not error when handed a symlink

            os.symlink('nonexistent', 'dangling')

            osutils.make_readonly('dangling')

            osutils.make_writable('dangling')

    def test_host_os_dereferences_symlinks(self):

        osutils.host_os_dereferences_symlinks()

class TestCanonicalRelPath(tests.TestCaseInTempDir):

    _test_needs_features = [tests.CaseInsCasePresFilenameFeature]

    def test_canonical_relpath_simple(self):

        f = file('MixedCaseName', 'w')

        f.close()

        actual = osutils.canonical_relpath(self.test_base_dir, 'mixedcasename')

        self.failUnlessEqual('work/MixedCaseName', actual)

    def test_canonical_relpath_missing_tail(self):

        os.mkdir('MixedCaseParent')

        actual = osutils.canonical_relpath(self.test_base_dir,

                                           'mixedcaseparent/nochild')

        self.failUnlessEqual('work/MixedCaseParent/nochild', actual)

class Test_CICPCanonicalRelpath(tests.TestCaseWithTransport):

    def assertRelpath(self, expected, base, path):

        actual = osutils._cicp_canonical_relpath(base, path)

        self.assertEqual(expected, actual)

    def test_simple(self):

        self.build_tree(['MixedCaseName'])

        base = osutils.realpath(self.get_transport('.').local_abspath('.'))

        self.assertRelpath('MixedCaseName', base, 'mixedcAsename')

    def test_subdir_missing_tail(self):

        self.build_tree(['MixedCaseParent/', 'MixedCaseParent/a_child'])

        base = osutils.realpath(self.get_transport('.').local_abspath('.'))

        self.assertRelpath('MixedCaseParent/a_child', base,

                           'MixedCaseParent/a_child')

        self.assertRelpath('MixedCaseParent/a_child', base,

                           'MixedCaseParent/A_Child')

        self.assertRelpath('MixedCaseParent/not_child', base,

                           'MixedCaseParent/not_child')

    def test_at_root_slash(self):

        # We can't test this on Windows, because it has a 'MIN_ABS_PATHLENGTH'

        # check...

        if osutils.MIN_ABS_PATHLENGTH > 1:

            raise tests.TestSkipped('relpath requires %d chars'

                                    % osutils.MIN_ABS_PATHLENGTH)

        self.assertRelpath('foo', '/', '/foo')

    def test_at_root_drive(self):

        if sys.platform != 'win32':

            raise tests.TestNotApplicable('we can only test drive-letter relative'

                                          ' paths on Windows where we have drive'

                                          ' letters.')

        # see bug #322807

        # The specific issue is that when at the root of a drive, 'abspath'

        # returns "C:/" or just "/". However, the code assumes that abspath

        # always returns something like "C:/foo" or "/foo" (no trailing slash).

        self.assertRelpath('foo', 'C:/', 'C:/foo')

        self.assertRelpath('foo', 'X:/', 'X:/foo')

        self.assertRelpath('foo', 'X:/', 'X://foo')

class TestPumpFile(tests.TestCase):

    """Test pumpfile method."""

    def setUp(self):

        tests.TestCase.setUp(self)

        # create a test datablock

        self.block_size = 512

        pattern = '0123456789ABCDEF'

        self.test_data = pattern * (3 * self.block_size / len(pattern))

        self.test_data_len = len(self.test_data)

    def test_bracket_block_size(self):

        """Read data in blocks with the requested read size bracketing the

        block size."""

        # make sure test data is larger than max read size

        self.assertTrue(self.test_data_len > self.block_size)

        from_file = file_utils.FakeReadFile(self.test_data)

        to_file = StringIO()

        # read (max / 2) bytes and verify read size wasn't affected

        num_bytes_to_read = self.block_size / 2

        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)

        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)

        self.assertEqual(from_file.get_read_count(), 1)

        # read (max) bytes and verify read size wasn't affected

        num_bytes_to_read = self.block_size

        from_file.reset_read_count()

        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)

        self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)

        self.assertEqual(from_file.get_read_count(), 1)

        # read (max + 1) bytes and verify read size was limited

        num_bytes_to_read = self.block_size + 1

        from_file.reset_read_count()

        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)

        self.assertEqual(from_file.get_max_read_size(), self.block_size)

        self.assertEqual(from_file.get_read_count(), 2)

        # finish reading the rest of the data

        num_bytes_to_read = self.test_data_len - to_file.tell()

        osutils.pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)

        # report error if the data wasn't equal (we only report the size due

        # to the length of the data)

        response_data = to_file.getvalue()

        if response_data != self.test_data:

            message = "Data not equal.  Expected %d bytes, received %d."

            self.fail(message % (len(response_data), self.test_data_len))

    def test_specified_size(self):

        """Request a transfer larger than the maximum block size and verify

        that the maximum read doesn't exceed the block_size."""

        # make sure test data is larger than max read size

        self.assertTrue(self.test_data_len > self.block_size)

        # retrieve data in blocks

        from_file = file_utils.FakeReadFile(self.test_data)

        to_file = StringIO()

        osutils.pumpfile(from_file, to_file, self.test_data_len,

                         self.block_size)

        # verify read size was equal to the maximum read size

        self.assertTrue(from_file.get_max_read_size() > 0)

        self.assertEqual(from_file.get_max_read_size(), self.block_size)

        self.assertEqual(from_file.get_read_count(), 3)

        # report error if the data wasn't equal (we only report the size due

        # to the length of the data)

        response_data = to_file.getvalue()

        if response_data != self.test_data:

            message = "Data not equal.  Expected %d bytes, received %d."

            self.fail(message % (len(response_data), self.test_data_len))

    def test_to_eof(self):

        """Read to end-of-file and verify that the reads are not larger than

        the maximum read size."""

        # make sure test data is larger than max read size

        self.assertTrue(self.test_data_len > self.block_size)

        # retrieve data to EOF

        from_file = file_utils.FakeReadFile(self.test_data)

        to_file = StringIO()

        osutils.pumpfile(from_file, to_file, -1, self.block_size)

        # verify read size was equal to the maximum read size

        self.assertEqual(from_file.get_max_read_size(), self.block_size)

        self.assertEqual(from_file.get_read_count(), 4)

        # report error if the data wasn't equal (we only report the size due

        # to the length of the data)

        response_data = to_file.getvalue()

        if response_data != self.test_data:

            message = "Data not equal.  Expected %d bytes, received %d."

            self.fail(message % (len(response_data), self.test_data_len))

    def test_defaults(self):

        """Verifies that the default arguments will read to EOF -- this

        test verifies that any existing usages of pumpfile will not be broken

        with this new version."""

        # retrieve data using default (old) pumpfile method

        from_file = file_utils.FakeReadFile(self.test_data)

        to_file = StringIO()

        osutils.pumpfile(from_file, to_file)

        # report error if the data wasn't equal (we only report the size due

        # to the length of the data)

        response_data = to_file.getvalue()

        if response_data != self.test_data:

            message = "Data not equal.  Expected %d bytes, received %d."

            self.fail(message % (len(response_data), self.test_data_len))