← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport_implementations.py

  • Committer: Robert Collins
  • Date: 2007-04-23 02:29:35 UTC
  • mfrom: (2441 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2442.
  • Revision ID: robertc@robertcollins.net-20070423022935-9hhongamvk6bfdso
Resolve conflicts with bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport_implementations.py
22
22
 
23
23
import os
24
24
from cStringIO import StringIO
 
25
from StringIO import StringIO as pyStringIO
25
26
import stat
26
27
import sys
27
28
 
30
31
    osutils,
31
32
    urlutils,
32
33
    )
33
 
from bzrlib.errors import (DirectoryNotEmpty, NoSuchFile, FileExists,
34
 
                           LockError, NoSmartServer, PathError,
35
 
                           TransportNotPossible, ConnectionError,
36
 
                           InvalidURL)
 
34
from bzrlib.errors import (ConnectionError,
 
35
                           DirectoryNotEmpty,
 
36
                           FileExists,
 
37
                           InvalidURL,
 
38
                           LockError,
 
39
                           NoSmartServer,
 
40
                           NoSuchFile,
 
41
                           NotLocalUrl,
 
42
                           PathError,
 
43
                           TransportNotPossible,
 
44
                           )
37
45
from bzrlib.osutils import getcwd
 
46
from bzrlib.smart import medium
38
47
from bzrlib.symbol_versioning import zero_eleven
39
48
from bzrlib.tests import TestCaseInTempDir, TestSkipped
40
49
from bzrlib.tests.test_transport import TestTransportImplementation
41
 
from bzrlib.transport import memory, smart, chroot
 
50
from bzrlib.transport import memory, remote
42
51
import bzrlib.transport
43
52
 
44
53
 
45
 
def _append(fn, txt):
46
 
    """Append the given text (file-like object) to the supplied filename."""
47
 
    f = open(fn, 'ab')
48
 
    try:
49
 
        f.write(txt.read())
50
 
    finally:
51
 
        f.close()
52
 
 
53
 
 
54
54
class TransportTests(TestTransportImplementation):
55
55
 
 
56
    def setUp(self):
 
57
        super(TransportTests, self).setUp()
 
58
        self._captureVar('BZR_NO_SMART_VFS', None)
 
59
 
56
60
    def check_transport_contents(self, content, transport, relpath):
57
61
        """Check that transport.get(relpath).read() == content."""
58
62
        self.assertEqualDiff(content, transport.get(relpath).read())
76
80
 
77
81
    def test_has_root_works(self):
78
82
        current_transport = self.get_transport()
79
 
        if isinstance(current_transport, chroot.ChrootTransportDecorator):
80
 
            raise TestSkipped("ChrootTransportDecorator disallows clone('..')")
81
83
        self.assertTrue(current_transport.has('/'))
82
84
        root = current_transport.clone('/')
83
85
        self.assertTrue(root.has(''))
375
377
                              dir_mode=0777, create_parent_dir=True)
376
378
        self.assertTransportMode(t, 'dir777', 0777)
377
379
 
 
380
    def test_put_bytes_unicode(self):
 
381
        # Expect put_bytes to raise AssertionError or UnicodeEncodeError if
 
382
        # given unicode "bytes".  UnicodeEncodeError doesn't really make sense
 
383
        # (we don't want to encode unicode here at all, callers should be
 
384
        # strictly passing bytes to put_bytes), but we allow it for backwards
 
385
        # compatibility.  At some point we should use a specific exception.
 
386
        # See https://bugs.launchpad.net/bzr/+bug/106898.
 
387
        t = self.get_transport()
 
388
        if t.is_readonly():
 
389
            return
 
390
        unicode_string = u'\u1234'
 
391
        self.assertRaises(
 
392
            (AssertionError, UnicodeEncodeError),
 
393
            t.put_bytes, 'foo', unicode_string)
 
394
 
 
395
    def test_put_file_unicode(self):
 
396
        # Like put_bytes, except with a StringIO.StringIO of a unicode string.
 
397
        # This situation can happen (and has) if code is careless about the type
 
398
        # of "string" they initialise/write to a StringIO with.  We cannot use
 
399
        # cStringIO, because it never returns unicode from read.
 
400
        # Like put_bytes, UnicodeEncodeError isn't quite the right exception to
 
401
        # raise, but we raise it for hysterical raisins.
 
402
        t = self.get_transport()
 
403
        if t.is_readonly():
 
404
            return
 
405
        unicode_file = pyStringIO(u'\u1234')
 
406
        self.assertRaises(UnicodeEncodeError, t.put_file, 'foo', unicode_file)
 
407
 
378
408
    def test_put_multi(self):
379
409
        t = self.get_transport()
380
410
 
759
789
        t.mkdir('adir/bdir')
760
790
        self.assertRaises(PathError, t.rmdir, 'adir')
761
791
 
 
792
    def test_rmdir_empty_but_similar_prefix(self):
 
