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):
148
TestCaseWithTransport.setUp(self)
149
self.remote_real = GitRepo.init('remote', mkdir=True)
150
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
151
self.permit_url(self.remote_url)
153
def test_set_last_revision_info(self):
154
c1 = self.remote_real.do_commit(
155
message=b'message 1',
156
committer=b'committer <committer@example.com>',
157
author=b'author <author@example.com>',
158
ref=b'refs/heads/newbranch')
159
c2 = self.remote_real.do_commit(
160
message=b'message 2',
161
committer=b'committer <committer@example.com>',
162
author=b'author <author@example.com>',
163
ref=b'refs/heads/newbranch')
165
remote = ControlDir.open(self.remote_url)
166
newbranch = remote.open_branch('newbranch')
167
self.assertEqual(newbranch.lookup_foreign_revision_id(c2),
168
newbranch.last_revision())
169
newbranch.set_last_revision_info(
170
1, newbranch.lookup_foreign_revision_id(c1))
171
self.assertEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
172
self.assertEqual(newbranch.last_revision(),
173
newbranch.lookup_foreign_revision_id(c1))
176
class FetchFromRemoteTestBase(object):
178
_test_needs_features = [ExecutableFeature('git')]
183
TestCaseWithTransport.setUp(self)
184
self.remote_real = GitRepo.init('remote', mkdir=True)
185
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
186
self.permit_url(self.remote_url)
188
def test_sprout_simple(self):
189
self.remote_real.do_commit(
191
committer=b'committer <committer@example.com>',
192
author=b'author <author@example.com>')
194
remote = ControlDir.open(self.remote_url)
195
self.make_controldir('local', format=self._to_format)
196
local = remote.sprout('local')
198
default_mapping.revision_id_foreign_to_bzr(
199
self.remote_real.head()),
200
local.open_branch().last_revision())
202
def test_sprout_with_tags(self):
203
c1 = self.remote_real.do_commit(
205
committer=b'committer <committer@example.com>',
206
author=b'author <author@example.com>')
207
c2 = self.remote_real.do_commit(
208
message=b'another commit',
209
committer=b'committer <committer@example.com>',
210
author=b'author <author@example.com>',
211
ref=b'refs/tags/another')
212
self.remote_real.refs[b'refs/tags/blah'] = self.remote_real.head()
214
remote = ControlDir.open(self.remote_url)
215
self.make_controldir('local', format=self._to_format)
216
local = remote.sprout('local')
217
local_branch = local.open_branch()
219
default_mapping.revision_id_foreign_to_bzr(c1),
220
local_branch.last_revision())
222
{'blah': local_branch.last_revision(),
223
'another': default_mapping.revision_id_foreign_to_bzr(c2)},
224
local_branch.tags.get_tag_dict())
226
def test_sprout_with_annotated_tag(self):
227
c1 = self.remote_real.do_commit(
229
committer=b'committer <committer@example.com>',
230
author=b'author <author@example.com>')
231
c2 = self.remote_real.do_commit(
232
message=b'another commit',
233
committer=b'committer <committer@example.com>',
234
author=b'author <author@example.com>',
235
ref=b'refs/heads/another')
236
porcelain.tag_create(
239
author=b'author <author@example.com>',
241
tag_time=int(time.time()),
244
message=b"Annotated tag")
246
remote = ControlDir.open(self.remote_url)
247
self.make_controldir('local', format=self._to_format)
248
local = remote.sprout(
249
'local', revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
250
local_branch = local.open_branch()
252
default_mapping.revision_id_foreign_to_bzr(c1),
253
local_branch.last_revision())
255
{'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
256
local_branch.tags.get_tag_dict())
258
def test_sprout_with_annotated_tag_unreferenced(self):
259
c1 = self.remote_real.do_commit(
261
committer=b'committer <committer@example.com>',
262
author=b'author <author@example.com>')
263
c2 = self.remote_real.do_commit(
264
message=b'another commit',
265
committer=b'committer <committer@example.com>',
266
author=b'author <author@example.com>')
267
porcelain.tag_create(
270
author=b'author <author@example.com>',
272
tag_time=int(time.time()),
275
message=b"Annotated tag")
277
remote = ControlDir.open(self.remote_url)
278
self.make_controldir('local', format=self._to_format)
279
local = remote.sprout(
281
revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
282
local_branch = local.open_branch()
284
default_mapping.revision_id_foreign_to_bzr(c1),
285
local_branch.last_revision())
287
{'blah': default_mapping.revision_id_foreign_to_bzr(c1)},
288
local_branch.tags.get_tag_dict())
291
class FetchFromRemoteToBzrTests(FetchFromRemoteTestBase, TestCaseWithTransport):
296
class FetchFromRemoteToGitTests(FetchFromRemoteTestBase, TestCaseWithTransport):
301
class PushToRemoteBase(object):
303
_test_needs_features = [ExecutableFeature('git')]
308
TestCaseWithTransport.setUp(self)
309
self.remote_real = GitRepo.init('remote', mkdir=True)
310
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
311
self.permit_url(self.remote_url)
313
def test_push_branch_new(self):
314
remote = ControlDir.open(self.remote_url)
315
wt = self.make_branch_and_tree('local', format=self._from_format)
316
self.build_tree(['local/blah'])
318
revid = wt.commit('blah')
320
if self._from_format == 'git':
321
result = remote.push_branch(wt.branch, name='newbranch')
323
result = remote.push_branch(
324
wt.branch, lossy=True, name='newbranch')
326
self.assertEqual(0, result.old_revno)
327
if self._from_format == 'git':
328
self.assertEqual(1, result.new_revno)
330
self.assertIs(None, result.new_revno)
332
result.report(BytesIO())
335
{b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
337
self.remote_real.get_refs())
340
c1 = self.remote_real.do_commit(
342
committer=b'committer <committer@example.com>',
343
author=b'author <author@example.com>')
345
remote = ControlDir.open(self.remote_url)
346
self.make_controldir('local', format=self._from_format)
347
local = remote.sprout('local')
348
self.build_tree(['local/blah'])
349
wt = local.open_workingtree()
351
revid = wt.commit('blah')
352
wt.branch.tags.set_tag('sometag', revid)
353
wt.branch.get_config_stack().set('branch.fetch_tags', True)
355
if self._from_format == 'git':
356
result = wt.branch.push(remote.create_branch('newbranch'))
358
result = wt.branch.push(
359
remote.create_branch('newbranch'), lossy=True)
361
self.assertEqual(0, result.old_revno)
362
self.assertEqual(2, result.new_revno)
364
result.report(BytesIO())
367
{b'refs/heads/master': self.remote_real.head(),
368
b'HEAD': self.remote_real.head(),
369
b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
370
b'refs/tags/sometag': self.remote_real.refs[b'refs/heads/newbranch'],
372
self.remote_real.get_refs())
374
def test_push_diverged(self):
375
c1 = self.remote_real.do_commit(
377
committer=b'committer <committer@example.com>',
378
author=b'author <author@example.com>',
379
ref=b'refs/heads/newbranch')
381
remote = ControlDir.open(self.remote_url)
382
wt = self.make_branch_and_tree('local', format=self._from_format)
383
self.build_tree(['local/blah'])
385
revid = wt.commit('blah')
387
newbranch = remote.open_branch('newbranch')
388
if self._from_format == 'git':
389
self.assertRaises(DivergedBranches, wt.branch.push, newbranch)
391
self.assertRaises(DivergedBranches, wt.branch.push,
392
newbranch, lossy=True)
395
{b'refs/heads/newbranch': c1},
396
self.remote_real.get_refs())
398
if self._from_format == 'git':
399
wt.branch.push(newbranch, overwrite=True)
401
wt.branch.push(newbranch, lossy=True, overwrite=True)
403
self.assertNotEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
406
class PushToRemoteFromBzrTests(PushToRemoteBase, TestCaseWithTransport):
411
class PushToRemoteFromGitTests(PushToRemoteBase, TestCaseWithTransport):
416
class RemoteControlDirTests(TestCaseWithTransport):
418
_test_needs_features = [ExecutableFeature('git')]
421
TestCaseWithTransport.setUp(self)
422
self.remote_real = GitRepo.init('remote', mkdir=True)
423
self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
424
self.permit_url(self.remote_url)
426
def test_remove_branch(self):
427
c1 = self.remote_real.do_commit(
429
committer=b'committer <committer@example.com>',
430
author=b'author <author@example.com>')
431
c2 = self.remote_real.do_commit(
432
message=b'another commit',
433
committer=b'committer <committer@example.com>',
434
author=b'author <author@example.com>',
435
ref=b'refs/heads/blah')
437
remote = ControlDir.open(self.remote_url)
438
remote.destroy_branch(name='blah')
440
self.remote_real.get_refs(),
441
{b'refs/heads/master': self.remote_real.head(),
442
b'HEAD': self.remote_real.head(),
445
def test_list_branches(self):
446
c1 = self.remote_real.do_commit(
448
committer=b'committer <committer@example.com>',
449
author=b'author <author@example.com>')
450
c2 = self.remote_real.do_commit(
451
message=b'another commit',
452
committer=b'committer <committer@example.com>',
453
author=b'author <author@example.com>',
454
ref=b'refs/heads/blah')
456
remote = ControlDir.open(self.remote_url)
458
set(['master', 'blah', 'master']),
459
set([b.name for b in remote.list_branches()]))
461
def test_get_branches(self):
462
c1 = self.remote_real.do_commit(
464
committer=b'committer <committer@example.com>',
465
author=b'author <author@example.com>')
466
c2 = self.remote_real.do_commit(
467
message=b'another commit',
468
committer=b'committer <committer@example.com>',
469
author=b'author <author@example.com>',
470
ref=b'refs/heads/blah')
472
remote = ControlDir.open(self.remote_url)
474
{'': 'master', 'blah': 'blah', 'master': 'master'},
475
{n: b.name for (n, b) in remote.get_branches().items()})
477
def test_remove_tag(self):
478
c1 = self.remote_real.do_commit(
480
committer=b'committer <committer@example.com>',
481
author=b'author <author@example.com>')
482
c2 = self.remote_real.do_commit(
483
message=b'another commit',
484
committer=b'committer <committer@example.com>',
485
author=b'author <author@example.com>',
486
ref=b'refs/tags/blah')
488
remote = ControlDir.open(self.remote_url)
489
remote_branch = remote.open_branch()
490
remote_branch.tags.delete_tag('blah')
491
self.assertRaises(NoSuchTag, remote_branch.tags.delete_tag, 'blah')
493
self.remote_real.get_refs(),
494
{b'refs/heads/master': self.remote_real.head(),
495
b'HEAD': self.remote_real.head(),
498
def test_set_tag(self):
499
c1 = self.remote_real.do_commit(
501
committer=b'committer <committer@example.com>',
502
author=b'author <author@example.com>')
503
c2 = self.remote_real.do_commit(
504
message=b'another commit',
505
committer=b'committer <committer@example.com>',
506
author=b'author <author@example.com>')
508
remote = ControlDir.open(self.remote_url)
509
remote.open_branch().tags.set_tag(
510
b'blah', default_mapping.revision_id_foreign_to_bzr(c1))
512
self.remote_real.get_refs(),
513
{b'refs/heads/master': self.remote_real.head(),
514
b'refs/tags/blah': c1,
515
b'HEAD': self.remote_real.head(),
518
def test_annotated_tag(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>')
528
porcelain.tag_create(
531
author=b'author <author@example.com>',
533
tag_time=int(time.time()),
536
message=b"Annotated tag")
538
remote = ControlDir.open(self.remote_url)
539
remote_branch = remote.open_branch()
541
'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
542
remote_branch.tags.get_tag_dict())
544
def test_get_branch_reference(self):
545
c1 = self.remote_real.do_commit(
547
committer=b'committer <committer@example.com>',
548
author=b'author <author@example.com>')
549
c2 = self.remote_real.do_commit(
550
message=b'another commit',
551
committer=b'committer <committer@example.com>',
552
author=b'author <author@example.com>')
554
remote = ControlDir.open(self.remote_url)
555
self.assertEqual(b'refs/heads/master', remote.get_branch_reference(''))
556
self.assertEqual(None, remote.get_branch_reference('master'))
558
def test_get_branch_nick(self):
559
c1 = self.remote_real.do_commit(
561
committer=b'committer <committer@example.com>',
562
author=b'author <author@example.com>')
563
remote = ControlDir.open(self.remote_url)
564
self.assertEqual('master', remote.open_branch().nick)