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_nopath(self):
69
self.assertEqual(("foo", None, None, "/"),
70
split_git_url("git://foo/"))
72
def test_slashpath(self):
73
self.assertEqual(("foo", None, None, "//bar"),
74
split_git_url("git://foo//bar"))
76
def test_homedir(self):
77
self.assertEqual(("foo", None, None, "~bar"),
78
split_git_url("git://foo/~bar"))
81
class ParseGitErrorTests(TestCase):
83
def test_unknown(self):
84
e = parse_git_error("url", "foo")
85
self.assertIsInstance(e, RemoteGitError)
87
def test_notbrancherror(self):
88
e = parse_git_error("url", "\n Could not find Repository foo/bar")
89
self.assertIsInstance(e, NotBranchError)
91
def test_notbrancherror_launchpad(self):
92
e = parse_git_error("url", "Repository 'foo/bar' not found.")
93
self.assertIsInstance(e, NotBranchError)
95
def test_notbrancherror_github(self):
96
e = parse_git_error("url", "Repository not found.\n")
97
self.assertIsInstance(e, NotBranchError)
99
def test_notbrancherror_normal(self):
101
"url", "fatal: '/srv/git/lintian-brush' does not appear to be a git repository")
102
self.assertIsInstance(e, NotBranchError)
104
def test_head_update(self):
105
e = parse_git_error("url", "HEAD failed to update\n")
106
self.assertIsInstance(e, HeadUpdateFailed)
108
def test_permission_dnied(self):
111
"access denied or repository not exported: /debian/altermime.git")
112
self.assertIsInstance(e, PermissionDenied)
114
def test_permission_denied_gitlab(self):
117
'GitLab: You are not allowed to push code to this project.\n')
118
self.assertIsInstance(e, PermissionDenied)
120
def test_permission_denied_github(self):
123
'Permission to porridge/gaduhistory.git denied to jelmer.')
124
self.assertIsInstance(e, PermissionDenied)
125
self.assertEqual(e.path, 'porridge/gaduhistory.git')
126
self.assertEqual(e.extra, ': denied to jelmer')
128
def test_invalid_repo_name(self):
131
"""Gregwar/fatcat/tree/debian is not a valid repository name
132
Email support@github.com for help
134
self.assertIsInstance(e, NotBranchError)
137
class TestRemoteGitBranchFormat(TestCase):
140
super(TestRemoteGitBranchFormat, self).setUp()
141
self.format = RemoteGitBranchFormat()
143
def test_get_format_description(self):
144
self.assertEqual("Remote Git Branch",
145
self.format.get_format_description())
147
def test_get_network_name(self):
148
self.assertEqual(b"git", self.format.network_name())
150
def test_supports_tags(self):
151
self.assertTrue(self.format.supports_tags())
154
class TestRemoteGitBranch(TestCaseWithTransport):
156
_test_needs_features = [ExecutableFeature('git')]
159
TestCaseWithTransport.setUp(self)
160
self.remote_real = GitRepo.init('remote', mkdir=True)
161
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
162
self.permit_url(self.remote_url)
164
def test_set_last_revision_info(self):
165
c1 = self.remote_real.do_commit(
166
message=b'message 1',
167
committer=b'committer <committer@example.com>',
168
author=b'author <author@example.com>',
169
ref=b'refs/heads/newbranch')
170
c2 = self.remote_real.do_commit(
171
message=b'message 2',
172
committer=b'committer <committer@example.com>',
173
author=b'author <author@example.com>',
174
ref=b'refs/heads/newbranch')
176
remote = ControlDir.open(self.remote_url)
177
newbranch = remote.open_branch('newbranch')
178
self.assertEqual(newbranch.lookup_foreign_revision_id(c2),
179
newbranch.last_revision())
180
newbranch.set_last_revision_info(
181
1, newbranch.lookup_foreign_revision_id(c1))
182
self.assertEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
183
self.assertEqual(newbranch.last_revision(),
184
newbranch.lookup_foreign_revision_id(c1))
187
class FetchFromRemoteTestBase(object):
189
_test_needs_features = [ExecutableFeature('git')]
194
TestCaseWithTransport.setUp(self)
195
self.remote_real = GitRepo.init('remote', mkdir=True)
196
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
197
self.permit_url(self.remote_url)
199
def test_sprout_simple(self):
200
self.remote_real.do_commit(
202
committer=b'committer <committer@example.com>',
203
author=b'author <author@example.com>')
205
remote = ControlDir.open(self.remote_url)
206
self.make_controldir('local', format=self._to_format)
207
local = remote.sprout('local')
209
default_mapping.revision_id_foreign_to_bzr(
210
self.remote_real.head()),
211
local.open_branch().last_revision())
213
def test_sprout_with_tags(self):
214
c1 = self.remote_real.do_commit(
216
committer=b'committer <committer@example.com>',
217
author=b'author <author@example.com>')
218
c2 = self.remote_real.do_commit(
219
message=b'another commit',
220
committer=b'committer <committer@example.com>',
221
author=b'author <author@example.com>',
222
ref=b'refs/tags/another')
223
self.remote_real.refs[b'refs/tags/blah'] = self.remote_real.head()
225
remote = ControlDir.open(self.remote_url)
226
self.make_controldir('local', format=self._to_format)
227
local = remote.sprout('local')
228
local_branch = local.open_branch()
230
default_mapping.revision_id_foreign_to_bzr(c1),
231
local_branch.last_revision())
233
{'blah': local_branch.last_revision(),
234
'another': default_mapping.revision_id_foreign_to_bzr(c2)},
235
local_branch.tags.get_tag_dict())
237
def test_sprout_with_annotated_tag(self):
238
c1 = self.remote_real.do_commit(
240
committer=b'committer <committer@example.com>',
241
author=b'author <author@example.com>')
242
c2 = self.remote_real.do_commit(
243
message=b'another commit',
244
committer=b'committer <committer@example.com>',
245
author=b'author <author@example.com>',
246
ref=b'refs/heads/another')
247
porcelain.tag_create(
250
author=b'author <author@example.com>',
252
tag_time=int(time.time()),
255
message=b"Annotated tag")
257
remote = ControlDir.open(self.remote_url)
258
self.make_controldir('local', format=self._to_format)
259
local = remote.sprout(
260
'local', revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
261
local_branch = local.open_branch()
263
default_mapping.revision_id_foreign_to_bzr(c1),
264
local_branch.last_revision())
266
{'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
267
local_branch.tags.get_tag_dict())
269
def test_sprout_with_annotated_tag_unreferenced(self):
270
c1 = self.remote_real.do_commit(
272
committer=b'committer <committer@example.com>',
273
author=b'author <author@example.com>')
274
c2 = self.remote_real.do_commit(
275
message=b'another commit',
276
committer=b'committer <committer@example.com>',
277
author=b'author <author@example.com>')
278
porcelain.tag_create(
281
author=b'author <author@example.com>',
283
tag_time=int(time.time()),
286
message=b"Annotated tag")
288
remote = ControlDir.open(self.remote_url)
289
self.make_controldir('local', format=self._to_format)
290
local = remote.sprout(
292
revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
293
local_branch = local.open_branch()
295
default_mapping.revision_id_foreign_to_bzr(c1),
296
local_branch.last_revision())
298
{'blah': default_mapping.revision_id_foreign_to_bzr(c1)},
299
local_branch.tags.get_tag_dict())
302
class FetchFromRemoteToBzrTests(FetchFromRemoteTestBase, TestCaseWithTransport):
307
class FetchFromRemoteToGitTests(FetchFromRemoteTestBase, TestCaseWithTransport):
312
class PushToRemoteBase(object):
314
_test_needs_features = [ExecutableFeature('git')]
319
TestCaseWithTransport.setUp(self)
320
self.remote_real = GitRepo.init('remote', mkdir=True)
321
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
322
self.permit_url(self.remote_url)
324
def test_push_branch_new(self):
325
remote = ControlDir.open(self.remote_url)
326
wt = self.make_branch_and_tree('local', format=self._from_format)
327
self.build_tree(['local/blah'])
329
revid = wt.commit('blah')
331
if self._from_format == 'git':
332
result = remote.push_branch(wt.branch, name='newbranch')
334
result = remote.push_branch(
335
wt.branch, lossy=True, name='newbranch')
337
self.assertEqual(0, result.old_revno)
338
if self._from_format == 'git':
339
self.assertEqual(1, result.new_revno)
341
self.assertIs(None, result.new_revno)
343
result.report(BytesIO())
346
{b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
348
self.remote_real.get_refs())
350
def test_push_branch_new_with_tags(self):
351
remote = ControlDir.open(self.remote_url)
352
builder = self.make_branch_builder('local', format=self._from_format)
353
builder.start_series()
354
rev_1 = builder.build_snapshot(None, [
355
('add', ('', None, 'directory', '')),
356
('add', ('filename', None, 'file', b'content'))])
357
rev_2 = builder.build_snapshot(
358
[rev_1], [('modify', ('filename', b'new-content\n'))])
359
rev_3 = builder.build_snapshot(
360
[rev_1], [('modify', ('filename', b'new-new-content\n'))])
361
builder.finish_series()
362
branch = builder.get_branch()
364
branch.tags.set_tag('atag', rev_2)
365
except TagsNotSupported:
366
raise TestNotApplicable('source format does not support tags')
368
branch.get_config_stack().set('branch.fetch_tags', True)
369
if self._from_format == 'git':
370
result = remote.push_branch(branch, name='newbranch')
372
result = remote.push_branch(
373
branch, lossy=True, name='newbranch')
375
self.assertEqual(0, result.old_revno)
376
if self._from_format == 'git':
377
self.assertEqual(2, result.new_revno)
379
self.assertIs(None, result.new_revno)
381
result.report(BytesIO())
384
{b'refs/heads/newbranch', b'refs/tags/atag'},
385
set(self.remote_real.get_refs().keys()))
388
c1 = self.remote_real.do_commit(
390
committer=b'committer <committer@example.com>',
391
author=b'author <author@example.com>')
393
remote = ControlDir.open(self.remote_url)
394
self.make_controldir('local', format=self._from_format)
395
local = remote.sprout('local')
396
self.build_tree(['local/blah'])
397
wt = local.open_workingtree()
399
revid = wt.commit('blah')
400
wt.branch.tags.set_tag('sometag', revid)
401
wt.branch.get_config_stack().set('branch.fetch_tags', True)
403
if self._from_format == 'git':
404
result = wt.branch.push(remote.create_branch('newbranch'))
406
result = wt.branch.push(
407
remote.create_branch('newbranch'), lossy=True)
409
self.assertEqual(0, result.old_revno)
410
self.assertEqual(2, result.new_revno)
412
result.report(BytesIO())
415
{b'refs/heads/master': self.remote_real.head(),
416
b'HEAD': self.remote_real.head(),
417
b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
418
b'refs/tags/sometag': self.remote_real.refs[b'refs/heads/newbranch'],
420
self.remote_real.get_refs())
422
def test_push_diverged(self):
423
c1 = self.remote_real.do_commit(
425
committer=b'committer <committer@example.com>',
426
author=b'author <author@example.com>',
427
ref=b'refs/heads/newbranch')
429
remote = ControlDir.open(self.remote_url)
430
wt = self.make_branch_and_tree('local', format=self._from_format)
431
self.build_tree(['local/blah'])
433
revid = wt.commit('blah')
435
newbranch = remote.open_branch('newbranch')
436
if self._from_format == 'git':
437
self.assertRaises(DivergedBranches, wt.branch.push, newbranch)
439
self.assertRaises(DivergedBranches, wt.branch.push,
440
newbranch, lossy=True)
443
{b'refs/heads/newbranch': c1},
444
self.remote_real.get_refs())
446
if self._from_format == 'git':
447
wt.branch.push(newbranch, overwrite=True)
449
wt.branch.push(newbranch, lossy=True, overwrite=True)
451
self.assertNotEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
454
class PushToRemoteFromBzrTests(PushToRemoteBase, TestCaseWithTransport):
459
class PushToRemoteFromGitTests(PushToRemoteBase, TestCaseWithTransport):
464
class RemoteControlDirTests(TestCaseWithTransport):
466
_test_needs_features = [ExecutableFeature('git')]
469
TestCaseWithTransport.setUp(self)
470
self.remote_real = GitRepo.init('remote', mkdir=True)
471
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
472
self.permit_url(self.remote_url)
474
def test_remove_branch(self):
475
c1 = self.remote_real.do_commit(
477
committer=b'committer <committer@example.com>',
478
author=b'author <author@example.com>')
479
c2 = self.remote_real.do_commit(
480
message=b'another commit',
481
committer=b'committer <committer@example.com>',
482
author=b'author <author@example.com>',
483
ref=b'refs/heads/blah')
485
remote = ControlDir.open(self.remote_url)
486
remote.destroy_branch(name='blah')
488
self.remote_real.get_refs(),
489
{b'refs/heads/master': self.remote_real.head(),
490
b'HEAD': self.remote_real.head(),
493
def test_list_branches(self):
494
c1 = self.remote_real.do_commit(
496
committer=b'committer <committer@example.com>',
497
author=b'author <author@example.com>')
498
c2 = self.remote_real.do_commit(
499
message=b'another commit',
500
committer=b'committer <committer@example.com>',
501
author=b'author <author@example.com>',
502
ref=b'refs/heads/blah')
504
remote = ControlDir.open(self.remote_url)
506
set(['master', 'blah', 'master']),
507
set([b.name for b in remote.list_branches()]))
509
def test_get_branches(self):
510
c1 = self.remote_real.do_commit(
512
committer=b'committer <committer@example.com>',
513
author=b'author <author@example.com>')
514
c2 = self.remote_real.do_commit(
515
message=b'another commit',
516
committer=b'committer <committer@example.com>',
517
author=b'author <author@example.com>',
518
ref=b'refs/heads/blah')
520
remote = ControlDir.open(self.remote_url)
522
{'': 'master', 'blah': 'blah', 'master': 'master'},
523
{n: b.name for (n, b) in remote.get_branches().items()})
525
def test_remove_tag(self):
526
c1 = self.remote_real.do_commit(
528
committer=b'committer <committer@example.com>',
529
author=b'author <author@example.com>')
530
c2 = self.remote_real.do_commit(
531
message=b'another commit',
532
committer=b'committer <committer@example.com>',
533
author=b'author <author@example.com>',
534
ref=b'refs/tags/blah')
536
remote = ControlDir.open(self.remote_url)
537
remote_branch = remote.open_branch()
538
remote_branch.tags.delete_tag('blah')
539
self.assertRaises(NoSuchTag, remote_branch.tags.delete_tag, 'blah')
541
self.remote_real.get_refs(),
542
{b'refs/heads/master': self.remote_real.head(),
543
b'HEAD': self.remote_real.head(),
546
def test_set_tag(self):
547
c1 = self.remote_real.do_commit(
549
committer=b'committer <committer@example.com>',
550
author=b'author <author@example.com>')
551
c2 = self.remote_real.do_commit(
552
message=b'another commit',
553
committer=b'committer <committer@example.com>',
554
author=b'author <author@example.com>')
556
remote = ControlDir.open(self.remote_url)
557
remote.open_branch().tags.set_tag(
558
b'blah', default_mapping.revision_id_foreign_to_bzr(c1))
560
self.remote_real.get_refs(),
561
{b'refs/heads/master': self.remote_real.head(),
562
b'refs/tags/blah': c1,
563
b'HEAD': self.remote_real.head(),
566
def test_annotated_tag(self):
567
c1 = self.remote_real.do_commit(
569
committer=b'committer <committer@example.com>',
570
author=b'author <author@example.com>')
571
c2 = self.remote_real.do_commit(
572
message=b'another commit',
573
committer=b'committer <committer@example.com>',
574
author=b'author <author@example.com>')
576
porcelain.tag_create(
579
author=b'author <author@example.com>',
581
tag_time=int(time.time()),
584
message=b"Annotated tag")
586
remote = ControlDir.open(self.remote_url)
587
remote_branch = remote.open_branch()
589
'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
590
remote_branch.tags.get_tag_dict())
592
def test_get_branch_reference(self):
593
c1 = self.remote_real.do_commit(
595
committer=b'committer <committer@example.com>',
596
author=b'author <author@example.com>')
597
c2 = self.remote_real.do_commit(
598
message=b'another commit',
599
committer=b'committer <committer@example.com>',
600
author=b'author <author@example.com>')
602
remote = ControlDir.open(self.remote_url)
603
self.assertEqual(b'refs/heads/master', remote.get_branch_reference(''))
604
self.assertEqual(None, remote.get_branch_reference('master'))
606
def test_get_branch_nick(self):
607
c1 = self.remote_real.do_commit(
609
committer=b'committer <committer@example.com>',
610
author=b'author <author@example.com>')
611
remote = ControlDir.open(self.remote_url)
612
self.assertEqual('master', remote.open_branch().nick)
615
class GitUrlAndPathFromTransportTests(TestCase):
618
split_url = _git_url_and_path_from_transport('file:///home/blah')
619
self.assertEqual(split_url.scheme, 'file')
620
self.assertEqual(split_url.path, '/home/blah')
622
def test_file_segment_params(self):
623
split_url = _git_url_and_path_from_transport('file:///home/blah,branch=master')
624
self.assertEqual(split_url.scheme, 'file')
625
self.assertEqual(split_url.path, '/home/blah')
627
def test_git_smart(self):
628
split_url = _git_url_and_path_from_transport(
629
'git://github.com/dulwich/dulwich,branch=master')
630
self.assertEqual(split_url.scheme, 'git')
631
self.assertEqual(split_url.path, '/dulwich/dulwich')
633
def test_https(self):
634
split_url = _git_url_and_path_from_transport(
635
'https://github.com/dulwich/dulwich')
636
self.assertEqual(split_url.scheme, 'https')
637
self.assertEqual(split_url.path, '/dulwich/dulwich')
639
def test_https_segment_params(self):
640
split_url = _git_url_and_path_from_transport(
641
'https://github.com/dulwich/dulwich,branch=master')
642
self.assertEqual(split_url.scheme, 'https')
643
self.assertEqual(split_url.path, '/dulwich/dulwich')