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,
49
from dulwich import porcelain
50
from dulwich.repo import Repo as GitRepo
53
class SplitUrlTests(TestCase):
55
def test_simple(self):
56
self.assertEqual(("foo", None, None, "/bar"),
57
split_git_url("git://foo/bar"))
60
self.assertEqual(("foo", 343, None, "/bar"),
61
split_git_url("git://foo:343/bar"))
63
def test_username(self):
64
self.assertEqual(("foo", None, "la", "/bar"),
65
split_git_url("git://la@foo/bar"))
67
def test_nopath(self):
68
self.assertEqual(("foo", None, None, "/"),
69
split_git_url("git://foo/"))
71
def test_slashpath(self):
72
self.assertEqual(("foo", None, None, "//bar"),
73
split_git_url("git://foo//bar"))
75
def test_homedir(self):
76
self.assertEqual(("foo", None, None, "~bar"),
77
split_git_url("git://foo/~bar"))
80
class ParseGitErrorTests(TestCase):
82
def test_unknown(self):
83
e = parse_git_error("url", "foo")
84
self.assertIsInstance(e, RemoteGitError)
86
def test_notbrancherror(self):
87
e = parse_git_error("url", "\n Could not find Repository foo/bar")
88
self.assertIsInstance(e, NotBranchError)
90
def test_notbrancherror_launchpad(self):
91
e = parse_git_error("url", "Repository 'foo/bar' not found.")
92
self.assertIsInstance(e, NotBranchError)
94
def test_notbrancherror_github(self):
95
e = parse_git_error("url", "Repository not found.\n")
96
self.assertIsInstance(e, NotBranchError)
98
def test_notbrancherror_normal(self):
100
"url", "fatal: '/srv/git/lintian-brush' does not appear to be a git repository")
101
self.assertIsInstance(e, NotBranchError)
103
def test_head_update(self):
104
e = parse_git_error("url", "HEAD failed to update\n")
105
self.assertIsInstance(e, HeadUpdateFailed)
107
def test_permission_dnied(self):
110
"access denied or repository not exported: /debian/altermime.git")
111
self.assertIsInstance(e, PermissionDenied)
113
def test_permission_denied_gitlab(self):
116
'GitLab: You are not allowed to push code to this project.\n')
117
self.assertIsInstance(e, PermissionDenied)
119
def test_permission_denied_github(self):
122
'Permission to porridge/gaduhistory.git denied to jelmer.')
123
self.assertIsInstance(e, PermissionDenied)
124
self.assertEqual(e.path, 'porridge/gaduhistory.git')
125
self.assertEqual(e.extra, ': denied to jelmer')
128
class TestRemoteGitBranchFormat(TestCase):
131
super(TestRemoteGitBranchFormat, self).setUp()
132
self.format = RemoteGitBranchFormat()
134
def test_get_format_description(self):
135
self.assertEqual("Remote Git Branch",
136
self.format.get_format_description())
138
def test_get_network_name(self):
139
self.assertEqual(b"git", self.format.network_name())
141
def test_supports_tags(self):
142
self.assertTrue(self.format.supports_tags())
145
class TestRemoteGitBranch(TestCaseWithTransport):
147
_test_needs_features = [ExecutableFeature('git')]
150
TestCaseWithTransport.setUp(self)
151
self.remote_real = GitRepo.init('remote', mkdir=True)
152
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
153
self.permit_url(self.remote_url)
155
def test_set_last_revision_info(self):
156
c1 = self.remote_real.do_commit(
157
message=b'message 1',
158
committer=b'committer <committer@example.com>',
159
author=b'author <author@example.com>',
160
ref=b'refs/heads/newbranch')
161
c2 = self.remote_real.do_commit(
162
message=b'message 2',
163
committer=b'committer <committer@example.com>',
164
author=b'author <author@example.com>',
165
ref=b'refs/heads/newbranch')
167
remote = ControlDir.open(self.remote_url)
168
newbranch = remote.open_branch('newbranch')
169
self.assertEqual(newbranch.lookup_foreign_revision_id(c2),
170
newbranch.last_revision())
171
newbranch.set_last_revision_info(
172
1, newbranch.lookup_foreign_revision_id(c1))
173
self.assertEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
174
self.assertEqual(newbranch.last_revision(),
175
newbranch.lookup_foreign_revision_id(c1))
178
class FetchFromRemoteTestBase(object):
180
_test_needs_features = [ExecutableFeature('git')]
185
TestCaseWithTransport.setUp(self)
186
self.remote_real = GitRepo.init('remote', mkdir=True)
187
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
188
self.permit_url(self.remote_url)
190
def test_sprout_simple(self):
191
self.remote_real.do_commit(
193
committer=b'committer <committer@example.com>',
194
author=b'author <author@example.com>')
196
remote = ControlDir.open(self.remote_url)
197
self.make_controldir('local', format=self._to_format)
198
local = remote.sprout('local')
200
default_mapping.revision_id_foreign_to_bzr(
201
self.remote_real.head()),
202
local.open_branch().last_revision())
204
def test_sprout_with_tags(self):
205
c1 = self.remote_real.do_commit(
207
committer=b'committer <committer@example.com>',
208
author=b'author <author@example.com>')
209
c2 = self.remote_real.do_commit(
210
message=b'another commit',
211
committer=b'committer <committer@example.com>',
212
author=b'author <author@example.com>',
213
ref=b'refs/tags/another')
214
self.remote_real.refs[b'refs/tags/blah'] = self.remote_real.head()
216
remote = ControlDir.open(self.remote_url)
217
self.make_controldir('local', format=self._to_format)
218
local = remote.sprout('local')
219
local_branch = local.open_branch()
221
default_mapping.revision_id_foreign_to_bzr(c1),
222
local_branch.last_revision())
224
{'blah': local_branch.last_revision(),
225
'another': default_mapping.revision_id_foreign_to_bzr(c2)},
226
local_branch.tags.get_tag_dict())
228
def test_sprout_with_annotated_tag(self):
229
c1 = self.remote_real.do_commit(
231
committer=b'committer <committer@example.com>',
232
author=b'author <author@example.com>')
233
c2 = self.remote_real.do_commit(
234
message=b'another commit',
235
committer=b'committer <committer@example.com>',
236
author=b'author <author@example.com>',
237
ref=b'refs/heads/another')
238
porcelain.tag_create(
241
author=b'author <author@example.com>',
243
tag_time=int(time.time()),
246
message=b"Annotated tag")
248
remote = ControlDir.open(self.remote_url)
249
self.make_controldir('local', format=self._to_format)
250
local = remote.sprout(
251
'local', revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
252
local_branch = local.open_branch()
254
default_mapping.revision_id_foreign_to_bzr(c1),
255
local_branch.last_revision())
257
{'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
258
local_branch.tags.get_tag_dict())
260
def test_sprout_with_annotated_tag_unreferenced(self):
261
c1 = self.remote_real.do_commit(
263
committer=b'committer <committer@example.com>',
264
author=b'author <author@example.com>')
265
c2 = self.remote_real.do_commit(
266
message=b'another commit',
267
committer=b'committer <committer@example.com>',
268
author=b'author <author@example.com>')
269
porcelain.tag_create(
272
author=b'author <author@example.com>',
274
tag_time=int(time.time()),
277
message=b"Annotated tag")
279
remote = ControlDir.open(self.remote_url)
280
self.make_controldir('local', format=self._to_format)
281
local = remote.sprout(
283
revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
284
local_branch = local.open_branch()
286
default_mapping.revision_id_foreign_to_bzr(c1),
287
local_branch.last_revision())
289
{'blah': default_mapping.revision_id_foreign_to_bzr(c1)},
290
local_branch.tags.get_tag_dict())
293
class FetchFromRemoteToBzrTests(FetchFromRemoteTestBase, TestCaseWithTransport):
298
class FetchFromRemoteToGitTests(FetchFromRemoteTestBase, TestCaseWithTransport):
303
class PushToRemoteBase(object):
305
_test_needs_features = [ExecutableFeature('git')]
310
TestCaseWithTransport.setUp(self)
311
self.remote_real = GitRepo.init('remote', mkdir=True)
312
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
313
self.permit_url(self.remote_url)
315
def test_push_branch_new(self):
316
remote = ControlDir.open(self.remote_url)
317
wt = self.make_branch_and_tree('local', format=self._from_format)
318
self.build_tree(['local/blah'])
320
revid = wt.commit('blah')
322
if self._from_format == 'git':
323
result = remote.push_branch(wt.branch, name='newbranch')
325
result = remote.push_branch(
326
wt.branch, lossy=True, name='newbranch')
328
self.assertEqual(0, result.old_revno)
329
if self._from_format == 'git':
330
self.assertEqual(1, result.new_revno)
332
self.assertIs(None, result.new_revno)
334
result.report(BytesIO())
337
{b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
339
self.remote_real.get_refs())
341
def test_push_branch_new_with_tags(self):
342
remote = ControlDir.open(self.remote_url)
343
builder = self.make_branch_builder('local', format=self._from_format)
344
builder.start_series()
345
rev_1 = builder.build_snapshot(None, [
346
('add', ('', None, 'directory', '')),
347
('add', ('filename', None, 'file', b'content'))])
348
rev_2 = builder.build_snapshot(
349
[rev_1], [('modify', ('filename', b'new-content\n'))])
350
rev_3 = builder.build_snapshot(
351
[rev_1], [('modify', ('filename', b'new-new-content\n'))])
352
builder.finish_series()
353
branch = builder.get_branch()
355
branch.tags.set_tag('atag', rev_2)
356
except TagsNotSupported:
357
raise TestNotApplicable('source format does not support tags')
359
branch.get_config_stack().set('branch.fetch_tags', True)
360
if self._from_format == 'git':
361
result = remote.push_branch(branch, name='newbranch')
363
result = remote.push_branch(
364
branch, lossy=True, name='newbranch')
366
self.assertEqual(0, result.old_revno)
367
if self._from_format == 'git':
368
self.assertEqual(2, result.new_revno)
370
self.assertIs(None, result.new_revno)
372
result.report(BytesIO())
375
{b'refs/heads/newbranch', b'refs/tags/atag'},
376
set(self.remote_real.get_refs().keys()))
379
c1 = self.remote_real.do_commit(
381
committer=b'committer <committer@example.com>',
382
author=b'author <author@example.com>')
384
remote = ControlDir.open(self.remote_url)
385
self.make_controldir('local', format=self._from_format)
386
local = remote.sprout('local')
387
self.build_tree(['local/blah'])
388
wt = local.open_workingtree()
390
revid = wt.commit('blah')
391
wt.branch.tags.set_tag('sometag', revid)
392
wt.branch.get_config_stack().set('branch.fetch_tags', True)
394
if self._from_format == 'git':
395
result = wt.branch.push(remote.create_branch('newbranch'))
397
result = wt.branch.push(
398
remote.create_branch('newbranch'), lossy=True)
400
self.assertEqual(0, result.old_revno)
401
self.assertEqual(2, result.new_revno)
403
result.report(BytesIO())
406
{b'refs/heads/master': self.remote_real.head(),
407
b'HEAD': self.remote_real.head(),
408
b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
409
b'refs/tags/sometag': self.remote_real.refs[b'refs/heads/newbranch'],
411
self.remote_real.get_refs())
413
def test_push_diverged(self):
414
c1 = self.remote_real.do_commit(
416
committer=b'committer <committer@example.com>',
417
author=b'author <author@example.com>',
418
ref=b'refs/heads/newbranch')
420
remote = ControlDir.open(self.remote_url)
421
wt = self.make_branch_and_tree('local', format=self._from_format)
422
self.build_tree(['local/blah'])
424
revid = wt.commit('blah')
426
newbranch = remote.open_branch('newbranch')
427
if self._from_format == 'git':
428
self.assertRaises(DivergedBranches, wt.branch.push, newbranch)
430
self.assertRaises(DivergedBranches, wt.branch.push,
431
newbranch, lossy=True)
434
{b'refs/heads/newbranch': c1},
435
self.remote_real.get_refs())
437
if self._from_format == 'git':
438
wt.branch.push(newbranch, overwrite=True)
440
wt.branch.push(newbranch, lossy=True, overwrite=True)
442
self.assertNotEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
445
class PushToRemoteFromBzrTests(PushToRemoteBase, TestCaseWithTransport):
450
class PushToRemoteFromGitTests(PushToRemoteBase, TestCaseWithTransport):
455
class RemoteControlDirTests(TestCaseWithTransport):
457
_test_needs_features = [ExecutableFeature('git')]
460
TestCaseWithTransport.setUp(self)
461
self.remote_real = GitRepo.init('remote', mkdir=True)
462
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
463
self.permit_url(self.remote_url)
465
def test_remove_branch(self):
466
c1 = self.remote_real.do_commit(
468
committer=b'committer <committer@example.com>',
469
author=b'author <author@example.com>')
470
c2 = self.remote_real.do_commit(
471
message=b'another commit',
472
committer=b'committer <committer@example.com>',
473
author=b'author <author@example.com>',
474
ref=b'refs/heads/blah')
476
remote = ControlDir.open(self.remote_url)
477
remote.destroy_branch(name='blah')
479
self.remote_real.get_refs(),
480
{b'refs/heads/master': self.remote_real.head(),
481
b'HEAD': self.remote_real.head(),
484
def test_list_branches(self):
485
c1 = self.remote_real.do_commit(
487
committer=b'committer <committer@example.com>',
488
author=b'author <author@example.com>')
489
c2 = self.remote_real.do_commit(
490
message=b'another commit',
491
committer=b'committer <committer@example.com>',
492
author=b'author <author@example.com>',
493
ref=b'refs/heads/blah')
495
remote = ControlDir.open(self.remote_url)
497
set(['master', 'blah', 'master']),
498
set([b.name for b in remote.list_branches()]))
500
def test_get_branches(self):
501
c1 = self.remote_real.do_commit(
503
committer=b'committer <committer@example.com>',
504
author=b'author <author@example.com>')
505
c2 = self.remote_real.do_commit(
506
message=b'another commit',
507
committer=b'committer <committer@example.com>',
508
author=b'author <author@example.com>',
509
ref=b'refs/heads/blah')
511
remote = ControlDir.open(self.remote_url)
513
{'': 'master', 'blah': 'blah', 'master': 'master'},
514
{n: b.name for (n, b) in remote.get_branches().items()})
516
def test_remove_tag(self):
517
c1 = self.remote_real.do_commit(
519
committer=b'committer <committer@example.com>',
520
author=b'author <author@example.com>')
521
c2 = self.remote_real.do_commit(
522
message=b'another commit',
523
committer=b'committer <committer@example.com>',
524
author=b'author <author@example.com>',
525
ref=b'refs/tags/blah')
527
remote = ControlDir.open(self.remote_url)
528
remote_branch = remote.open_branch()
529
remote_branch.tags.delete_tag('blah')
530
self.assertRaises(NoSuchTag, remote_branch.tags.delete_tag, '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_set_tag(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>')
547
remote = ControlDir.open(self.remote_url)
548
remote.open_branch().tags.set_tag(
549
b'blah', default_mapping.revision_id_foreign_to_bzr(c1))
551
self.remote_real.get_refs(),
552
{b'refs/heads/master': self.remote_real.head(),
553
b'refs/tags/blah': c1,
554
b'HEAD': self.remote_real.head(),
557
def test_annotated_tag(self):
558
c1 = self.remote_real.do_commit(
560
committer=b'committer <committer@example.com>',
561
author=b'author <author@example.com>')
562
c2 = self.remote_real.do_commit(
563
message=b'another commit',
564
committer=b'committer <committer@example.com>',
565
author=b'author <author@example.com>')
567
porcelain.tag_create(
570
author=b'author <author@example.com>',
572
tag_time=int(time.time()),
575
message=b"Annotated tag")
577
remote = ControlDir.open(self.remote_url)
578
remote_branch = remote.open_branch()
580
'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
581
remote_branch.tags.get_tag_dict())
583
def test_get_branch_reference(self):
584
c1 = self.remote_real.do_commit(
586
committer=b'committer <committer@example.com>',
587
author=b'author <author@example.com>')
588
c2 = self.remote_real.do_commit(
589
message=b'another commit',
590
committer=b'committer <committer@example.com>',
591
author=b'author <author@example.com>')
593
remote = ControlDir.open(self.remote_url)
594
self.assertEqual(b'refs/heads/master', remote.get_branch_reference(''))
595
self.assertEqual(None, remote.get_branch_reference('master'))
597
def test_get_branch_nick(self):
598
c1 = self.remote_real.do_commit(
600
committer=b'committer <committer@example.com>',
601
author=b'author <author@example.com>')
602
remote = ControlDir.open(self.remote_url)
603
self.assertEqual('master', remote.open_branch().nick)