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_pre_receive_hook_declined(self):
141
'pre-receive hook declined')
142
self.assertIsInstance(e, PermissionDenied)
143
self.assertEqual(e.path, "url")
144
self.assertEqual(e.extra, ': pre-receive hook declined')
146
def test_invalid_repo_name(self):
149
"""Gregwar/fatcat/tree/debian is not a valid repository name
150
Email support@github.com for help
152
self.assertIsInstance(e, NotBranchError)
154
def test_invalid_git_error(self):
158
'GitLab: You are not allowed to push code to protected '
159
'branches on this project.'),
163
'GitLab: You are not allowed to push code to '
164
'protected branches on this project.')))
167
class ParseHangupTests(TestCase):
170
super(ParseHangupTests, self).setUp()
172
HangupException([b'foo'])
174
self.skipTest('dulwich version too old')
176
def test_not_set(self):
177
self.assertIsInstance(
178
parse_git_hangup('http://', HangupException()), HangupException)
180
def test_single_line(self):
182
RemoteGitError('foo bar'),
183
parse_git_hangup('http://', HangupException([b'foo bar'])))
185
def test_multi_lines(self):
187
RemoteGitError('foo bar\nbla bla'),
189
'http://', HangupException([b'foo bar', b'bla bla'])))
191
def test_filter_boring(self):
193
RemoteGitError('foo bar'), parse_git_hangup('http://', HangupException(
194
[b'=======', b'foo bar', b'======'])))
196
RemoteGitError('foo bar'), parse_git_hangup('http://', HangupException(
197
[b'remote: =======', b'remote: foo bar', b'remote: ======'])))
199
def test_permission_denied(self):
201
PermissionDenied('http://', 'You are not allowed to push code to this project.'),
206
b'You are not allowed to push code to this project.', b'', b'======'])))
209
class TestRemoteGitBranchFormat(TestCase):
212
super(TestRemoteGitBranchFormat, self).setUp()
213
self.format = RemoteGitBranchFormat()
215
def test_get_format_description(self):
216
self.assertEqual("Remote Git Branch",
217
self.format.get_format_description())
219
def test_get_network_name(self):
220
self.assertEqual(b"git", self.format.network_name())
222
def test_supports_tags(self):
223
self.assertTrue(self.format.supports_tags())
226
class TestRemoteGitBranch(TestCaseWithTransport):
228
_test_needs_features = [ExecutableFeature('git')]
231
TestCaseWithTransport.setUp(self)
232
self.remote_real = GitRepo.init('remote', mkdir=True)
233
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
234
self.permit_url(self.remote_url)
236
def test_set_last_revision_info(self):
237
c1 = self.remote_real.do_commit(
238
message=b'message 1',
239
committer=b'committer <committer@example.com>',
240
author=b'author <author@example.com>',
241
ref=b'refs/heads/newbranch')
242
c2 = self.remote_real.do_commit(
243
message=b'message 2',
244
committer=b'committer <committer@example.com>',
245
author=b'author <author@example.com>',
246
ref=b'refs/heads/newbranch')
248
remote = ControlDir.open(self.remote_url)
249
newbranch = remote.open_branch('newbranch')
250
self.assertEqual(newbranch.lookup_foreign_revision_id(c2),
251
newbranch.last_revision())
252
newbranch.set_last_revision_info(
253
1, newbranch.lookup_foreign_revision_id(c1))
254
self.assertEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
255
self.assertEqual(newbranch.last_revision(),
256
newbranch.lookup_foreign_revision_id(c1))
259
class FetchFromRemoteTestBase(object):
261
_test_needs_features = [ExecutableFeature('git')]
266
TestCaseWithTransport.setUp(self)
267
self.remote_real = GitRepo.init('remote', mkdir=True)
268
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
269
self.permit_url(self.remote_url)
271
def test_sprout_simple(self):
272
self.remote_real.do_commit(
274
committer=b'committer <committer@example.com>',
275
author=b'author <author@example.com>')
277
remote = ControlDir.open(self.remote_url)
278
self.make_controldir('local', format=self._to_format)
279
local = remote.sprout('local')
281
default_mapping.revision_id_foreign_to_bzr(
282
self.remote_real.head()),
283
local.open_branch().last_revision())
285
def test_sprout_with_tags(self):
286
c1 = self.remote_real.do_commit(
288
committer=b'committer <committer@example.com>',
289
author=b'author <author@example.com>')
290
c2 = self.remote_real.do_commit(
291
message=b'another commit',
292
committer=b'committer <committer@example.com>',
293
author=b'author <author@example.com>',
294
ref=b'refs/tags/another')
295
self.remote_real.refs[b'refs/tags/blah'] = self.remote_real.head()
297
remote = ControlDir.open(self.remote_url)
298
self.make_controldir('local', format=self._to_format)
299
local = remote.sprout('local')
300
local_branch = local.open_branch()
302
default_mapping.revision_id_foreign_to_bzr(c1),
303
local_branch.last_revision())
305
{'blah': local_branch.last_revision(),
306
'another': default_mapping.revision_id_foreign_to_bzr(c2)},
307
local_branch.tags.get_tag_dict())
309
def test_sprout_with_annotated_tag(self):
310
c1 = self.remote_real.do_commit(
312
committer=b'committer <committer@example.com>',
313
author=b'author <author@example.com>')
314
c2 = self.remote_real.do_commit(
315
message=b'another commit',
316
committer=b'committer <committer@example.com>',
317
author=b'author <author@example.com>',
318
ref=b'refs/heads/another')
319
porcelain.tag_create(
322
author=b'author <author@example.com>',
324
tag_time=int(time.time()),
327
message=b"Annotated tag")
329
remote = ControlDir.open(self.remote_url)
330
self.make_controldir('local', format=self._to_format)
331
local = remote.sprout(
332
'local', revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
333
local_branch = local.open_branch()
335
default_mapping.revision_id_foreign_to_bzr(c1),
336
local_branch.last_revision())
338
{'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
339
local_branch.tags.get_tag_dict())
341
def test_sprout_with_annotated_tag_unreferenced(self):
342
c1 = self.remote_real.do_commit(
344
committer=b'committer <committer@example.com>',
345
author=b'author <author@example.com>')
346
c2 = self.remote_real.do_commit(
347
message=b'another commit',
348
committer=b'committer <committer@example.com>',
349
author=b'author <author@example.com>')
350
porcelain.tag_create(
353
author=b'author <author@example.com>',
355
tag_time=int(time.time()),
358
message=b"Annotated tag")
360
remote = ControlDir.open(self.remote_url)
361
self.make_controldir('local', format=self._to_format)
362
local = remote.sprout(
364
revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
365
local_branch = local.open_branch()
367
default_mapping.revision_id_foreign_to_bzr(c1),
368
local_branch.last_revision())
370
{'blah': default_mapping.revision_id_foreign_to_bzr(c1)},
371
local_branch.tags.get_tag_dict())
374
class FetchFromRemoteToBzrTests(FetchFromRemoteTestBase, TestCaseWithTransport):
379
class FetchFromRemoteToGitTests(FetchFromRemoteTestBase, TestCaseWithTransport):
384
class PushToRemoteBase(object):
386
_test_needs_features = [ExecutableFeature('git')]
391
TestCaseWithTransport.setUp(self)
392
self.remote_real = GitRepo.init('remote', mkdir=True)
393
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
394
self.permit_url(self.remote_url)
396
def test_push_branch_new(self):
397
remote = ControlDir.open(self.remote_url)
398
wt = self.make_branch_and_tree('local', format=self._from_format)
399
self.build_tree(['local/blah'])
401
revid = wt.commit('blah')
403
if self._from_format == 'git':
404
result = remote.push_branch(wt.branch, name='newbranch')
406
result = remote.push_branch(
407
wt.branch, lossy=True, name='newbranch')
409
self.assertEqual(0, result.old_revno)
410
if self._from_format == 'git':
411
self.assertEqual(1, result.new_revno)
413
self.assertIs(None, result.new_revno)
415
result.report(BytesIO())
418
{b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
420
self.remote_real.get_refs())
422
def test_push_branch_symref(self):
423
cfg = self.remote_real.get_config()
424
cfg.set((b'core', ), b'bare', True)
426
self.remote_real.refs.set_symbolic_ref(b'HEAD', b'refs/heads/master')
427
c1 = self.remote_real.do_commit(
429
committer=b'committer <committer@example.com>',
430
author=b'author <author@example.com>',
431
ref=b'refs/heads/master')
432
remote = ControlDir.open(self.remote_url)
433
wt = self.make_branch_and_tree('local', format=self._from_format)
434
self.build_tree(['local/blah'])
436
revid = wt.commit('blah')
438
if self._from_format == 'git':
439
result = remote.push_branch(wt.branch, overwrite=True)
441
result = remote.push_branch(wt.branch, lossy=True, overwrite=True)
443
self.assertEqual(None, result.old_revno)
444
if self._from_format == 'git':
445
self.assertEqual(1, result.new_revno)
447
self.assertIs(None, result.new_revno)
449
result.report(BytesIO())
453
b'HEAD': self.remote_real.refs[b'refs/heads/master'],
454
b'refs/heads/master': self.remote_real.refs[b'refs/heads/master'],
456
self.remote_real.get_refs())
458
def test_push_branch_new_with_tags(self):
459
remote = ControlDir.open(self.remote_url)
460
builder = self.make_branch_builder('local', format=self._from_format)
461
builder.start_series()
462
rev_1 = builder.build_snapshot(None, [
463
('add', ('', None, 'directory', '')),
464
('add', ('filename', None, 'file', b'content'))])
465
rev_2 = builder.build_snapshot(
466
[rev_1], [('modify', ('filename', b'new-content\n'))])
467
rev_3 = builder.build_snapshot(
468
[rev_1], [('modify', ('filename', b'new-new-content\n'))])
469
builder.finish_series()
470
branch = builder.get_branch()
472
branch.tags.set_tag('atag', rev_2)
473
except TagsNotSupported:
474
raise TestNotApplicable('source format does not support tags')
476
branch.get_config_stack().set('branch.fetch_tags', True)
477
if self._from_format == 'git':
478
result = remote.push_branch(branch, name='newbranch')
480
result = remote.push_branch(
481
branch, lossy=True, name='newbranch')
483
self.assertEqual(0, result.old_revno)
484
if self._from_format == 'git':
485
self.assertEqual(2, result.new_revno)
487
self.assertIs(None, result.new_revno)
489
result.report(BytesIO())
492
{b'refs/heads/newbranch', b'refs/tags/atag'},
493
set(self.remote_real.get_refs().keys()))
496
c1 = self.remote_real.do_commit(
498
committer=b'committer <committer@example.com>',
499
author=b'author <author@example.com>')
501
remote = ControlDir.open(self.remote_url)
502
self.make_controldir('local', format=self._from_format)
503
local = remote.sprout('local')
504
self.build_tree(['local/blah'])
505
wt = local.open_workingtree()
507
revid = wt.commit('blah')
508
wt.branch.tags.set_tag('sometag', revid)
509
wt.branch.get_config_stack().set('branch.fetch_tags', True)
511
if self._from_format == 'git':
512
result = wt.branch.push(remote.create_branch('newbranch'))
514
result = wt.branch.push(
515
remote.create_branch('newbranch'), lossy=True)
517
self.assertEqual(0, result.old_revno)
518
self.assertEqual(2, result.new_revno)
520
result.report(BytesIO())
523
{b'refs/heads/master': self.remote_real.head(),
524
b'HEAD': self.remote_real.head(),
525
b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
526
b'refs/tags/sometag': self.remote_real.refs[b'refs/heads/newbranch'],
528
self.remote_real.get_refs())
530
def test_push_diverged(self):
531
c1 = self.remote_real.do_commit(
533
committer=b'committer <committer@example.com>',
534
author=b'author <author@example.com>',
535
ref=b'refs/heads/newbranch')
537
remote = ControlDir.open(self.remote_url)
538
wt = self.make_branch_and_tree('local', format=self._from_format)
539
self.build_tree(['local/blah'])
541
revid = wt.commit('blah')
543
newbranch = remote.open_branch('newbranch')
544
if self._from_format == 'git':
545
self.assertRaises(DivergedBranches, wt.branch.push, newbranch)
547
self.assertRaises(DivergedBranches, wt.branch.push,
548
newbranch, lossy=True)
551
{b'refs/heads/newbranch': c1},
552
self.remote_real.get_refs())
554
if self._from_format == 'git':
555
wt.branch.push(newbranch, overwrite=True)
557
wt.branch.push(newbranch, lossy=True, overwrite=True)
559
self.assertNotEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
562
class PushToRemoteFromBzrTests(PushToRemoteBase, TestCaseWithTransport):
567
class PushToRemoteFromGitTests(PushToRemoteBase, TestCaseWithTransport):
572
class RemoteControlDirTests(TestCaseWithTransport):
574
_test_needs_features = [ExecutableFeature('git')]
577
TestCaseWithTransport.setUp(self)
578
self.remote_real = GitRepo.init('remote', mkdir=True)
579
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
580
self.permit_url(self.remote_url)
582
def test_remove_branch(self):
583
c1 = self.remote_real.do_commit(
585
committer=b'committer <committer@example.com>',
586
author=b'author <author@example.com>')
587
c2 = self.remote_real.do_commit(
588
message=b'another commit',
589
committer=b'committer <committer@example.com>',
590
author=b'author <author@example.com>',
591
ref=b'refs/heads/blah')
593
remote = ControlDir.open(self.remote_url)
594
remote.destroy_branch(name='blah')
596
self.remote_real.get_refs(),
597
{b'refs/heads/master': self.remote_real.head(),
598
b'HEAD': self.remote_real.head(),
601
def test_list_branches(self):
602
c1 = self.remote_real.do_commit(
604
committer=b'committer <committer@example.com>',
605
author=b'author <author@example.com>')
606
c2 = self.remote_real.do_commit(
607
message=b'another commit',
608
committer=b'committer <committer@example.com>',
609
author=b'author <author@example.com>',
610
ref=b'refs/heads/blah')
612
remote = ControlDir.open(self.remote_url)
614
set(['master', 'blah', 'master']),
615
set([b.name for b in remote.list_branches()]))
617
def test_get_branches(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/heads/blah')
628
remote = ControlDir.open(self.remote_url)
630
{'': 'master', 'blah': 'blah', 'master': 'master'},
631
{n: b.name for (n, b) in remote.get_branches().items()})
633
set(['', 'blah', 'master']), set(remote.branch_names()))
635
def test_remove_tag(self):
636
c1 = self.remote_real.do_commit(
638
committer=b'committer <committer@example.com>',
639
author=b'author <author@example.com>')
640
c2 = self.remote_real.do_commit(
641
message=b'another commit',
642
committer=b'committer <committer@example.com>',
643
author=b'author <author@example.com>',
644
ref=b'refs/tags/blah')
646
remote = ControlDir.open(self.remote_url)
647
remote_branch = remote.open_branch()
648
remote_branch.tags.delete_tag('blah')
649
self.assertRaises(NoSuchTag, remote_branch.tags.delete_tag, 'blah')
651
self.remote_real.get_refs(),
652
{b'refs/heads/master': self.remote_real.head(),
653
b'HEAD': self.remote_real.head(),
656
def test_set_tag(self):
657
c1 = self.remote_real.do_commit(
659
committer=b'committer <committer@example.com>',
660
author=b'author <author@example.com>')
661
c2 = self.remote_real.do_commit(
662
message=b'another commit',
663
committer=b'committer <committer@example.com>',
664
author=b'author <author@example.com>')
666
remote = ControlDir.open(self.remote_url)
667
remote.open_branch().tags.set_tag(
668
b'blah', default_mapping.revision_id_foreign_to_bzr(c1))
670
self.remote_real.get_refs(),
671
{b'refs/heads/master': self.remote_real.head(),
672
b'refs/tags/blah': c1,
673
b'HEAD': self.remote_real.head(),
676
def test_annotated_tag(self):
677
c1 = self.remote_real.do_commit(
679
committer=b'committer <committer@example.com>',
680
author=b'author <author@example.com>')
681
c2 = self.remote_real.do_commit(
682
message=b'another commit',
683
committer=b'committer <committer@example.com>',
684
author=b'author <author@example.com>')
686
porcelain.tag_create(
689
author=b'author <author@example.com>',
691
tag_time=int(time.time()),
694
message=b"Annotated tag")
696
remote = ControlDir.open(self.remote_url)
697
remote_branch = remote.open_branch()
699
'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
700
remote_branch.tags.get_tag_dict())
702
def test_get_branch_reference(self):
703
c1 = self.remote_real.do_commit(
705
committer=b'committer <committer@example.com>',
706
author=b'author <author@example.com>')
707
c2 = self.remote_real.do_commit(
708
message=b'another commit',
709
committer=b'committer <committer@example.com>',
710
author=b'author <author@example.com>')
712
remote = ControlDir.open(self.remote_url)
713
self.assertEqual(b'refs/heads/master', remote.get_branch_reference(''))
714
self.assertEqual(None, remote.get_branch_reference('master'))
716
def test_get_branch_nick(self):
717
c1 = self.remote_real.do_commit(
719
committer=b'committer <committer@example.com>',
720
author=b'author <author@example.com>')
721
remote = ControlDir.open(self.remote_url)
722
self.assertEqual('master', remote.open_branch().nick)
725
class GitUrlAndPathFromTransportTests(TestCase):
728
split_url = _git_url_and_path_from_transport('file:///home/blah')
729
self.assertEqual(split_url.scheme, 'file')
730
self.assertEqual(split_url.path, '/home/blah')
732
def test_file_segment_params(self):
733
split_url = _git_url_and_path_from_transport('file:///home/blah,branch=master')
734
self.assertEqual(split_url.scheme, 'file')
735
self.assertEqual(split_url.path, '/home/blah')
737
def test_git_smart(self):
738
split_url = _git_url_and_path_from_transport(
739
'git://github.com/dulwich/dulwich,branch=master')
740
self.assertEqual(split_url.scheme, 'git')
741
self.assertEqual(split_url.path, '/dulwich/dulwich')
743
def test_https(self):
744
split_url = _git_url_and_path_from_transport(
745
'https://github.com/dulwich/dulwich')
746
self.assertEqual(split_url.scheme, 'https')
747
self.assertEqual(split_url.path, '/dulwich/dulwich')
749
def test_https_segment_params(self):
750
split_url = _git_url_and_path_from_transport(
751
'https://github.com/dulwich/dulwich,branch=master')
752
self.assertEqual(split_url.scheme, 'https')
753
self.assertEqual(split_url.path, '/dulwich/dulwich')
added
removed
breezy/git/tests/test_remote.py
