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 io import BytesIO
26
from ...controldir import ControlDir
27
from ...errors import (
35
from ...tests import (
37
TestCaseWithTransport,
39
from ...tests.features import ExecutableFeature
41
from ..mapping import default_mapping
42
from ..remote import (
47
RemoteGitBranchFormat,
50
from dulwich import porcelain
51
from dulwich.repo import Repo as GitRepo
54
class SplitUrlTests(TestCase):
56
def test_simple(self):
57
self.assertEqual(("foo", None, None, "/bar"),
58
split_git_url("git://foo/bar"))
61
self.assertEqual(("foo", 343, None, "/bar"),
62
split_git_url("git://foo:343/bar"))
64
def test_username(self):
65
self.assertEqual(("foo", None, "la", "/bar"),
66
split_git_url("git://la@foo/bar"))
68
def test_nopath(self):
69
self.assertEqual(("foo", None, None, "/"),
70
split_git_url("git://foo/"))
72
def test_slashpath(self):
73
self.assertEqual(("foo", None, None, "//bar"),
74
split_git_url("git://foo//bar"))
76
def test_homedir(self):
77
self.assertEqual(("foo", None, None, "~bar"),
78
split_git_url("git://foo/~bar"))
81
class ParseGitErrorTests(TestCase):
83
def test_unknown(self):
84
e = parse_git_error("url", "foo")
85
self.assertIsInstance(e, RemoteGitError)
87
def test_notbrancherror(self):
88
e = parse_git_error("url", "\n Could not find Repository foo/bar")
89
self.assertIsInstance(e, NotBranchError)
91
def test_notbrancherror_launchpad(self):
92
e = parse_git_error("url", "Repository 'foo/bar' not found.")
93
self.assertIsInstance(e, NotBranchError)
95
def test_notbrancherror_github(self):
96
e = parse_git_error("url", "Repository not found.\n")
97
self.assertIsInstance(e, NotBranchError)
99
def test_head_update(self):
100
e = parse_git_error("url", "HEAD failed to update\n")
101
self.assertIsInstance(e, HeadUpdateFailed)
103
def test_permission_dnied(self):
106
"access denied or repository not exported: /debian/altermime.git")
107
self.assertIsInstance(e, PermissionDenied)
110
class TestRemoteGitBranchFormat(TestCase):
113
super(TestRemoteGitBranchFormat, self).setUp()
114
self.format = RemoteGitBranchFormat()
116
def test_get_format_description(self):
117
self.assertEqual("Remote Git Branch", self.format.get_format_description())
119
def test_get_network_name(self):
120
self.assertEqual(b"git", self.format.network_name())
122
def test_supports_tags(self):
123
self.assertTrue(self.format.supports_tags())
126
class FetchFromRemoteTestBase(object):
128
_test_needs_features = [ExecutableFeature('git')]
133
TestCaseWithTransport.setUp(self)
134
self.remote_real = GitRepo.init('remote', mkdir=True)
135
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
136
self.permit_url(self.remote_url)
138
def test_sprout_simple(self):
139
self.remote_real.do_commit(
141
committer=b'committer <committer@example.com>',
142
author=b'author <author@example.com>')
144
remote = ControlDir.open(self.remote_url)
145
self.make_controldir('local', format=self._to_format)
146
local = remote.sprout('local')
148
default_mapping.revision_id_foreign_to_bzr(self.remote_real.head()),
149
local.open_branch().last_revision())
151
def test_sprout_with_tags(self):
152
c1 = self.remote_real.do_commit(
154
committer=b'committer <committer@example.com>',
155
author=b'author <author@example.com>')
156
c2 = self.remote_real.do_commit(
157
message=b'another commit',
158
committer=b'committer <committer@example.com>',
159
author=b'author <author@example.com>',
160
ref=b'refs/tags/another')
161
self.remote_real.refs[b'refs/tags/blah'] = self.remote_real.head()
163
remote = ControlDir.open(self.remote_url)
164
self.make_controldir('local', format=self._to_format)
165
local = remote.sprout('local')
166
local_branch = local.open_branch()
168
default_mapping.revision_id_foreign_to_bzr(c1),
169
local_branch.last_revision())
171
{'blah': local_branch.last_revision(),
172
'another': default_mapping.revision_id_foreign_to_bzr(c2)},
173
local_branch.tags.get_tag_dict())
175
def test_sprout_with_annotated_tag(self):
176
c1 = self.remote_real.do_commit(
178
committer=b'committer <committer@example.com>',
179
author=b'author <author@example.com>')
180
c2 = self.remote_real.do_commit(
181
message=b'another commit',
182
committer=b'committer <committer@example.com>',
183
author=b'author <author@example.com>',
184
ref=b'refs/heads/another')
185
porcelain.tag_create(
188
author=b'author <author@example.com>',
190
tag_time=int(time.time()),
193
message=b"Annotated tag")
195
remote = ControlDir.open(self.remote_url)
196
self.make_controldir('local', format=self._to_format)
197
local = remote.sprout('local', revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
198
local_branch = local.open_branch()
200
default_mapping.revision_id_foreign_to_bzr(c1),
201
local_branch.last_revision())
203
{'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
204
local_branch.tags.get_tag_dict())
207
class FetchFromRemoteToBzrTests(FetchFromRemoteTestBase,TestCaseWithTransport):
212
class FetchFromRemoteToGitTests(FetchFromRemoteTestBase,TestCaseWithTransport):
217
class PushToRemoteBase(object):
219
_test_needs_features = [ExecutableFeature('git')]
224
TestCaseWithTransport.setUp(self)
225
self.remote_real = GitRepo.init('remote', mkdir=True)
226
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
227
self.permit_url(self.remote_url)
229
def test_push_branch_new(self):
230
remote = ControlDir.open(self.remote_url)
231
wt = self.make_branch_and_tree('local', format=self._from_format)
232
self.build_tree(['local/blah'])
234
revid = wt.commit('blah')
236
if self._from_format == 'git':
237
result = remote.push_branch(wt.branch, name='newbranch')
239
result = remote.push_branch(wt.branch, lossy=True, name='newbranch')
241
self.assertEqual(0, result.old_revno)
242
if self._from_format == 'git':
243
self.assertEqual(1, result.new_revno)
245
self.assertIs(None, result.new_revno)
247
result.report(BytesIO())
250
{b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
252
self.remote_real.get_refs())
255
c1 = self.remote_real.do_commit(
257
committer=b'committer <committer@example.com>',
258
author=b'author <author@example.com>')
260
remote = ControlDir.open(self.remote_url)
261
self.make_controldir('local', format=self._from_format)
262
local = remote.sprout('local')
263
self.build_tree(['local/blah'])
264
wt = local.open_workingtree()
266
revid = wt.commit('blah')
267
wt.branch.tags.set_tag('sometag', revid)
268
wt.branch.get_config_stack().set('branch.fetch_tags', True)
270
if self._from_format == 'git':
271
result = wt.branch.push(remote.create_branch('newbranch'))
273
result = wt.branch.push(remote.create_branch('newbranch'), lossy=True)
275
self.assertEqual(0, result.old_revno)
276
self.assertEqual(2, result.new_revno)
278
result.report(BytesIO())
281
{b'refs/heads/master': self.remote_real.head(),
282
b'HEAD': self.remote_real.head(),
283
b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
284
b'refs/tags/sometag': self.remote_real.refs[b'refs/heads/newbranch'],
286
self.remote_real.get_refs())
288
def test_push_diverged(self):
289
c1 = self.remote_real.do_commit(
291
committer=b'committer <committer@example.com>',
292
author=b'author <author@example.com>',
293
ref=b'refs/heads/newbranch')
295
remote = ControlDir.open(self.remote_url)
296
wt = self.make_branch_and_tree('local', format=self._from_format)
297
self.build_tree(['local/blah'])
299
revid = wt.commit('blah')
301
newbranch = remote.open_branch('newbranch')
302
if self._from_format == 'git':
303
self.assertRaises(DivergedBranches, wt.branch.push, newbranch)
305
self.assertRaises(DivergedBranches, wt.branch.push, newbranch, lossy=True)
308
{b'refs/heads/newbranch': c1 },
309
self.remote_real.get_refs())
311
if self._from_format == 'git':
312
wt.branch.push(newbranch, overwrite=True)
314
wt.branch.push(newbranch, lossy=True, overwrite=True)
316
self.assertNotEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
319
class PushToRemoteFromBzrTests(PushToRemoteBase,TestCaseWithTransport):
324
class PushToRemoteFromGitTests(PushToRemoteBase,TestCaseWithTransport):
329
class RemoteControlDirTests(TestCaseWithTransport):
331
_test_needs_features = [ExecutableFeature('git')]
334
TestCaseWithTransport.setUp(self)
335
self.remote_real = GitRepo.init('remote', mkdir=True)
336
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
337
self.permit_url(self.remote_url)
339
def test_remove_branch(self):
340
c1 = self.remote_real.do_commit(
342
committer=b'committer <committer@example.com>',
343
author=b'author <author@example.com>')
344
c2 = self.remote_real.do_commit(
345
message=b'another commit',
346
committer=b'committer <committer@example.com>',
347
author=b'author <author@example.com>',
348
ref=b'refs/heads/blah')
350
remote = ControlDir.open(self.remote_url)
351
remote.destroy_branch(name='blah')
353
self.remote_real.get_refs(),
354
{b'refs/heads/master': self.remote_real.head(),
355
b'HEAD': self.remote_real.head(),
358
def test_list_branches(self):
359
c1 = self.remote_real.do_commit(
361
committer=b'committer <committer@example.com>',
362
author=b'author <author@example.com>')
363
c2 = self.remote_real.do_commit(
364
message=b'another commit',
365
committer=b'committer <committer@example.com>',
366
author=b'author <author@example.com>',
367
ref=b'refs/heads/blah')
369
remote = ControlDir.open(self.remote_url)
371
set(['master', 'blah', 'master']),
372
set([b.name for b in remote.list_branches()]))
374
def test_get_branches(self):
375
c1 = self.remote_real.do_commit(
377
committer=b'committer <committer@example.com>',
378
author=b'author <author@example.com>')
379
c2 = self.remote_real.do_commit(
380
message=b'another commit',
381
committer=b'committer <committer@example.com>',
382
author=b'author <author@example.com>',
383
ref=b'refs/heads/blah')
385
remote = ControlDir.open(self.remote_url)
387
{'': 'master', 'blah': 'blah', 'master': 'master'},
388
{n: b.name for (n, b) in remote.get_branches().items()})
390
def test_remove_tag(self):
391
c1 = self.remote_real.do_commit(
393
committer=b'committer <committer@example.com>',
394
author=b'author <author@example.com>')
395
c2 = self.remote_real.do_commit(
396
message=b'another commit',
397
committer=b'committer <committer@example.com>',
398
author=b'author <author@example.com>',
399
ref=b'refs/tags/blah')
401
remote = ControlDir.open(self.remote_url)
402
remote_branch = remote.open_branch()
403
remote_branch.tags.delete_tag('blah')
404
self.assertRaises(NoSuchTag, remote_branch.tags.delete_tag, 'blah')
406
self.remote_real.get_refs(),
407
{b'refs/heads/master': self.remote_real.head(),
408
b'HEAD': self.remote_real.head(),
411
def test_set_tag(self):
412
c1 = self.remote_real.do_commit(
414
committer=b'committer <committer@example.com>',
415
author=b'author <author@example.com>')
416
c2 = self.remote_real.do_commit(
417
message=b'another commit',
418
committer=b'committer <committer@example.com>',
419
author=b'author <author@example.com>')
421
remote = ControlDir.open(self.remote_url)
422
remote.open_branch().tags.set_tag(
423
b'blah', default_mapping.revision_id_foreign_to_bzr(c1))
425
self.remote_real.get_refs(),
426
{b'refs/heads/master': self.remote_real.head(),
427
b'refs/tags/blah': c1,
428
b'HEAD': self.remote_real.head(),
431
def test_annotated_tag(self):
432
c1 = self.remote_real.do_commit(
434
committer=b'committer <committer@example.com>',
435
author=b'author <author@example.com>')
436
c2 = self.remote_real.do_commit(
437
message=b'another commit',
438
committer=b'committer <committer@example.com>',
439
author=b'author <author@example.com>')
441
porcelain.tag_create(
444
author=b'author <author@example.com>',
446
tag_time=int(time.time()),
449
message=b"Annotated tag")
451
remote = ControlDir.open(self.remote_url)
452
remote_branch = remote.open_branch()
454
'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
455
remote_branch.tags.get_tag_dict())
457
def tetst_get_branch_reference(self):
458
c1 = self.remote_real.do_commit(
460
committer=b'committer <committer@example.com>',
461
author=b'author <author@example.com>')
462
c2 = self.remote_real.do_commit(
463
message=b'another commit',
464
committer=b'committer <committer@example.com>',
465
author=b'author <author@example.com>')
467
remote = ControlDir.open(self.remote_url)
468
self.assertEqual(b'refs/heads/master', remote.get_branch_reference(''))
469
self.assertEqual(None, remote.get_branch_reference('master'))