~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/per_branch/test_push.py

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

Show diffs side-by-side

added

removed

bzrlib/tests/per_branch/test_push.py

1

 

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

 

1

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

26

26

    check,

27

27

    debug,

28

28

    errors,

 

29

    memorytree,

29

30

    push,

30

31

    repository,

 

32

    revision,

31

33

    tests,

32

 

    )

33

 

from bzrlib.branch import Branch

34

 

from bzrlib.bzrdir import BzrDir

35

 

from bzrlib.memorytree import MemoryTree

36

 

from bzrlib.revision import NULL_REVISION

37

 

from bzrlib.smart import client, server

38

 

from bzrlib.smart.repository import SmartServerRepositoryGetParentMap

39

 

from bzrlib.tests.per_branch.test_branch import TestCaseWithBranch

40

 

from bzrlib.transport import get_transport

41

 

from bzrlib.transport.local import LocalURLServer

42

 

 

43

 

 

44

 

class TestPush(TestCaseWithBranch):

 

34

    transport,

 

35

    )

 

36

from bzrlib.smart import (

 

37

    client,

 

38

    server,

 

39

    repository as _mod_smart_repo,

 

40

    )

 

41

from bzrlib.tests import per_branch

 

42

from bzrlib.transport import local

 

43

 

 

44

 

 

45

class TestPush(per_branch.TestCaseWithBranch):

45

46

 

46

47

    def test_push_convergence_simple(self):

47

48

        # when revisions are pushed, the left-most accessible parents must

148

149

        try:

149

150

            tree = a_branch.bzrdir.create_workingtree()

150

151

        except errors.NotLocalUrl:

151

 

            if self.vfs_transport_factory is LocalURLServer:

 

152

            if self.vfs_transport_factory is local.LocalURLServer:

152

153

                # the branch is colocated on disk, we cannot create a checkout.

153

154

                # hopefully callers will expect this.

154

 

                local_controldir= bzrdir.BzrDir.open(self.get_vfs_only_url('repo/tree'))

 

155

                local_controldir= bzrdir.BzrDir.open(

 

156

                    self.get_vfs_only_url('repo/tree'))

155

157

                tree = local_controldir.create_workingtree()

156

158

            else:

157

159

                tree = a_branch.create_checkout('repo/tree', lightweight=True)

223

225

        push._show_push_branch(trunk, 'rev-2', self.get_url('remote'), output)

224

226

        # Push rev-3 onto "remote".  If "remote" not stacked and is missing the

225

227

        # fulltext record for f-id @ rev-1, then this will fail.

226

 

        remote_branch = Branch.open(self.get_url('remote'))

 

228

        remote_branch = branch.Branch.open(self.get_url('remote'))

227

229

        trunk.push(remote_branch)

228

230

        check.check_dwim(remote_branch.base, False, True, True)

229

231

 

277

279

        # for or receiving more data than the caller asked for.

278

280

        self.overrideAttr(repository.InterRepository,

279

281

                          '_walk_to_common_revisions_batch_size', 1)

