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 (
34
from ...tests import (
36
TestCaseWithTransport,
38
from ...tests.features import ExecutableFeature
40
from ..mapping import default_mapping
41
from ..remote import (
46
RemoteGitBranchFormat,
49
from dulwich import porcelain
50
from dulwich.repo import Repo as GitRepo
53
class SplitUrlTests(TestCase):
55
def test_simple(self):
56
self.assertEqual(("foo", None, None, "/bar"),
57
split_git_url("git://foo/bar"))
60
self.assertEqual(("foo", 343, None, "/bar"),
61
split_git_url("git://foo:343/bar"))
63
def test_username(self):
64
self.assertEqual(("foo", None, "la", "/bar"),
65
split_git_url("git://la@foo/bar"))
67
def test_nopath(self):
68
self.assertEqual(("foo", None, None, "/"),
69
split_git_url("git://foo/"))
71
def test_slashpath(self):
72
self.assertEqual(("foo", None, None, "//bar"),
73
split_git_url("git://foo//bar"))
75
def test_homedir(self):
76
self.assertEqual(("foo", None, None, "~bar"),
77
split_git_url("git://foo/~bar"))
80
class ParseGitErrorTests(TestCase):
82
def test_unknown(self):
83
e = parse_git_error("url", "foo")
84
self.assertIsInstance(e, RemoteGitError)
86
def test_notbrancherror(self):
87
e = parse_git_error("url", "\n Could not find Repository foo/bar")
88
self.assertIsInstance(e, NotBranchError)
90
def test_notbrancherror_launchpad(self):
91
e = parse_git_error("url", "Repository 'foo/bar' not found.")
92
self.assertIsInstance(e, NotBranchError)
94
def test_notbrancherror_github(self):
95
e = parse_git_error("url", "Repository not found.\n")
96
self.assertIsInstance(e, NotBranchError)
98
def test_notbrancherror_normal(self):
100
"url", "fatal: '/srv/git/lintian-brush' does not appear to be a git repository")
101
self.assertIsInstance(e, NotBranchError)
103
def test_head_update(self):
104
e = parse_git_error("url", "HEAD failed to update\n")
105
self.assertIsInstance(e, HeadUpdateFailed)
107
def test_permission_dnied(self):
110
"access denied or repository not exported: /debian/altermime.git")
111
self.assertIsInstance(e, PermissionDenied)
113
def test_permission_denied_gitlab(self):
116
'GitLab: You are not allowed to push code to this project.\n')
117
self.assertIsInstance(e, PermissionDenied)
119
def test_permission_denied_github(self):
122
'Permission to porridge/gaduhistory.git denied to jelmer.')
123
self.assertIsInstance(e, PermissionDenied)
124
self.assertEqual(e.path, 'porridge/gaduhistory.git')
125
self.assertEqual(e.extra, ': denied to jelmer')
128
class TestRemoteGitBranchFormat(TestCase):
131
super(TestRemoteGitBranchFormat, self).setUp()
132
self.format = RemoteGitBranchFormat()
134
def test_get_format_description(self):
135
self.assertEqual("Remote Git Branch",
136
self.format.get_format_description())
138
def test_get_network_name(self):
139
self.assertEqual(b"git", self.format.network_name())
141
def test_supports_tags(self):
142
self.assertTrue(self.format.supports_tags())
145
class FetchFromRemoteTestBase(object):
147
_test_needs_features = [ExecutableFeature('git')]
152
TestCaseWithTransport.setUp(self)
153
self.remote_real = GitRepo.init('remote', mkdir=True)
154
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
155
self.permit_url(self.remote_url)
157
def test_sprout_simple(self):
158
self.remote_real.do_commit(
160
committer=b'committer <committer@example.com>',
161
author=b'author <author@example.com>')
163
remote = ControlDir.open(self.remote_url)
164
self.make_controldir('local', format=self._to_format)
165
local = remote.sprout('local')
167
default_mapping.revision_id_foreign_to_bzr(
168
self.remote_real.head()),
169
local.open_branch().last_revision())
171
def test_sprout_with_tags(self):
172
c1 = self.remote_real.do_commit(
174
committer=b'committer <committer@example.com>',
175
author=b'author <author@example.com>')
176
c2 = self.remote_real.do_commit(
177
message=b'another commit',
178
committer=b'committer <committer@example.com>',
179
author=b'author <author@example.com>',
180
ref=b'refs/tags/another')
181
self.remote_real.refs[b'refs/tags/blah'] = self.remote_real.head()
183
remote = ControlDir.open(self.remote_url)
184
self.make_controldir('local', format=self._to_format)
185
local = remote.sprout('local')
186
local_branch = local.open_branch()
188
default_mapping.revision_id_foreign_to_bzr(c1),
189
local_branch.last_revision())
191
{'blah': local_branch.last_revision(),
192
'another': default_mapping.revision_id_foreign_to_bzr(c2)},
193
local_branch.tags.get_tag_dict())
195
def test_sprout_with_annotated_tag(self):
196
c1 = self.remote_real.do_commit(
198
committer=b'committer <committer@example.com>',
199
author=b'author <author@example.com>')
200
c2 = self.remote_real.do_commit(
201
message=b'another commit',
202
committer=b'committer <committer@example.com>',
203
author=b'author <author@example.com>',
204
ref=b'refs/heads/another')
205
porcelain.tag_create(
208
author=b'author <author@example.com>',
210
tag_time=int(time.time()),
213
message=b"Annotated tag")
215
remote = ControlDir.open(self.remote_url)
216
self.make_controldir('local', format=self._to_format)
217
local = remote.sprout(
218
'local', revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
219
local_branch = local.open_branch()
221
default_mapping.revision_id_foreign_to_bzr(c1),
222
local_branch.last_revision())
224
{'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
225
local_branch.tags.get_tag_dict())
228
class FetchFromRemoteToBzrTests(FetchFromRemoteTestBase, TestCaseWithTransport):
233
class FetchFromRemoteToGitTests(FetchFromRemoteTestBase, TestCaseWithTransport):
238
class PushToRemoteBase(object):
240
_test_needs_features = [ExecutableFeature('git')]
245
TestCaseWithTransport.setUp(self)
246
self.remote_real = GitRepo.init('remote', mkdir=True)
247
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
248
self.permit_url(self.remote_url)
250
def test_push_branch_new(self):
251
remote = ControlDir.open(self.remote_url)
252
wt = self.make_branch_and_tree('local', format=self._from_format)
253
self.build_tree(['local/blah'])
255
revid = wt.commit('blah')
257
if self._from_format == 'git':
258
result = remote.push_branch(wt.branch, name='newbranch')
260
result = remote.push_branch(
261
wt.branch, lossy=True, name='newbranch')
263
self.assertEqual(0, result.old_revno)
264
if self._from_format == 'git':
265
self.assertEqual(1, result.new_revno)
267
self.assertIs(None, result.new_revno)
269
result.report(BytesIO())
272
{b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
274
self.remote_real.get_refs())
277
c1 = self.remote_real.do_commit(
279
committer=b'committer <committer@example.com>',
280
author=b'author <author@example.com>')
282
remote = ControlDir.open(self.remote_url)
283
self.make_controldir('local', format=self._from_format)
284
local = remote.sprout('local')
285
self.build_tree(['local/blah'])
286
wt = local.open_workingtree()
288
revid = wt.commit('blah')
289
wt.branch.tags.set_tag('sometag', revid)
290
wt.branch.get_config_stack().set('branch.fetch_tags', True)
292
if self._from_format == 'git':
293
result = wt.branch.push(remote.create_branch('newbranch'))
295
result = wt.branch.push(
296
remote.create_branch('newbranch'), lossy=True)
298
self.assertEqual(0, result.old_revno)
299
self.assertEqual(2, result.new_revno)
301
result.report(BytesIO())
304
{b'refs/heads/master': self.remote_real.head(),
305
b'HEAD': self.remote_real.head(),
306
b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
307
b'refs/tags/sometag': self.remote_real.refs[b'refs/heads/newbranch'],
309
self.remote_real.get_refs())
311
def test_push_diverged(self):
312
c1 = self.remote_real.do_commit(
314
committer=b'committer <committer@example.com>',
315
author=b'author <author@example.com>',
316
ref=b'refs/heads/newbranch')
318
remote = ControlDir.open(self.remote_url)
319
wt = self.make_branch_and_tree('local', format=self._from_format)
320
self.build_tree(['local/blah'])
322
revid = wt.commit('blah')
324
newbranch = remote.open_branch('newbranch')
325
if self._from_format == 'git':
326
self.assertRaises(DivergedBranches, wt.branch.push, newbranch)
328
self.assertRaises(DivergedBranches, wt.branch.push,
329
newbranch, lossy=True)
332
{b'refs/heads/newbranch': c1},
333
self.remote_real.get_refs())
335
if self._from_format == 'git':
336
wt.branch.push(newbranch, overwrite=True)
338
wt.branch.push(newbranch, lossy=True, overwrite=True)
340
self.assertNotEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
343
class PushToRemoteFromBzrTests(PushToRemoteBase, TestCaseWithTransport):
348
class PushToRemoteFromGitTests(PushToRemoteBase, TestCaseWithTransport):
353
class RemoteControlDirTests(TestCaseWithTransport):
355
_test_needs_features = [ExecutableFeature('git')]
358
TestCaseWithTransport.setUp(self)
359
self.remote_real = GitRepo.init('remote', mkdir=True)
360
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
361
self.permit_url(self.remote_url)
363
def test_remove_branch(self):
364
c1 = self.remote_real.do_commit(
366
committer=b'committer <committer@example.com>',
367
author=b'author <author@example.com>')
368
c2 = self.remote_real.do_commit(
369
message=b'another commit',
370
committer=b'committer <committer@example.com>',
371
author=b'author <author@example.com>',
372
ref=b'refs/heads/blah')
374
remote = ControlDir.open(self.remote_url)
375
remote.destroy_branch(name='blah')
377
self.remote_real.get_refs(),
378
{b'refs/heads/master': self.remote_real.head(),
379
b'HEAD': self.remote_real.head(),
382
def test_list_branches(self):
383
c1 = self.remote_real.do_commit(
385
committer=b'committer <committer@example.com>',
386
author=b'author <author@example.com>')
387
c2 = self.remote_real.do_commit(
388
message=b'another commit',
389
committer=b'committer <committer@example.com>',
390
author=b'author <author@example.com>',
391
ref=b'refs/heads/blah')
393
remote = ControlDir.open(self.remote_url)
395
set(['master', 'blah', 'master']),
396
set([b.name for b in remote.list_branches()]))
398
def test_get_branches(self):
399
c1 = self.remote_real.do_commit(
401
committer=b'committer <committer@example.com>',
402
author=b'author <author@example.com>')
403
c2 = self.remote_real.do_commit(
404
message=b'another commit',
405
committer=b'committer <committer@example.com>',
406
author=b'author <author@example.com>',
407
ref=b'refs/heads/blah')
409
remote = ControlDir.open(self.remote_url)
411
{'': 'master', 'blah': 'blah', 'master': 'master'},
412
{n: b.name for (n, b) in remote.get_branches().items()})
414
def test_remove_tag(self):
415
c1 = self.remote_real.do_commit(
417
committer=b'committer <committer@example.com>',
418
author=b'author <author@example.com>')
419
c2 = self.remote_real.do_commit(
420
message=b'another commit',
421
committer=b'committer <committer@example.com>',
422
author=b'author <author@example.com>',
423
ref=b'refs/tags/blah')
425
remote = ControlDir.open(self.remote_url)
426
remote_branch = remote.open_branch()
427
remote_branch.tags.delete_tag('blah')
428
self.assertRaises(NoSuchTag, remote_branch.tags.delete_tag, 'blah')
430
self.remote_real.get_refs(),
431
{b'refs/heads/master': self.remote_real.head(),
432
b'HEAD': self.remote_real.head(),
435
def test_set_tag(self):
436
c1 = self.remote_real.do_commit(
438
committer=b'committer <committer@example.com>',
439
author=b'author <author@example.com>')
440
c2 = self.remote_real.do_commit(
441
message=b'another commit',
442
committer=b'committer <committer@example.com>',
443
author=b'author <author@example.com>')
445
remote = ControlDir.open(self.remote_url)
446
remote.open_branch().tags.set_tag(
447
b'blah', default_mapping.revision_id_foreign_to_bzr(c1))
449
self.remote_real.get_refs(),
450
{b'refs/heads/master': self.remote_real.head(),
451
b'refs/tags/blah': c1,
452
b'HEAD': self.remote_real.head(),
455
def test_annotated_tag(self):
456
c1 = self.remote_real.do_commit(
458
committer=b'committer <committer@example.com>',
459
author=b'author <author@example.com>')
460
c2 = self.remote_real.do_commit(
461
message=b'another commit',
462
committer=b'committer <committer@example.com>',
463
author=b'author <author@example.com>')
465
porcelain.tag_create(
468
author=b'author <author@example.com>',
470
tag_time=int(time.time()),
473
message=b"Annotated tag")
475
remote = ControlDir.open(self.remote_url)
476
remote_branch = remote.open_branch()
478
'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
479
remote_branch.tags.get_tag_dict())
481
def tetst_get_branch_reference(self):
482
c1 = self.remote_real.do_commit(
484
committer=b'committer <committer@example.com>',
485
author=b'author <author@example.com>')
486
c2 = self.remote_real.do_commit(
487
message=b'another commit',
488
committer=b'committer <committer@example.com>',
489
author=b'author <author@example.com>')
491
remote = ControlDir.open(self.remote_url)
492
self.assertEqual(b'refs/heads/master', remote.get_branch_reference(''))
493
self.assertEqual(None, remote.get_branch_reference('master'))