/brz/remove-bazaar

#

# 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

 

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

 

import errno

import os

import re

import socket

import sys

import time

 

    errors,

    osutils,

    tests,

    trace,

    win32utils,

    )

    features,

    file_utils,

    test__walkdirs_win32,

    )

 

    def _probe(self):

        try:

            self.reader = _readdir_pyx.UTF8DirReader

            return True

        except ImportError:

            return False

 

 

 

def _already_unicode(s):

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

    # available.

    if UTF8DirReaderFeature.available():

        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:

            scenarios.append(

                ('win32',

                 dict(_dir_reader_class=_walkdirs_win32.Win32ReadDir,

    return scenarios

 

 

 

 

class TestContainsWhitespace(tests.TestCase):

 

    def test_contains_whitespace(self):

 

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

        # is whitespace, but we do not.

 

 

class TestRename(tests.TestCaseInTempDir):

 

    def test_fancy_rename(self):

        # This should work everywhere

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

 

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

 

 

    def test_fancy_rename_fails_source_missing(self):

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

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

                          'missingsource', 'target')

 

    def test_fancy_rename_fails_if_source_and_target_missing(self):

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

 

    def test_rename(self):

        # Rename should be semi-atomic on all platforms

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

 

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

 

 

    # TODO: test fancy_rename using a MemoryTransport

 

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

        # so try to check shape of the tree

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

 

 

 

class TestRandChars(tests.TestCase):

                         (['src'], SRC_FOO_C),

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

                         ]:

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

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

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

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

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

                         ]:

 

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

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

            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')

        # 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

 

        osutils.rmtree('dir')

 

 

 

class TestDeleteAny(tests.TestCaseInTempDir):

 

    def test_file_kind(self):

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

        if osutils.has_symlinks():

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

 

        # TODO: jam 20060529 Test a block device

        try:

            os.lstat('/dev/null')

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

                raise

        else:

 

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

        if mkfifo:

            mkfifo('fifo')

            try:

            finally:

                os.remove('fifo')

 

            s = socket.socket(AF_UNIX)

            s.bind('socket')

            try:

            finally:

                os.remove('socket')

 

 

        orig_umask = osutils.get_umask()

        self.addCleanup(os.umask, orig_umask)

 

 

class TestDateTime(tests.TestCase):

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

 

    def test_format_date(self):

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

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

        # 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

        self.assertTrue(-eighteen_hours < offset < eighteen_hours)

 

 

class TestLinks(tests.TestCaseInTempDir):

 

    def test_dereference_path(self):

        cwd = osutils.realpath('.')

        os.mkdir('bar')

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

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

 

    def test_changing_access(self):

 

        # Make a file readonly

        osutils.make_readonly('file')

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

 

        # Make a file writable

        osutils.make_writable('file')

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

 

        if osutils.has_symlinks():

            # should not error when handed a symlink

 

class TestCanonicalRelPath(tests.TestCaseInTempDir):

 

 

    def test_canonical_relpath_simple(self):

        f.close()

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

 

    def test_canonical_relpath_missing_tail(self):

        os.mkdir('MixedCaseParent')

        actual = osutils.canonical_relpath(self.test_base_dir,

                                           'mixedcaseparent/nochild')

 

 

class Test_CICPCanonicalRelpath(tests.TestCaseWithTransport):

    """Test pumpfile method."""

 

    def setUp(self):

        # create a test datablock

        self.block_size = 512

        self.test_data_len = len(self.test_data)

 

    def test_bracket_block_size(self):

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

 

        from_file = file_utils.FakeReadFile(self.test_data)

 

        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)

 

        # retrieve data in blocks

        from_file = file_utils.FakeReadFile(self.test_data)

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

                         self.block_size)

 

 

        # retrieve data to EOF

        from_file = file_utils.FakeReadFile(self.test_data)

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

 

        # verify read size was equal to the maximum read size

        with this new version."""

        # retrieve data using default (old) pumpfile method

        from_file = file_utils.FakeReadFile(self.test_data)

        osutils.pumpfile(from_file, to_file)

 

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

        activity = []

        def log_activity(length, direction):

            activity.append((length, direction))

        osutils.pumpfile(from_file, to_file, buff_size=500,

                         report_activity=log_activity, direction='read')

        self.assertEqual([(500, 'read'), (500, 'read'), (500, 'read'),

                          (36, 'read')], activity)

 

        del activity[:]

        osutils.pumpfile(from_file, to_file, buff_size=500,

                         report_activity=log_activity, direction='write')

                          (36, 'write')], activity)

 

        # And with a limited amount of data

        del activity[:]

        osutils.pumpfile(from_file, to_file, buff_size=500, read_length=1028,

                         report_activity=log_activity, direction='read')

class TestPumpStringFile(tests.TestCase):

 

    def test_empty(self):

 

    def test_more_than_segment_size(self):

 

    def test_segment_size(self):

 

    def test_segment_size_multiple(self):

 

 

class TestRelpath(tests.TestCase):

class TestSafeUnicode(tests.TestCase):

 

    def test_from_ascii_string(self):

 

    def test_from_unicode_string_ascii_contents(self):

        self.assertEqual(u'bargam', osutils.safe_unicode(u'bargam'))

        self.assertEqual(u'bargam\xae', osutils.safe_unicode(u'bargam\xae'))

 

    def test_from_utf8_string(self):

 

    def test_bad_utf8_string(self):

        self.assertRaises(errors.BzrBadParameterNotUnicode,

                          osutils.safe_unicode,

 

 

class TestSafeUtf8(tests.TestCase):

 

    def test_from_ascii_string(self):

 

    def test_from_unicode_string_ascii_contents(self):

 

    def test_from_unicode_string_unicode_contents(self):

 

    def test_from_utf8_string(self):