~bzr-pqm/bzr/bzr.dev : revision 6039.1.6

16

17

18

from cStringIO import StringIO

19

import os

19

20

import subprocess

20

21

import sys

21

22

import threading

90

91

transport.register_lazy_transport(

91

92

'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')

92

93

try:

93

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

94

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

94

95

except errors.UnsupportedProtocol, e:

95

96

e_str = str(e)

96

97

self.assertEquals('Unsupported protocol'

110

111

'foo', 'bzrlib.tests.test_transport', 'BackupTransportHandler')

111

112

transport.register_lazy_transport(

112

113

'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')

113

t = transport.get_transport('foo://fooserver/foo')

114

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

114

115

self.assertTrue(isinstance(t, BackupTransportHandler))

115

116

117

def test_ssh_hints(self):

117

118

"""Transport ssh:// should raise an error pointing out bzr+ssh://"""

118

119

try:

119

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

120

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

120

121

except errors.UnsupportedProtocol, e:

121

122

e_str = str(e)

122

123

self.assertEquals('Unsupported protocol'

246

247

server.start_server()

247

248

url = server.get_url()

248

249

self.assertTrue(url in transport.transport_list_registry)

249

t = transport.get_transport(url)

250

t = transport.get_transport_from_url(url)

250

251

del t

251

252

server.stop_server()

252

253

self.assertFalse(url in transport.transport_list_registry)

367

368

def test_abspath(self):

368

369

# The abspath is always relative to the chroot_url.

369

370

server = chroot.ChrootServer(

370

transport.get_transport('memory:///foo/bar/'))

371

transport.get_transport_from_url('memory:///foo/bar/'))

371

372

self.start_server(server)

372

t = transport.get_transport(server.get_url())

373

t = transport.get_transport_from_url(server.get_url())

373

374

self.assertEqual(server.get_url(), t.abspath('/'))

374

375

376

subdir_t = t.clone('subdir')

377

378

379

def test_clone(self):

379

380

server = chroot.ChrootServer(

380

transport.get_transport('memory:///foo/bar/'))

381

transport.get_transport_from_url('memory:///foo/bar/'))

381

382

self.start_server(server)

382

t = transport.get_transport(server.get_url())

383

t = transport.get_transport_from_url(server.get_url())

383

384

# relpath from root and root path are the same

384

385

relpath_cloned = t.clone('foo')

385

386

abspath_cloned = t.clone('/foo')

394

395

This is so that it is not possible to escape a chroot by doing::

395

396

url = chroot_transport.base

396

397

parent_url = urlutils.join(url, '..')

397

new_t = transport.get_transport(parent_url)

398

new_t = transport.get_transport_from_url(parent_url)

398

399

"""

399

400

server = chroot.ChrootServer(

400

transport.get_transport('memory:///path/subpath'))

401

transport.get_transport_from_url('memory:///path/subpath'))

401

402

self.start_server(server)

402

t = transport.get_transport(server.get_url())

403

new_t = transport.get_transport(t.base)

403

t = transport.get_transport_from_url(server.get_url())

404

new_t = transport.get_transport_from_url(t.base)

404

405

self.assertEqual(t.server, new_t.server)

405

406

self.assertEqual(t.base, new_t.base)

406

407

411

412

This is so that it is not possible to escape a chroot by doing::

412

413

url = chroot_transport.base

413

414

parent_url = urlutils.join(url, '..')

414

new_t = transport.get_transport(parent_url)

415

new_t = transport.get_transport_from_url(parent_url)

415

416

"""

416

417

server = chroot.ChrootServer(

417

transport.get_transport('memory:///path/'))

418

transport.get_transport_from_url('memory:///path/'))

418

419

self.start_server(server)

419

t = transport.get_transport(server.get_url())

420

t = transport.get_transport_from_url(server.get_url())

420

421

self.assertRaises(

421

422

errors.InvalidURLJoin, urlutils.join, t.base, '..')

422

423

458

459

def test_abspath(self):

459

460

# The abspath is always relative to the base of the backing transport.

460

461

server = pathfilter.PathFilteringServer(

461

transport.get_transport('memory:///foo/bar/'),

462

transport.get_transport_from_url('memory:///foo/bar/'),

462

463

lambda x: x)

463

464

server.start_server()

464

t = transport.get_transport(server.get_url())

465

t = transport.get_transport_from_url(server.get_url())

465

466

self.assertEqual(server.get_url(), t.abspath('/'))

466

467

468

subdir_t = t.clone('subdir')

476

477

if filter_func is None:

477

478

filter_func = lambda x: x

478

479

server = pathfilter.PathFilteringServer(

479

transport.get_transport('memory:///foo/bar/'), filter_func)

480

transport.get_transport_from_url('memory:///foo/bar/'), filter_func)

480

481

server.start_server()

481

482

self.addCleanup(server.stop_server)

482

return transport.get_transport(server.get_url())

483

return transport.get_transport_from_url(server.get_url())

483

484

485

def test__filter(self):

485

486

# _filter (with an identity func as filter_func) always returns

529

530

otherwise) the filtering by doing::

530

531

url = filtered_transport.base

531

532

parent_url = urlutils.join(url, '..')

532

new_t = transport.get_transport(parent_url)

533

new_t = transport.get_transport_from_url(parent_url)

