1
# Copyright (C) 2010-2018 Jelmer Vernooij <jelmer@jelmer.uk>
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Test the smart client."""
19
from __future__ import absolute_import
21
from StringIO import StringIO
26
from ....controldir import ControlDir
27
from ....errors import (
34
from ....tests import (
36
TestCaseWithTransport,
38
from ....tests.features import ExecutableFeature
40
from ..mapping import default_mapping
41
from ..remote import (
44
RemoteGitBranchFormat,
47
from dulwich import porcelain
48
from dulwich.repo import Repo as GitRepo
51
class SplitUrlTests(TestCase):
53
def test_simple(self):
54
self.assertEquals(("foo", None, None, "/bar"),
55
split_git_url("git://foo/bar"))
58
self.assertEquals(("foo", 343, None, "/bar"),
59
split_git_url("git://foo:343/bar"))
61
def test_username(self):
62
self.assertEquals(("foo", None, "la", "/bar"),
63
split_git_url("git://la@foo/bar"))
65
def test_nopath(self):
66
self.assertEquals(("foo", None, None, "/"),
67
split_git_url("git://foo/"))
69
def test_slashpath(self):
70
self.assertEquals(("foo", None, None, "//bar"),
71
split_git_url("git://foo//bar"))
73
def test_homedir(self):
74
self.assertEquals(("foo", None, None, "~bar"),
75
split_git_url("git://foo/~bar"))
78
class ParseGitErrorTests(TestCase):
80
def test_unknown(self):
81
e = parse_git_error("url", "foo")
82
self.assertIsInstance(e, BzrError)
84
def test_notbrancherror(self):
85
e = parse_git_error("url", "\n Could not find Repository foo/bar")
86
self.assertIsInstance(e, NotBranchError)
89
class TestRemoteGitBranchFormat(TestCase):
92
super(TestRemoteGitBranchFormat, self).setUp()
93
self.format = RemoteGitBranchFormat()
95
def test_get_format_description(self):
96
self.assertEquals("Remote Git Branch", self.format.get_format_description())
98
def test_get_network_name(self):
99
self.assertEquals("git", self.format.network_name())
101
def test_supports_tags(self):
102
self.assertTrue(self.format.supports_tags())
105
class FetchFromRemoteTestBase(object):
107
_test_needs_features = [ExecutableFeature('git')]
112
TestCaseWithTransport.setUp(self)
113
self.remote_real = GitRepo.init('remote', mkdir=True)
114
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
115
self.permit_url(self.remote_url)
117
def test_sprout_simple(self):
118
self.remote_real.do_commit(
120
committer='committer <committer@example.com>',
121
author='author <author@example.com>')
123
remote = ControlDir.open(self.remote_url)
124
self.make_controldir('local', format=self._to_format)
125
local = remote.sprout('local')
127
default_mapping.revision_id_foreign_to_bzr(self.remote_real.head()),
128
local.open_branch().last_revision())
130
def test_sprout_with_tags(self):
131
c1 = self.remote_real.do_commit(
133
committer='committer <committer@example.com>',
134
author='author <author@example.com>')
135
c2 = self.remote_real.do_commit(
136
message='another commit',
137
committer='committer <committer@example.com>',
138
author='author <author@example.com>',
139
ref='refs/tags/another')
140
self.remote_real.refs['refs/tags/blah'] = self.remote_real.head()
142
remote = ControlDir.open(self.remote_url)
143
self.make_controldir('local', format=self._to_format)
144
local = remote.sprout('local')
145
local_branch = local.open_branch()
147
default_mapping.revision_id_foreign_to_bzr(c1),
148
local_branch.last_revision())
150
{'blah': local_branch.last_revision(),
151
'another': default_mapping.revision_id_foreign_to_bzr(c2)},
152
local_branch.tags.get_tag_dict())
154
def test_sprout_with_annotated_tag(self):
155
c1 = self.remote_real.do_commit(
157
committer='committer <committer@example.com>',
158
author='author <author@example.com>')
159
c2 = self.remote_real.do_commit(
160
message='another commit',
161
committer='committer <committer@example.com>',
162
author='author <author@example.com>',
163
ref='refs/heads/another')
164
porcelain.tag_create(
167
author='author <author@example.com>',
169
tag_time=int(time.time()),
172
message="Annotated tag")
174
remote = ControlDir.open(self.remote_url)
175
self.make_controldir('local', format=self._to_format)
176
local = remote.sprout('local', revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
177
local_branch = local.open_branch()
179
default_mapping.revision_id_foreign_to_bzr(c1),
180
local_branch.last_revision())
182
{'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
183
local_branch.tags.get_tag_dict())
186
class FetchFromRemoteToBzrTests(FetchFromRemoteTestBase,TestCaseWithTransport):
191
class FetchFromRemoteToGitTests(FetchFromRemoteTestBase,TestCaseWithTransport):
196
class PushToRemoteBase(object):
198
_test_needs_features = [ExecutableFeature('git')]
203
TestCaseWithTransport.setUp(self)
204
self.remote_real = GitRepo.init('remote', mkdir=True)
205
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
206
self.permit_url(self.remote_url)
208
def test_push_branch_new(self):
209
remote = ControlDir.open(self.remote_url)
210
wt = self.make_branch_and_tree('local', format=self._from_format)
211
self.build_tree(['local/blah'])
213
revid = wt.commit('blah')
215
if self._from_format == 'git':
216
result = remote.push_branch(wt.branch, name='newbranch')
218
result = remote.push_branch(wt.branch, lossy=True, name='newbranch')
220
self.assertEqual(0, result.old_revno)
221
if self._from_format == 'git':
222
self.assertEqual(1, result.new_revno)
224
self.assertIs(None, result.new_revno)
226
result.report(StringIO())
229
{'refs/heads/newbranch': self.remote_real.refs['refs/heads/newbranch'],
231
self.remote_real.get_refs())
234
c1 = self.remote_real.do_commit(
236
committer='committer <committer@example.com>',
237
author='author <author@example.com>')
239
remote = ControlDir.open(self.remote_url)
240
self.make_controldir('local', format=self._from_format)
241
local = remote.sprout('local')
242
self.build_tree(['local/blah'])
243
wt = local.open_workingtree()
245
revid = wt.commit('blah')
246
wt.branch.tags.set_tag('sometag', revid)
247
wt.branch.get_config_stack().set('branch.fetch_tags', True)
249
if self._from_format == 'git':
250
result = wt.branch.push(remote.create_branch('newbranch'))
252
result = wt.branch.push(remote.create_branch('newbranch'), lossy=True)
254
self.assertEqual(0, result.old_revno)
255
self.assertEqual(2, result.new_revno)
257
result.report(StringIO())
260
{'refs/heads/master': self.remote_real.head(),
261
'HEAD': self.remote_real.head(),
262
'refs/heads/newbranch': self.remote_real.refs['refs/heads/newbranch'],
263
'refs/tags/sometag': self.remote_real.refs['refs/heads/newbranch'],
265
self.remote_real.get_refs())
267
def test_push_diverged(self):
268
c1 = self.remote_real.do_commit(
270
committer='committer <committer@example.com>',
271
author='author <author@example.com>',
272
ref='refs/heads/newbranch')
274
remote = ControlDir.open(self.remote_url)
275
wt = self.make_branch_and_tree('local', format=self._from_format)
276
self.build_tree(['local/blah'])
278
revid = wt.commit('blah')
280
newbranch = remote.open_branch('newbranch')
281
if self._from_format == 'git':
282
self.assertRaises(DivergedBranches, wt.branch.push, newbranch)
284
self.assertRaises(DivergedBranches, wt.branch.push, newbranch, lossy=True)
287
{'refs/heads/newbranch': c1 },
288
self.remote_real.get_refs())
290
if self._from_format == 'git':
291
wt.branch.push(newbranch, overwrite=True)
293
wt.branch.push(newbranch, lossy=True, overwrite=True)
295
self.assertNotEqual(c1, self.remote_real.refs['refs/heads/newbranch'])
298
class PushToRemoteFromBzrTests(PushToRemoteBase,TestCaseWithTransport):
303
class PushToRemoteFromGitTests(PushToRemoteBase,TestCaseWithTransport):
308
class RemoteControlDirTests(TestCaseWithTransport):
310
_test_needs_features = [ExecutableFeature('git')]
313
TestCaseWithTransport.setUp(self)
314
self.remote_real = GitRepo.init('remote', mkdir=True)
315
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
316
self.permit_url(self.remote_url)
318
def test_remove_branch(self):
319
c1 = self.remote_real.do_commit(
321
committer='committer <committer@example.com>',
322
author='author <author@example.com>')
323
c2 = self.remote_real.do_commit(
324
message='another commit',
325
committer='committer <committer@example.com>',
326
author='author <author@example.com>',
327
ref='refs/heads/blah')
329
remote = ControlDir.open(self.remote_url)
330
remote.destroy_branch(name='blah')
332
self.remote_real.get_refs(),
333
{'refs/heads/master': self.remote_real.head(),
334
'HEAD': self.remote_real.head(),
337
def test_list_branches(self):
338
c1 = self.remote_real.do_commit(
340
committer='committer <committer@example.com>',
341
author='author <author@example.com>')
342
c2 = self.remote_real.do_commit(
343
message='another commit',
344
committer='committer <committer@example.com>',
345
author='author <author@example.com>',
346
ref='refs/heads/blah')
348
remote = ControlDir.open(self.remote_url)
350
['master', 'blah', 'master'],
351
[b.name for b in remote.list_branches()])
353
def test_get_branches(self):
354
c1 = self.remote_real.do_commit(
356
committer='committer <committer@example.com>',
357
author='author <author@example.com>')
358
c2 = self.remote_real.do_commit(
359
message='another commit',
360
committer='committer <committer@example.com>',
361
author='author <author@example.com>',
362
ref='refs/heads/blah')
364
remote = ControlDir.open(self.remote_url)
366
{'': 'master', 'blah': 'blah', 'master': 'master'},
367
{n: b.name for (n, b) in remote.get_branches().items()})
369
def test_remove_tag(self):
370
c1 = self.remote_real.do_commit(
372
committer='committer <committer@example.com>',
373
author='author <author@example.com>')
374
c2 = self.remote_real.do_commit(
375
message='another commit',
376
committer='committer <committer@example.com>',
377
author='author <author@example.com>',
378
ref='refs/tags/blah')
380
remote = ControlDir.open(self.remote_url)
381
remote_branch = remote.open_branch()
382
remote_branch.tags.delete_tag('blah')
383
self.assertRaises(NoSuchTag, remote_branch.tags.delete_tag, 'blah')
385
self.remote_real.get_refs(),
386
{'refs/heads/master': self.remote_real.head(),
387
'HEAD': self.remote_real.head(),
390
def test_set_tag(self):
391
c1 = self.remote_real.do_commit(
393
committer='committer <committer@example.com>',
394
author='author <author@example.com>')
395
c2 = self.remote_real.do_commit(
396
message='another commit',
397
committer='committer <committer@example.com>',
398
author='author <author@example.com>')
400
remote = ControlDir.open(self.remote_url)
401
remote.open_branch().tags.set_tag(
402
'blah', default_mapping.revision_id_foreign_to_bzr(c1))
404
self.remote_real.get_refs(),
405
{'refs/heads/master': self.remote_real.head(),
406
'refs/tags/blah': c1,
407
'HEAD': self.remote_real.head(),
410
def test_annotated_tag(self):
411
c1 = self.remote_real.do_commit(
413
committer='committer <committer@example.com>',
414
author='author <author@example.com>')
415
c2 = self.remote_real.do_commit(
416
message='another commit',
417
committer='committer <committer@example.com>',
418
author='author <author@example.com>')
420
porcelain.tag_create(
423
author='author <author@example.com>',
425
tag_time=int(time.time()),
428
message="Annotated tag")
430
remote = ControlDir.open(self.remote_url)
431
remote_branch = remote.open_branch()
433
'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
434
remote_branch.tags.get_tag_dict())
436
def tetst_get_branch_reference(self):
437
c1 = self.remote_real.do_commit(
439
committer='committer <committer@example.com>',
440
author='author <author@example.com>')
441
c2 = self.remote_real.do_commit(
442
message='another commit',
443
committer='committer <committer@example.com>',
444
author='author <author@example.com>')
446
remote = ControlDir.open(self.remote_url)
447
self.assertEqual('refs/heads/master', remote.get_branch_reference(''))
448
self.assertEqual(None, remote.get_branch_reference('master'))