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_notbrancherror_normal(self):
100
e = parse_git_error("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", self.format.get_format_description())
137
def test_get_network_name(self):
138
self.assertEqual(b"git", self.format.network_name())
140
def test_supports_tags(self):
141
self.assertTrue(self.format.supports_tags())
144
class FetchFromRemoteTestBase(object):
146
_test_needs_features = [ExecutableFeature('git')]
151
TestCaseWithTransport.setUp(self)
152
self.remote_real = GitRepo.init('remote', mkdir=True)
153
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
154
self.permit_url(self.remote_url)
156
def test_sprout_simple(self):
157
self.remote_real.do_commit(
159
committer=b'committer <committer@example.com>',
160
author=b'author <author@example.com>')
162
remote = ControlDir.open(self.remote_url)
163
self.make_controldir('local', format=self._to_format)
164
local = remote.sprout('local')
166
default_mapping.revision_id_foreign_to_bzr(self.remote_real.head()),
167
local.open_branch().last_revision())
169
def test_sprout_with_tags(self):
170
c1 = self.remote_real.do_commit(
172
committer=b'committer <committer@example.com>',
173
author=b'author <author@example.com>')
174
c2 = self.remote_real.do_commit(
175
message=b'another commit',
176
committer=b'committer <committer@example.com>',
177
author=b'author <author@example.com>',
178
ref=b'refs/tags/another')
179
self.remote_real.refs[b'refs/tags/blah'] = self.remote_real.head()
181
remote = ControlDir.open(self.remote_url)
182
self.make_controldir('local', format=self._to_format)
183
local = remote.sprout('local')
184
local_branch = local.open_branch()
186
default_mapping.revision_id_foreign_to_bzr(c1),
187
local_branch.last_revision())
189
{'blah': local_branch.last_revision(),
190
'another': default_mapping.revision_id_foreign_to_bzr(c2)},
191
local_branch.tags.get_tag_dict())
193
def test_sprout_with_annotated_tag(self):
194
c1 = self.remote_real.do_commit(
196
committer=b'committer <committer@example.com>',
197
author=b'author <author@example.com>')
198
c2 = self.remote_real.do_commit(
199
message=b'another commit',
200
committer=b'committer <committer@example.com>',
201
author=b'author <author@example.com>',
202
ref=b'refs/heads/another')
203
porcelain.tag_create(
206
author=b'author <author@example.com>',
208
tag_time=int(time.time()),
211
message=b"Annotated tag")
213
remote = ControlDir.open(self.remote_url)
214
self.make_controldir('local', format=self._to_format)
215
local = remote.sprout('local', revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
216
local_branch = local.open_branch()
218
default_mapping.revision_id_foreign_to_bzr(c1),
219
local_branch.last_revision())
221
{'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
222
local_branch.tags.get_tag_dict())
225
class FetchFromRemoteToBzrTests(FetchFromRemoteTestBase,TestCaseWithTransport):
230
class FetchFromRemoteToGitTests(FetchFromRemoteTestBase,TestCaseWithTransport):
235
class PushToRemoteBase(object):
237
_test_needs_features = [ExecutableFeature('git')]
242
TestCaseWithTransport.setUp(self)
243
self.remote_real = GitRepo.init('remote', mkdir=True)
244
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
245
self.permit_url(self.remote_url)
247
def test_push_branch_new(self):
248
remote = ControlDir.open(self.remote_url)
249
wt = self.make_branch_and_tree('local', format=self._from_format)
250
self.build_tree(['local/blah'])
252
revid = wt.commit('blah')
254
if self._from_format == 'git':
255
result = remote.push_branch(wt.branch, name='newbranch')
257
result = remote.push_branch(wt.branch, lossy=True, name='newbranch')
259
self.assertEqual(0, result.old_revno)
260
if self._from_format == 'git':
261
self.assertEqual(1, result.new_revno)
263
self.assertIs(None, result.new_revno)
265
result.report(BytesIO())
268
{b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
270
self.remote_real.get_refs())
273
c1 = self.remote_real.do_commit(
275
committer=b'committer <committer@example.com>',
276
author=b'author <author@example.com>')
278
remote = ControlDir.open(self.remote_url)
279
self.make_controldir('local', format=self._from_format)
280
local = remote.sprout('local')
281
self.build_tree(['local/blah'])
282
wt = local.open_workingtree()
284
revid = wt.commit('blah')
285
wt.branch.tags.set_tag('sometag', revid)
286
wt.branch.get_config_stack().set('branch.fetch_tags', True)
288
if self._from_format == 'git':
289
result = wt.branch.push(remote.create_branch('newbranch'))
291
result = wt.branch.push(remote.create_branch('newbranch'), lossy=True)
293
self.assertEqual(0, result.old_revno)
294
self.assertEqual(2, result.new_revno)
296
result.report(BytesIO())
299
{b'refs/heads/master': self.remote_real.head(),
300
b'HEAD': self.remote_real.head(),
301
b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
302
b'refs/tags/sometag': self.remote_real.refs[b'refs/heads/newbranch'],
304
self.remote_real.get_refs())
306
def test_push_diverged(self):
307
c1 = self.remote_real.do_commit(
309
committer=b'committer <committer@example.com>',
310
author=b'author <author@example.com>',
311
ref=b'refs/heads/newbranch')
313
remote = ControlDir.open(self.remote_url)
314
wt = self.make_branch_and_tree('local', format=self._from_format)
315
self.build_tree(['local/blah'])
317
revid = wt.commit('blah')
319
newbranch = remote.open_branch('newbranch')
320
if self._from_format == 'git':
321
self.assertRaises(DivergedBranches, wt.branch.push, newbranch)
323
self.assertRaises(DivergedBranches, wt.branch.push, newbranch, lossy=True)
326
{b'refs/heads/newbranch': c1 },
327
self.remote_real.get_refs())
329
if self._from_format == 'git':
330
wt.branch.push(newbranch, overwrite=True)
332
wt.branch.push(newbranch, lossy=True, overwrite=True)
334
self.assertNotEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
337
class PushToRemoteFromBzrTests(PushToRemoteBase,TestCaseWithTransport):
342
class PushToRemoteFromGitTests(PushToRemoteBase,TestCaseWithTransport):
347
class RemoteControlDirTests(TestCaseWithTransport):
349
_test_needs_features = [ExecutableFeature('git')]
352
TestCaseWithTransport.setUp(self)
353
self.remote_real = GitRepo.init('remote', mkdir=True)
354
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
355
self.permit_url(self.remote_url)
357
def test_remove_branch(self):
358
c1 = self.remote_real.do_commit(
360
committer=b'committer <committer@example.com>',
361
author=b'author <author@example.com>')
362
c2 = self.remote_real.do_commit(
363
message=b'another commit',
364
committer=b'committer <committer@example.com>',
365
author=b'author <author@example.com>',
366
ref=b'refs/heads/blah')
368
remote = ControlDir.open(self.remote_url)
369
remote.destroy_branch(name='blah')
371
self.remote_real.get_refs(),
372
{b'refs/heads/master': self.remote_real.head(),
373
b'HEAD': self.remote_real.head(),
376
def test_list_branches(self):
377
c1 = self.remote_real.do_commit(
379
committer=b'committer <committer@example.com>',
380
author=b'author <author@example.com>')
381
c2 = self.remote_real.do_commit(
382
message=b'another commit',
383
committer=b'committer <committer@example.com>',
384
author=b'author <author@example.com>',
385
ref=b'refs/heads/blah')
387
remote = ControlDir.open(self.remote_url)
389
set(['master', 'blah', 'master']),
390
set([b.name for b in remote.list_branches()]))
392
def test_get_branches(self):
393
c1 = self.remote_real.do_commit(
395
committer=b'committer <committer@example.com>',
396
author=b'author <author@example.com>')
397
c2 = self.remote_real.do_commit(
398
message=b'another commit',
399
committer=b'committer <committer@example.com>',
400
author=b'author <author@example.com>',
401
ref=b'refs/heads/blah')
403
remote = ControlDir.open(self.remote_url)
405
{'': 'master', 'blah': 'blah', 'master': 'master'},
406
{n: b.name for (n, b) in remote.get_branches().items()})
408
def test_remove_tag(self):
409
c1 = self.remote_real.do_commit(
411
committer=b'committer <committer@example.com>',
412
author=b'author <author@example.com>')
413
c2 = self.remote_real.do_commit(
414
message=b'another commit',
415
committer=b'committer <committer@example.com>',
416
author=b'author <author@example.com>',
417
ref=b'refs/tags/blah')
419
remote = ControlDir.open(self.remote_url)
420
remote_branch = remote.open_branch()
421
remote_branch.tags.delete_tag('blah')
422
self.assertRaises(NoSuchTag, remote_branch.tags.delete_tag, 'blah')
424
self.remote_real.get_refs(),
425
{b'refs/heads/master': self.remote_real.head(),
426
b'HEAD': self.remote_real.head(),
429
def test_set_tag(self):
430
c1 = self.remote_real.do_commit(
432
committer=b'committer <committer@example.com>',
433
author=b'author <author@example.com>')
434
c2 = self.remote_real.do_commit(
435
message=b'another commit',
436
committer=b'committer <committer@example.com>',
437
author=b'author <author@example.com>')
439
remote = ControlDir.open(self.remote_url)
440
remote.open_branch().tags.set_tag(
441
b'blah', default_mapping.revision_id_foreign_to_bzr(c1))
443
self.remote_real.get_refs(),
444
{b'refs/heads/master': self.remote_real.head(),
445
b'refs/tags/blah': c1,
446
b'HEAD': self.remote_real.head(),
449
def test_annotated_tag(self):
450
c1 = self.remote_real.do_commit(
452
committer=b'committer <committer@example.com>',
453
author=b'author <author@example.com>')
454
c2 = self.remote_real.do_commit(
455
message=b'another commit',
456
committer=b'committer <committer@example.com>',
457
author=b'author <author@example.com>')
459
porcelain.tag_create(
462
author=b'author <author@example.com>',
464
tag_time=int(time.time()),
467
message=b"Annotated tag")
469
remote = ControlDir.open(self.remote_url)
470
remote_branch = remote.open_branch()
472
'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
473
remote_branch.tags.get_tag_dict())
475
def tetst_get_branch_reference(self):
476
c1 = self.remote_real.do_commit(
478
committer=b'committer <committer@example.com>',
479
author=b'author <author@example.com>')
480
c2 = self.remote_real.do_commit(
481
message=b'another commit',
482
committer=b'committer <committer@example.com>',
483
author=b'author <author@example.com>')
485
remote = ControlDir.open(self.remote_url)
486
self.assertEqual(b'refs/heads/master', remote.get_branch_reference(''))
487
self.assertEqual(None, remote.get_branch_reference('master'))