1
# Copyright (C) 2004, 2005, 2007 Canonical Ltd
1
# Copyright (C) 2007-2010 Canonical Ltd
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
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
44
class TestPush(TestCaseWithBranch):
36
from bzrlib.smart import (
39
repository as _mod_smart_repo,
41
from bzrlib.tests import (
47
class TestPush(per_branch.TestCaseWithBranch):
46
49
def test_push_convergence_simple(self):
47
50
# when revisions are pushed, the left-most accessible parents must
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()
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)
277
281
# for or receiving more data than the caller asked for.
278
282
self.overrideAttr(repository.InterRepository,
279
283
'_walk_to_common_revisions_batch_size', 1)
280
self.overrideAttr(SmartServerRepositoryGetParentMap,
284
self.overrideAttr(_mod_smart_repo.SmartServerRepositoryGetParentMap,
281
285
'no_extra_results', True)
284
class TestPushHook(TestCaseWithBranch):
288
class TestPushHook(per_branch.TestCaseWithBranch):
287
291
self.hook_calls = []
288
TestCaseWithBranch.setUp(self)
292
super(TestPushHook, self).setUp()
290
294
def capture_post_push_hook(self, result):
291
295
"""Capture post push hook calls to self.hook_calls.
309
313
def test_post_push_empty_history(self):
310
314
target = self.make_branch('target')
311
315
source = self.make_branch('source')
312
Branch.hooks.install_named_hook('post_push',
313
self.capture_post_push_hook, None)
316
branch.Branch.hooks.install_named_hook(
317
'post_push', self.capture_post_push_hook, None)
314
318
source.push(target)
315
319
# with nothing there we should still get a notification, and
316
320
# have both branches locked at the notification time.
317
321
self.assertEqual([
318
('post_push', source, None, target.base, 0, NULL_REVISION,
319
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)
335
339
# remotebranches can't be bound. Let's instead make a new local
336
340
# branch of the default type, which does allow binding.
337
341
# See https://bugs.launchpad.net/bzr/+bug/112020
338
local = BzrDir.create_branch_convenience('local2')
342
local = bzrdir.BzrDir.create_branch_convenience('local2')
339
343
local.bind(target)
340
344
source = self.make_branch('source')
341
Branch.hooks.install_named_hook('post_push',
342
self.capture_post_push_hook, None)
345
branch.Branch.hooks.install_named_hook(
346
'post_push', self.capture_post_push_hook, None)
343
347
source.push(local)
344
348
# with nothing there we should still get a notification, and
345
349
# have both branches locked at the notification time.
346
350
self.assertEqual([
347
('post_push', source, local.base, target.base, 0, NULL_REVISION,
348
0, NULL_REVISION, True, True, True)
351
('post_push', source, local.base, target.base, 0,
352
revision.NULL_REVISION, 0, revision.NULL_REVISION,
356
361
rev1 = target.commit('rev 1')
358
363
sourcedir = target.bzrdir.clone(self.get_url('source'))
359
source = MemoryTree.create_on_branch(sourcedir.open_branch())
364
source = memorytree.MemoryTree.create_on_branch(sourcedir.open_branch())
360
365
rev2 = source.commit('rev 2')
361
Branch.hooks.install_named_hook('post_push',
362
self.capture_post_push_hook, None)
366
branch.Branch.hooks.install_named_hook(
367
'post_push', self.capture_post_push_hook, None)
363
368
source.branch.push(target.branch)
364
369
# with nothing there we should still get a notification, and
365
370
# have both branches locked at the notification time.
388
393
super(EmptyPushSmartEffortTests, self).setUp()
389
394
# Create a smart server that publishes whatever the backing VFS server
391
self.smart_server = server.SmartTCPServer_for_testing()
396
self.smart_server = test_server.SmartTCPServer_for_testing()
392
397
self.start_server(self.smart_server, self.get_server())
393
398
# Make two empty branches, 'empty' and 'target'.
394
399
self.empty_branch = self.make_branch('empty')
404
409
def test_empty_branch_api(self):
405
410
"""The branch_obj.push API should make a limited number of HPSS calls.
407
transport = get_transport(self.smart_server.get_url()).clone('target')
408
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)
409
414
self.empty_branch.push(target)
410
415
self.assertEqual(
411
416
['BzrDir.open_2.1',
432
437
self.assertTrue(len(self.hpss_calls) <= 9, self.hpss_calls)
435
class TestLossyPush(TestCaseWithBranch):
440
class TestLossyPush(per_branch.TestCaseWithBranch):
438
443
self.hook_calls = []
439
TestCaseWithBranch.setUp(self)
444
super(TestLossyPush, self).setUp()
441
446
def test_lossy_push_raises_same_vcs(self):
442
447
target = self.make_branch('target')
added
removed
bzrlib/tests/per_branch/test_push.py
