~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

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

Show diffs side-by-side

added

removed

bzrlib/tests/test_transport.py

1

 

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

 

1

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

25

24

from bzrlib import (

26

25

    errors,

27

26

    osutils,

28

27

    tests,

 

28

    transport as _mod_transport,

29

29

    urlutils,

30

30

    )

 

31

from bzrlib.transport import (

 

32

    fakenfs,

 

33

    memory,

 

34

    readonly,

 

35

    )

31

36

from bzrlib.errors import (DependencyNotPresent,

32

37

                           FileExists,

33

38

                           InvalidURLJoin,

50

55

                              Transport,

51

56

                              )

52

57

from bzrlib.transport.chroot import ChrootServer

53

 

from bzrlib.transport.memory import MemoryTransport

54

58

from bzrlib.transport.local import (LocalTransport,

55

59

                                    EmulatedWin32LocalTransport)

56

60

from bzrlib.transport.pathfilter import PathFilteringServer

166

170

 

167

171

    def test_local_abspath_non_local_transport(self):

168

172

        # the base implementation should throw

169

 

        t = MemoryTransport()

 

173

        t = memory.MemoryTransport()

170

174

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

171

175

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

172

176

 

256

260

                   max_size=1*1024*1024*1024)

257

261

 

258

262

 

 

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

 

259

278

class TestMemoryTransport(TestCase):

260

279

 

261

280

    def test_get_transport(self):

262

 

        MemoryTransport()

 

281

        memory.MemoryTransport()

263

282

 

264

283

    def test_clone(self):

265

 

        transport = MemoryTransport()

266

 

        self.assertTrue(isinstance(transport, MemoryTransport))

 

284

        transport = memory.MemoryTransport()

 

285

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

267

286

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

268

287

 

269

288

    def test_abspath(self):

270

 

        transport = MemoryTransport()

 

289

        transport = memory.MemoryTransport()

271

290

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

272

291

 

273

292

    def test_abspath_of_root(self):

274

 

        transport = MemoryTransport()

 

293

        transport = memory.MemoryTransport()

275

294

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

276

295

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

277

296

 

278

297

    def test_abspath_of_relpath_starting_at_root(self):

279

 

        transport = MemoryTransport()

 

298

        transport = memory.MemoryTransport()

280

299

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

281

300

 

282

301

    def test_append_and_get(self):

283

 

        transport = MemoryTransport()

 

302

        transport = memory.MemoryTransport()

284

303

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

285

304

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

286

305

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

287

306

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

288

307

 

289

308

    def test_put_and_get(self):

290

 

        transport = MemoryTransport()

 

309

        transport = memory.MemoryTransport()

291

310

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

292

311

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

293

312

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

294

313

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

295

314

 

296

315

    def test_append_without_dir_fails(self):

297

 

        transport = MemoryTransport()

 

316

        transport = memory.MemoryTransport()

298