533

534

"""

534

535

t = self.make_pf_transport()

535

new_t = transport.get_transport(t.base)

536

new_t = transport.get_transport_from_url(t.base)

536

537

self.assertEqual(t.server, new_t.server)

537

538

self.assertEqual(t.base, new_t.base)

538

539

551

552

# connect to '.' via http which is not listable

552

553

server = HttpServer()

553

554

self.start_server(server)

554

t = transport.get_transport('readonly+' + server.get_url())

555

t = transport.get_transport_from_url('readonly+' + server.get_url())

555

556

self.assertIsInstance(t, readonly.ReadonlyTransportDecorator)

556

557

self.assertEqual(False, t.listable())

557

558

self.assertEqual(True, t.is_readonly())

590

591

# the url should be decorated appropriately

591

592

self.assertStartsWith(server.get_url(), 'fakenfs+')

592

593

# and we should be able to get a transport for it

593

t = transport.get_transport(server.get_url())

594

t = transport.get_transport_from_url(server.get_url())

594

595

# which must be a FakeNFSTransportDecorator instance.

595

596

self.assertIsInstance(t, fakenfs.FakeNFSTransportDecorator)

596

597

684

685

return t

685

686

687

688

class TestTransportFromPath(tests.TestCaseInTempDir):

689

690

def test_with_path(self):

691

t = transport.get_transport_from_path(self.test_dir)

692

self.assertIsInstance(t, local.LocalTransport)

693

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

694

urlutils.local_path_to_url(self.test_dir))

695

696

def test_with_url(self):

697

t = transport.get_transport_from_path("file:")

698

self.assertIsInstance(t, local.LocalTransport)

699

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

700

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

701

702

703

class TestTransportFromUrl(tests.TestCaseInTempDir):

704

705

def test_with_path(self):

706

self.assertRaises(errors.InvalidURL, transport.get_transport_from_url,

707

self.test_dir)

708

709

def test_with_url(self):

710

url = urlutils.local_path_to_url(self.test_dir)

711

t = transport.get_transport_from_url(url)

712

self.assertIsInstance(t, local.LocalTransport)

713

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

714

715

687

716

class TestLocalTransports(tests.TestCase):

688

717

689

718

def test_get_transport_from_abspath(self):

816

845

817

846

def test_reuse_same_transport(self):

818

847

possible_transports = []

819

t1 = transport.get_transport('http://foo/',

848

t1 = transport.get_transport_from_url('http://foo/',

820

849

possible_transports=possible_transports)

821

850

self.assertEqual([t1], possible_transports)

822

t2 = transport.get_transport('http://foo/',

851

t2 = transport.get_transport_from_url('http://foo/',

823

852

possible_transports=[t1])

824

853

self.assertIs(t1, t2)

825

854

826

855

# Also check that final '/' are handled correctly

827

t3 = transport.get_transport('http://foo/path/')

828

t4 = transport.get_transport('http://foo/path',

856

t3 = transport.get_transport_from_url('http://foo/path/')

857

t4 = transport.get_transport_from_url('http://foo/path',

829

858

possible_transports=[t3])

830

859

self.assertIs(t3, t4)

831

860

832

t5 = transport.get_transport('http://foo/path')

833

t6 = transport.get_transport('http://foo/path/',

861

t5 = transport.get_transport_from_url('http://foo/path')

862

t6 = transport.get_transport_from_url('http://foo/path/',

834

863

possible_transports=[t5])

835

864

self.assertIs(t5, t6)

836

865

837

866

def test_don_t_reuse_different_transport(self):

838

t1 = transport.get_transport('http://foo/path')

839

t2 = transport.get_transport('http://bar/path',

867

t1 = transport.get_transport_from_url('http://foo/path')

868

t2 = transport.get_transport_from_url('http://bar/path',

840

869

possible_transports=[t1])

841

870

self.assertIsNot(t1, t2)

842

871

844

873

class TestTransportTrace(tests.TestCase):

845

874

846

875

def test_decorator(self):

847

t = transport.get_transport('trace+memory://')

876

t = transport.get_transport_from_url('trace+memory://')

848

877

self.assertIsInstance(

849

878

t, bzrlib.transport.trace.TransportTraceDecorator)

850

879

851

880

def test_clone_preserves_activity(self):

852

t = transport.get_transport('trace+memory://')

881

t = transport.get_transport_from_url('trace+memory://')

853

882

t2 = t.clone('.')

854

883

self.assertTrue(t is not t2)

855

884

self.assertTrue(t._activity is t2._activity)

859

888

# still won't cause a test failure when the top level Transport API

860

889

# changes; so there is little return doing that.

861

890

def test_get(self):

862

t = transport.get_transport('trace+memory:///')

891

t = transport.get_transport_from_url('trace+memory:///')

863

892

t.put_bytes('foo', 'barish')

864

893

t.get('foo')

865

894

expected_result = []

871

900

self.assertEqual(expected_result, t._activity)

872

901

873

902

def test_readv(self):

874

t = transport.get_transport('trace+memory:///')

903

t = transport.get_transport_from_url('trace+memory:///')

875

904

t.put_bytes('foo', 'barish')

876

905

list(t.readv('foo', [(0, 1), (3, 2)],

877

906

adjust_for_latency=True, upper_limit=6))