← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

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

  • Committer: Robert Collins
  • Date: 2010-04-08 04:34:03 UTC
  • mfrom: (5138 +trunk)
  • mto: This revision was merged to the branch mainline in revision 5139.
  • Revision ID: robertc@robertcollins.net-20100408043403-56z0d07vdqrx7f3t
Update bugfix for 528114 to trunk.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/per_branch/test_push.py
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 (
 
42
    per_branch,
 
43
    test_server,
 
44
    )
 
45
 
 
46
 
 
47
class TestPush(per_branch.TestCaseWithBranch):
45
48
 
46
49
    def test_push_convergence_simple(self):
47
50
        # when revisions are pushed, the left-most accessible parents must
148
151
        try:
149
152
            tree = a_branch.bzrdir.create_workingtree()
150
153
        except errors.NotLocalUrl:
151
 
            if self.vfs_transport_factory is LocalURLServer:
 
154
            if self.vfs_transport_factory is test_server.LocalURLServer:
152
155
                # the branch is colocated on disk, we cannot create a checkout.
153
156
                # hopefully callers will expect this.
154
 
                local_controldir= bzrdir.BzrDir.open(self.get_vfs_only_url('repo/tree'))
 
157
                local_controldir= bzrdir.BzrDir.open(
 
158
                    self.get_vfs_only_url('repo/tree'))
155
159
                tree = local_controldir.create_workingtree()
156
160
            else:
157
161
                tree = a_branch.create_checkout('repo/tree', lightweight=True)
223
227
        push._show_push_branch(trunk, 'rev-2', self.get_url('remote'), output)
224
228
        # Push rev-3 onto "remote".  If "remote" not stacked and is missing the
225
229
        # fulltext record for f-id @ rev-1, then this will fail.
226
 
        remote_branch = Branch.open(self.get_url('remote'))
 
230
        remote_branch = branch.Branch.open(self.get_url('remote'))
227
231
        trunk.push(remote_branch)
228
232
        check.check_dwim(remote_branch.base, False, True, True)
229
233
 
275
279
    def disableOptimisticGetParentMap(self):
276
280
        # Tweak some class variables to stop remote get_parent_map calls asking
277
281
        # for or receiving more data than the caller asked for.
278
 
        old_flag = SmartServerRepositoryGetParentMap.no_extra_results
279
 
        inter_class = repository.InterRepository
280
 
        old_batch_size = inter_class._walk_to_common_revisions_batch_size
281
 
        inter_class._walk_to_common_revisions_batch_size = 1
282
 
        SmartServerRepositoryGetParentMap.no_extra_results = True
283
 
        def reset_values():
284
 
            SmartServerRepositoryGetParentMap.no_extra_results = old_flag
285
 
            inter_class._walk_to_common_revisions_batch_size = old_batch_size
286
 
        self.addCleanup(reset_values)
287
 
 
288
 
 
289
 
class TestPushHook(TestCaseWithBranch):
 
282
        self.overrideAttr(repository.InterRepository,
 
283
                          '_walk_to_common_revisions_batch_size', 1)
 
284
        self.overrideAttr(_mod_smart_repo.SmartServerRepositoryGetParentMap,
 
285
                          'no_extra_results', True)
 
286
 
 
287
 
 
288
class TestPushHook(per_branch.TestCaseWithBranch):
290
289
 
291
290
    def setUp(self):
292
291
        self.hook_calls = []
293
 
        TestCaseWithBranch.setUp(self)
 
292
        super(TestPushHook, self).setUp()
294
293
 
295
294
    def capture_post_push_hook(self, result):
