~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

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

• reverse the comparison (4805 to 4964)
• stop comparing with revision 4964

Show diffs side-by-side

added

removed

bzrlib/tests/test_transport.py

1

 

# Copyright (C) 2004, 2005, 2006, 2007, 2008, 2009, 2010 Canonical Ltd

 

1

# Copyright (C) 2004, 2005, 2006, 2007 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

21

21

import sys

22

22

import threading

23

23

 

 

24

import bzrlib

24

25

from bzrlib import (

25

26

    errors,

26

27

    osutils,

27

28

    tests,

28

 

    transport as _mod_transport,

29

29

    urlutils,

30

30

    )

31

 

from bzrlib.transport import (

32

 

    fakenfs,

33

 

    memory,

34

 

    readonly,

35

 

    )

36

31

from bzrlib.errors import (DependencyNotPresent,

37

32

                           FileExists,

38

33

                           InvalidURLJoin,

41

36

                           ReadError,

42

37

                           UnsupportedProtocol,

43

38

                           )

44

 

from bzrlib.tests import features, TestCase, TestCaseInTempDir

 

39

from bzrlib.tests import ParamikoFeature, TestCase, TestCaseInTempDir

45

40

from bzrlib.transport import (_clear_protocol_handlers,

46

41

                              _CoalescedOffset,

47

42

                              ConnectedTransport,

55

50

                              Transport,

56

51

                              )

57

52

from bzrlib.transport.chroot import ChrootServer

 

53

from bzrlib.transport.memory import MemoryTransport

58

54

from bzrlib.transport.local import (LocalTransport,

59

55

                                    EmulatedWin32LocalTransport)

60

56

from bzrlib.transport.pathfilter import PathFilteringServer

170

166

 

171

167

    def test_local_abspath_non_local_transport(self):

172

168

        # the base implementation should throw

173

 

        t = memory.MemoryTransport()

 

169

        t = MemoryTransport()

174

170

        e = self.assertRaises(errors.NotLocalUrl, t.local_abspath, 't')

175

171

        self.assertEqual('memory:///t is not a local path.', str(e))

176

172

 

260

256

                   max_size=1*1024*1024*1024)

261

257

 

262

258

 

263

 

class TestMemoryServer(TestCase):

264

 

 

265

 

    def test_create_server(self):

266

 

        server = memory.MemoryServer()

267

 

        server.start_server()

268

 

        url = server.get_url()

269

 

        self.assertTrue(url in _mod_transport.transport_list_registry)

270

 

        t = _mod_transport.get_transport(url)

271

 

        del t

272

 

        server.stop_server()

273

 

        self.assertFalse(url in _mod_transport.transport_list_registry)

274

 

        self.assertRaises(errors.UnsupportedProtocol,

275

 

                          _mod_transport.get_transport, url)

276

 

 

277

 

 

278

259

class TestMemoryTransport(TestCase):

279

260

 

280

261

    def test_get_transport(self):

281

 

        memory.MemoryTransport()

 

262

        MemoryTransport()

282

263

 

283

264

    def test_clone(self):

284

 

        transport = memory.MemoryTransport()

285

 

        self.assertTrue(isinstance(transport, memory.MemoryTransport))

 

265

        transport = MemoryTransport()

 

266

        self.assertTrue(isinstance(transport, MemoryTransport))

286

267

        self.assertEqual("memory:///", transport.clone("/").base)

287

268

 

288

269

    def test_abspath(self):

289

 

        transport = memory.MemoryTransport()

 

270

        transport = MemoryTransport()

290

271

        self.assertEqual("memory:///relpath", transport.abspath('relpath'))

291

272

 

292

273

    def test_abspath_of_root(self):

293

 

        transport = memory.MemoryTransport()

 

274

        transport = MemoryTransport()

294

275

        self.assertEqual("memory:///", transport.base)

295

276

        self.assertEqual("memory:///", transport.abspath('/'))

296

277

 

297

278

    def test_abspath_of_relpath_starting_at_root(self):

298

 

        transport = memory.MemoryTransport()

 

279

        transport = MemoryTransport()

299

280

        self.assertEqual("memory:///foo", transport.abspath('/foo'))

300

281

 

301

282

    def test_append_and_get(self):

302

 

        transport = memory.MemoryTransport()

 

283

        transport = MemoryTransport()

303

284

        transport.append_bytes('path', 'content')

304

285

        self.assertEqual(transport.get('path').read(), 'content')

305

286

        transport.append_file('path', StringIO('content'))

306

287

        self.assertEqual(transport.get('path').read(), 'contentcontent')

307

288

 

308

289

    def test_put_and_get(self):

309

 

        transport = memory.MemoryTransport()

 

290

        transport = MemoryTransport()

310

291

        transport.put_file('path', StringIO('content'))

311

