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 (
44
RemoteGitBranchFormat,
45
_git_url_and_path_from_transport,
48
from dulwich import porcelain
49
from dulwich.repo import Repo as GitRepo
52
class SplitUrlTests(TestCase):
54
def test_simple(self):
55
self.assertEqual(("foo", None, None, "/bar"),
56
split_git_url("git://foo/bar"))
59
self.assertEqual(("foo", 343, None, "/bar"),
60
split_git_url("git://foo:343/bar"))
62
def test_username(self):
63
self.assertEqual(("foo", None, "la", "/bar"),
64
split_git_url("git://la@foo/bar"))
66
def test_username_password(self):
68
("foo", None, "la", "/bar"),
69
split_git_url("git://la:passwd@foo/bar"))
71
def test_nopath(self):
72
self.assertEqual(("foo", None, None, "/"),
73
split_git_url("git://foo/"))
75
def test_slashpath(self):
76
self.assertEqual(("foo", None, None, "//bar"),
77
split_git_url("git://foo//bar"))
79
def test_homedir(self):
80
self.assertEqual(("foo", None, None, "~bar"),
81
split_git_url("git://foo/~bar"))
85
("", None, None, "/bar"),
86
split_git_url("file:///bar"))
89
class ParseGitErrorTests(TestCase):
91
def test_unknown(self):
92
e = parse_git_error("url", "foo")
93
self.assertIsInstance(e, RemoteGitError)
95
def test_notbrancherror(self):
96
e = parse_git_error("url", "\n Could not find Repository foo/bar")
97
self.assertIsInstance(e, NotBranchError)
99
def test_notbrancherror_launchpad(self):
100
e = parse_git_error("url", "Repository 'foo/bar' not found.")
101
self.assertIsInstance(e, NotBranchError)
103
def test_notbrancherror_github(self):
104
e = parse_git_error("url", "Repository not found.\n")
105
self.assertIsInstance(e, NotBranchError)
107
def test_notbrancherror_normal(self):
109
"url", "fatal: '/srv/git/lintian-brush' does not appear to be a git repository")
110
self.assertIsInstance(e, NotBranchError)
112
def test_head_update(self):
113
e = parse_git_error("url", "HEAD failed to update\n")
114
self.assertIsInstance(e, HeadUpdateFailed)
116
def test_permission_dnied(self):
119
"access denied or repository not exported: /debian/altermime.git")
120
self.assertIsInstance(e, PermissionDenied)
122
def test_permission_denied_gitlab(self):
125
'GitLab: You are not allowed to push code to this project.\n')
126
self.assertIsInstance(e, PermissionDenied)
128
def test_permission_denied_github(self):
131
'Permission to porridge/gaduhistory.git denied to jelmer.')
132
self.assertIsInstance(e, PermissionDenied)
133
self.assertEqual(e.path, 'porridge/gaduhistory.git')
134
self.assertEqual(e.extra, ': denied to jelmer')
136
def test_invalid_repo_name(self):
139
"""Gregwar/fatcat/tree/debian is not a valid repository name
140
Email support@github.com for help
142
self.assertIsInstance(e, NotBranchError)
145
class TestRemoteGitBranchFormat(TestCase):
148
super(TestRemoteGitBranchFormat, self).setUp()
149
self.format = RemoteGitBranchFormat()
151
def test_get_format_description(self):
152
self.assertEqual("Remote Git Branch",
153
self.format.get_format_description())
155
def test_get_network_name(self):
156
self.assertEqual(b"git", self.format.network_name())
158
def test_supports_tags(self):
159
self.assertTrue(self.format.supports_tags())
162
class TestRemoteGitBranch(TestCaseWithTransport):
164
_test_needs_features = [ExecutableFeature('git')]
167
TestCaseWithTransport.setUp(self)
168
self.remote_real = GitRepo.init('remote', mkdir=True)
169
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
170
self.permit_url(self.remote_url)
172
def test_set_last_revision_info(self):
173
c1 = self.remote_real.do_commit(
174
message=b'message 1',
175
committer=b'committer <committer@example.com>',
176
author=b'author <author@example.com>',
177
ref=b'refs/heads/newbranch')
178
c2 = self.remote_real.do_commit(
179
message=b'message 2',
180
committer=b'committer <committer@example.com>',
181
author=b'author <author@example.com>',
182
ref=b'refs/heads/newbranch')
184
remote = ControlDir.open(self.remote_url)
185
newbranch = remote.open_branch('newbranch')
186
self.assertEqual(newbranch.lookup_foreign_revision_id(c2),
187
newbranch.last_revision())
188
newbranch.set_last_revision_info(
189
1, newbranch.lookup_foreign_revision_id(c1))
190
self.assertEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
191
self.assertEqual(newbranch.last_revision(),
192
newbranch.lookup_foreign_revision_id(c1))
195
class FetchFromRemoteTestBase(object):
197
_test_needs_features = [ExecutableFeature('git')]
202
TestCaseWithTransport.setUp(self)
203
self.remote_real = GitRepo.init('remote', mkdir=True)
204
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
205
self.permit_url(self.remote_url)
207
def test_sprout_simple(self):
208
self.remote_real.do_commit(
210
committer=b'committer <committer@example.com>',
211
author=b'author <author@example.com>')
213
remote = ControlDir.open(self.remote_url)
214
self.make_controldir('local', format=self._to_format)
215
local = remote.sprout('local')
217
default_mapping.revision_id_foreign_to_bzr(
218
self.remote_real.head()),
219
local.open_branch().last_revision())
221
def test_sprout_with_tags(self):
222
c1 = self.remote_real.do_commit(
224
committer=b'committer <committer@example.com>',
225
author=b'author <author@example.com>')
226
c2 = self.remote_real.do_commit(
227
message=b'another commit',
228
committer=b'committer <committer@example.com>',
229
author=b'author <author@example.com>',
230
ref=b'refs/tags/another')
231
self.remote_real.refs[b'refs/tags/blah'] = self.remote_real.head()
233
remote = ControlDir.open(self.remote_url)
234
self.make_controldir('local', format=self._to_format)
235
local = remote.sprout('local')
236
local_branch = local.open_branch()
238
default_mapping.revision_id_foreign_to_bzr(c1),
239
local_branch.last_revision())
241
{'blah': local_branch.last_revision(),
242
'another': default_mapping.revision_id_foreign_to_bzr(c2)},
243
local_branch.tags.get_tag_dict())
245
def test_sprout_with_annotated_tag(self):
246
c1 = self.remote_real.do_commit(
248
committer=b'committer <committer@example.com>',
249
author=b'author <author@example.com>')
250
c2 = self.remote_real.do_commit(
251
message=b'another commit',
252
committer=b'committer <committer@example.com>',
253
author=b'author <author@example.com>',
254
ref=b'refs/heads/another')
255
porcelain.tag_create(
258
author=b'author <author@example.com>',
260
tag_time=int(time.time()),
263
message=b"Annotated tag")
265
remote = ControlDir.open(self.remote_url)
266
self.make_controldir('local', format=self._to_format)
267
local = remote.sprout(
268
'local', revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
269
local_branch = local.open_branch()
271
default_mapping.revision_id_foreign_to_bzr(c1),
272
local_branch.last_revision())
274
{'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
275
local_branch.tags.get_tag_dict())
277
def test_sprout_with_annotated_tag_unreferenced(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
porcelain.tag_create(
289
author=b'author <author@example.com>',
291
tag_time=int(time.time()),
294
message=b"Annotated tag")
296
remote = ControlDir.open(self.remote_url)
297
self.make_controldir('local', format=self._to_format)
298
local = remote.sprout(
300
revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
301
local_branch = local.open_branch()
303
default_mapping.revision_id_foreign_to_bzr(c1),
304
local_branch.last_revision())
306
{'blah': default_mapping.revision_id_foreign_to_bzr(c1)},
307
local_branch.tags.get_tag_dict())
310
class FetchFromRemoteToBzrTests(FetchFromRemoteTestBase, TestCaseWithTransport):
315
class FetchFromRemoteToGitTests(FetchFromRemoteTestBase, TestCaseWithTransport):
320
class PushToRemoteBase(object):
322
_test_needs_features = [ExecutableFeature('git')]
327
TestCaseWithTransport.setUp(self)
328
self.remote_real = GitRepo.init('remote', mkdir=True)
329
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
330
self.permit_url(self.remote_url)
332
def test_push_branch_new(self):
333
remote = ControlDir.open(self.remote_url)
334
wt = self.make_branch_and_tree('local', format=self._from_format)
335
self.build_tree(['local/blah'])
337
revid = wt.commit('blah')
339
if self._from_format == 'git':
340
result = remote.push_branch(wt.branch, name='newbranch')
342
result = remote.push_branch(
343
wt.branch, lossy=True, name='newbranch')
345
self.assertEqual(0, result.old_revno)
346
if self._from_format == 'git':
347
self.assertEqual(1, result.new_revno)
349
self.assertIs(None, result.new_revno)
351
result.report(BytesIO())
354
{b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
356
self.remote_real.get_refs())
358
def test_push_branch_symref(self):
359
cfg = self.remote_real.get_config()
360
cfg.set((b'core', ), b'bare', True)
362
self.remote_real.refs.set_symbolic_ref(b'HEAD', b'refs/heads/master')
363
c1 = self.remote_real.do_commit(
365
committer=b'committer <committer@example.com>',
366
author=b'author <author@example.com>',
367
ref=b'refs/heads/master')
368
remote = ControlDir.open(self.remote_url)
369
wt = self.make_branch_and_tree('local', format=self._from_format)
370
self.build_tree(['local/blah'])
372
revid = wt.commit('blah')
374
if self._from_format == 'git':
375
result = remote.push_branch(wt.branch, overwrite=True)
377
result = remote.push_branch(wt.branch, lossy=True, overwrite=True)
379
self.assertEqual(None, result.old_revno)
380
if self._from_format == 'git':
381
self.assertEqual(1, result.new_revno)
383
self.assertIs(None, result.new_revno)
385
result.report(BytesIO())
389
b'HEAD': self.remote_real.refs[b'refs/heads/master'],
390
b'refs/heads/master': self.remote_real.refs[b'refs/heads/master'],
392
self.remote_real.get_refs())
394
def test_push_branch_new_with_tags(self):
395
remote = ControlDir.open(self.remote_url)
396
builder = self.make_branch_builder('local', format=self._from_format)
397
builder.start_series()
398
rev_1 = builder.build_snapshot(None, [
399
('add', ('', None, 'directory', '')),
400
('add', ('filename', None, 'file', b'content'))])
401
rev_2 = builder.build_snapshot(
402
[rev_1], [('modify', ('filename', b'new-content\n'))])
403
rev_3 = builder.build_snapshot(
404
[rev_1], [('modify', ('filename', b'new-new-content\n'))])
405
builder.finish_series()
406
branch = builder.get_branch()
408
branch.tags.set_tag('atag', rev_2)
409
except TagsNotSupported:
410
raise TestNotApplicable('source format does not support tags')
412
branch.get_config_stack().set('branch.fetch_tags', True)
413
if self._from_format == 'git':
414
result = remote.push_branch(branch, name='newbranch')
416
result = remote.push_branch(
417
branch, lossy=True, name='newbranch')
419
self.assertEqual(0, result.old_revno)
420
if self._from_format == 'git':
421
self.assertEqual(2, result.new_revno)
423
self.assertIs(None, result.new_revno)
425
result.report(BytesIO())
428
{b'refs/heads/newbranch', b'refs/tags/atag'},
429
set(self.remote_real.get_refs().keys()))
432
c1 = self.remote_real.do_commit(
434
committer=b'committer <committer@example.com>',
435
author=b'author <author@example.com>')
437
remote = ControlDir.open(self.remote_url)
438
self.make_controldir('local', format=self._from_format)
439
local = remote.sprout('local')
440
self.build_tree(['local/blah'])
441
wt = local.open_workingtree()
443
revid = wt.commit('blah')
444
wt.branch.tags.set_tag('sometag', revid)
445
wt.branch.get_config_stack().set('branch.fetch_tags', True)
447
if self._from_format == 'git':
448
result = wt.branch.push(remote.create_branch('newbranch'))
450
result = wt.branch.push(
451
remote.create_branch('newbranch'), lossy=True)
453
self.assertEqual(0, result.old_revno)
454
self.assertEqual(2, result.new_revno)
456
result.report(BytesIO())
459
{b'refs/heads/master': self.remote_real.head(),
460
b'HEAD': self.remote_real.head(),
461
b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
462
b'refs/tags/sometag': self.remote_real.refs[b'refs/heads/newbranch'],
464
self.remote_real.get_refs())
466
def test_push_diverged(self):
467
c1 = self.remote_real.do_commit(
469
committer=b'committer <committer@example.com>',
470
author=b'author <author@example.com>',
471
ref=b'refs/heads/newbranch')
473
remote = ControlDir.open(self.remote_url)
474
wt = self.make_branch_and_tree('local', format=self._from_format)
475
self.build_tree(['local/blah'])
477
revid = wt.commit('blah')
479
newbranch = remote.open_branch('newbranch')
480
if self._from_format == 'git':
481
self.assertRaises(DivergedBranches, wt.branch.push, newbranch)
483
self.assertRaises(DivergedBranches, wt.branch.push,
484
newbranch, lossy=True)
487
{b'refs/heads/newbranch': c1},
488
self.remote_real.get_refs())
490
if self._from_format == 'git':
491
wt.branch.push(newbranch, overwrite=True)
493
wt.branch.push(newbranch, lossy=True, overwrite=True)
495
self.assertNotEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
498
class PushToRemoteFromBzrTests(PushToRemoteBase, TestCaseWithTransport):
503
class PushToRemoteFromGitTests(PushToRemoteBase, TestCaseWithTransport):
508
class RemoteControlDirTests(TestCaseWithTransport):
510
_test_needs_features = [ExecutableFeature('git')]
513
TestCaseWithTransport.setUp(self)
514
self.remote_real = GitRepo.init('remote', mkdir=True)
515
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
516
self.permit_url(self.remote_url)
518
def test_remove_branch(self):
519
c1 = self.remote_real.do_commit(
521
committer=b'committer <committer@example.com>',
522
author=b'author <author@example.com>')
523
c2 = self.remote_real.do_commit(
524
message=b'another commit',
525
committer=b'committer <committer@example.com>',
526
author=b'author <author@example.com>',
527
ref=b'refs/heads/blah')
529
remote = ControlDir.open(self.remote_url)
530
remote.destroy_branch(name='blah')
532
self.remote_real.get_refs(),
533
{b'refs/heads/master': self.remote_real.head(),
534
b'HEAD': self.remote_real.head(),
537
def test_list_branches(self):
538
c1 = self.remote_real.do_commit(
540
committer=b'committer <committer@example.com>',
541
author=b'author <author@example.com>')
542
c2 = self.remote_real.do_commit(
543
message=b'another commit',
544
committer=b'committer <committer@example.com>',
545
author=b'author <author@example.com>',
546
ref=b'refs/heads/blah')
548
remote = ControlDir.open(self.remote_url)
550
set(['master', 'blah', 'master']),
551
set([b.name for b in remote.list_branches()]))
553
def test_get_branches(self):
554
c1 = self.remote_real.do_commit(
556
committer=b'committer <committer@example.com>',
557
author=b'author <author@example.com>')
558
c2 = self.remote_real.do_commit(
559
message=b'another commit',
560
committer=b'committer <committer@example.com>',
561
author=b'author <author@example.com>',
562
ref=b'refs/heads/blah')
564
remote = ControlDir.open(self.remote_url)
566
{'': 'master', 'blah': 'blah', 'master': 'master'},
567
{n: b.name for (n, b) in remote.get_branches().items()})
569
def test_remove_tag(self):
570
c1 = self.remote_real.do_commit(
572
committer=b'committer <committer@example.com>',
573
author=b'author <author@example.com>')
574
c2 = self.remote_real.do_commit(
575
message=b'another commit',
576
committer=b'committer <committer@example.com>',
577
author=b'author <author@example.com>',
578
ref=b'refs/tags/blah')
580
remote = ControlDir.open(self.remote_url)
581
remote_branch = remote.open_branch()
582
remote_branch.tags.delete_tag('blah')
583
self.assertRaises(NoSuchTag, remote_branch.tags.delete_tag, 'blah')
585
self.remote_real.get_refs(),
586
{b'refs/heads/master': self.remote_real.head(),
587
b'HEAD': self.remote_real.head(),
590
def test_set_tag(self):
591
c1 = self.remote_real.do_commit(
593
committer=b'committer <committer@example.com>',
594
author=b'author <author@example.com>')
595
c2 = self.remote_real.do_commit(
596
message=b'another commit',
597
committer=b'committer <committer@example.com>',
598
author=b'author <author@example.com>')
600
remote = ControlDir.open(self.remote_url)
601
remote.open_branch().tags.set_tag(
602
b'blah', default_mapping.revision_id_foreign_to_bzr(c1))
604
self.remote_real.get_refs(),
605
{b'refs/heads/master': self.remote_real.head(),
606
b'refs/tags/blah': c1,
607
b'HEAD': self.remote_real.head(),
610
def test_annotated_tag(self):
611
c1 = self.remote_real.do_commit(
613
committer=b'committer <committer@example.com>',
614
author=b'author <author@example.com>')
615
c2 = self.remote_real.do_commit(
616
message=b'another commit',
617
committer=b'committer <committer@example.com>',
618
author=b'author <author@example.com>')
620
porcelain.tag_create(
623
author=b'author <author@example.com>',
625
tag_time=int(time.time()),
628
message=b"Annotated tag")
630
remote = ControlDir.open(self.remote_url)
631
remote_branch = remote.open_branch()
633
'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
634
remote_branch.tags.get_tag_dict())
636
def test_get_branch_reference(self):
637
c1 = self.remote_real.do_commit(
639
committer=b'committer <committer@example.com>',
640
author=b'author <author@example.com>')
641
c2 = self.remote_real.do_commit(
642
message=b'another commit',
643
committer=b'committer <committer@example.com>',
644
author=b'author <author@example.com>')
646
remote = ControlDir.open(self.remote_url)
647
self.assertEqual(b'refs/heads/master', remote.get_branch_reference(''))
648
self.assertEqual(None, remote.get_branch_reference('master'))
650
def test_get_branch_nick(self):
651
c1 = self.remote_real.do_commit(
653
committer=b'committer <committer@example.com>',
654
author=b'author <author@example.com>')
655
remote = ControlDir.open(self.remote_url)
656
self.assertEqual('master', remote.open_branch().nick)
659
class GitUrlAndPathFromTransportTests(TestCase):
662
split_url = _git_url_and_path_from_transport('file:///home/blah')
663
self.assertEqual(split_url.scheme, 'file')
664
self.assertEqual(split_url.path, '/home/blah')
666
def test_file_segment_params(self):
667
split_url = _git_url_and_path_from_transport('file:///home/blah,branch=master')
668
self.assertEqual(split_url.scheme, 'file')
669
self.assertEqual(split_url.path, '/home/blah')
671
def test_git_smart(self):
672
split_url = _git_url_and_path_from_transport(
673
'git://github.com/dulwich/dulwich,branch=master')
674
self.assertEqual(split_url.scheme, 'git')
675
self.assertEqual(split_url.path, '/dulwich/dulwich')
677
def test_https(self):
678
split_url = _git_url_and_path_from_transport(
679
'https://github.com/dulwich/dulwich')
680
self.assertEqual(split_url.scheme, 'https')
681
self.assertEqual(split_url.path, '/dulwich/dulwich')
683
def test_https_segment_params(self):
684
split_url = _git_url_and_path_from_transport(
685
'https://github.com/dulwich/dulwich,branch=master')
686
self.assertEqual(split_url.scheme, 'https')
687
self.assertEqual(split_url.path, '/dulwich/dulwich')