~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

• browse files at revision 6437.26.4
• compare with another revision
• download diff
• download tarball
• view history from revision 6437.26.4

• reverse the comparison (6437.26.4 to 6617)
• stop comparing with revision 6617

Show diffs side-by-side

added

removed

bzrlib/tests/test_transport.py

1

 

# Copyright (C) 2005-2011, 2015, 2016 Canonical Ltd

 

1

# Copyright (C) 2005-2011 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

95

95

            transport.get_transport_from_url('foo://fooserver/foo')

96

96

        except errors.UnsupportedProtocol, e:

97

97

            e_str = str(e)

98

 

            self.assertEqual('Unsupported protocol'

 

98

            self.assertEquals('Unsupported protocol'

99

99

                                ' for url "foo://fooserver/foo":'

100

100

                                ' Unable to import library "some_lib":'

101

101

                                ' testing missing dependency', str(e))

121

121

            transport.get_transport_from_url('ssh://fooserver/foo')

122

122

        except errors.UnsupportedProtocol, e:

123

123

            e_str = str(e)

124

 

            self.assertEqual('Unsupported protocol'

 

124

            self.assertEquals('Unsupported protocol'

125

125

                              ' for url "ssh://fooserver/foo":'

126

126

                              ' bzr supports bzr+ssh to operate over ssh,'

127

127

                              ' use "bzr+ssh://fooserver/foo".',

298

298

 

299

299

    def test_has_missing(self):

300

300

        t = memory.MemoryTransport()

301

 

        self.assertEqual(False, t.has('foo'))

 

301

        self.assertEquals(False, t.has('foo'))

302

302

 

303

303

    def test_has_present(self):

304

304

        t = memory.MemoryTransport()

305

305

        t.append_bytes('foo', 'content')

306

 

        self.assertEqual(True, t.has('foo'))

 

306

        self.assertEquals(True, t.has('foo'))

307

307

 

308

308

    def test_list_dir(self):

309

309

        t = memory.MemoryTransport()

312

312

        t.put_bytes('dir/subfoo', 'content')

313

313

        t.put_bytes('dirlike', 'content')

314

314

 

315

 

        self.assertEqual(['dir', 'dirlike', 'foo'], sorted(t.list_dir('.')))

316

 

        self.assertEqual(['subfoo'], sorted(t.list_dir('dir')))

 

315

        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(t.list_dir('.')))

 

316

        self.assertEquals(['subfoo'], sorted(t.list_dir('dir')))

317

317

 

318

318

    def test_mkdir(self):

319

319

        t = memory.MemoryTransport()

703

703

    def test_with_path(self):

704

704

        t = transport.get_transport_from_path(self.test_dir)

705

705

        self.assertIsInstance(t, local.LocalTransport)

706

 

        self.assertEqual(t.base.rstrip("/"),

 

706

        self.assertEquals(t.base.rstrip("/"),

707

707

            urlutils.local_path_to_url(self.test_dir))

708

708

 

709

709

    def test_with_url(self):

710

710

        t = transport.get_transport_from_path("file:")

711

711

        self.assertIsInstance(t, local.LocalTransport)

712

 

        self.assertEqual(t.base.rstrip("/"),

 

712

        self.assertEquals(t.base.rstrip("/"),

713

713

            urlutils.local_path_to_url(os.path.join(self.test_dir, "file:")))

714

714

 

715

715

 

723

723

        url = urlutils.local_path_to_url(self.test_dir)

724

724

        t = transport.get_transport_from_url(url)

725

725

        self.assertIsInstance(t, local.LocalTransport)

726

 

        self.assertEqual(t.base.rstrip("/"), url)

 

726

        self.assertEquals(t.base.rstrip("/"), url)

727

727

 

728

728

    def test_with_url_and_segment_parameters(self):

729

729

        url = urlutils.local_path_to_url(self.test_dir)+",branch=foo"

730

730

        t = transport.get_transport_from_url(url)

731

731

        self.assertIsInstance(t, local.LocalTransport)

732

 

        self.assertEqual(t.base.rstrip("/"), url)

 

732

        self.assertEquals(t.base.rstrip("/"), url)

733

733

        with open(os.path.join(self.test_dir, "afile"), 'w') as f:

734

734

            f.write("data")

735

735

        self.assertTrue(t.has("afile"))

741

741

        here = osutils.abspath('.')

742

742

        t = transport.get_transport(here)

743

743

        self.assertIsInstance(t, local.LocalTransport)

744

 

        self.assertEqual(t.base, urlutils.local_path_to_url(here) + '/')

 

744

        self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')

745

745

 

746

746

    def test_get_transport_from_relpath(self):

747

747

        here = osutils.abspath('.')

748

748

        t = transport.get_transport('.')

749

749

        self.assertIsInstance(t, local.LocalTransport)

750

 

        self.assertEqual(t.base, urlutils.local_path_to_url('.') + '/')

 

750

        self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')

751

751

 

752

752

    def test_get_transport_from_local_url(self):

753

753

        here = osutils.abspath('.')

754

754

        here_url = urlutils.local_path_to_url(here) + '/'

755

755

        t = transport.get_transport(here_url)

756

756

        self.assertIsInstance(t, local.LocalTransport)

757

 

        self.assertEqual(t.base, here_url)

 

757

        self.assertEquals(t.base, here_url)

758

758

 

759

759

    def test_local_abspath(self):

760

760

        here = osutils.abspath('.')

761

761

        t = transport.get_transport(here)

762

 

        self.assertEqual(t.local_abspath(''), here)

 

762

        self.assertEquals(t.local_abspath(''), here)

763

763

 

764

764

 

765

765

class TestLocalTransportMutation(tests.TestCaseInTempDir):

804

804

        w.fdatasync()

805

805

        with open('out', 'rb') as f:

806

806

            # Should have been flushed.

807

 

            self.assertEqual(f.read(), 'foo')

808

 

        self.assertEqual(len(calls), 1, calls)

 

807

            self.assertEquals(f.read(), 'foo')

 

808

        self.assertEquals(len(calls), 1, calls)

809

809

 

810

810

    def test_missing_directory(self):

811

811

        t = self.get_transport('.')

815

815

class TestWin32LocalTransport(tests.TestCase):

816

816

 

817

817

    def test_unc_clone_to_root(self):

818

 

        self.requireFeature(features.win32_feature)

819

818

        # Win32 UNC path like \\HOST\path

820

819

        # clone to root should stop at least at \\HOST part

821

820

        # not on \\

822

821

        t = local.EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')

823

822

        for i in xrange(4):

824

823

            t = t.clone('..')

825

 

        self.assertEqual(t.base, 'file://HOST/')

 

824

        self.assertEquals(t.base, 'file://HOST/')

826

825

        # make sure we reach the root

827

826

        t = t.clone('..')

828

 

        self.assertEqual(t.base, 'file://HOST/')

 

827

        self.assertEquals(t.base, 'file://HOST/')

829

828

 

830

829

 

831

830

class TestConnectedTransport(tests.TestCase):

834

833

    def test_parse_url(self):

835

834

        t = transport.ConnectedTransport(

836

835

            'http://simple.example.com/home/source')

837

 

        self.assertEqual(t._parsed_url.host, 'simple.example.com')

838

 

        self.assertEqual(t._parsed_url.port, None)

839

 

        self.assertEqual(t._parsed_url.path, '/home/source/')

 

836

        self.assertEquals(t._parsed_url.host, 'simple.example.com')

 

837

        self.assertEquals(t._parsed_url.port, None)

 

838

        self.assertEquals(t._parsed_url.path, '/home/source/')

840

839

        self.assertTrue(t._parsed_url.user is None)

841

840

        self.assertTrue(t._parsed_url.password is None)

842

841

 

843

 

        self.assertEqual(t.base, 'http://simple.example.com/home/source/')

 

842

        self.assertEquals(t.base, 'http://simple.example.com/home/source/')

844

843

 

845

844

    def test_parse_url_with_at_in_user(self):

846

845

        # Bug 228058

847

846

        t = transport.ConnectedTransport('ftp://user@host.com@www.host.com/')

848

 

        self.assertEqual(t._parsed_url.user, 'user@host.com')

 

847

        self.assertEquals(t._parsed_url.user, 'user@host.com')

849

848

 

850

849

    def test_parse_quoted_url(self):

851

850

        t = transport.ConnectedTransport(

852

851

            'http://ro%62ey:h%40t@ex%41mple.com:2222/path')

853

 

        self.assertEqual(t._parsed_url.host, 'exAmple.com')

854

 

        self.assertEqual(t._parsed_url.port, 2222)

855

 

        self.assertEqual(t._parsed_url.user, 'robey')

856

 

        self.assertEqual(t._parsed_url.password, 'h@t')

857

 

        self.assertEqual(t._parsed_url.path, '/path/')

 

852

        self.assertEquals(t._parsed_url.host, 'exAmple.com')

 

853

        self.assertEquals(t._parsed_url.port, 2222)

 

854

        self.assertEquals(t._parsed_url.user, 'robey')

 

855

        self.assertEquals(t._parsed_url.password, 'h@t')

 

856

        self.assertEquals(t._parsed_url.path, '/path/')

858

857

 

859

858

        # Base should not keep track of the password

860

 

        self.assertEqual(t.base, 'http://ro%62ey@ex%41mple.com:2222/path/')

 

859

        self.assertEquals(t.base, 'http://ro%62ey@ex%41mple.com:2222/path/')

861

860

 

862

861

    def test_parse_invalid_url(self):

863

862

        self.assertRaises(errors.InvalidURL,

867

866

    def test_relpath(self):

868

867

        t = transport.ConnectedTransport('sftp://user@host.com/abs/path')

869

868

 

870

 

        self.assertEqual(t.relpath('sftp://user@host.com/abs/path/sub'),

 

869

        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'),

871

870

            'sub')

872

871

        self.assertRaises(errors.PathNotChild, t.relpath,

873

872

                          'http://user@host.com/abs/path/sub')

879

878

                          'sftp://user@host.com:33/abs/path/sub')

880

879

        # Make sure it works when we don't supply a username

881

880

        t = transport.ConnectedTransport('sftp://host.com/abs/path')

882

 

        self.assertEqual(t.relpath('sftp://host.com/abs/path/sub'), 'sub')

 

881

        self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')

883

882

 

884

883

        # Make sure it works when parts of the path will be url encoded

885

884

        t = transport.ConnectedTransport('sftp://host.com/dev/%path')

886

 

        self.assertEqual(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')

 

885

        self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')

887

886

 

888

887

    def test_connection_sharing_propagate_credentials(self):

889

888

        t = transport.ConnectedTransport('ftp://user@host.com/abs/path')

890

 

        self.assertEqual('user', t._parsed_url.user)

891

 

        self.assertEqual('host.com', t._parsed_url.host)

 

889

        self.assertEquals('user', t._parsed_url.user)

 

890

        self.assertEquals('host.com', t._parsed_url.host)

892

891

        self.assertIs(None, t._get_connection())

893

892

        self.assertIs(None, t._parsed_url.password)

894

893

        c = t.clone('subdir')

1096

1095

    def test_truncation(self):

1097

1096

        fake_html = "<p>something!\n" * 1000

1098

1097

        result = http.unhtml_roughly(fake_html)

1099

 

        self.assertEqual(len(result), 1000)

 

1098

        self.assertEquals(len(result), 1000)

1100

1099

        self.assertStartsWith(result, " something!")

1101

1100

 

1102

1101

 

1119

1118

        return path, url

1120

1119

 

1121

1120

    def test_regular_url(self):

1122

 

        self.assertEqual("file://foo", location_to_url("file://foo"))

 

1121

        self.assertEquals("file://foo", location_to_url("file://foo"))

1123

1122

 

1124

1123

    def test_directory(self):

1125

1124

        directories.register("bar:", SomeDirectory, "Dummy directory")

1126

1125

        self.addCleanup(directories.remove, "bar:")

1127

 

        self.assertEqual("http://bar", location_to_url("bar:"))

 

1126

        self.assertEquals("http://bar", location_to_url("bar:"))

1128

1127

 

1129

1128

    def test_unicode_url(self):

1130

1129

        self.assertRaises(errors.InvalidURL, location_to_url,

1134

1133

        path, url = self.get_base_location()

1135

1134

        location = path + "\xc3\xaf".decode("utf-8")

1136

1135

        url += '%C3%AF'

1137

 

        self.assertEqual(url, location_to_url(location))

 

1136

        self.assertEquals(url, location_to_url(location))

1138

1137

 

1139

1138

    def test_path(self):

1140

1139

        path, url = self.get_base_location()

1141

 

        self.assertEqual(url, location_to_url(path))

 

1140

        self.assertEquals(url, location_to_url(path))

1142

1141

 

1143

1142

    def test_relative_file_url(self):

1144

 

        self.assertEqual(urlutils.local_path_to_url(".") + "/bar",

 

1143

        self.assertEquals(urlutils.local_path_to_url(".") + "/bar",

1145

1144

            location_to_url("file:bar"))

1146

1145

 

1147

1146

    def test_absolute_file_url(self):

1148

 

        self.assertEqual("file:///bar", location_to_url("file:/bar"))

 

1147

        self.assertEquals("file:///bar", location_to_url("file:/bar"))

• Older »

Loggerhead is a web-based interface for Breezy
Version: 2.0.1