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)
148
def test_invalid_git_error(self):
152
'GitLab: You are not allowed to push code to protected '
153
'branches on this project.'),
157
'GitLab: You are not allowed to push code to '
158
'protected branches on this project.')))
161
class ParseHangupTests(TestCase):
164
super(ParseHangupTests, self).setUp()
166
HangupException([b'foo'])
168
self.skipTest('dulwich version too old')
170
def test_not_set(self):
171
self.assertIsInstance(
172
parse_git_hangup('http://', HangupException()), HangupException)
174
def test_single_line(self):
176
RemoteGitError('foo bar'),
177
parse_git_hangup('http://', HangupException([b'foo bar'])))
179
def test_multi_lines(self):
181
RemoteGitError('foo bar\nbla bla'),
183
'http://', HangupException([b'foo bar', b'bla bla'])))
185
def test_filter_boring(self):
187
RemoteGitError('foo bar'), parse_git_hangup('http://', HangupException(
188
[b'=======', b'foo bar', b'======'])))
190
RemoteGitError('foo bar'), parse_git_hangup('http://', HangupException(
191
[b'remote: =======', b'remote: foo bar', b'remote: ======'])))
193
def test_permission_denied(self):
195
PermissionDenied('http://', 'You are not allowed to push code to this project.'),
200
b'You are not allowed to push code to this project.', b'', b'======'])))
203
class TestRemoteGitBranchFormat(TestCase):
206
super(TestRemoteGitBranchFormat, self).setUp()
207
self.format = RemoteGitBranchFormat()
209
def test_get_format_description(self):
210
self.assertEqual("Remote Git Branch",
211
self.format.get_format_description())
213
def test_get_network_name(self):
214
self.assertEqual(b"git", self.format.network_name())
216
def test_supports_tags(self):
217
self.assertTrue(self.format.supports_tags())
220
class TestRemoteGitBranch(TestCaseWithTransport):
222
_test_needs_features = [ExecutableFeature('git')]
225
TestCaseWithTransport.setUp(self)
226
self.remote_real = GitRepo.init('remote', mkdir=True)
227
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
228
self.permit_url(self.remote_url)
230
def test_set_last_revision_info(self):
231
c1 = self.remote_real.do_commit(
232
message=b'message 1',
233
committer=b'committer <committer@example.com>',
234
author=b'author <author@example.com>',
235
ref=b'refs/heads/newbranch')
236
c2 = self.remote_real.do_commit(
237
message=b'message 2',
238
committer=b'committer <committer@example.com>',
239
author=b'author <author@example.com>',
240
ref=b'refs/heads/newbranch')
242
remote = ControlDir.open(self.remote_url)
243
newbranch = remote.open_branch('newbranch')
244
self.assertEqual(newbranch.lookup_foreign_revision_id(c2),
245
newbranch.last_revision())
246
newbranch.set_last_revision_info(
247
1, newbranch.lookup_foreign_revision_id(c1))
248
self.assertEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
249
self.assertEqual(newbranch.last_revision(),
250
newbranch.lookup_foreign_revision_id(c1))
253
class FetchFromRemoteTestBase(object):
255
_test_needs_features = [ExecutableFeature('git')]
260
TestCaseWithTransport.setUp(self)
261
self.remote_real = GitRepo.init('remote', mkdir=True)
262
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
263
self.permit_url(self.remote_url)
265
def test_sprout_simple(self):
266
self.remote_real.do_commit(
268
committer=b'committer <committer@example.com>',
269
author=b'author <author@example.com>')
271
remote = ControlDir.open(self.remote_url)
272
self.make_controldir('local', format=self._to_format)
273
local = remote.sprout('local')
275
default_mapping.revision_id_foreign_to_bzr(
276
self.remote_real.head()),
277
local.open_branch().last_revision())
279
def test_sprout_with_tags(self):
280
c1 = self.remote_real.do_commit(
282
committer=b'committer <committer@example.com>',
283
author=b'author <author@example.com>')
284
c2 = self.remote_real.do_commit(
285
message=b'another commit',
286
committer=b'committer <committer@example.com>',
287
author=b'author <author@example.com>',
288
ref=b'refs/tags/another')
289
self.remote_real.refs[b'refs/tags/blah'] = self.remote_real.head()
291
remote = ControlDir.open(self.remote_url)
292
self.make_controldir('local', format=self._to_format)
293
local = remote.sprout('local')
294
local_branch = local.open_branch()
296
default_mapping.revision_id_foreign_to_bzr(c1),
297
local_branch.last_revision())
299
{'blah': local_branch.last_revision(),
300
'another': default_mapping.revision_id_foreign_to_bzr(c2)},
301
local_branch.tags.get_tag_dict())
303
def test_sprout_with_annotated_tag(self):
304
c1 = self.remote_real.do_commit(
306
committer=b'committer <committer@example.com>',
307
author=b'author <author@example.com>')
308
c2 = self.remote_real.do_commit(
309
message=b'another commit',
310
committer=b'committer <committer@example.com>',
311
author=b'author <author@example.com>',
312
ref=b'refs/heads/another')
313
porcelain.tag_create(
316
author=b'author <author@example.com>',
318
tag_time=int(time.time()),
321
message=b"Annotated tag")
323
remote = ControlDir.open(self.remote_url)
324
self.make_controldir('local', format=self._to_format)
325
local = remote.sprout(
326
'local', revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
327
local_branch = local.open_branch()
329
default_mapping.revision_id_foreign_to_bzr(c1),
330
local_branch.last_revision())
332
{'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
333
local_branch.tags.get_tag_dict())
335
def test_sprout_with_annotated_tag_unreferenced(self):
336
c1 = self.remote_real.do_commit(
338
committer=b'committer <committer@example.com>',
339
author=b'author <author@example.com>')
340
c2 = self.remote_real.do_commit(
341
message=b'another commit',
342
committer=b'committer <committer@example.com>',
343
author=b'author <author@example.com>')
344
porcelain.tag_create(
347
author=b'author <author@example.com>',
349
tag_time=int(time.time()),
352
message=b"Annotated tag")
354
remote = ControlDir.open(self.remote_url)
355
self.make_controldir('local', format=self._to_format)
356
local = remote.sprout(
358
revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
359
local_branch = local.open_branch()
361
default_mapping.revision_id_foreign_to_bzr(c1),
362
local_branch.last_revision())
364
{'blah': default_mapping.revision_id_foreign_to_bzr(c1)},
365
local_branch.tags.get_tag_dict())
368
class FetchFromRemoteToBzrTests(FetchFromRemoteTestBase, TestCaseWithTransport):
373
class FetchFromRemoteToGitTests(FetchFromRemoteTestBase, TestCaseWithTransport):
378
class PushToRemoteBase(object):
380
_test_needs_features = [ExecutableFeature('git')]
385
TestCaseWithTransport.setUp(self)
386
self.remote_real = GitRepo.init('remote', mkdir=True)
387
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
388
self.permit_url(self.remote_url)
390
def test_push_branch_new(self):
391
remote = ControlDir.open(self.remote_url)
392
wt = self.make_branch_and_tree('local', format=self._from_format)
393
self.build_tree(['local/blah'])
395
revid = wt.commit('blah')
397
if self._from_format == 'git':
398
result = remote.push_branch(wt.branch, name='newbranch')
400
result = remote.push_branch(
401
wt.branch, lossy=True, name='newbranch')
403
self.assertEqual(0, result.old_revno)
404
if self._from_format == 'git':
405
self.assertEqual(1, result.new_revno)
407
self.assertIs(None, result.new_revno)
409
result.report(BytesIO())
412
{b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
414
self.remote_real.get_refs())
416
def test_push_branch_symref(self):
417
cfg = self.remote_real.get_config()
418
cfg.set((b'core', ), b'bare', True)
420
self.remote_real.refs.set_symbolic_ref(b'HEAD', b'refs/heads/master')
421
c1 = self.remote_real.do_commit(
423
committer=b'committer <committer@example.com>',
424
author=b'author <author@example.com>',
425
ref=b'refs/heads/master')
426
remote = ControlDir.open(self.remote_url)
427
wt = self.make_branch_and_tree('local', format=self._from_format)
428
self.build_tree(['local/blah'])
430
revid = wt.commit('blah')
432
if self._from_format == 'git':
433
result = remote.push_branch(wt.branch, overwrite=True)
435
result = remote.push_branch(wt.branch, lossy=True, overwrite=True)
437
self.assertEqual(None, result.old_revno)
438
if self._from_format == 'git':
439
self.assertEqual(1, result.new_revno)
441
self.assertIs(None, result.new_revno)
443
result.report(BytesIO())
447
b'HEAD': self.remote_real.refs[b'refs/heads/master'],
448
b'refs/heads/master': self.remote_real.refs[b'refs/heads/master'],
450
self.remote_real.get_refs())
452
def test_push_branch_new_with_tags(self):
453
remote = ControlDir.open(self.remote_url)
454
builder = self.make_branch_builder('local', format=self._from_format)
455
builder.start_series()
456
rev_1 = builder.build_snapshot(None, [
457
('add', ('', None, 'directory', '')),
458
('add', ('filename', None, 'file', b'content'))])
459
rev_2 = builder.build_snapshot(
460
[rev_1], [('modify', ('filename', b'new-content\n'))])
461
rev_3 = builder.build_snapshot(
462
[rev_1], [('modify', ('filename', b'new-new-content\n'))])
463
builder.finish_series()
464
branch = builder.get_branch()
466
branch.tags.set_tag('atag', rev_2)
467
except TagsNotSupported:
468
raise TestNotApplicable('source format does not support tags')
470
branch.get_config_stack().set('branch.fetch_tags', True)
471
if self._from_format == 'git':
472
result = remote.push_branch(branch, name='newbranch')
474
result = remote.push_branch(
475
branch, lossy=True, name='newbranch')
477
self.assertEqual(0, result.old_revno)
478
if self._from_format == 'git':
479
self.assertEqual(2, result.new_revno)
481
self.assertIs(None, result.new_revno)
483
result.report(BytesIO())
486
{b'refs/heads/newbranch', b'refs/tags/atag'},
487
set(self.remote_real.get_refs().keys()))
490
c1 = self.remote_real.do_commit(
492
committer=b'committer <committer@example.com>',
493
author=b'author <author@example.com>')
495
remote = ControlDir.open(self.remote_url)
496
self.make_controldir('local', format=self._from_format)
497
local = remote.sprout('local')
498
self.build_tree(['local/blah'])
499
wt = local.open_workingtree()
501
revid = wt.commit('blah')
502
wt.branch.tags.set_tag('sometag', revid)
503
wt.branch.get_config_stack().set('branch.fetch_tags', True)
505
if self._from_format == 'git':
506
result = wt.branch.push(remote.create_branch('newbranch'))
508
result = wt.branch.push(
509
remote.create_branch('newbranch'), lossy=True)
511
self.assertEqual(0, result.old_revno)
512
self.assertEqual(2, result.new_revno)
514
result.report(BytesIO())
517
{b'refs/heads/master': self.remote_real.head(),
518
b'HEAD': self.remote_real.head(),
519
b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
520
b'refs/tags/sometag': self.remote_real.refs[b'refs/heads/newbranch'],
522
self.remote_real.get_refs())
524
def test_push_diverged(self):
525
c1 = self.remote_real.do_commit(
527
committer=b'committer <committer@example.com>',
528
author=b'author <author@example.com>',
529
ref=b'refs/heads/newbranch')
531
remote = ControlDir.open(self.remote_url)
532
wt = self.make_branch_and_tree('local', format=self._from_format)
533
self.build_tree(['local/blah'])
535
revid = wt.commit('blah')
537
newbranch = remote.open_branch('newbranch')
538
if self._from_format == 'git':
539
self.assertRaises(DivergedBranches, wt.branch.push, newbranch)
541
self.assertRaises(DivergedBranches, wt.branch.push,
542
newbranch, lossy=True)
545
{b'refs/heads/newbranch': c1},
546
self.remote_real.get_refs())
548
if self._from_format == 'git':
549
wt.branch.push(newbranch, overwrite=True)
551
wt.branch.push(newbranch, lossy=True, overwrite=True)
553
self.assertNotEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
556
class PushToRemoteFromBzrTests(PushToRemoteBase, TestCaseWithTransport):
561
class PushToRemoteFromGitTests(PushToRemoteBase, TestCaseWithTransport):
566
class RemoteControlDirTests(TestCaseWithTransport):
568
_test_needs_features = [ExecutableFeature('git')]
571
TestCaseWithTransport.setUp(self)
572
self.remote_real = GitRepo.init('remote', mkdir=True)
573
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
574
self.permit_url(self.remote_url)
576
def test_remove_branch(self):
577
c1 = self.remote_real.do_commit(
579
committer=b'committer <committer@example.com>',
580
author=b'author <author@example.com>')
581
c2 = self.remote_real.do_commit(
582
message=b'another commit',
583
committer=b'committer <committer@example.com>',
584
author=b'author <author@example.com>',
585
ref=b'refs/heads/blah')
587
remote = ControlDir.open(self.remote_url)
588
remote.destroy_branch(name='blah')
590
self.remote_real.get_refs(),
591
{b'refs/heads/master': self.remote_real.head(),
592
b'HEAD': self.remote_real.head(),
595
def test_list_branches(self):
596
c1 = self.remote_real.do_commit(
598
committer=b'committer <committer@example.com>',
599
author=b'author <author@example.com>')
600
c2 = self.remote_real.do_commit(
601
message=b'another commit',
602
committer=b'committer <committer@example.com>',
603
author=b'author <author@example.com>',
604
ref=b'refs/heads/blah')
606
remote = ControlDir.open(self.remote_url)
608
set(['master', 'blah', 'master']),
609
set([b.name for b in remote.list_branches()]))
611
def test_get_branches(self):
612
c1 = self.remote_real.do_commit(
614
committer=b'committer <committer@example.com>',
615
author=b'author <author@example.com>')
616
c2 = self.remote_real.do_commit(
617
message=b'another commit',
618
committer=b'committer <committer@example.com>',
619
author=b'author <author@example.com>',
620
ref=b'refs/heads/blah')
622
remote = ControlDir.open(self.remote_url)
624
{'': 'master', 'blah': 'blah', 'master': 'master'},
625
{n: b.name for (n, b) in remote.get_branches().items()})
627
set(['', 'blah', 'master']), set(remote.branch_names()))
629
def test_remove_tag(self):
630
c1 = self.remote_real.do_commit(
632
committer=b'committer <committer@example.com>',
633
author=b'author <author@example.com>')
634
c2 = self.remote_real.do_commit(
635
message=b'another commit',
636
committer=b'committer <committer@example.com>',
637
author=b'author <author@example.com>',
638
ref=b'refs/tags/blah')
640
remote = ControlDir.open(self.remote_url)
641
remote_branch = remote.open_branch()
642
remote_branch.tags.delete_tag('blah')
643
self.assertRaises(NoSuchTag, remote_branch.tags.delete_tag, 'blah')
645
self.remote_real.get_refs(),
646
{b'refs/heads/master': self.remote_real.head(),
647
b'HEAD': self.remote_real.head(),
650
def test_set_tag(self):
651
c1 = self.remote_real.do_commit(
653
committer=b'committer <committer@example.com>',
654
author=b'author <author@example.com>')
655
c2 = self.remote_real.do_commit(
656
message=b'another commit',
657
committer=b'committer <committer@example.com>',
658
author=b'author <author@example.com>')
660
remote = ControlDir.open(self.remote_url)
661
remote.open_branch().tags.set_tag(
662
b'blah', default_mapping.revision_id_foreign_to_bzr(c1))
664
self.remote_real.get_refs(),
665
{b'refs/heads/master': self.remote_real.head(),
666
b'refs/tags/blah': c1,
667
b'HEAD': self.remote_real.head(),
670
def test_annotated_tag(self):
671
c1 = self.remote_real.do_commit(
673
committer=b'committer <committer@example.com>',
674
author=b'author <author@example.com>')
675
c2 = self.remote_real.do_commit(
676
message=b'another commit',
677
committer=b'committer <committer@example.com>',
678
author=b'author <author@example.com>')
680
porcelain.tag_create(
683
author=b'author <author@example.com>',
685
tag_time=int(time.time()),
688
message=b"Annotated tag")
690
remote = ControlDir.open(self.remote_url)
691
remote_branch = remote.open_branch()
693
'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
694
remote_branch.tags.get_tag_dict())
696
def test_get_branch_reference(self):
697
c1 = self.remote_real.do_commit(
699
committer=b'committer <committer@example.com>',
700
author=b'author <author@example.com>')
701
c2 = self.remote_real.do_commit(
702
message=b'another commit',
703
committer=b'committer <committer@example.com>',
704
author=b'author <author@example.com>')
706
remote = ControlDir.open(self.remote_url)
707
self.assertEqual(b'refs/heads/master', remote.get_branch_reference(''))
708
self.assertEqual(None, remote.get_branch_reference('master'))
710
def test_get_branch_nick(self):
711
c1 = self.remote_real.do_commit(
713
committer=b'committer <committer@example.com>',
714
author=b'author <author@example.com>')
715
remote = ControlDir.open(self.remote_url)
716
self.assertEqual('master', remote.open_branch().nick)
719
class GitUrlAndPathFromTransportTests(TestCase):
722
split_url = _git_url_and_path_from_transport('file:///home/blah')
723
self.assertEqual(split_url.scheme, 'file')
724
self.assertEqual(split_url.path, '/home/blah')
726
def test_file_segment_params(self):
727
split_url = _git_url_and_path_from_transport('file:///home/blah,branch=master')
728
self.assertEqual(split_url.scheme, 'file')
729
self.assertEqual(split_url.path, '/home/blah')
731
def test_git_smart(self):
732
split_url = _git_url_and_path_from_transport(
733
'git://github.com/dulwich/dulwich,branch=master')
734
self.assertEqual(split_url.scheme, 'git')
735
self.assertEqual(split_url.path, '/dulwich/dulwich')
737
def test_https(self):
738
split_url = _git_url_and_path_from_transport(
739
'https://github.com/dulwich/dulwich')
740
self.assertEqual(split_url.scheme, 'https')
741
self.assertEqual(split_url.path, '/dulwich/dulwich')
743
def test_https_segment_params(self):
744
split_url = _git_url_and_path_from_transport(
745
'https://github.com/dulwich/dulwich,branch=master')
746
self.assertEqual(split_url.scheme, 'https')
747
self.assertEqual(split_url.path, '/dulwich/dulwich')