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 io import BytesIO
24
from ...controldir import ControlDir
25
from ...errors import (
32
from ...tests import (
34
TestCaseWithTransport,
36
from ...tests.features import ExecutableFeature
38
from ..mapping import default_mapping
39
from ..remote import (
45
RemoteGitBranchFormat,
46
_git_url_and_path_from_transport,
49
from dulwich import porcelain
50
from dulwich.errors import HangupException
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_username_password(self):
70
("foo", None, "la", "/bar"),
71
split_git_url("git://la:passwd@foo/bar"))
73
def test_nopath(self):
74
self.assertEqual(("foo", None, None, "/"),
75
split_git_url("git://foo/"))
77
def test_slashpath(self):
78
self.assertEqual(("foo", None, None, "//bar"),
79
split_git_url("git://foo//bar"))
81
def test_homedir(self):
82
self.assertEqual(("foo", None, None, "~bar"),
83
split_git_url("git://foo/~bar"))
87
("", None, None, "/bar"),
88
split_git_url("file:///bar"))
91
class ParseGitErrorTests(TestCase):
93
def test_unknown(self):
94
e = parse_git_error("url", "foo")
95
self.assertIsInstance(e, RemoteGitError)
97
def test_notbrancherror(self):
98
e = parse_git_error("url", "\n Could not find Repository foo/bar")
99
self.assertIsInstance(e, NotBranchError)
101
def test_notbrancherror_launchpad(self):
102
e = parse_git_error("url", "Repository 'foo/bar' not found.")
103
self.assertIsInstance(e, NotBranchError)
105
def test_notbrancherror_github(self):
106
e = parse_git_error("url", "Repository not found.\n")
107
self.assertIsInstance(e, NotBranchError)
109
def test_notbrancherror_normal(self):
111
"url", "fatal: '/srv/git/lintian-brush' does not appear to be a git repository")
112
self.assertIsInstance(e, NotBranchError)
114
def test_head_update(self):
115
e = parse_git_error("url", "HEAD failed to update\n")
116
self.assertIsInstance(e, HeadUpdateFailed)
118
def test_permission_dnied(self):
121
"access denied or repository not exported: /debian/altermime.git")
122
self.assertIsInstance(e, PermissionDenied)
124
def test_permission_denied_gitlab(self):
127
'GitLab: You are not allowed to push code to this project.\n')
128
self.assertIsInstance(e, PermissionDenied)
130
def test_permission_denied_github(self):
133
'Permission to porridge/gaduhistory.git denied to jelmer.')
134
self.assertIsInstance(e, PermissionDenied)
135
self.assertEqual(e.path, 'porridge/gaduhistory.git')
136
self.assertEqual(e.extra, ': denied to jelmer')
138
def test_invalid_repo_name(self):
141
"""Gregwar/fatcat/tree/debian is not a valid repository name
142
Email support@github.com for help
144
self.assertIsInstance(e, NotBranchError)
146
def test_invalid_git_error(self):
150
'GitLab: You are not allowed to push code to protected '
151
'branches on this project.'),
155
'GitLab: You are not allowed to push code to '
156
'protected branches on this project.')))
159
class ParseHangupTests(TestCase):
162
super(ParseHangupTests, self).setUp()
164
HangupException([b'foo'])
166
self.skipTest('dulwich version too old')
168
def test_not_set(self):
169
self.assertIsInstance(
170
parse_git_hangup('http://', HangupException()), HangupException)
172
def test_single_line(self):
174
RemoteGitError('foo bar'),
175
parse_git_hangup('http://', HangupException([b'foo bar'])))
177
def test_multi_lines(self):
179
RemoteGitError('foo bar\nbla bla'),
181
'http://', HangupException([b'foo bar', b'bla bla'])))
183
def test_filter_boring(self):
185
RemoteGitError('foo bar'), parse_git_hangup('http://', HangupException(
186
[b'=======', b'foo bar', b'======'])))
188
RemoteGitError('foo bar'), parse_git_hangup('http://', HangupException(
189
[b'remote: =======', b'remote: foo bar', b'remote: ======'])))
191
def test_permission_denied(self):
193
PermissionDenied('http://', 'You are not allowed to push code to this project.'),
198
b'You are not allowed to push code to this project.', b'', b'======'])))
201
class TestRemoteGitBranchFormat(TestCase):
204
super(TestRemoteGitBranchFormat, self).setUp()
205
self.format = RemoteGitBranchFormat()
207
def test_get_format_description(self):
208
self.assertEqual("Remote Git Branch",
209
self.format.get_format_description())
211
def test_get_network_name(self):
212
self.assertEqual(b"git", self.format.network_name())
214
def test_supports_tags(self):
215
self.assertTrue(self.format.supports_tags())
218
class TestRemoteGitBranch(TestCaseWithTransport):
220
_test_needs_features = [ExecutableFeature('git')]
223
TestCaseWithTransport.setUp(self)
224
self.remote_real = GitRepo.init('remote', mkdir=True)
225
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
226
self.permit_url(self.remote_url)
228
def test_set_last_revision_info(self):
229
c1 = self.remote_real.do_commit(
230
message=b'message 1',
231
committer=b'committer <committer@example.com>',
232
author=b'author <author@example.com>',
233
ref=b'refs/heads/newbranch')
234
c2 = self.remote_real.do_commit(
235
message=b'message 2',
236
committer=b'committer <committer@example.com>',
237
author=b'author <author@example.com>',
238
ref=b'refs/heads/newbranch')
240
remote = ControlDir.open(self.remote_url)
241
newbranch = remote.open_branch('newbranch')
242
self.assertEqual(newbranch.lookup_foreign_revision_id(c2),
243
newbranch.last_revision())
244
newbranch.set_last_revision_info(
245
1, newbranch.lookup_foreign_revision_id(c1))
246
self.assertEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
247
self.assertEqual(newbranch.last_revision(),
248
newbranch.lookup_foreign_revision_id(c1))
251
class FetchFromRemoteTestBase(object):
253
_test_needs_features = [ExecutableFeature('git')]
258
TestCaseWithTransport.setUp(self)
259
self.remote_real = GitRepo.init('remote', mkdir=True)
260
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
261
self.permit_url(self.remote_url)
263
def test_sprout_simple(self):
264
self.remote_real.do_commit(
266
committer=b'committer <committer@example.com>',
267
author=b'author <author@example.com>')
269
remote = ControlDir.open(self.remote_url)
270
self.make_controldir('local', format=self._to_format)
271
local = remote.sprout('local')
273
default_mapping.revision_id_foreign_to_bzr(
274
self.remote_real.head()),
275
local.open_branch().last_revision())
277
def test_sprout_with_tags(self):
278
c1 = self.remote_real.do_commit(
280
committer=b'committer <committer@example.com>',
281
author=b'author <author@example.com>')
282
c2 = self.remote_real.do_commit(
283
message=b'another commit',
284
committer=b'committer <committer@example.com>',
285
author=b'author <author@example.com>',
286
ref=b'refs/tags/another')
287
self.remote_real.refs[b'refs/tags/blah'] = self.remote_real.head()
289
remote = ControlDir.open(self.remote_url)
290
self.make_controldir('local', format=self._to_format)
291
local = remote.sprout('local')
292
local_branch = local.open_branch()
294
default_mapping.revision_id_foreign_to_bzr(c1),
295
local_branch.last_revision())
297
{'blah': local_branch.last_revision(),
298
'another': default_mapping.revision_id_foreign_to_bzr(c2)},
299
local_branch.tags.get_tag_dict())
301
def test_sprout_with_annotated_tag(self):
302
c1 = self.remote_real.do_commit(
304
committer=b'committer <committer@example.com>',
305
author=b'author <author@example.com>')
306
c2 = self.remote_real.do_commit(
307
message=b'another commit',
308
committer=b'committer <committer@example.com>',
309
author=b'author <author@example.com>',
310
ref=b'refs/heads/another')
311
porcelain.tag_create(
314
author=b'author <author@example.com>',
316
tag_time=int(time.time()),
319
message=b"Annotated tag")
321
remote = ControlDir.open(self.remote_url)
322
self.make_controldir('local', format=self._to_format)
323
local = remote.sprout(
324
'local', revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
325
local_branch = local.open_branch()
327
default_mapping.revision_id_foreign_to_bzr(c1),
328
local_branch.last_revision())
330
{'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
331
local_branch.tags.get_tag_dict())
333
def test_sprout_with_annotated_tag_unreferenced(self):
334
c1 = self.remote_real.do_commit(
336
committer=b'committer <committer@example.com>',
337
author=b'author <author@example.com>')
338
c2 = self.remote_real.do_commit(
339
message=b'another commit',
340
committer=b'committer <committer@example.com>',
341
author=b'author <author@example.com>')
342
porcelain.tag_create(
345
author=b'author <author@example.com>',
347
tag_time=int(time.time()),
350
message=b"Annotated tag")
352
remote = ControlDir.open(self.remote_url)
353
self.make_controldir('local', format=self._to_format)
354
local = remote.sprout(
356
revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
357
local_branch = local.open_branch()
359
default_mapping.revision_id_foreign_to_bzr(c1),
360
local_branch.last_revision())
362
{'blah': default_mapping.revision_id_foreign_to_bzr(c1)},
363
local_branch.tags.get_tag_dict())
366
class FetchFromRemoteToBzrTests(FetchFromRemoteTestBase, TestCaseWithTransport):
371
class FetchFromRemoteToGitTests(FetchFromRemoteTestBase, TestCaseWithTransport):
376
class PushToRemoteBase(object):
378
_test_needs_features = [ExecutableFeature('git')]
383
TestCaseWithTransport.setUp(self)
384
self.remote_real = GitRepo.init('remote', mkdir=True)
385
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
386
self.permit_url(self.remote_url)
388
def test_push_branch_new(self):
389
remote = ControlDir.open(self.remote_url)
390
wt = self.make_branch_and_tree('local', format=self._from_format)
391
self.build_tree(['local/blah'])
393
revid = wt.commit('blah')
395
if self._from_format == 'git':
396
result = remote.push_branch(wt.branch, name='newbranch')
398
result = remote.push_branch(
399
wt.branch, lossy=True, name='newbranch')
401
self.assertEqual(0, result.old_revno)
402
if self._from_format == 'git':
403
self.assertEqual(1, result.new_revno)
405
self.assertIs(None, result.new_revno)
407
result.report(BytesIO())
410
{b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
412
self.remote_real.get_refs())
414
def test_push_branch_symref(self):
415
cfg = self.remote_real.get_config()
416
cfg.set((b'core', ), b'bare', True)
418
self.remote_real.refs.set_symbolic_ref(b'HEAD', b'refs/heads/master')
419
c1 = self.remote_real.do_commit(
421
committer=b'committer <committer@example.com>',
422
author=b'author <author@example.com>',
423
ref=b'refs/heads/master')
424
remote = ControlDir.open(self.remote_url)
425
wt = self.make_branch_and_tree('local', format=self._from_format)
426
self.build_tree(['local/blah'])
428
revid = wt.commit('blah')
430
if self._from_format == 'git':
431
result = remote.push_branch(wt.branch, overwrite=True)
433
result = remote.push_branch(wt.branch, lossy=True, overwrite=True)
435
self.assertEqual(None, result.old_revno)
436
if self._from_format == 'git':
437
self.assertEqual(1, result.new_revno)
439
self.assertIs(None, result.new_revno)
441
result.report(BytesIO())
445
b'HEAD': self.remote_real.refs[b'refs/heads/master'],
446
b'refs/heads/master': self.remote_real.refs[b'refs/heads/master'],
448
self.remote_real.get_refs())
450
def test_push_branch_new_with_tags(self):
451
remote = ControlDir.open(self.remote_url)
452
builder = self.make_branch_builder('local', format=self._from_format)
453
builder.start_series()
454
rev_1 = builder.build_snapshot(None, [
455
('add', ('', None, 'directory', '')),
456
('add', ('filename', None, 'file', b'content'))])
457
rev_2 = builder.build_snapshot(
458
[rev_1], [('modify', ('filename', b'new-content\n'))])
459
rev_3 = builder.build_snapshot(
460
[rev_1], [('modify', ('filename', b'new-new-content\n'))])
461
builder.finish_series()
462
branch = builder.get_branch()
464
branch.tags.set_tag('atag', rev_2)
465
except TagsNotSupported:
466
raise TestNotApplicable('source format does not support tags')
468
branch.get_config_stack().set('branch.fetch_tags', True)
469
if self._from_format == 'git':
470
result = remote.push_branch(branch, name='newbranch')
472
result = remote.push_branch(
473
branch, lossy=True, name='newbranch')
475
self.assertEqual(0, result.old_revno)
476
if self._from_format == 'git':
477
self.assertEqual(2, result.new_revno)
479
self.assertIs(None, result.new_revno)
481
result.report(BytesIO())
484
{b'refs/heads/newbranch', b'refs/tags/atag'},
485
set(self.remote_real.get_refs().keys()))
488
c1 = self.remote_real.do_commit(
490
committer=b'committer <committer@example.com>',
491
author=b'author <author@example.com>')
493
remote = ControlDir.open(self.remote_url)
494
self.make_controldir('local', format=self._from_format)
495
local = remote.sprout('local')
496
self.build_tree(['local/blah'])
497
wt = local.open_workingtree()
499
revid = wt.commit('blah')
500
wt.branch.tags.set_tag('sometag', revid)
501
wt.branch.get_config_stack().set('branch.fetch_tags', True)
503
if self._from_format == 'git':
504
result = wt.branch.push(remote.create_branch('newbranch'))
506
result = wt.branch.push(
507
remote.create_branch('newbranch'), lossy=True)
509
self.assertEqual(0, result.old_revno)
510
self.assertEqual(2, result.new_revno)
512
result.report(BytesIO())
515
{b'refs/heads/master': self.remote_real.head(),
516
b'HEAD': self.remote_real.head(),
517
b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
518
b'refs/tags/sometag': self.remote_real.refs[b'refs/heads/newbranch'],
520
self.remote_real.get_refs())
522
def test_push_diverged(self):
523
c1 = self.remote_real.do_commit(
525
committer=b'committer <committer@example.com>',
526
author=b'author <author@example.com>',
527
ref=b'refs/heads/newbranch')
529
remote = ControlDir.open(self.remote_url)
530
wt = self.make_branch_and_tree('local', format=self._from_format)
531
self.build_tree(['local/blah'])
533
revid = wt.commit('blah')
535
newbranch = remote.open_branch('newbranch')
536
if self._from_format == 'git':
537
self.assertRaises(DivergedBranches, wt.branch.push, newbranch)
539
self.assertRaises(DivergedBranches, wt.branch.push,
540
newbranch, lossy=True)
543
{b'refs/heads/newbranch': c1},
544
self.remote_real.get_refs())
546
if self._from_format == 'git':
547
wt.branch.push(newbranch, overwrite=True)
549
wt.branch.push(newbranch, lossy=True, overwrite=True)
551
self.assertNotEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
554
class PushToRemoteFromBzrTests(PushToRemoteBase, TestCaseWithTransport):
559
class PushToRemoteFromGitTests(PushToRemoteBase, TestCaseWithTransport):
564
class RemoteControlDirTests(TestCaseWithTransport):
566
_test_needs_features = [ExecutableFeature('git')]
569
TestCaseWithTransport.setUp(self)
570
self.remote_real = GitRepo.init('remote', mkdir=True)
571
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
572
self.permit_url(self.remote_url)
574
def test_remove_branch(self):
575
c1 = self.remote_real.do_commit(
577
committer=b'committer <committer@example.com>',
578
author=b'author <author@example.com>')
579
c2 = self.remote_real.do_commit(
580
message=b'another commit',
581
committer=b'committer <committer@example.com>',
582
author=b'author <author@example.com>',
583
ref=b'refs/heads/blah')
585
remote = ControlDir.open(self.remote_url)
586
remote.destroy_branch(name='blah')
588
self.remote_real.get_refs(),
589
{b'refs/heads/master': self.remote_real.head(),
590
b'HEAD': self.remote_real.head(),
593
def test_list_branches(self):
594
c1 = self.remote_real.do_commit(
596
committer=b'committer <committer@example.com>',
597
author=b'author <author@example.com>')
598
c2 = self.remote_real.do_commit(
599
message=b'another commit',
600
committer=b'committer <committer@example.com>',
601
author=b'author <author@example.com>',
602
ref=b'refs/heads/blah')
604
remote = ControlDir.open(self.remote_url)
606
set(['master', 'blah', 'master']),
607
set([b.name for b in remote.list_branches()]))
609
def test_get_branches(self):
610
c1 = self.remote_real.do_commit(
612
committer=b'committer <committer@example.com>',
613
author=b'author <author@example.com>')
614
c2 = self.remote_real.do_commit(
615
message=b'another commit',
616
committer=b'committer <committer@example.com>',
617
author=b'author <author@example.com>',
618
ref=b'refs/heads/blah')
620
remote = ControlDir.open(self.remote_url)
622
{'': 'master', 'blah': 'blah', 'master': 'master'},
623
{n: b.name for (n, b) in remote.get_branches().items()})
625
set(['', 'blah', 'master']), set(remote.branch_names()))
627
def test_remove_tag(self):
628
c1 = self.remote_real.do_commit(
630
committer=b'committer <committer@example.com>',
631
author=b'author <author@example.com>')
632
c2 = self.remote_real.do_commit(
633
message=b'another commit',
634
committer=b'committer <committer@example.com>',
635
author=b'author <author@example.com>',
636
ref=b'refs/tags/blah')
638
remote = ControlDir.open(self.remote_url)
639
remote_branch = remote.open_branch()
640
remote_branch.tags.delete_tag('blah')
641
self.assertRaises(NoSuchTag, remote_branch.tags.delete_tag, 'blah')
643
self.remote_real.get_refs(),
644
{b'refs/heads/master': self.remote_real.head(),
645
b'HEAD': self.remote_real.head(),
648
def test_set_tag(self):
649
c1 = self.remote_real.do_commit(
651
committer=b'committer <committer@example.com>',
652
author=b'author <author@example.com>')
653
c2 = self.remote_real.do_commit(
654
message=b'another commit',
655
committer=b'committer <committer@example.com>',
656
author=b'author <author@example.com>')
658
remote = ControlDir.open(self.remote_url)
659
remote.open_branch().tags.set_tag(
660
b'blah', default_mapping.revision_id_foreign_to_bzr(c1))
662
self.remote_real.get_refs(),
663
{b'refs/heads/master': self.remote_real.head(),
664
b'refs/tags/blah': c1,
665
b'HEAD': self.remote_real.head(),
668
def test_annotated_tag(self):
669
c1 = self.remote_real.do_commit(
671
committer=b'committer <committer@example.com>',
672
author=b'author <author@example.com>')
673
c2 = self.remote_real.do_commit(
674
message=b'another commit',
675
committer=b'committer <committer@example.com>',
676
author=b'author <author@example.com>')
678
porcelain.tag_create(
681
author=b'author <author@example.com>',
683
tag_time=int(time.time()),
686
message=b"Annotated tag")
688
remote = ControlDir.open(self.remote_url)
689
remote_branch = remote.open_branch()
691
'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
692
remote_branch.tags.get_tag_dict())
694
def test_get_branch_reference(self):
695
c1 = self.remote_real.do_commit(
697
committer=b'committer <committer@example.com>',
698
author=b'author <author@example.com>')
699
c2 = self.remote_real.do_commit(
700
message=b'another commit',
701
committer=b'committer <committer@example.com>',
702
author=b'author <author@example.com>')
704
remote = ControlDir.open(self.remote_url)
705
self.assertEqual(b'refs/heads/master', remote.get_branch_reference(''))
706
self.assertEqual(None, remote.get_branch_reference('master'))
708
def test_get_branch_nick(self):
709
c1 = self.remote_real.do_commit(
711
committer=b'committer <committer@example.com>',
712
author=b'author <author@example.com>')
713
remote = ControlDir.open(self.remote_url)
714
self.assertEqual('master', remote.open_branch().nick)
717
class GitUrlAndPathFromTransportTests(TestCase):
720
split_url = _git_url_and_path_from_transport('file:///home/blah')
721
self.assertEqual(split_url.scheme, 'file')
722
self.assertEqual(split_url.path, '/home/blah')
724
def test_file_segment_params(self):
725
split_url = _git_url_and_path_from_transport('file:///home/blah,branch=master')
726
self.assertEqual(split_url.scheme, 'file')
727
self.assertEqual(split_url.path, '/home/blah')
729
def test_git_smart(self):
730
split_url = _git_url_and_path_from_transport(
731
'git://github.com/dulwich/dulwich,branch=master')
732
self.assertEqual(split_url.scheme, 'git')
733
self.assertEqual(split_url.path, '/dulwich/dulwich')
735
def test_https(self):
736
split_url = _git_url_and_path_from_transport(
737
'https://github.com/dulwich/dulwich')
738
self.assertEqual(split_url.scheme, 'https')
739
self.assertEqual(split_url.path, '/dulwich/dulwich')
741
def test_https_segment_params(self):
742
split_url = _git_url_and_path_from_transport(
743
'https://github.com/dulwich/dulwich,branch=master')
744
self.assertEqual(split_url.scheme, 'https')
745
self.assertEqual(split_url.path, '/dulwich/dulwich')
added
removed
breezy/git/tests/test_remote.py