317

        self.assertRaises(NoSuchFile,

299

318

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

300

319

 

301

320

    def test_put_without_dir_fails(self):

302

 

        transport = MemoryTransport()

 

321

        transport = memory.MemoryTransport()

303

322

        self.assertRaises(NoSuchFile,

304

323

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

305

324

 

306

325

    def test_get_missing(self):

307

 

        transport = MemoryTransport()

 

326

        transport = memory.MemoryTransport()

308

327

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

309

328

 

310

329

    def test_has_missing(self):

311

 

        transport = MemoryTransport()

 

330

        transport = memory.MemoryTransport()

312

331

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

313

332

 

314

333

    def test_has_present(self):

315

 

        transport = MemoryTransport()

 

334

        transport = memory.MemoryTransport()

316

335

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

317

336

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

318

337

 

319

338

    def test_list_dir(self):

320

 

        transport = MemoryTransport()

 

339

        transport = memory.MemoryTransport()

321

340

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

322

341

        transport.mkdir('dir')

323

342

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

327

346

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

328

347

 

329

348

    def test_mkdir(self):

330

 

        transport = MemoryTransport()

 

349

        transport = memory.MemoryTransport()

331

350

        transport.mkdir('dir')

332

351

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

333

352

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

334

353

 

335

354

    def test_mkdir_missing_parent(self):

336

 

        transport = MemoryTransport()

 

355

        transport = memory.MemoryTransport()

337

356

        self.assertRaises(NoSuchFile,

338

357

                          transport.mkdir, 'dir/dir')

339

358

 

340

359

    def test_mkdir_twice(self):

341

 

        transport = MemoryTransport()

 

360

        transport = memory.MemoryTransport()

342

361

        transport.mkdir('dir')

343

362

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

344

363

 

345

364

    def test_parameters(self):

346

 

        transport = MemoryTransport()

 

365

        transport = memory.MemoryTransport()

347

366

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

348

367

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

349

368

 

350

369

    def test_iter_files_recursive(self):

351

 

        transport = MemoryTransport()

 

370

        transport = memory.MemoryTransport()

352

371

        transport.mkdir('dir')

353

372

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

354

373

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

357

376

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

358

377

 

359

378

    def test_stat(self):

360

 

        transport = MemoryTransport()

 

379

        transport = memory.MemoryTransport()

361

380

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

362

381

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

363

382

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

423

442

class ChrootServerTest(TestCase):

424

443

 

425

444

    def test_construct(self):

426

 

        backing_transport = MemoryTransport()

 

445

        backing_transport = memory.MemoryTransport()

427

446

        server = ChrootServer(backing_transport)

428

447

        self.assertEqual(backing_transport, server.backing_transport)

429

448

 

430

449

    def test_setUp(self):

431

 

        backing_transport = MemoryTransport()

 

450

        backing_transport = memory.MemoryTransport()

432

451

        server = ChrootServer(backing_transport)

433

 

        server.setUp()

 

452

        server.start_server()

434

453

        try:

435

454

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

436

455

        finally:

437

 

            server.tearDown()

 

456

            server.stop_server()

438

457

 

439

 

    def test_tearDown(self):

440

 

        backing_transport = MemoryTransport()

 

458

    def test_stop_server(self):

 

459

        backing_transport = memory.MemoryTransport()

441

460

        server = ChrootServer(backing_transport)

442

 

        server.setUp()

443

 

        server.tearDown()

 

461

        server.start_server()

 

462

        server.stop_server()

444

463

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

445

464

 

446

465

    def test_get_url(self):

447

 

        backing_transport = MemoryTransport()

 

466

        backing_transport = memory.MemoryTransport()

448

467

        server = ChrootServer(backing_transport)

449

 

        server.setUp()

 

468

        server.start_server()

450

469

        try:

451

470

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

452

471

        finally:

453

 

            server.tearDown()

 

472

            server.stop_server()

454

473

 

455

474

 

456

475

class PathFilteringDecoratorTransportTest(TestCase):

460

479

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

461

480

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

462

481

            lambda x: x)

463

 

        server.setUp()

 

482

        server.start_server()

464

483

        transport = get_transport(server.get_url())

465

484

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

466

485

 

467

486

        subdir_transport = transport.clone('subdir')

468

487

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

469

 

        server.tearDown()

 

488

        server.stop_server()

470

489

 

471

490

    def make_pf_transport(self, filter_func=None):

472

491

        """Make a PathFilteringTransport backed by a MemoryTransport.

477

496

            filter_func = lambda x: x

478

497

        server = PathFilteringServer(

479

498

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

480

 

        server.setUp()

481

 

        self.addCleanup(server.tearDown)

 

499

        server.start_server()

 

500

        self.addCleanup(server.stop_server)

482

501

        return get_transport(server.get_url())

483

502

 

484

503

    def test__filter(self):

541

560

    """Readonly decoration specific tests."""

542

561

 

543

562

    def test_local_parameters(self):

544

 

        import bzrlib.transport.readonly as readonly

545

563

        # connect to . in readonly mode

546

564

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

547

565

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

549

567

 

550

568

    def test_http_parameters(self):

551

569

        from bzrlib.tests.http_server import HttpServer

552

 

        import bzrlib.transport.readonly as readonly

553

570

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

554

571

        server = HttpServer()

555

572

        self.start_server(server)

564

581

    """NFS decorator specific tests."""

565

582

 

566

583

    def get_nfs_transport(self, url):

567

 

        import bzrlib.transport.fakenfs as fakenfs

568

584

        # connect to url with nfs decoration

569

585

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

570

586

 

584

600

        self.start_server(server)

585

601

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

586

602

        self.assertIsInstance(

587

 

            transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)

 

603

            transport, fakenfs.FakeNFSTransportDecorator)

588

604

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

589

605

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

590

606

 

591

607

    def test_fakenfs_server_default(self):

592

608

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

593

 

        import bzrlib.transport.fakenfs as fakenfs

594

609

        server = fakenfs.FakeNFSServer()

595

610

        self.start_server(server)

596

611

        # the url should be decorated appropriately

• Older »

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