← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_osutils.py

(gz) Create working tree at specified revision when doing a local push
 (Martin Packman)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_osutils.py
23
23
import select
24
24
import socket
25
25
import sys
26
 
import tempfile
27
26
import time
28
27
 
29
28
from bzrlib import (
437
436
        self.assertTrue(-eighteen_hours < offset < eighteen_hours)
438
437
 
439
438
 
440
 
class TestFdatasync(tests.TestCaseInTempDir):
441
 
 
442
 
    def do_fdatasync(self):
443
 
        f = tempfile.NamedTemporaryFile()
444
 
        osutils.fdatasync(f.fileno())
445
 
        f.close()
446
 
 
447
 
    @staticmethod
448
 
    def raise_eopnotsupp(*args, **kwargs):
449
 
        raise IOError(errno.EOPNOTSUPP, os.strerror(errno.EOPNOTSUPP))
450
 
 
451
 
    @staticmethod
452
 
    def raise_enotsup(*args, **kwargs):
453
 
        raise IOError(errno.ENOTSUP, os.strerror(errno.ENOTSUP))
454
 
 
455
 
    def test_fdatasync_handles_system_function(self):
456
 
        self.overrideAttr(os, "fdatasync")
457
 
        self.do_fdatasync()
458
 
 
459
 
    def test_fdatasync_handles_no_fdatasync_no_fsync(self):
460
 
        self.overrideAttr(os, "fdatasync")
461
 
        self.overrideAttr(os, "fsync")
462
 
        self.do_fdatasync()
463
 
 
464
 
    def test_fdatasync_handles_no_EOPNOTSUPP(self):
465
 
        self.overrideAttr(errno, "EOPNOTSUPP")
466
 
        self.do_fdatasync()
467
 
 
468
 
    def test_fdatasync_catches_ENOTSUP(self):
469
 
        enotsup = getattr(errno, "ENOTSUP", None)
470
 
        if enotsup is None:
471
 
            raise tests.TestNotApplicable("No ENOTSUP on this platform")
472
 
        self.overrideAttr(os, "fdatasync", self.raise_enotsup)
473
 
        self.do_fdatasync()
474
 
 
475
 
    def test_fdatasync_catches_EOPNOTSUPP(self):
476
 
        enotsup = getattr(errno, "EOPNOTSUPP", None)
477
 
        if enotsup is None:
478
 
            raise tests.TestNotApplicable("No EOPNOTSUPP on this platform")
479
 
        self.overrideAttr(os, "fdatasync", self.raise_eopnotsupp)
480
 
        self.do_fdatasync()
481
 
 
482
 
 
483
439
class TestLinks(tests.TestCaseInTempDir):
484
440
 
485
441
    def test_dereference_path(self):
598
554
    """Test pumpfile method."""
599
555
 
600
556
    def setUp(self):
601
 
        super(TestPumpFile, self).setUp()
 
557
        tests.TestCase.setUp(self)
602
558
        # create a test datablock
603
559
        self.block_size = 512
604
560
        pattern = '0123456789ABCDEF'
872
828
        self.assertEqual(None, osutils.safe_file_id(None))
873
829
 
874
830
 
875
 
class TestSendAll(tests.TestCase):
876
 
 
877
 
    def test_send_with_disconnected_socket(self):
878
 
        class DisconnectedSocket(object):
879
 
            def __init__(self, err):
880
 
                self.err = err
881
 
            def send(self, content):
882
 
                raise self.err
883
 
            def close(self):
884
 
                pass
885
 
        # All of these should be treated as ConnectionReset
886
 
        errs = []
887
 
        for err_cls in (IOError, socket.error):
888
 
            for errnum in osutils._end_of_stream_errors:
889
 
                errs.append(err_cls(errnum))
890
 
        for err in errs:
891
 
            sock = DisconnectedSocket(err)
892
 
            self.assertRaises(errors.ConnectionReset,
893
 
                osutils.send_all, sock, 'some more content')
894
 
 
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):
900
 
            def __init__(self):
901
 
                self.call_count = 0
902
 
            def send(self, bytes):
903
 
                self.call_count += 1
904
 
                if self.call_count > 100:
905
 
                    # Prevent the test suite from hanging
906
 
                    raise RuntimeError('too many calls')
907
 
                return 0
908
 
        sock = NoSendingSocket()
909
 
        self.assertRaises(errors.ConnectionReset,
910
 
                          osutils.send_all, sock, 'content')
911
 
        self.assertEqual(1, sock.call_count)
912
 
 
913
 
 
914
831
class TestPosixFuncs(tests.TestCase):
915
832
    """Test that the posix version of normpath returns an appropriate path
916
833
       when used with 2 leading slashes."""
943
860
                         osutils._win32_pathjoin('path/to', 'C:/foo'))
944
861
        self.assertEqual('path/to/foo',
945
862
                         osutils._win32_pathjoin('path/to/', 'foo'))
946
 
 
947
 
    def test_pathjoin_late_bugfix(self):
948
 
        if sys.version_info < (2, 7, 6):
949
 
            expected = '/foo'
950
 
        else:
951
 
            expected = 'C:/foo'
952
 
        self.assertEqual(expected,
 
863
        self.assertEqual('/foo',
953
864
                         osutils._win32_pathjoin('C:/path/to/', '/foo'))
954
 
        self.assertEqual(expected,
 
865
        self.assertEqual('/foo',
955
866
                         osutils._win32_pathjoin('C:\\path\\to\\', '\\foo'))
956
867
 
957
868
    def test_normpath(self):
1860
1771
    _native_to_unicode = None
1861
1772
 
1862
1773
    def setUp(self):
1863
 
        super(TestDirReader, self).setUp()
 
1774
        tests.TestCaseInTempDir.setUp(self)
1864
1775
        self.overrideAttr(osutils,
1865
1776
                          '_selected_dir_reader', self._dir_reader_class())
1866
1777
 
2074
1985
class TestTerminalWidth(tests.TestCase):
2075
1986
 
2076
1987
    def setUp(self):
2077
 
        super(TestTerminalWidth, self).setUp()
 
1988
        tests.TestCase.setUp(self)
2078
1989
        self._orig_terminal_size_state = osutils._terminal_size_state
2079
1990
        self._orig_first_terminal_size = osutils._first_terminal_size
2080
1991
        self.addCleanup(self.restore_osutils_globals)
2161
2072
    _test_needs_features = [features.chown_feature]
2162
2073
 
2163
2074
    def setUp(self):
2164
 
        super(TestCreationOps, self).setUp()
 
2075
        tests.TestCaseInTempDir.setUp(self)
2165
2076
        self.overrideAttr(os, 'chown', self._dummy_chown)
2166
2077
 
2167
2078
        # params set by call to _dummy_chown