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 (
46
RemoteGitBranchFormat,
47
_git_url_and_path_from_transport,
50
from dulwich import porcelain
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)
147
class TestRemoteGitBranchFormat(TestCase):
150
super(TestRemoteGitBranchFormat, self).setUp()
151
self.format = RemoteGitBranchFormat()
153
def test_get_format_description(self):
154
self.assertEqual("Remote Git Branch",
155
self.format.get_format_description())
157
def test_get_network_name(self):
158
self.assertEqual(b"git", self.format.network_name())
160
def test_supports_tags(self):
161
self.assertTrue(self.format.supports_tags())
164
class TestRemoteGitBranch(TestCaseWithTransport):
166
_test_needs_features = [ExecutableFeature('git')]
169
TestCaseWithTransport.setUp(self)
170
self.remote_real = GitRepo.init('remote', mkdir=True)
171
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
172
self.permit_url(self.remote_url)
174
def test_set_last_revision_info(self):
175
c1 = self.remote_real.do_commit(
176
message=b'message 1',
177
committer=b'committer <committer@example.com>',
178
author=b'author <author@example.com>',
179
ref=b'refs/heads/newbranch')
180
c2 = self.remote_real.do_commit(
181
message=b'message 2',
182
committer=b'committer <committer@example.com>',
183
author=b'author <author@example.com>',
184
ref=b'refs/heads/newbranch')
186
remote = ControlDir.open(self.remote_url)
187
newbranch = remote.open_branch('newbranch')
188
self.assertEqual(newbranch.lookup_foreign_revision_id(c2),
189
newbranch.last_revision())
190
newbranch.set_last_revision_info(
191
1, newbranch.lookup_foreign_revision_id(c1))
192
self.assertEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
193
self.assertEqual(newbranch.last_revision(),
194
newbranch.lookup_foreign_revision_id(c1))
197
class FetchFromRemoteTestBase(object):
199
_test_needs_features = [ExecutableFeature('git')]
204
TestCaseWithTransport.setUp(self)
205
self.remote_real = GitRepo.init('remote', mkdir=True)
206
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
207
self.permit_url(self.remote_url)
209
def test_sprout_simple(self):
210
self.remote_real.do_commit(
212
committer=b'committer <committer@example.com>',
213
author=b'author <author@example.com>')
215
remote = ControlDir.open(self.remote_url)
216
self.make_controldir('local', format=self._to_format)
217
local = remote.sprout('local')
219
default_mapping.revision_id_foreign_to_bzr(
220
self.remote_real.head()),
221
local.open_branch().last_revision())
223
def test_sprout_with_tags(self):
224
c1 = self.remote_real.do_commit(
226
committer=b'committer <committer@example.com>',
227
author=b'author <author@example.com>')
228
c2 = self.remote_real.do_commit(
229
message=b'another commit',
230
committer=b'committer <committer@example.com>',
231
author=b'author <author@example.com>',
232
ref=b'refs/tags/another')
233
self.remote_real.refs[b'refs/tags/blah'] = self.remote_real.head()
235
remote = ControlDir.open(self.remote_url)
236
self.make_controldir('local', format=self._to_format)
237
local = remote.sprout('local')
238
local_branch = local.open_branch()
240
default_mapping.revision_id_foreign_to_bzr(c1),
241
local_branch.last_revision())
243
{'blah': local_branch.last_revision(),
244
'another': default_mapping.revision_id_foreign_to_bzr(c2)},
245
local_branch.tags.get_tag_dict())
247
def test_sprout_with_annotated_tag(self):
248
c1 = self.remote_real.do_commit(
250
committer=b'committer <committer@example.com>',
251
author=b'author <author@example.com>')
252
c2 = self.remote_real.do_commit(
253
message=b'another commit',
254
committer=b'committer <committer@example.com>',
255
author=b'author <author@example.com>',
256
ref=b'refs/heads/another')
257
porcelain.tag_create(
260
author=b'author <author@example.com>',
262
tag_time=int(time.time()),
265
message=b"Annotated tag")
267
remote = ControlDir.open(self.remote_url)
268
self.make_controldir('local', format=self._to_format)
269
local = remote.sprout(
270
'local', revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
271
local_branch = local.open_branch()
273
default_mapping.revision_id_foreign_to_bzr(c1),
274
local_branch.last_revision())
276
{'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
277
local_branch.tags.get_tag_dict())
279
def test_sprout_with_annotated_tag_unreferenced(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
porcelain.tag_create(
291
author=b'author <author@example.com>',
293
tag_time=int(time.time()),
296
message=b"Annotated tag")
298
remote = ControlDir.open(self.remote_url)
299
self.make_controldir('local', format=self._to_format)
300
local = remote.sprout(
302
revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
303
local_branch = local.open_branch()
305
default_mapping.revision_id_foreign_to_bzr(c1),
306
local_branch.last_revision())
308
{'blah': default_mapping.revision_id_foreign_to_bzr(c1)},
309
local_branch.tags.get_tag_dict())
312
class FetchFromRemoteToBzrTests(FetchFromRemoteTestBase, TestCaseWithTransport):
317
class FetchFromRemoteToGitTests(FetchFromRemoteTestBase, TestCaseWithTransport):
322
class PushToRemoteBase(object):
324
_test_needs_features = [ExecutableFeature('git')]
329
TestCaseWithTransport.setUp(self)
330
self.remote_real = GitRepo.init('remote', mkdir=True)
331
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
332
self.permit_url(self.remote_url)
334
def test_push_branch_new(self):
335
remote = ControlDir.open(self.remote_url)
336
wt = self.make_branch_and_tree('local', format=self._from_format)
337
self.build_tree(['local/blah'])
339
revid = wt.commit('blah')
341
if self._from_format == 'git':
342
result = remote.push_branch(wt.branch, name='newbranch')
344
result = remote.push_branch(
345
wt.branch, lossy=True, name='newbranch')
347
self.assertEqual(0, result.old_revno)
348
if self._from_format == 'git':
349
self.assertEqual(1, result.new_revno)
351
self.assertIs(None, result.new_revno)
353
result.report(BytesIO())
356
{b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
358
self.remote_real.get_refs())
360
def test_push_branch_symref(self):
361
cfg = self.remote_real.get_config()
362
cfg.set((b'core', ), b'bare', True)
364
self.remote_real.refs.set_symbolic_ref(b'HEAD', b'refs/heads/master')
365
c1 = self.remote_real.do_commit(
367
committer=b'committer <committer@example.com>',
368
author=b'author <author@example.com>',
369
ref=b'refs/heads/master')
370
remote = ControlDir.open(self.remote_url)
371
wt = self.make_branch_and_tree('local', format=self._from_format)
372
self.build_tree(['local/blah'])
374
revid = wt.commit('blah')
376
if self._from_format == 'git':
377
result = remote.push_branch(wt.branch, overwrite=True)
379
result = remote.push_branch(wt.branch, lossy=True, overwrite=True)
381
self.assertEqual(None, result.old_revno)
382
if self._from_format == 'git':
383
self.assertEqual(1, result.new_revno)
385
self.assertIs(None, result.new_revno)
387
result.report(BytesIO())
391
b'HEAD': self.remote_real.refs[b'refs/heads/master'],
392
b'refs/heads/master': self.remote_real.refs[b'refs/heads/master'],
394
self.remote_real.get_refs())
396
def test_push_branch_new_with_tags(self):
397
remote = ControlDir.open(self.remote_url)
398
builder = self.make_branch_builder('local', format=self._from_format)
399
builder.start_series()
400
rev_1 = builder.build_snapshot(None, [
401
('add', ('', None, 'directory', '')),
402
('add', ('filename', None, 'file', b'content'))])
403
rev_2 = builder.build_snapshot(
404
[rev_1], [('modify', ('filename', b'new-content\n'))])
405
rev_3 = builder.build_snapshot(
406
[rev_1], [('modify', ('filename', b'new-new-content\n'))])
407
builder.finish_series()
408
branch = builder.get_branch()
410
branch.tags.set_tag('atag', rev_2)
411
except TagsNotSupported:
412
raise TestNotApplicable('source format does not support tags')
414
branch.get_config_stack().set('branch.fetch_tags', True)
415
if self._from_format == 'git':
416
result = remote.push_branch(branch, name='newbranch')
418
result = remote.push_branch(
419
branch, lossy=True, name='newbranch')
421
self.assertEqual(0, result.old_revno)
422
if self._from_format == 'git':
423
self.assertEqual(2, result.new_revno)
425
self.assertIs(None, result.new_revno)
427
result.report(BytesIO())
430
{b'refs/heads/newbranch', b'refs/tags/atag'},
431
set(self.remote_real.get_refs().keys()))
434
c1 = self.remote_real.do_commit(
436
committer=b'committer <committer@example.com>',
437
author=b'author <author@example.com>')
439
remote = ControlDir.open(self.remote_url)
440
self.make_controldir('local', format=self._from_format)
441
local = remote.sprout('local')
442
self.build_tree(['local/blah'])
443
wt = local.open_workingtree()
445
revid = wt.commit('blah')
446
wt.branch.tags.set_tag('sometag', revid)
447
wt.branch.get_config_stack().set('branch.fetch_tags', True)
449
if self._from_format == 'git':
450
result = wt.branch.push(remote.create_branch('newbranch'))
452
result = wt.branch.push(
453
remote.create_branch('newbranch'), lossy=True)
455
self.assertEqual(0, result.old_revno)
456
self.assertEqual(2, result.new_revno)
458
result.report(BytesIO())
461
{b'refs/heads/master': self.remote_real.head(),
462
b'HEAD': self.remote_real.head(),
463
b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
464
b'refs/tags/sometag': self.remote_real.refs[b'refs/heads/newbranch'],
466
self.remote_real.get_refs())
468
def test_push_diverged(self):
469
c1 = self.remote_real.do_commit(
471
committer=b'committer <committer@example.com>',
472
author=b'author <author@example.com>',
473
ref=b'refs/heads/newbranch')
475
remote = ControlDir.open(self.remote_url)
476
wt = self.make_branch_and_tree('local', format=self._from_format)
477
self.build_tree(['local/blah'])
479
revid = wt.commit('blah')
481
newbranch = remote.open_branch('newbranch')
482
if self._from_format == 'git':
483
self.assertRaises(DivergedBranches, wt.branch.push, newbranch)
485
self.assertRaises(DivergedBranches, wt.branch.push,
486
newbranch, lossy=True)
489
{b'refs/heads/newbranch': c1},
490
self.remote_real.get_refs())
492
if self._from_format == 'git':
493
wt.branch.push(newbranch, overwrite=True)
495
wt.branch.push(newbranch, lossy=True, overwrite=True)
497
self.assertNotEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
500
class PushToRemoteFromBzrTests(PushToRemoteBase, TestCaseWithTransport):
505
class PushToRemoteFromGitTests(PushToRemoteBase, TestCaseWithTransport):
510
class RemoteControlDirTests(TestCaseWithTransport):
512
_test_needs_features = [ExecutableFeature('git')]
515
TestCaseWithTransport.setUp(self)
516
self.remote_real = GitRepo.init('remote', mkdir=True)
517
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
518
self.permit_url(self.remote_url)
520
def test_remove_branch(self):
521
c1 = self.remote_real.do_commit(
523
committer=b'committer <committer@example.com>',
524
author=b'author <author@example.com>')
525
c2 = self.remote_real.do_commit(
526
message=b'another commit',
527
committer=b'committer <committer@example.com>',
528
author=b'author <author@example.com>',
529
ref=b'refs/heads/blah')
531
remote = ControlDir.open(self.remote_url)
532
remote.destroy_branch(name='blah')
534
self.remote_real.get_refs(),
535
{b'refs/heads/master': self.remote_real.head(),
536
b'HEAD': self.remote_real.head(),
539
def test_list_branches(self):
540
c1 = self.remote_real.do_commit(
542
committer=b'committer <committer@example.com>',
543
author=b'author <author@example.com>')
544
c2 = self.remote_real.do_commit(
545
message=b'another commit',
546
committer=b'committer <committer@example.com>',
547
author=b'author <author@example.com>',
548
ref=b'refs/heads/blah')
550
remote = ControlDir.open(self.remote_url)
552
set(['master', 'blah', 'master']),
553
set([b.name for b in remote.list_branches()]))
555
def test_get_branches(self):
556
c1 = self.remote_real.do_commit(
558
committer=b'committer <committer@example.com>',
559
author=b'author <author@example.com>')
560
c2 = self.remote_real.do_commit(
561
message=b'another commit',
562
committer=b'committer <committer@example.com>',
563
author=b'author <author@example.com>',
564
ref=b'refs/heads/blah')
566
remote = ControlDir.open(self.remote_url)
568
{'': 'master', 'blah': 'blah', 'master': 'master'},
569
{n: b.name for (n, b) in remote.get_branches().items()})
571
set(['', 'blah', 'master']), set(remote.branch_names()))
573
def test_remove_tag(self):
574
c1 = self.remote_real.do_commit(
576
committer=b'committer <committer@example.com>',
577
author=b'author <author@example.com>')
578
c2 = self.remote_real.do_commit(
579
message=b'another commit',
580
committer=b'committer <committer@example.com>',
581
author=b'author <author@example.com>',
582
ref=b'refs/tags/blah')
584
remote = ControlDir.open(self.remote_url)
585
remote_branch = remote.open_branch()
586
remote_branch.tags.delete_tag('blah')
587
self.assertRaises(NoSuchTag, remote_branch.tags.delete_tag, 'blah')
589
self.remote_real.get_refs(),
590
{b'refs/heads/master': self.remote_real.head(),
591
b'HEAD': self.remote_real.head(),
594
def test_set_tag(self):
595
c1 = self.remote_real.do_commit(
597
committer=b'committer <committer@example.com>',
598
author=b'author <author@example.com>')
599
c2 = self.remote_real.do_commit(
600
message=b'another commit',
601
committer=b'committer <committer@example.com>',
602
author=b'author <author@example.com>')
604
remote = ControlDir.open(self.remote_url)
605
remote.open_branch().tags.set_tag(
606
b'blah', default_mapping.revision_id_foreign_to_bzr(c1))
608
self.remote_real.get_refs(),
609
{b'refs/heads/master': self.remote_real.head(),
610
b'refs/tags/blah': c1,
611
b'HEAD': self.remote_real.head(),
614
def test_annotated_tag(self):
615
c1 = self.remote_real.do_commit(
617
committer=b'committer <committer@example.com>',
618
author=b'author <author@example.com>')
619
c2 = self.remote_real.do_commit(
620
message=b'another commit',
621
committer=b'committer <committer@example.com>',
622
author=b'author <author@example.com>')
624
porcelain.tag_create(
627
author=b'author <author@example.com>',
629
tag_time=int(time.time()),
632
message=b"Annotated tag")
634
remote = ControlDir.open(self.remote_url)
635
remote_branch = remote.open_branch()
637
'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
638
remote_branch.tags.get_tag_dict())
640
def test_get_branch_reference(self):
641
c1 = self.remote_real.do_commit(
643
committer=b'committer <committer@example.com>',
644
author=b'author <author@example.com>')
645
c2 = self.remote_real.do_commit(
646
message=b'another commit',
647
committer=b'committer <committer@example.com>',
648
author=b'author <author@example.com>')
650
remote = ControlDir.open(self.remote_url)
651
self.assertEqual(b'refs/heads/master', remote.get_branch_reference(''))
652
self.assertEqual(None, remote.get_branch_reference('master'))
654
def test_get_branch_nick(self):
655
c1 = self.remote_real.do_commit(
657
committer=b'committer <committer@example.com>',
658
author=b'author <author@example.com>')
659
remote = ControlDir.open(self.remote_url)
660
self.assertEqual('master', remote.open_branch().nick)
663
class GitUrlAndPathFromTransportTests(TestCase):
666
split_url = _git_url_and_path_from_transport('file:///home/blah')
667
self.assertEqual(split_url.scheme, 'file')
668
self.assertEqual(split_url.path, '/home/blah')
670
def test_file_segment_params(self):
671
split_url = _git_url_and_path_from_transport('file:///home/blah,branch=master')
672
self.assertEqual(split_url.scheme, 'file')
673
self.assertEqual(split_url.path, '/home/blah')
675
def test_git_smart(self):
676
split_url = _git_url_and_path_from_transport(
677
'git://github.com/dulwich/dulwich,branch=master')
678
self.assertEqual(split_url.scheme, 'git')
679
self.assertEqual(split_url.path, '/dulwich/dulwich')
681
def test_https(self):
682
split_url = _git_url_and_path_from_transport(
683
'https://github.com/dulwich/dulwich')
684
self.assertEqual(split_url.scheme, 'https')
685
self.assertEqual(split_url.path, '/dulwich/dulwich')
687
def test_https_segment_params(self):
688
split_url = _git_url_and_path_from_transport(
689
'https://github.com/dulwich/dulwich,branch=master')
690
self.assertEqual(split_url.scheme, 'https')
691
self.assertEqual(split_url.path, '/dulwich/dulwich')
added
removed
breezy/git/tests/test_remote.py
