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):
99
e = parse_git_error("url", "fatal: '/srv/git/lintian-brush' does not appear to be a git repository")
100
self.assertIsInstance(e, NotBranchError)
102
def test_head_update(self):
103
e = parse_git_error("url", "HEAD failed to update\n")
104
self.assertIsInstance(e, HeadUpdateFailed)
106
def test_permission_dnied(self):
109
"access denied or repository not exported: /debian/altermime.git")
110
self.assertIsInstance(e, PermissionDenied)
112
def test_permission_denied_gitlab(self):
115
'GitLab: You are not allowed to push code to this project.\n')
116
self.assertIsInstance(e, PermissionDenied)
118
def test_permission_denied_github(self):
121
'Permission to porridge/gaduhistory.git denied to jelmer.')
122
self.assertIsInstance(e, PermissionDenied)
123
self.assertEqual(e.path, 'porridge/gaduhistory.git')
124
self.assertEqual(e.extra, ': denied to jelmer')
127
class TestRemoteGitBranchFormat(TestCase):
130
super(TestRemoteGitBranchFormat, self).setUp()
131
self.format = RemoteGitBranchFormat()
133
def test_get_format_description(self):
134
self.assertEqual("Remote Git Branch", self.format.get_format_description())
136
def test_get_network_name(self):
137
self.assertEqual(b"git", self.format.network_name())
139
def test_supports_tags(self):
140
self.assertTrue(self.format.supports_tags())
143
class TestRemoteGitBranch(TestCaseWithTransport):
146
TestCaseWithTransport.setUp(self)
147
self.remote_real = GitRepo.init('remote', mkdir=True)
148
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
149
self.permit_url(self.remote_url)
151
def test_set_last_revision_info(self):
152
c1 = self.remote_real.do_commit(
153
message=b'message 1',
154
committer=b'committer <committer@example.com>',
155
author=b'author <author@example.com>',
156
ref=b'refs/heads/newbranch')
157
c2 = self.remote_real.do_commit(
158
message=b'message 2',
159
committer=b'committer <committer@example.com>',
160
author=b'author <author@example.com>',
161
ref=b'refs/heads/newbranch')
163
remote = ControlDir.open(self.remote_url)
164
newbranch = remote.open_branch('newbranch')
165
self.assertEqual(newbranch.lookup_foreign_revision_id(c2),
166
newbranch.last_revision())
167
newbranch.set_last_revision_info(
168
1, newbranch.lookup_foreign_revision_id(c1))
169
self.assertEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
170
self.assertEqual(newbranch.last_revision(),
171
newbranch.lookup_foreign_revision_id(c1))
174
class FetchFromRemoteTestBase(object):
176
_test_needs_features = [ExecutableFeature('git')]
181
TestCaseWithTransport.setUp(self)
182
self.remote_real = GitRepo.init('remote', mkdir=True)
183
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
184
self.permit_url(self.remote_url)
186
def test_sprout_simple(self):
187
self.remote_real.do_commit(
189
committer=b'committer <committer@example.com>',
190
author=b'author <author@example.com>')
192
remote = ControlDir.open(self.remote_url)
193
self.make_controldir('local', format=self._to_format)
194
local = remote.sprout('local')
196
default_mapping.revision_id_foreign_to_bzr(self.remote_real.head()),
197
local.open_branch().last_revision())
199
def test_sprout_with_tags(self):
200
c1 = self.remote_real.do_commit(
202
committer=b'committer <committer@example.com>',
203
author=b'author <author@example.com>')
204
c2 = self.remote_real.do_commit(
205
message=b'another commit',
206
committer=b'committer <committer@example.com>',
207
author=b'author <author@example.com>',
208
ref=b'refs/tags/another')
209
self.remote_real.refs[b'refs/tags/blah'] = self.remote_real.head()
211
remote = ControlDir.open(self.remote_url)
212
self.make_controldir('local', format=self._to_format)
213
local = remote.sprout('local')
214
local_branch = local.open_branch()
216
default_mapping.revision_id_foreign_to_bzr(c1),
217
local_branch.last_revision())
219
{'blah': local_branch.last_revision(),
220
'another': default_mapping.revision_id_foreign_to_bzr(c2)},
221
local_branch.tags.get_tag_dict())
223
def test_sprout_with_annotated_tag(self):
224
c1 = self.remote_real.do_commit(
226
committer=b'committer <committer@example.com>',
227
author=b'author <author@example.com>')
228
c2 = self.remote_real.do_commit(
229
message=b'another commit',
230
committer=b'committer <committer@example.com>',
231
author=b'author <author@example.com>',
232
ref=b'refs/heads/another')
233
porcelain.tag_create(
236
author=b'author <author@example.com>',
238
tag_time=int(time.time()),
241
message=b"Annotated tag")
243
remote = ControlDir.open(self.remote_url)
244
self.make_controldir('local', format=self._to_format)
245
local = remote.sprout('local', revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
246
local_branch = local.open_branch()
248
default_mapping.revision_id_foreign_to_bzr(c1),
249
local_branch.last_revision())
251
{'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
252
local_branch.tags.get_tag_dict())
254
def test_sprout_with_annotated_tag_unreferenced(self):
255
c1 = self.remote_real.do_commit(
257
committer=b'committer <committer@example.com>',
258
author=b'author <author@example.com>')
259
c2 = self.remote_real.do_commit(
260
message=b'another commit',
261
committer=b'committer <committer@example.com>',
262
author=b'author <author@example.com>')
263
porcelain.tag_create(
266
author=b'author <author@example.com>',
268
tag_time=int(time.time()),
271
message=b"Annotated tag")
273
remote = ControlDir.open(self.remote_url)
274
self.make_controldir('local', format=self._to_format)
275
local = remote.sprout(
277
revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
278
local_branch = local.open_branch()
280
default_mapping.revision_id_foreign_to_bzr(c1),
281
local_branch.last_revision())
283
{'blah': default_mapping.revision_id_foreign_to_bzr(c1)},
284
local_branch.tags.get_tag_dict())
287
class FetchFromRemoteToBzrTests(FetchFromRemoteTestBase,TestCaseWithTransport):
292
class FetchFromRemoteToGitTests(FetchFromRemoteTestBase,TestCaseWithTransport):
297
class PushToRemoteBase(object):
299
_test_needs_features = [ExecutableFeature('git')]
304
TestCaseWithTransport.setUp(self)
305
self.remote_real = GitRepo.init('remote', mkdir=True)
306
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
307
self.permit_url(self.remote_url)
309
def test_push_branch_new(self):
310
remote = ControlDir.open(self.remote_url)
311
wt = self.make_branch_and_tree('local', format=self._from_format)
312
self.build_tree(['local/blah'])
314
revid = wt.commit('blah')
316
if self._from_format == 'git':
317
result = remote.push_branch(wt.branch, name='newbranch')
319
result = remote.push_branch(wt.branch, lossy=True, name='newbranch')
321
self.assertEqual(0, result.old_revno)
322
if self._from_format == 'git':
323
self.assertEqual(1, result.new_revno)
325
self.assertIs(None, result.new_revno)
327
result.report(BytesIO())
330
{b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
332
self.remote_real.get_refs())
335
c1 = self.remote_real.do_commit(
337
committer=b'committer <committer@example.com>',
338
author=b'author <author@example.com>')
340
remote = ControlDir.open(self.remote_url)
341
self.make_controldir('local', format=self._from_format)
342
local = remote.sprout('local')
343
self.build_tree(['local/blah'])
344
wt = local.open_workingtree()
346
revid = wt.commit('blah')
347
wt.branch.tags.set_tag('sometag', revid)
348
wt.branch.get_config_stack().set('branch.fetch_tags', True)
350
if self._from_format == 'git':
351
result = wt.branch.push(remote.create_branch('newbranch'))
353
result = wt.branch.push(remote.create_branch('newbranch'), lossy=True)
355
self.assertEqual(0, result.old_revno)
356
self.assertEqual(2, result.new_revno)
358
result.report(BytesIO())
361
{b'refs/heads/master': self.remote_real.head(),
362
b'HEAD': self.remote_real.head(),
363
b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
364
b'refs/tags/sometag': self.remote_real.refs[b'refs/heads/newbranch'],
366
self.remote_real.get_refs())
368
def test_push_diverged(self):
369
c1 = self.remote_real.do_commit(
371
committer=b'committer <committer@example.com>',
372
author=b'author <author@example.com>',
373
ref=b'refs/heads/newbranch')
375
remote = ControlDir.open(self.remote_url)
376
wt = self.make_branch_and_tree('local', format=self._from_format)
377
self.build_tree(['local/blah'])
379
revid = wt.commit('blah')
381
newbranch = remote.open_branch('newbranch')
382
if self._from_format == 'git':
383
self.assertRaises(DivergedBranches, wt.branch.push, newbranch)
385
self.assertRaises(DivergedBranches, wt.branch.push, newbranch, lossy=True)
388
{b'refs/heads/newbranch': c1 },
389
self.remote_real.get_refs())
391
if self._from_format == 'git':
392
wt.branch.push(newbranch, overwrite=True)
394
wt.branch.push(newbranch, lossy=True, overwrite=True)
396
self.assertNotEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
399
class PushToRemoteFromBzrTests(PushToRemoteBase,TestCaseWithTransport):
404
class PushToRemoteFromGitTests(PushToRemoteBase,TestCaseWithTransport):
409
class RemoteControlDirTests(TestCaseWithTransport):
411
_test_needs_features = [ExecutableFeature('git')]
414
TestCaseWithTransport.setUp(self)
415
self.remote_real = GitRepo.init('remote', mkdir=True)
416
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
417
self.permit_url(self.remote_url)
419
def test_remove_branch(self):
420
c1 = self.remote_real.do_commit(
422
committer=b'committer <committer@example.com>',
423
author=b'author <author@example.com>')
424
c2 = self.remote_real.do_commit(
425
message=b'another commit',
426
committer=b'committer <committer@example.com>',
427
author=b'author <author@example.com>',
428
ref=b'refs/heads/blah')
430
remote = ControlDir.open(self.remote_url)
431
remote.destroy_branch(name='blah')
433
self.remote_real.get_refs(),
434
{b'refs/heads/master': self.remote_real.head(),
435
b'HEAD': self.remote_real.head(),
438
def test_list_branches(self):
439
c1 = self.remote_real.do_commit(
441
committer=b'committer <committer@example.com>',
442
author=b'author <author@example.com>')
443
c2 = self.remote_real.do_commit(
444
message=b'another commit',
445
committer=b'committer <committer@example.com>',
446
author=b'author <author@example.com>',
447
ref=b'refs/heads/blah')
449
remote = ControlDir.open(self.remote_url)
451
set(['master', 'blah', 'master']),
452
set([b.name for b in remote.list_branches()]))
454
def test_get_branches(self):
455
c1 = self.remote_real.do_commit(
457
committer=b'committer <committer@example.com>',
458
author=b'author <author@example.com>')
459
c2 = self.remote_real.do_commit(
460
message=b'another commit',
461
committer=b'committer <committer@example.com>',
462
author=b'author <author@example.com>',
463
ref=b'refs/heads/blah')
465
remote = ControlDir.open(self.remote_url)
467
{'': 'master', 'blah': 'blah', 'master': 'master'},
468
{n: b.name for (n, b) in remote.get_branches().items()})
470
def test_remove_tag(self):
471
c1 = self.remote_real.do_commit(
473
committer=b'committer <committer@example.com>',
474
author=b'author <author@example.com>')
475
c2 = self.remote_real.do_commit(
476
message=b'another commit',
477
committer=b'committer <committer@example.com>',
478
author=b'author <author@example.com>',
479
ref=b'refs/tags/blah')
481
remote = ControlDir.open(self.remote_url)
482
remote_branch = remote.open_branch()
483
remote_branch.tags.delete_tag('blah')
484
self.assertRaises(NoSuchTag, remote_branch.tags.delete_tag, 'blah')
486
self.remote_real.get_refs(),
487
{b'refs/heads/master': self.remote_real.head(),
488
b'HEAD': self.remote_real.head(),
491
def test_set_tag(self):
492
c1 = self.remote_real.do_commit(
494
committer=b'committer <committer@example.com>',
495
author=b'author <author@example.com>')
496
c2 = self.remote_real.do_commit(
497
message=b'another commit',
498
committer=b'committer <committer@example.com>',
499
author=b'author <author@example.com>')
501
remote = ControlDir.open(self.remote_url)
502
remote.open_branch().tags.set_tag(
503
b'blah', default_mapping.revision_id_foreign_to_bzr(c1))
505
self.remote_real.get_refs(),
506
{b'refs/heads/master': self.remote_real.head(),
507
b'refs/tags/blah': c1,
508
b'HEAD': self.remote_real.head(),
511
def test_annotated_tag(self):
512
c1 = self.remote_real.do_commit(
514
committer=b'committer <committer@example.com>',
515
author=b'author <author@example.com>')
516
c2 = self.remote_real.do_commit(
517
message=b'another commit',
518
committer=b'committer <committer@example.com>',
519
author=b'author <author@example.com>')
521
porcelain.tag_create(
524
author=b'author <author@example.com>',
526
tag_time=int(time.time()),
529
message=b"Annotated tag")
531
remote = ControlDir.open(self.remote_url)
532
remote_branch = remote.open_branch()
534
'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
535
remote_branch.tags.get_tag_dict())
537
def test_get_branch_reference(self):
538
c1 = self.remote_real.do_commit(
540
committer=b'committer <committer@example.com>',
541
author=b'author <author@example.com>')
542
c2 = self.remote_real.do_commit(
543
message=b'another commit',
544
committer=b'committer <committer@example.com>',
545
author=b'author <author@example.com>')
547
remote = ControlDir.open(self.remote_url)
548
self.assertEqual(b'refs/heads/master', remote.get_branch_reference(''))
549
self.assertEqual(None, remote.get_branch_reference('master'))
551
def test_get_branch_nick(self):
552
c1 = self.remote_real.do_commit(
554
committer=b'committer <committer@example.com>',
555
author=b'author <author@example.com>')
556
remote = ControlDir.open(self.remote_url)
557
self.assertEqual('master', remote.open_branch().nick)