296
295
        """Capture post push hook calls to self.hook_calls.
314
313
    def test_post_push_empty_history(self):
315
314
        target = self.make_branch('target')
316
315
        source = self.make_branch('source')
317
 
        Branch.hooks.install_named_hook('post_push',
318
 
                                        self.capture_post_push_hook, None)
 
316
        branch.Branch.hooks.install_named_hook(
 
317
            'post_push', self.capture_post_push_hook, None)
319
318
        source.push(target)
320
319
        # with nothing there we should still get a notification, and
321
320
        # have both branches locked at the notification time.
322
321
        self.assertEqual([
323
 
            ('post_push', source, None, target.base, 0, NULL_REVISION,
324
 
             0, NULL_REVISION, True, None, True)
 
322
            ('post_push', source, None, target.base, 0, revision.NULL_REVISION,
 
323
             0, revision.NULL_REVISION, True, None, True)
325
324
            ],
326
325
            self.hook_calls)
327
326
 
340
339
            # remotebranches can't be bound.  Let's instead make a new local
341
340
            # branch of the default type, which does allow binding.
342
341
            # See https://bugs.launchpad.net/bzr/+bug/112020
343
 
            local = BzrDir.create_branch_convenience('local2')
 
342
            local = bzrdir.BzrDir.create_branch_convenience('local2')
344
343
            local.bind(target)
345
344
        source = self.make_branch('source')
346
 
        Branch.hooks.install_named_hook('post_push',
347
 
                                        self.capture_post_push_hook, None)
 
345
        branch.Branch.hooks.install_named_hook(
 
346
            'post_push', self.capture_post_push_hook, None)
348
347
        source.push(local)
349
348
        # with nothing there we should still get a notification, and
350
349
        # have both branches locked at the notification time.
351
350
        self.assertEqual([
352
 
            ('post_push', source, local.base, target.base, 0, NULL_REVISION,
353
 
             0, NULL_REVISION, True, True, True)
 
351
            ('post_push', source, local.base, target.base, 0,
 
352
             revision.NULL_REVISION, 0, revision.NULL_REVISION,
 
353
             True, True, True)
354
354
            ],
355
355
            self.hook_calls)
356
356
 
361
361
        rev1 = target.commit('rev 1')
362
362
        target.unlock()
363
363
        sourcedir = target.bzrdir.clone(self.get_url('source'))
364
 
        source = MemoryTree.create_on_branch(sourcedir.open_branch())
 
364
        source = memorytree.MemoryTree.create_on_branch(sourcedir.open_branch())
365
365
        rev2 = source.commit('rev 2')
366
 
        Branch.hooks.install_named_hook('post_push',
367
 
                                        self.capture_post_push_hook, None)
 
366
        branch.Branch.hooks.install_named_hook(
 
367
            'post_push', self.capture_post_push_hook, None)
368
368
        source.branch.push(target.branch)
369
369
        # with nothing there we should still get a notification, and
370
370
        # have both branches locked at the notification time.
375
375
            self.hook_calls)
376
376
 
377
377
 
378
 
class EmptyPushSmartEffortTests(TestCaseWithBranch):
 
378
class EmptyPushSmartEffortTests(per_branch.TestCaseWithBranch):
379
379
    """Tests that a push of 0 revisions should make a limited number of smart
380
380
    protocol RPCs.
381
381
    """
393
393
        super(EmptyPushSmartEffortTests, self).setUp()
394
394
        # Create a smart server that publishes whatever the backing VFS server
395
395
        # does.
396
 
        self.smart_server = server.SmartTCPServer_for_testing()
 
396
        self.smart_server = test_server.SmartTCPServer_for_testing()
397
397
        self.start_server(self.smart_server, self.get_server())
398
398
        # Make two empty branches, 'empty' and 'target'.
399
399
        self.empty_branch = self.make_branch('empty')
409
409
    def test_empty_branch_api(self):
410
410
        """The branch_obj.push API should make a limited number of HPSS calls.
411
411
        """
412
 
        transport = get_transport(self.smart_server.get_url()).clone('target')
413
 
        target = Branch.open_from_transport(transport)
 
412
        t = transport.get_transport(self.smart_server.get_url()).clone('target')
 
413
        target = branch.Branch.open_from_transport(t)
414
414
        self.empty_branch.push(target)
415
415
        self.assertEqual(
416
416
            ['BzrDir.open_2.1',
437
437
        self.assertTrue(len(self.hpss_calls) <= 9, self.hpss_calls)
438
438
 
439
439
 
440
 
class TestLossyPush(TestCaseWithBranch):
 
440
class TestLossyPush(per_branch.TestCaseWithBranch):
441
441
 
442
442
    def setUp(self):
443
443
        self.hook_calls = []
444
 
        TestCaseWithBranch.setUp(self)
 
444
        super(TestLossyPush, self).setUp()
445
445
 
446
446
    def test_lossy_push_raises_same_vcs(self):
447
447
        target = self.make_branch('target')