793
        """rmdir does not get confused by sibling paths.
 
794
        
 
795
        A naive implementation of MemoryTransport would refuse to rmdir
 
796
        ".bzr/branch" if there is a ".bzr/branch-format" directory, because it
 
797
        uses "path.startswith(dir)" on all file paths to determine if directory
 
798
        is empty.
 
799
        """
 
800
        t = self.get_transport()
 
801
        if t.is_readonly():
 
802
            return
 
803
        t.mkdir('foo')
 
804
        t.put_bytes('foo-bar', '')
 
805
        t.mkdir('foo-baz')
 
806
        t.rmdir('foo')
 
807
        self.assertRaises((NoSuchFile, PathError), t.rmdir, 'foo')
 
808
        self.failUnless(t.has('foo-bar'))
 
809
 
762
810
    def test_rename_dir_succeeds(self):
763
811
        t = self.get_transport()
764
812
        if t.is_readonly():
990
1038
    def test_clone(self):
991
1039
        # TODO: Test that clone moves up and down the filesystem
992
1040
        t1 = self.get_transport()
993
 
        if isinstance(t1, chroot.ChrootTransportDecorator):
994
 
            raise TestSkipped("ChrootTransportDecorator disallows clone('..')")
995
1041
 
996
1042
        self.build_tree(['a', 'b/', 'b/c'], transport=t1)
997
1043
 
1024
1070
 
1025
1071
    def test_clone_to_root(self):
1026
1072
        orig_transport = self.get_transport()
1027
 
        if isinstance(orig_transport, chroot.ChrootTransportDecorator):
1028
 
            raise TestSkipped("ChrootTransportDecorator disallows clone('..')")
1029
1073
        # Repeatedly go up to a parent directory until we're at the root
1030
1074
        # directory of this transport
1031
1075
        root_transport = orig_transport
1054
1098
    def test_clone_from_root(self):
1055
1099
        """At the root, cloning to a simple dir should just do string append."""
1056
1100
        orig_transport = self.get_transport()
1057
 
        if isinstance(orig_transport, chroot.ChrootTransportDecorator):
1058
 
            raise TestSkipped("ChrootTransportDecorator disallows clone('/')")
1059
1101
        root_transport = orig_transport.clone('/')
1060
1102
        self.assertEqual(root_transport.base + '.bzr/',
1061
1103
            root_transport.clone('.bzr').base)
1076
1118
 
1077
1119
    def test_relpath_at_root(self):
1078
1120
        t = self.get_transport()
1079
 
        if isinstance(t, chroot.ChrootTransportDecorator):
1080
 
            raise TestSkipped("ChrootTransportDecorator disallows clone('..')")
1081
1121
        # clone all the way to the top
1082
1122
        new_transport = t.clone('..')
1083
1123
        while new_transport.base != t.base:
1093
1133
        # that have aliasing problems like symlinks should go in backend
1094
1134
        # specific test cases.
1095
1135
        transport = self.get_transport()
1096
 
        if isinstance(transport, chroot.ChrootTransportDecorator):
1097
 
            raise TestSkipped("ChrootTransportDecorator disallows clone('..')")
1098
1136
        
1099
1137
        self.assertEqual(transport.base + 'relpath',
1100
1138
                         transport.abspath('relpath'))
1112
1150
        transport = self.get_transport()
1113
1151
        try:
1114
1152
            p = transport.local_abspath('.')
1115
 
        except TransportNotPossible:
 
1153
        except NotLocalUrl:
1116
1154
            pass # This is not a local transport
1117
1155
        else:
1118
1156
            self.assertEqual(getcwd(), p)
1119
1157
 
1120
1158
    def test_abspath_at_root(self):
1121
1159
        t = self.get_transport()
1122
 
        if isinstance(t, chroot.ChrootTransportDecorator):
1123
 
            raise TestSkipped("ChrootTransportDecorator disallows clone('..')")
1124
1160
        # clone all the way to the top
1125
1161
        new_transport = t.clone('..')
1126
1162
        while new_transport.base != t.base:
1225
1261
        # check that our server (whatever it is) is accessable reliably
1226
1262
        # via get_transport and multiple connections share content.
1227
1263
        transport = self.get_transport()
1228
 
        if isinstance(transport, chroot.ChrootTransportDecorator):
1229
 
            raise TestSkipped("ChrootTransportDecorator disallows clone('..')")
1230
1264
        if transport.is_readonly():
1231
1265
            return
1232
1266
        transport.put_bytes('foo', 'bar')
1312
1346
        """
1313
1347
        transport = self.get_transport()
1314
1348
        try:
1315
 
            medium = transport.get_smart_medium()
1316
 
            self.assertIsInstance(medium, smart.SmartClientMedium)
 
1349
            client_medium = transport.get_smart_medium()
 
1350
            self.assertIsInstance(client_medium, medium.SmartClientMedium)
1317
1351
        except errors.NoSmartMedium:
1318
1352
            # as long as we got it we're fine
1319
1353
            pass