280

 

        self.overrideAttr(SmartServerRepositoryGetParentMap,

 

282

        self.overrideAttr(_mod_smart_repo.SmartServerRepositoryGetParentMap,

281

283

                          'no_extra_results', True)

282

284

 

283

285

 

284

 

class TestPushHook(TestCaseWithBranch):

 

286

class TestPushHook(per_branch.TestCaseWithBranch):

285

287

 

286

288

    def setUp(self):

287

289

        self.hook_calls = []

288

 

        TestCaseWithBranch.setUp(self)

 

290

        super(TestPushHook, self).setUp()

289

291

 

290

292

    def capture_post_push_hook(self, result):

291

293

        """Capture post push hook calls to self.hook_calls.

309

311

    def test_post_push_empty_history(self):

310

312

        target = self.make_branch('target')

311

313

        source = self.make_branch('source')

312

 

        Branch.hooks.install_named_hook('post_push',

313

 

                                        self.capture_post_push_hook, None)

 

314

        branch.Branch.hooks.install_named_hook(

 

315

            'post_push', self.capture_post_push_hook, None)

314

316

        source.push(target)

315

317

        # with nothing there we should still get a notification, and

316

318

        # have both branches locked at the notification time.

317

319

        self.assertEqual([

318

 

            ('post_push', source, None, target.base, 0, NULL_REVISION,

319

 

             0, NULL_REVISION, True, None, True)

 

320

            ('post_push', source, None, target.base, 0, revision.NULL_REVISION,

 

321

             0, revision.NULL_REVISION, True, None, True)

320

322

            ],

321

323

            self.hook_calls)

322

324

 

335

337

            # remotebranches can't be bound.  Let's instead make a new local

336

338

            # branch of the default type, which does allow binding.

337

339

            # See https://bugs.launchpad.net/bzr/+bug/112020

338

 

            local = BzrDir.create_branch_convenience('local2')

 

340

            local = bzrdir.BzrDir.create_branch_convenience('local2')

339

341

            local.bind(target)

340

342

        source = self.make_branch('source')

341

 

        Branch.hooks.install_named_hook('post_push',

342

 

                                        self.capture_post_push_hook, None)

 

343

        branch.Branch.hooks.install_named_hook(

 

344

            'post_push', self.capture_post_push_hook, None)

343

345

        source.push(local)

344

346

        # with nothing there we should still get a notification, and

345

347

        # have both branches locked at the notification time.

346

348

        self.assertEqual([

347

 

            ('post_push', source, local.base, target.base, 0, NULL_REVISION,

348

 

             0, NULL_REVISION, True, True, True)

 

349

            ('post_push', source, local.base, target.base, 0,

 

350

             revision.NULL_REVISION, 0, revision.NULL_REVISION,

 

351

             True, True, True)

349

352

            ],

350

353

            self.hook_calls)

351

354

 

356

359

        rev1 = target.commit('rev 1')

357

360

        target.unlock()

358

361

        sourcedir = target.bzrdir.clone(self.get_url('source'))

359

 

        source = MemoryTree.create_on_branch(sourcedir.open_branch())

 

362

        source = memorytree.MemoryTree.create_on_branch(sourcedir.open_branch())

360

363

        rev2 = source.commit('rev 2')

361

 

        Branch.hooks.install_named_hook('post_push',

362

 

                                        self.capture_post_push_hook, None)

 

364

        branch.Branch.hooks.install_named_hook(

 

365

            'post_push', self.capture_post_push_hook, None)

363

366

        source.branch.push(target.branch)

364

367

        # with nothing there we should still get a notification, and

365

368

        # have both branches locked at the notification time.

370

373

            self.hook_calls)

371

374

 

372

375

 

373

 

class EmptyPushSmartEffortTests(TestCaseWithBranch):

 

376

class EmptyPushSmartEffortTests(per_branch.TestCaseWithBranch):

374

377

    """Tests that a push of 0 revisions should make a limited number of smart

375

378

    protocol RPCs.

376

379

    """

404

407

    def test_empty_branch_api(self):

405

408

        """The branch_obj.push API should make a limited number of HPSS calls.

406

409

        """

407

 

        transport = get_transport(self.smart_server.get_url()).clone('target')

408

 

        target = Branch.open_from_transport(transport)

 

410

        t = transport.get_transport(self.smart_server.get_url()).clone('target')

 

411

        target = branch.Branch.open_from_transport(t)

409

412

        self.empty_branch.push(target)

410

413

        self.assertEqual(

411

414

            ['BzrDir.open_2.1',

432

435

        self.assertTrue(len(self.hpss_calls) <= 9, self.hpss_calls)

433

436

 

434

437

 

435

 

class TestLossyPush(TestCaseWithBranch):

 

438

class TestLossyPush(per_branch.TestCaseWithBranch):

436

439

 

437

440

    def setUp(self):

438

441

        self.hook_calls = []

439

 

        TestCaseWithBranch.setUp(self)

 

442

        super(TestLossyPush, self).setUp()

440

443

 

441

444

    def test_lossy_push_raises_same_vcs(self):

442

445

        target = self.make_branch('target')

• Older »

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