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 (
47
RemoteGitBranchFormat,
48
_git_url_and_path_from_transport,
51
from dulwich import porcelain
52
from dulwich.errors import HangupException
53
from dulwich.repo import Repo as GitRepo
56
class SplitUrlTests(TestCase):
58
def test_simple(self):
59
self.assertEqual(("foo", None, None, "/bar"),
60
split_git_url("git://foo/bar"))
63
self.assertEqual(("foo", 343, None, "/bar"),
64
split_git_url("git://foo:343/bar"))
66
def test_username(self):
67
self.assertEqual(("foo", None, "la", "/bar"),
68
split_git_url("git://la@foo/bar"))
70
def test_username_password(self):
72
("foo", None, "la", "/bar"),
73
split_git_url("git://la:passwd@foo/bar"))
75
def test_nopath(self):
76
self.assertEqual(("foo", None, None, "/"),
77
split_git_url("git://foo/"))
79
def test_slashpath(self):
80
self.assertEqual(("foo", None, None, "//bar"),
81
split_git_url("git://foo//bar"))
83
def test_homedir(self):
84
self.assertEqual(("foo", None, None, "~bar"),
85
split_git_url("git://foo/~bar"))
89
("", None, None, "/bar"),
90
split_git_url("file:///bar"))
93
class ParseGitErrorTests(TestCase):
95
def test_unknown(self):
96
e = parse_git_error("url", "foo")
97
self.assertIsInstance(e, RemoteGitError)
99
def test_notbrancherror(self):
100
e = parse_git_error("url", "\n Could not find Repository foo/bar")
101
self.assertIsInstance(e, NotBranchError)
103
def test_notbrancherror_launchpad(self):
104
e = parse_git_error("url", "Repository 'foo/bar' not found.")
105
self.assertIsInstance(e, NotBranchError)
107
def test_notbrancherror_github(self):
108
e = parse_git_error("url", "Repository not found.\n")
109
self.assertIsInstance(e, NotBranchError)
111
def test_notbrancherror_normal(self):
113
"url", "fatal: '/srv/git/lintian-brush' does not appear to be a git repository")
114
self.assertIsInstance(e, NotBranchError)
116
def test_head_update(self):
117
e = parse_git_error("url", "HEAD failed to update\n")
118
self.assertIsInstance(e, HeadUpdateFailed)
120
def test_permission_dnied(self):
123
"access denied or repository not exported: /debian/altermime.git")
124
self.assertIsInstance(e, PermissionDenied)
126
def test_permission_denied_gitlab(self):
129
'GitLab: You are not allowed to push code to this project.\n')
130
self.assertIsInstance(e, PermissionDenied)
132
def test_permission_denied_github(self):
135
'Permission to porridge/gaduhistory.git denied to jelmer.')
136
self.assertIsInstance(e, PermissionDenied)
137
self.assertEqual(e.path, 'porridge/gaduhistory.git')
138
self.assertEqual(e.extra, ': denied to jelmer')
140
def test_invalid_repo_name(self):
143
"""Gregwar/fatcat/tree/debian is not a valid repository name
144
Email support@github.com for help
146
self.assertIsInstance(e, NotBranchError)
149
class ParseHangupTests(TestCase):
152
super(ParseHangupTests, self).setUp()
154
HangupException([b'foo'])
156
self.skipTest('dulwich version too old')
158
def test_not_set(self):
159
self.assertIsInstance(
160
parse_git_hangup('http://', HangupException()), HangupException)
162
def test_single_line(self):
164
RemoteGitError('foo bar'),
165
parse_git_hangup('http://', HangupException([b'foo bar'])))
167
def test_multi_lines(self):
169
RemoteGitError('foo bar\nbla bla'),
171
'http://', HangupException([b'foo bar', b'bla bla'])))
173
def test_filter_boring(self):
175
RemoteGitError('foo bar'), parse_git_hangup('http://', HangupException(
176
[b'=======', b'foo bar', b'======'])))
178
RemoteGitError('foo bar'), parse_git_hangup('http://', HangupException(
179
[b'remote: =======', b'remote: foo bar', b'remote: ======'])))
181
def test_permission_denied(self):
183
PermissionDenied('http://', 'You are not allowed to push code to this project.'),
188
b'You are not allowed to push code to this project.', b'', b'======'])))
191
class TestRemoteGitBranchFormat(TestCase):
194
super(TestRemoteGitBranchFormat, self).setUp()
195
self.format = RemoteGitBranchFormat()
197
def test_get_format_description(self):
198
self.assertEqual("Remote Git Branch",
199
self.format.get_format_description())
201
def test_get_network_name(self):
202
self.assertEqual(b"git", self.format.network_name())
204
def test_supports_tags(self):
205
self.assertTrue(self.format.supports_tags())
208
class TestRemoteGitBranch(TestCaseWithTransport):
210
_test_needs_features = [ExecutableFeature('git')]
213
TestCaseWithTransport.setUp(self)
214
self.remote_real = GitRepo.init('remote', mkdir=True)
215
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
216
self.permit_url(self.remote_url)
218
def test_set_last_revision_info(self):
219
c1 = self.remote_real.do_commit(
220
message=b'message 1',
221
committer=b'committer <committer@example.com>',
222
author=b'author <author@example.com>',
223
ref=b'refs/heads/newbranch')
224
c2 = self.remote_real.do_commit(
225
message=b'message 2',
226
committer=b'committer <committer@example.com>',
227
author=b'author <author@example.com>',
228
ref=b'refs/heads/newbranch')
230
remote = ControlDir.open(self.remote_url)
231
newbranch = remote.open_branch('newbranch')
232
self.assertEqual(newbranch.lookup_foreign_revision_id(c2),
233
newbranch.last_revision())
234
newbranch.set_last_revision_info(
235
1, newbranch.lookup_foreign_revision_id(c1))
236
self.assertEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
237
self.assertEqual(newbranch.last_revision(),
238
newbranch.lookup_foreign_revision_id(c1))
241
class FetchFromRemoteTestBase(object):
243
_test_needs_features = [ExecutableFeature('git')]
248
TestCaseWithTransport.setUp(self)
249
self.remote_real = GitRepo.init('remote', mkdir=True)
250
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
251
self.permit_url(self.remote_url)
253
def test_sprout_simple(self):
254
self.remote_real.do_commit(
256
committer=b'committer <committer@example.com>',
257
author=b'author <author@example.com>')
259
remote = ControlDir.open(self.remote_url)
260
self.make_controldir('local', format=self._to_format)
261
local = remote.sprout('local')
263
default_mapping.revision_id_foreign_to_bzr(
264
self.remote_real.head()),
265
local.open_branch().last_revision())
267
def test_sprout_with_tags(self):
268
c1 = self.remote_real.do_commit(
270
committer=b'committer <committer@example.com>',
271
author=b'author <author@example.com>')
272
c2 = self.remote_real.do_commit(
273
message=b'another commit',
274
committer=b'committer <committer@example.com>',
275
author=b'author <author@example.com>',
276
ref=b'refs/tags/another')
277
self.remote_real.refs[b'refs/tags/blah'] = self.remote_real.head()
279
remote = ControlDir.open(self.remote_url)
280
self.make_controldir('local', format=self._to_format)
281
local = remote.sprout('local')
282
local_branch = local.open_branch()
284
default_mapping.revision_id_foreign_to_bzr(c1),
285
local_branch.last_revision())
287
{'blah': local_branch.last_revision(),
288
'another': default_mapping.revision_id_foreign_to_bzr(c2)},
289
local_branch.tags.get_tag_dict())
291
def test_sprout_with_annotated_tag(self):
292
c1 = self.remote_real.do_commit(
294
committer=b'committer <committer@example.com>',
295
author=b'author <author@example.com>')
296
c2 = self.remote_real.do_commit(
297
message=b'another commit',
298
committer=b'committer <committer@example.com>',
299
author=b'author <author@example.com>',
300
ref=b'refs/heads/another')
301
porcelain.tag_create(
304
author=b'author <author@example.com>',
306
tag_time=int(time.time()),
309
message=b"Annotated tag")
311
remote = ControlDir.open(self.remote_url)
312
self.make_controldir('local', format=self._to_format)
313
local = remote.sprout(
314
'local', revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
315
local_branch = local.open_branch()
317
default_mapping.revision_id_foreign_to_bzr(c1),
318
local_branch.last_revision())
320
{'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
321
local_branch.tags.get_tag_dict())
323
def test_sprout_with_annotated_tag_unreferenced(self):
324
c1 = self.remote_real.do_commit(
326
committer=b'committer <committer@example.com>',
327
author=b'author <author@example.com>')
328
c2 = self.remote_real.do_commit(
329
message=b'another commit',
330
committer=b'committer <committer@example.com>',
331
author=b'author <author@example.com>')
332
porcelain.tag_create(
335
author=b'author <author@example.com>',
337
tag_time=int(time.time()),
340
message=b"Annotated tag")
342
remote = ControlDir.open(self.remote_url)
343
self.make_controldir('local', format=self._to_format)
344
local = remote.sprout(
346
revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
347
local_branch = local.open_branch()
349
default_mapping.revision_id_foreign_to_bzr(c1),
350
local_branch.last_revision())
352
{'blah': default_mapping.revision_id_foreign_to_bzr(c1)},
353
local_branch.tags.get_tag_dict())
356
class FetchFromRemoteToBzrTests(FetchFromRemoteTestBase, TestCaseWithTransport):
361
class FetchFromRemoteToGitTests(FetchFromRemoteTestBase, TestCaseWithTransport):
366
class PushToRemoteBase(object):
368
_test_needs_features = [ExecutableFeature('git')]
373
TestCaseWithTransport.setUp(self)
374
self.remote_real = GitRepo.init('remote', mkdir=True)
375
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
376
self.permit_url(self.remote_url)
378
def test_push_branch_new(self):
379
remote = ControlDir.open(self.remote_url)
380
wt = self.make_branch_and_tree('local', format=self._from_format)
381
self.build_tree(['local/blah'])
383
revid = wt.commit('blah')
385
if self._from_format == 'git':
386
result = remote.push_branch(wt.branch, name='newbranch')
388
result = remote.push_branch(
389
wt.branch, lossy=True, name='newbranch')
391
self.assertEqual(0, result.old_revno)
392
if self._from_format == 'git':
393
self.assertEqual(1, result.new_revno)
395
self.assertIs(None, result.new_revno)
397
result.report(BytesIO())
400
{b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
402
self.remote_real.get_refs())
404
def test_push_branch_symref(self):
405
cfg = self.remote_real.get_config()
406
cfg.set((b'core', ), b'bare', True)
408
self.remote_real.refs.set_symbolic_ref(b'HEAD', b'refs/heads/master')
409
c1 = self.remote_real.do_commit(
411
committer=b'committer <committer@example.com>',
412
author=b'author <author@example.com>',
413
ref=b'refs/heads/master')
414
remote = ControlDir.open(self.remote_url)
415
wt = self.make_branch_and_tree('local', format=self._from_format)
416
self.build_tree(['local/blah'])
418
revid = wt.commit('blah')
420
if self._from_format == 'git':
421
result = remote.push_branch(wt.branch, overwrite=True)
423
result = remote.push_branch(wt.branch, lossy=True, overwrite=True)
425
self.assertEqual(None, result.old_revno)
426
if self._from_format == 'git':
427
self.assertEqual(1, result.new_revno)
429
self.assertIs(None, result.new_revno)
431
result.report(BytesIO())
435
b'HEAD': self.remote_real.refs[b'refs/heads/master'],
436
b'refs/heads/master': self.remote_real.refs[b'refs/heads/master'],
438
self.remote_real.get_refs())
440
def test_push_branch_new_with_tags(self):
441
remote = ControlDir.open(self.remote_url)
442
builder = self.make_branch_builder('local', format=self._from_format)
443
builder.start_series()
444
rev_1 = builder.build_snapshot(None, [
445
('add', ('', None, 'directory', '')),
446
('add', ('filename', None, 'file', b'content'))])
447
rev_2 = builder.build_snapshot(
448
[rev_1], [('modify', ('filename', b'new-content\n'))])
449
rev_3 = builder.build_snapshot(
450
[rev_1], [('modify', ('filename', b'new-new-content\n'))])
451
builder.finish_series()
452
branch = builder.get_branch()
454
branch.tags.set_tag('atag', rev_2)
455
except TagsNotSupported:
456
raise TestNotApplicable('source format does not support tags')
458
branch.get_config_stack().set('branch.fetch_tags', True)
459
if self._from_format == 'git':
460
result = remote.push_branch(branch, name='newbranch')
462
result = remote.push_branch(
463
branch, lossy=True, name='newbranch')
465
self.assertEqual(0, result.old_revno)
466
if self._from_format == 'git':
467
self.assertEqual(2, result.new_revno)
469
self.assertIs(None, result.new_revno)
471
result.report(BytesIO())
474
{b'refs/heads/newbranch', b'refs/tags/atag'},
475
set(self.remote_real.get_refs().keys()))
478
c1 = self.remote_real.do_commit(
480
committer=b'committer <committer@example.com>',
481
author=b'author <author@example.com>')
483
remote = ControlDir.open(self.remote_url)
484
self.make_controldir('local', format=self._from_format)
485
local = remote.sprout('local')
486
self.build_tree(['local/blah'])
487
wt = local.open_workingtree()
489
revid = wt.commit('blah')
490
wt.branch.tags.set_tag('sometag', revid)
491
wt.branch.get_config_stack().set('branch.fetch_tags', True)
493
if self._from_format == 'git':
494
result = wt.branch.push(remote.create_branch('newbranch'))
496
result = wt.branch.push(
497
remote.create_branch('newbranch'), lossy=True)
499
self.assertEqual(0, result.old_revno)
500
self.assertEqual(2, result.new_revno)
502
result.report(BytesIO())
505
{b'refs/heads/master': self.remote_real.head(),
506
b'HEAD': self.remote_real.head(),
507
b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
508
b'refs/tags/sometag': self.remote_real.refs[b'refs/heads/newbranch'],
510
self.remote_real.get_refs())
512
def test_push_diverged(self):
513
c1 = self.remote_real.do_commit(
515
committer=b'committer <committer@example.com>',
516
author=b'author <author@example.com>',
517
ref=b'refs/heads/newbranch')
519
remote = ControlDir.open(self.remote_url)
520
wt = self.make_branch_and_tree('local', format=self._from_format)
521
self.build_tree(['local/blah'])
523
revid = wt.commit('blah')
525
newbranch = remote.open_branch('newbranch')
526
if self._from_format == 'git':
527
self.assertRaises(DivergedBranches, wt.branch.push, newbranch)
529
self.assertRaises(DivergedBranches, wt.branch.push,
530
newbranch, lossy=True)
533
{b'refs/heads/newbranch': c1},
534
self.remote_real.get_refs())
536
if self._from_format == 'git':
537
wt.branch.push(newbranch, overwrite=True)
539
wt.branch.push(newbranch, lossy=True, overwrite=True)
541
self.assertNotEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
544
class PushToRemoteFromBzrTests(PushToRemoteBase, TestCaseWithTransport):
549
class PushToRemoteFromGitTests(PushToRemoteBase, TestCaseWithTransport):
554
class RemoteControlDirTests(TestCaseWithTransport):
556
_test_needs_features = [ExecutableFeature('git')]
559
TestCaseWithTransport.setUp(self)
560
self.remote_real = GitRepo.init('remote', mkdir=True)
561
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
562
self.permit_url(self.remote_url)
564
def test_remove_branch(self):
565
c1 = self.remote_real.do_commit(
567
committer=b'committer <committer@example.com>',
568
author=b'author <author@example.com>')
569
c2 = self.remote_real.do_commit(
570
message=b'another commit',
571
committer=b'committer <committer@example.com>',
572
author=b'author <author@example.com>',
573
ref=b'refs/heads/blah')
575
remote = ControlDir.open(self.remote_url)
576
remote.destroy_branch(name='blah')
578
self.remote_real.get_refs(),
579
{b'refs/heads/master': self.remote_real.head(),
580
b'HEAD': self.remote_real.head(),
583
def test_list_branches(self):
584
c1 = self.remote_real.do_commit(
586
committer=b'committer <committer@example.com>',
587
author=b'author <author@example.com>')
588
c2 = self.remote_real.do_commit(
589
message=b'another commit',
590
committer=b'committer <committer@example.com>',
591
author=b'author <author@example.com>',
592
ref=b'refs/heads/blah')
594
remote = ControlDir.open(self.remote_url)
596
set(['master', 'blah', 'master']),
597
set([b.name for b in remote.list_branches()]))
599
def test_get_branches(self):
600
c1 = self.remote_real.do_commit(
602
committer=b'committer <committer@example.com>',
603
author=b'author <author@example.com>')
604
c2 = self.remote_real.do_commit(
605
message=b'another commit',
606
committer=b'committer <committer@example.com>',
607
author=b'author <author@example.com>',
608
ref=b'refs/heads/blah')
610
remote = ControlDir.open(self.remote_url)
612
{'': 'master', 'blah': 'blah', 'master': 'master'},
613
{n: b.name for (n, b) in remote.get_branches().items()})
615
set(['', 'blah', 'master']), set(remote.branch_names()))
617
def test_remove_tag(self):
618
c1 = self.remote_real.do_commit(
620
committer=b'committer <committer@example.com>',
621
author=b'author <author@example.com>')
622
c2 = self.remote_real.do_commit(
623
message=b'another commit',
624
committer=b'committer <committer@example.com>',
625
author=b'author <author@example.com>',
626
ref=b'refs/tags/blah')
628
remote = ControlDir.open(self.remote_url)
629
remote_branch = remote.open_branch()
630
remote_branch.tags.delete_tag('blah')
631
self.assertRaises(NoSuchTag, remote_branch.tags.delete_tag, 'blah')
633
self.remote_real.get_refs(),
634
{b'refs/heads/master': self.remote_real.head(),
635
b'HEAD': self.remote_real.head(),
638
def test_set_tag(self):
639
c1 = self.remote_real.do_commit(
641
committer=b'committer <committer@example.com>',
642
author=b'author <author@example.com>')
643
c2 = self.remote_real.do_commit(
644
message=b'another commit',
645
committer=b'committer <committer@example.com>',
646
author=b'author <author@example.com>')
648
remote = ControlDir.open(self.remote_url)
649
remote.open_branch().tags.set_tag(
650
b'blah', default_mapping.revision_id_foreign_to_bzr(c1))
652
self.remote_real.get_refs(),
653
{b'refs/heads/master': self.remote_real.head(),
654
b'refs/tags/blah': c1,
655
b'HEAD': self.remote_real.head(),
658
def test_annotated_tag(self):
659
c1 = self.remote_real.do_commit(
661
committer=b'committer <committer@example.com>',
662
author=b'author <author@example.com>')
663
c2 = self.remote_real.do_commit(
664
message=b'another commit',
665
committer=b'committer <committer@example.com>',
666
author=b'author <author@example.com>')
668
porcelain.tag_create(
671
author=b'author <author@example.com>',
673
tag_time=int(time.time()),
676
message=b"Annotated tag")
678
remote = ControlDir.open(self.remote_url)
679
remote_branch = remote.open_branch()
681
'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
682
remote_branch.tags.get_tag_dict())
684
def test_get_branch_reference(self):
685
c1 = self.remote_real.do_commit(
687
committer=b'committer <committer@example.com>',
688
author=b'author <author@example.com>')
689
c2 = self.remote_real.do_commit(
690
message=b'another commit',
691
committer=b'committer <committer@example.com>',
692
author=b'author <author@example.com>')
694
remote = ControlDir.open(self.remote_url)
695
self.assertEqual(b'refs/heads/master', remote.get_branch_reference(''))
696
self.assertEqual(None, remote.get_branch_reference('master'))
698
def test_get_branch_nick(self):
699
c1 = self.remote_real.do_commit(
701
committer=b'committer <committer@example.com>',
702
author=b'author <author@example.com>')
703
remote = ControlDir.open(self.remote_url)
704
self.assertEqual('master', remote.open_branch().nick)
707
class GitUrlAndPathFromTransportTests(TestCase):
710
split_url = _git_url_and_path_from_transport('file:///home/blah')
711
self.assertEqual(split_url.scheme, 'file')
712
self.assertEqual(split_url.path, '/home/blah')
714
def test_file_segment_params(self):
715
split_url = _git_url_and_path_from_transport('file:///home/blah,branch=master')
716
self.assertEqual(split_url.scheme, 'file')
717
self.assertEqual(split_url.path, '/home/blah')
719
def test_git_smart(self):
720
split_url = _git_url_and_path_from_transport(
721
'git://github.com/dulwich/dulwich,branch=master')
722
self.assertEqual(split_url.scheme, 'git')
723
self.assertEqual(split_url.path, '/dulwich/dulwich')
725
def test_https(self):
726
split_url = _git_url_and_path_from_transport(
727
'https://github.com/dulwich/dulwich')
728
self.assertEqual(split_url.scheme, 'https')
729
self.assertEqual(split_url.path, '/dulwich/dulwich')
731
def test_https_segment_params(self):
732
split_url = _git_url_and_path_from_transport(
733
'https://github.com/dulwich/dulwich,branch=master')
734
self.assertEqual(split_url.scheme, 'https')
735
self.assertEqual(split_url.path, '/dulwich/dulwich')