292

        self.assertEqual(transport.get('path').read(), 'content')

312

293

        transport.put_bytes('path', 'content')

313

294

        self.assertEqual(transport.get('path').read(), 'content')

314

295

 

315

296

    def test_append_without_dir_fails(self):

316

 

        transport = memory.MemoryTransport()

 

297

        transport = MemoryTransport()

317

298

        self.assertRaises(NoSuchFile,

318

299

                          transport.append_bytes, 'dir/path', 'content')

319

300

 

320

301

    def test_put_without_dir_fails(self):

321

 

        transport = memory.MemoryTransport()

 

302

        transport = MemoryTransport()

322

303

        self.assertRaises(NoSuchFile,

323

304

                          transport.put_file, 'dir/path', StringIO('content'))

324

305

 

325

306

    def test_get_missing(self):

326

 

        transport = memory.MemoryTransport()

 

307

        transport = MemoryTransport()

327

308

        self.assertRaises(NoSuchFile, transport.get, 'foo')

328

309

 

329

310

    def test_has_missing(self):

330

 

        transport = memory.MemoryTransport()

 

311

        transport = MemoryTransport()

331

312

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

332

313

 

333

314

    def test_has_present(self):

334

 

        transport = memory.MemoryTransport()

 

315

        transport = MemoryTransport()

335

316

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

336

317

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

337

318

 

338

319

    def test_list_dir(self):

339

 

        transport = memory.MemoryTransport()

 

320

        transport = MemoryTransport()

340

321

        transport.put_bytes('foo', 'content')

341

322

        transport.mkdir('dir')

342

323

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

346

327

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

347

328

 

348

329

    def test_mkdir(self):

349

 

        transport = memory.MemoryTransport()

 

330

        transport = MemoryTransport()

350

331

        transport.mkdir('dir')

351

332

        transport.append_bytes('dir/path', 'content')

352

333

        self.assertEqual(transport.get('dir/path').read(), 'content')

353

334

 

354

335

    def test_mkdir_missing_parent(self):

355

 

        transport = memory.MemoryTransport()

 

336

        transport = MemoryTransport()

356

337

        self.assertRaises(NoSuchFile,

357

338

                          transport.mkdir, 'dir/dir')

358

339

 

359

340

    def test_mkdir_twice(self):

360

 

        transport = memory.MemoryTransport()

 

341

        transport = MemoryTransport()

361

342

        transport.mkdir('dir')

362

343

        self.assertRaises(FileExists, transport.mkdir, 'dir')

363

344

 

364

345

    def test_parameters(self):

365

 

        transport = memory.MemoryTransport()

 

346

        transport = MemoryTransport()

366

347

        self.assertEqual(True, transport.listable())

367

348

        self.assertEqual(False, transport.is_readonly())

368

349

 

369

350

    def test_iter_files_recursive(self):

370

 

        transport = memory.MemoryTransport()

 

351

        transport = MemoryTransport()

371

352

        transport.mkdir('dir')

372

353

        transport.put_bytes('dir/foo', 'content')

373

354

        transport.put_bytes('dir/bar', 'content')

376

357

        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)

377

358

 

378

359

    def test_stat(self):

379

 

        transport = memory.MemoryTransport()

 

360

        transport = MemoryTransport()

380

361

        transport.put_bytes('foo', 'content')

381

362

        transport.put_bytes('bar', 'phowar')

382

363

        self.assertEqual(7, transport.stat('foo').st_size)

442

423

class ChrootServerTest(TestCase):

443

424

 

444

425

    def test_construct(self):

445

 

        backing_transport = memory.MemoryTransport()

 

426

        backing_transport = MemoryTransport()

446

427

        server = ChrootServer(backing_transport)

447

428

        self.assertEqual(backing_transport, server.backing_transport)

448

429

 

449

430

    def test_setUp(self):

450

 

        backing_transport = memory.MemoryTransport()

 

431

        backing_transport = MemoryTransport()

451

432

        server = ChrootServer(backing_transport)

452

 

        server.start_server()

 

433

        server.setUp()

453

434

        try:

454

435

            self.assertTrue(server.scheme in _get_protocol_handlers().keys())

455

436

        finally:

456

 

            server.stop_server()

 

437

            server.tearDown()

457

438

 

458

 

    def test_stop_server(self):

459

 

        backing_transport = memory.MemoryTransport()

 

439

    def test_tearDown(self):

 

440

        backing_transport = MemoryTransport()

460

441

        server = ChrootServer(backing_transport)

461

 

        server.start_server()

462

 

        server.stop_server()

 

442

        server.setUp()

 

443

        server.tearDown()

463

444

        self.assertFalse(server.scheme in _get_protocol_handlers().keys())

464

445

 

465

446

    def test_get_url(self):

466

 

        backing_transport = memory.MemoryTransport()

 

