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_pre_receive_hook_declined(self):
143
'pre-receive hook declined')
144
self.assertIsInstance(e, PermissionDenied)
145
self.assertEqual(e.path, "url")
146
self.assertEqual(e.extra, ': pre-receive hook declined')
148
def test_invalid_repo_name(self):
151
"""Gregwar/fatcat/tree/debian is not a valid repository name
152
Email support@github.com for help
154
self.assertIsInstance(e, NotBranchError)
156
def test_invalid_git_error(self):
160
'GitLab: You are not allowed to push code to protected '
161
'branches on this project.'),
165
'GitLab: You are not allowed to push code to '
166
'protected branches on this project.')))
169
class ParseHangupTests(TestCase):
172
super(ParseHangupTests, self).setUp()
174
HangupException([b'foo'])
176
self.skipTest('dulwich version too old')
178
def test_not_set(self):
179
self.assertIsInstance(
180
parse_git_hangup('http://', HangupException()), HangupException)
182
def test_single_line(self):
184
RemoteGitError('foo bar'),
185
parse_git_hangup('http://', HangupException([b'foo bar'])))
187
def test_multi_lines(self):
189
RemoteGitError('foo bar\nbla bla'),
191
'http://', HangupException([b'foo bar', b'bla bla'])))
193
def test_filter_boring(self):
195
RemoteGitError('foo bar'), parse_git_hangup('http://', HangupException(
196
[b'=======', b'foo bar', b'======'])))
198
RemoteGitError('foo bar'), parse_git_hangup('http://', HangupException(
199
[b'remote: =======', b'remote: foo bar', b'remote: ======'])))
201
def test_permission_denied(self):
203
PermissionDenied('http://', 'You are not allowed to push code to this project.'),
208
b'You are not allowed to push code to this project.', b'', b'======'])))
211
class TestRemoteGitBranchFormat(TestCase):
214
super(TestRemoteGitBranchFormat, self).setUp()
215
self.format = RemoteGitBranchFormat()
217
def test_get_format_description(self):
218
self.assertEqual("Remote Git Branch",
219
self.format.get_format_description())
221
def test_get_network_name(self):
222
self.assertEqual(b"git", self.format.network_name())
224
def test_supports_tags(self):
225
self.assertTrue(self.format.supports_tags())
228
class TestRemoteGitBranch(TestCaseWithTransport):
230
_test_needs_features = [ExecutableFeature('git')]
233
TestCaseWithTransport.setUp(self)
234
self.remote_real = GitRepo.init('remote', mkdir=True)
235
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
236
self.permit_url(self.remote_url)
238
def test_set_last_revision_info(self):
239
c1 = self.remote_real.do_commit(
240
message=b'message 1',
241
committer=b'committer <committer@example.com>',
242
author=b'author <author@example.com>',
243
ref=b'refs/heads/newbranch')
244
c2 = self.remote_real.do_commit(
245
message=b'message 2',
246
committer=b'committer <committer@example.com>',
247
author=b'author <author@example.com>',
248
ref=b'refs/heads/newbranch')
250
remote = ControlDir.open(self.remote_url)
251
newbranch = remote.open_branch('newbranch')
252
self.assertEqual(newbranch.lookup_foreign_revision_id(c2),
253
newbranch.last_revision())
254
newbranch.set_last_revision_info(
255
1, newbranch.lookup_foreign_revision_id(c1))
256
self.assertEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
257
self.assertEqual(newbranch.last_revision(),
258
newbranch.lookup_foreign_revision_id(c1))
261
class FetchFromRemoteTestBase(object):
263
_test_needs_features = [ExecutableFeature('git')]
268
TestCaseWithTransport.setUp(self)
269
self.remote_real = GitRepo.init('remote', mkdir=True)
270
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
271
self.permit_url(self.remote_url)
273
def test_sprout_simple(self):
274
self.remote_real.do_commit(
276
committer=b'committer <committer@example.com>',
277
author=b'author <author@example.com>')
279
remote = ControlDir.open(self.remote_url)
280
self.make_controldir('local', format=self._to_format)
281
local = remote.sprout('local')
283
default_mapping.revision_id_foreign_to_bzr(
284
self.remote_real.head()),
285
local.open_branch().last_revision())
287
def test_sprout_with_tags(self):
288
c1 = self.remote_real.do_commit(
290
committer=b'committer <committer@example.com>',
291
author=b'author <author@example.com>')
292
c2 = self.remote_real.do_commit(
293
message=b'another commit',
294
committer=b'committer <committer@example.com>',
295
author=b'author <author@example.com>',
296
ref=b'refs/tags/another')
297
self.remote_real.refs[b'refs/tags/blah'] = self.remote_real.head()
299
remote = ControlDir.open(self.remote_url)
300
self.make_controldir('local', format=self._to_format)
301
local = remote.sprout('local')
302
local_branch = local.open_branch()
304
default_mapping.revision_id_foreign_to_bzr(c1),
305
local_branch.last_revision())
307
{'blah': local_branch.last_revision(),
308
'another': default_mapping.revision_id_foreign_to_bzr(c2)},
309
local_branch.tags.get_tag_dict())
311
def test_sprout_with_annotated_tag(self):
312
c1 = self.remote_real.do_commit(
314
committer=b'committer <committer@example.com>',
315
author=b'author <author@example.com>')
316
c2 = self.remote_real.do_commit(
317
message=b'another commit',
318
committer=b'committer <committer@example.com>',
319
author=b'author <author@example.com>',
320
ref=b'refs/heads/another')
321
porcelain.tag_create(
324
author=b'author <author@example.com>',
326
tag_time=int(time.time()),
329
message=b"Annotated tag")
331
remote = ControlDir.open(self.remote_url)
332
self.make_controldir('local', format=self._to_format)
333
local = remote.sprout(
334
'local', revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
335
local_branch = local.open_branch()
337
default_mapping.revision_id_foreign_to_bzr(c1),
338
local_branch.last_revision())
340
{'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
341
local_branch.tags.get_tag_dict())
343
def test_sprout_with_annotated_tag_unreferenced(self):
344
c1 = self.remote_real.do_commit(
346
committer=b'committer <committer@example.com>',
347
author=b'author <author@example.com>')
348
c2 = self.remote_real.do_commit(
349
message=b'another commit',
350
committer=b'committer <committer@example.com>',
351
author=b'author <author@example.com>')
352
porcelain.tag_create(
355
author=b'author <author@example.com>',
357
tag_time=int(time.time()),
360
message=b"Annotated tag")
362
remote = ControlDir.open(self.remote_url)
363
self.make_controldir('local', format=self._to_format)
364
local = remote.sprout(
366
revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
367
local_branch = local.open_branch()
369
default_mapping.revision_id_foreign_to_bzr(c1),
370
local_branch.last_revision())
372
{'blah': default_mapping.revision_id_foreign_to_bzr(c1)},
373
local_branch.tags.get_tag_dict())
376
class FetchFromRemoteToBzrTests(FetchFromRemoteTestBase, TestCaseWithTransport):
381
class FetchFromRemoteToGitTests(FetchFromRemoteTestBase, TestCaseWithTransport):
386
class PushToRemoteBase(object):
388
_test_needs_features = [ExecutableFeature('git')]
393
TestCaseWithTransport.setUp(self)
394
self.remote_real = GitRepo.init('remote', mkdir=True)
395
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
396
self.permit_url(self.remote_url)
398
def test_push_branch_new(self):
399
remote = ControlDir.open(self.remote_url)
400
wt = self.make_branch_and_tree('local', format=self._from_format)
401
self.build_tree(['local/blah'])
403
revid = wt.commit('blah')
405
if self._from_format == 'git':
406
result = remote.push_branch(wt.branch, name='newbranch')
408
result = remote.push_branch(
409
wt.branch, lossy=True, name='newbranch')
411
self.assertEqual(0, result.old_revno)
412
if self._from_format == 'git':
413
self.assertEqual(1, result.new_revno)
415
self.assertIs(None, result.new_revno)
417
result.report(BytesIO())
420
{b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
422
self.remote_real.get_refs())
424
def test_push_branch_symref(self):
425
cfg = self.remote_real.get_config()
426
cfg.set((b'core', ), b'bare', True)
428
self.remote_real.refs.set_symbolic_ref(b'HEAD', b'refs/heads/master')
429
c1 = self.remote_real.do_commit(
431
committer=b'committer <committer@example.com>',
432
author=b'author <author@example.com>',
433
ref=b'refs/heads/master')
434
remote = ControlDir.open(self.remote_url)
435
wt = self.make_branch_and_tree('local', format=self._from_format)
436
self.build_tree(['local/blah'])
438
revid = wt.commit('blah')
440
if self._from_format == 'git':
441
result = remote.push_branch(wt.branch, overwrite=True)
443
result = remote.push_branch(wt.branch, lossy=True, overwrite=True)
445
self.assertEqual(None, result.old_revno)
446
if self._from_format == 'git':
447
self.assertEqual(1, result.new_revno)
449
self.assertIs(None, result.new_revno)
451
result.report(BytesIO())
455
b'HEAD': self.remote_real.refs[b'refs/heads/master'],
456
b'refs/heads/master': self.remote_real.refs[b'refs/heads/master'],
458
self.remote_real.get_refs())
460
def test_push_branch_new_with_tags(self):
461
remote = ControlDir.open(self.remote_url)
462
builder = self.make_branch_builder('local', format=self._from_format)
463
builder.start_series()
464
rev_1 = builder.build_snapshot(None, [
465
('add', ('', None, 'directory', '')),
466
('add', ('filename', None, 'file', b'content'))])
467
rev_2 = builder.build_snapshot(
468
[rev_1], [('modify', ('filename', b'new-content\n'))])
469
rev_3 = builder.build_snapshot(
470
[rev_1], [('modify', ('filename', b'new-new-content\n'))])
471
builder.finish_series()
472
branch = builder.get_branch()
474
branch.tags.set_tag('atag', rev_2)
475
except TagsNotSupported:
476
raise TestNotApplicable('source format does not support tags')
478
branch.get_config_stack().set('branch.fetch_tags', True)
479
if self._from_format == 'git':
480
result = remote.push_branch(branch, name='newbranch')
482
result = remote.push_branch(
483
branch, lossy=True, name='newbranch')
485
self.assertEqual(0, result.old_revno)
486
if self._from_format == 'git':
487
self.assertEqual(2, result.new_revno)
489
self.assertIs(None, result.new_revno)
491
result.report(BytesIO())
494
{b'refs/heads/newbranch', b'refs/tags/atag'},
495
set(self.remote_real.get_refs().keys()))
498
c1 = self.remote_real.do_commit(
500
committer=b'committer <committer@example.com>',
501
author=b'author <author@example.com>')
503
remote = ControlDir.open(self.remote_url)
504
self.make_controldir('local', format=self._from_format)
505
local = remote.sprout('local')
506
self.build_tree(['local/blah'])
507
wt = local.open_workingtree()
509
revid = wt.commit('blah')
510
wt.branch.tags.set_tag('sometag', revid)
511
wt.branch.get_config_stack().set('branch.fetch_tags', True)
513
if self._from_format == 'git':
514
result = wt.branch.push(remote.create_branch('newbranch'))
516
result = wt.branch.push(
517
remote.create_branch('newbranch'), lossy=True)
519
self.assertEqual(0, result.old_revno)
520
self.assertEqual(2, result.new_revno)
522
result.report(BytesIO())
525
{b'refs/heads/master': self.remote_real.head(),
526
b'HEAD': self.remote_real.head(),
527
b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
528
b'refs/tags/sometag': self.remote_real.refs[b'refs/heads/newbranch'],
530
self.remote_real.get_refs())
532
def test_push_diverged(self):
533
c1 = self.remote_real.do_commit(
535
committer=b'committer <committer@example.com>',
536
author=b'author <author@example.com>',
537
ref=b'refs/heads/newbranch')
539
remote = ControlDir.open(self.remote_url)
540
wt = self.make_branch_and_tree('local', format=self._from_format)
541
self.build_tree(['local/blah'])
543
revid = wt.commit('blah')
545
newbranch = remote.open_branch('newbranch')
546
if self._from_format == 'git':
547
self.assertRaises(DivergedBranches, wt.branch.push, newbranch)
549
self.assertRaises(DivergedBranches, wt.branch.push,
550
newbranch, lossy=True)
553
{b'refs/heads/newbranch': c1},
554
self.remote_real.get_refs())
556
if self._from_format == 'git':
557
wt.branch.push(newbranch, overwrite=True)
559
wt.branch.push(newbranch, lossy=True, overwrite=True)
561
self.assertNotEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
564
class PushToRemoteFromBzrTests(PushToRemoteBase, TestCaseWithTransport):
569
class PushToRemoteFromGitTests(PushToRemoteBase, TestCaseWithTransport):
574
class RemoteControlDirTests(TestCaseWithTransport):
576
_test_needs_features = [ExecutableFeature('git')]
579
TestCaseWithTransport.setUp(self)
580
self.remote_real = GitRepo.init('remote', mkdir=True)
581
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
582
self.permit_url(self.remote_url)
584
def test_remove_branch(self):
585
c1 = self.remote_real.do_commit(
587
committer=b'committer <committer@example.com>',
588
author=b'author <author@example.com>')
589
c2 = self.remote_real.do_commit(
590
message=b'another commit',
591
committer=b'committer <committer@example.com>',
592
author=b'author <author@example.com>',
593
ref=b'refs/heads/blah')
595
remote = ControlDir.open(self.remote_url)
596
remote.destroy_branch(name='blah')
598
self.remote_real.get_refs(),
599
{b'refs/heads/master': self.remote_real.head(),
600
b'HEAD': self.remote_real.head(),
603
def test_list_branches(self):
604
c1 = self.remote_real.do_commit(
606
committer=b'committer <committer@example.com>',
607
author=b'author <author@example.com>')
608
c2 = self.remote_real.do_commit(
609
message=b'another commit',
610
committer=b'committer <committer@example.com>',
611
author=b'author <author@example.com>',
612
ref=b'refs/heads/blah')
614
remote = ControlDir.open(self.remote_url)
616
set(['master', 'blah', 'master']),
617
set([b.name for b in remote.list_branches()]))
619
def test_get_branches(self):
620
c1 = self.remote_real.do_commit(
622
committer=b'committer <committer@example.com>',
623
author=b'author <author@example.com>')
624
c2 = self.remote_real.do_commit(
625
message=b'another commit',
626
committer=b'committer <committer@example.com>',
627
author=b'author <author@example.com>',
628
ref=b'refs/heads/blah')
630
remote = ControlDir.open(self.remote_url)
632
{'': 'master', 'blah': 'blah', 'master': 'master'},
633
{n: b.name for (n, b) in remote.get_branches().items()})
635
set(['', 'blah', 'master']), set(remote.branch_names()))
637
def test_remove_tag(self):
638
c1 = self.remote_real.do_commit(
640
committer=b'committer <committer@example.com>',
641
author=b'author <author@example.com>')
642
c2 = self.remote_real.do_commit(
643
message=b'another commit',
644
committer=b'committer <committer@example.com>',
645
author=b'author <author@example.com>',
646
ref=b'refs/tags/blah')
648
remote = ControlDir.open(self.remote_url)
649
remote_branch = remote.open_branch()
650
remote_branch.tags.delete_tag('blah')
651
self.assertRaises(NoSuchTag, remote_branch.tags.delete_tag, 'blah')
653
self.remote_real.get_refs(),
654
{b'refs/heads/master': self.remote_real.head(),
655
b'HEAD': self.remote_real.head(),
658
def test_set_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
remote = ControlDir.open(self.remote_url)
669
remote.open_branch().tags.set_tag(
670
b'blah', default_mapping.revision_id_foreign_to_bzr(c1))
672
self.remote_real.get_refs(),
673
{b'refs/heads/master': self.remote_real.head(),
674
b'refs/tags/blah': c1,
675
b'HEAD': self.remote_real.head(),
678
def test_annotated_tag(self):
679
c1 = self.remote_real.do_commit(
681
committer=b'committer <committer@example.com>',
682
author=b'author <author@example.com>')
683
c2 = self.remote_real.do_commit(
684
message=b'another commit',
685
committer=b'committer <committer@example.com>',
686
author=b'author <author@example.com>')
688
porcelain.tag_create(
691
author=b'author <author@example.com>',
693
tag_time=int(time.time()),
696
message=b"Annotated tag")
698
remote = ControlDir.open(self.remote_url)
699
remote_branch = remote.open_branch()
701
'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
702
remote_branch.tags.get_tag_dict())
704
def test_get_branch_reference(self):
705
c1 = self.remote_real.do_commit(
707
committer=b'committer <committer@example.com>',
708
author=b'author <author@example.com>')
709
c2 = self.remote_real.do_commit(
710
message=b'another commit',
711
committer=b'committer <committer@example.com>',
712
author=b'author <author@example.com>')
714
remote = ControlDir.open(self.remote_url)
715
self.assertEqual(b'refs/heads/master', remote.get_branch_reference(''))
716
self.assertEqual(None, remote.get_branch_reference('master'))
718
def test_get_branch_nick(self):
719
c1 = self.remote_real.do_commit(
721
committer=b'committer <committer@example.com>',
722
author=b'author <author@example.com>')
723
remote = ControlDir.open(self.remote_url)
724
self.assertEqual('master', remote.open_branch().nick)
727
class GitUrlAndPathFromTransportTests(TestCase):
730
split_url = _git_url_and_path_from_transport('file:///home/blah')
731
self.assertEqual(split_url.scheme, 'file')
732
self.assertEqual(split_url.path, '/home/blah')
734
def test_file_segment_params(self):
735
split_url = _git_url_and_path_from_transport('file:///home/blah,branch=master')
736
self.assertEqual(split_url.scheme, 'file')
737
self.assertEqual(split_url.path, '/home/blah')
739
def test_git_smart(self):
740
split_url = _git_url_and_path_from_transport(
741
'git://github.com/dulwich/dulwich,branch=master')
742
self.assertEqual(split_url.scheme, 'git')
743
self.assertEqual(split_url.path, '/dulwich/dulwich')
745
def test_https(self):
746
split_url = _git_url_and_path_from_transport(
747
'https://github.com/dulwich/dulwich')
748
self.assertEqual(split_url.scheme, 'https')
749
self.assertEqual(split_url.path, '/dulwich/dulwich')
751
def test_https_segment_params(self):
752
split_url = _git_url_and_path_from_transport(
753
'https://github.com/dulwich/dulwich,branch=master')
754
self.assertEqual(split_url.scheme, 'https')
755
self.assertEqual(split_url.path, '/dulwich/dulwich')