447

        backing_transport = MemoryTransport()

467

448

        server = ChrootServer(backing_transport)

468

 

        server.start_server()

 

449

        server.setUp()

469

450

        try:

470

451

            self.assertEqual('chroot-%d:///' % id(server), server.get_url())

471

452

        finally:

472

 

            server.stop_server()

 

453

            server.tearDown()

473

454

 

474

455

 

475

456

class PathFilteringDecoratorTransportTest(TestCase):

479

460

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

480

461

        server = PathFilteringServer(get_transport('memory:///foo/bar/'),

481

462

            lambda x: x)

482

 

        server.start_server()

 

463

        server.setUp()

483

464

        transport = get_transport(server.get_url())

484

465

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

485

466

 

486

467

        subdir_transport = transport.clone('subdir')

487

468

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

488

 

        server.stop_server()

 

469

        server.tearDown()

489

470

 

490

471

    def make_pf_transport(self, filter_func=None):

491

472

        """Make a PathFilteringTransport backed by a MemoryTransport.

496

477

            filter_func = lambda x: x

497

478

        server = PathFilteringServer(

498

479

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

499

 

        server.start_server()

500

 

        self.addCleanup(server.stop_server)

 

480

        server.setUp()

 

481

        self.addCleanup(server.tearDown)

501

482

        return get_transport(server.get_url())

502

483

 

503

484

    def test__filter(self):

560

541

    """Readonly decoration specific tests."""

561

542

 

562

543

    def test_local_parameters(self):

 

544

        import bzrlib.transport.readonly as readonly

563

545

        # connect to . in readonly mode

564

546

        transport = readonly.ReadonlyTransportDecorator('readonly+.')

565

547

        self.assertEqual(True, transport.listable())

567

549

 

568

550

    def test_http_parameters(self):

569

551

        from bzrlib.tests.http_server import HttpServer

 

552

        import bzrlib.transport.readonly as readonly

570

553

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

571

554

        server = HttpServer()

572

555

        self.start_server(server)

581

564

    """NFS decorator specific tests."""

582

565

 

583

566

    def get_nfs_transport(self, url):

 

567

        import bzrlib.transport.fakenfs as fakenfs

584

568

        # connect to url with nfs decoration

585

569

        return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)

586

570

 

600

584

        self.start_server(server)

601

585

        transport = self.get_nfs_transport(server.get_url())

602

586

        self.assertIsInstance(

603

 

            transport, fakenfs.FakeNFSTransportDecorator)

 

587

            transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)

604

588

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

605

589

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

606

590

 

607

591

    def test_fakenfs_server_default(self):

608

592

        # a FakeNFSServer() should bring up a local relpath server for itself

 

593

        import bzrlib.transport.fakenfs as fakenfs

609

594

        server = fakenfs.FakeNFSServer()

610

595

        self.start_server(server)

611

596

        # the url should be decorated appropriately

910

895

        # A reasonable evolution for this would be to simply check inside

911

896

        # check_channel_exec_request that the command is appropriate, and then

912

897

        # satisfy requests in-process.

913

 

        self.requireFeature(features.paramiko)

 

898

        self.requireFeature(ParamikoFeature)

914

899

        # SFTPFullAbsoluteServer has a get_url method, and doesn't

915

900

        # override the interface (doesn't change self._vendor).

916

901

        # Note that this does encryption, so can be slow.

923

908

        # executes commands, and manage the hooking up of stdin/out/err to the

924

909

        # SSH channel ourselves.  Surely this has already been implemented

925

910

        # elsewhere?

926

 

        started = []

927

911

        class StubSSHServer(StubServer):

928

912

 

929

913

            test = self

949

933

                    (channel.recv, proc.stdin.write, proc.stdin.close),

950

934

                    (proc.stdout.read, channel.sendall, channel.close),

951

935

                    (proc.stderr.read, channel.sendall_stderr, channel.close)]

952

 

                started.append(proc)

953

936

                for read, write, close in file_functions:

954

937

                    t = threading.Thread(

955

938

                        target=ferry_bytes, args=(read, write, close))

956

939

                    t.start()

957

 

                    started.append(t)

958

940

 

959

941

                return True

960

942

 

985

967

        self.assertEqual(

986

968

            ['%s serve --inet --directory=/ --allow-writes' % bzr_remote_path],

987

969

            self.command_executed)

988

 

        # Make sure to disconnect, so that the remote process can stop, and we

989

 

        # can cleanup. Then pause the test until everything is shutdown

990

 

        t._client._medium.disconnect()

991

 

        if not started:

992

 

            return

993

 

        # First wait for the subprocess

994

 

        started[0].wait()

995

 

        # And the rest are threads

996

 

        for t in started[1:]:

997

 

            t.join()

• Older »

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