1# Copyright (C) 2010-2018 Jelmer Vernooij <jelmer@jelmer.uk>
2#
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.
7#
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.
12#
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
16
17"""Test the smart client."""
18
19from io import BytesIO
20
21import os
22import time
23
24from ...controldir import ControlDir
25from ...errors import (
26    DivergedBranches,
27    NotBranchError,
28    NoSuchTag,
29    PermissionDenied,
30    )
31
32from ...tests import (
33    TestCase,
34    TestCaseWithTransport,
35    )
36from ...tests.features import ExecutableFeature
37
38from ..mapping import default_mapping
39from ..remote import (
40    split_git_url,
41    parse_git_error,
42    parse_git_hangup,
43    HeadUpdateFailed,
44    RemoteGitError,
45    RemoteGitBranchFormat,
46    _git_url_and_path_from_transport,
47    )
48
49from dulwich import porcelain
50from dulwich.errors import HangupException
51from dulwich.repo import Repo as GitRepo
52
53
54class SplitUrlTests(TestCase):
55
56    def test_simple(self):
57        self.assertEqual(("foo", None, None, "/bar"),
58                         split_git_url("git://foo/bar"))
59
60    def test_port(self):
61        self.assertEqual(("foo", 343, None, "/bar"),
62                         split_git_url("git://foo:343/bar"))
63
64    def test_username(self):
65        self.assertEqual(("foo", None, "la", "/bar"),
66                         split_git_url("git://la@foo/bar"))
67
68    def test_username_password(self):
69        self.assertEqual(
70            ("foo", None, "la", "/bar"),
71            split_git_url("git://la:passwd@foo/bar"))
72
73    def test_nopath(self):
74        self.assertEqual(("foo", None, None, "/"),
75                         split_git_url("git://foo/"))
76
77    def test_slashpath(self):
78        self.assertEqual(("foo", None, None, "//bar"),
79                         split_git_url("git://foo//bar"))
80
81    def test_homedir(self):
82        self.assertEqual(("foo", None, None, "~bar"),
83                         split_git_url("git://foo/~bar"))
84
85    def test_file(self):
86        self.assertEqual(
87            ("", None, None, "/bar"),
88            split_git_url("file:///bar"))
89
90
91class ParseGitErrorTests(TestCase):
92
93    def test_unknown(self):
94        e = parse_git_error("url", "foo")
95        self.assertIsInstance(e, RemoteGitError)
96
97    def test_notbrancherror(self):
98        e = parse_git_error("url", "\n Could not find Repository foo/bar")
99        self.assertIsInstance(e, NotBranchError)
100
101    def test_notbrancherror_launchpad(self):
102        e = parse_git_error("url", "Repository 'foo/bar' not found.")
103        self.assertIsInstance(e, NotBranchError)
104
105    def test_notbrancherror_github(self):
106        e = parse_git_error("url", "Repository not found.\n")
107        self.assertIsInstance(e, NotBranchError)
108
109    def test_notbrancherror_normal(self):
110        e = parse_git_error(
111            "url", "fatal: '/srv/git/lintian-brush' does not appear to be a git repository")
112        self.assertIsInstance(e, NotBranchError)
113
114    def test_head_update(self):
115        e = parse_git_error("url", "HEAD failed to update\n")
116        self.assertIsInstance(e, HeadUpdateFailed)
117
118    def test_permission_dnied(self):
119        e = parse_git_error(
120            "url",
121            "access denied or repository not exported: /debian/altermime.git")
122        self.assertIsInstance(e, PermissionDenied)
123
124    def test_permission_denied_gitlab(self):
125        e = parse_git_error(
126            "url",
127            'GitLab: You are not allowed to push code to this project.\n')
128        self.assertIsInstance(e, PermissionDenied)
129
130    def test_permission_denied_github(self):
131        e = parse_git_error(
132            "url",
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')
137
138    def test_pre_receive_hook_declined(self):
139        e = parse_git_error(
140            "url",
141            'pre-receive hook declined')
142        self.assertIsInstance(e, PermissionDenied)
143        self.assertEqual(e.path, "url")
144        self.assertEqual(e.extra, ': pre-receive hook declined')
145
146    def test_invalid_repo_name(self):
147        e = parse_git_error(
148            "url",
149            """Gregwar/fatcat/tree/debian is not a valid repository name
150Email support@github.com for help
151""")
152        self.assertIsInstance(e, NotBranchError)
153
154    def test_invalid_git_error(self):
155        self.assertEqual(
156            PermissionDenied(
157                'url',
158                'GitLab: You are not allowed to push code to protected '
159                'branches on this project.'),
160            parse_git_error(
161                'url',
162                RemoteGitError(
163                    'GitLab: You are not allowed to push code to '
164                    'protected branches on this project.')))
165
166
167class ParseHangupTests(TestCase):
168
169    def setUp(self):
170        super(ParseHangupTests, self).setUp()
171        try:
172            HangupException([b'foo'])
173        except TypeError:
174            self.skipTest('dulwich version too old')
175
176    def test_not_set(self):
177        self.assertIsInstance(
178            parse_git_hangup('http://', HangupException()), HangupException)
179
180    def test_single_line(self):
181        self.assertEqual(
182            RemoteGitError('foo bar'),
183            parse_git_hangup('http://', HangupException([b'foo bar'])))
184
185    def test_multi_lines(self):
186        self.assertEqual(
187            RemoteGitError('foo bar\nbla bla'),
188            parse_git_hangup(
189                'http://', HangupException([b'foo bar', b'bla bla'])))
190
191    def test_filter_boring(self):
192        self.assertEqual(
193            RemoteGitError('foo bar'), parse_git_hangup('http://', HangupException(
194                [b'=======', b'foo bar', b'======'])))
195        self.assertEqual(
196            RemoteGitError('foo bar'), parse_git_hangup('http://', HangupException(
197                [b'remote: =======', b'remote: foo bar', b'remote: ======'])))
198
199    def test_permission_denied(self):
200        self.assertEqual(
201            PermissionDenied('http://', 'You are not allowed to push code to this project.'),
202            parse_git_hangup(
203                'http://',
204                HangupException(
205                    [b'=======',
206                     b'You are not allowed to push code to this project.', b'', b'======'])))
207
208    def test_notbrancherror_yet(self):
209        self.assertEqual(
210            NotBranchError('http://', 'A repository for this project does not exist yet.'),
211            parse_git_hangup(
212                'http://',
213                HangupException(
214                    [b'=======',
215                     b'',
216                     b'A repository for this project does not exist yet.', b'', b'======'])))
217
218
219class TestRemoteGitBranchFormat(TestCase):
220
221    def setUp(self):
222        super(TestRemoteGitBranchFormat, self).setUp()
223        self.format = RemoteGitBranchFormat()
224
225    def test_get_format_description(self):
226        self.assertEqual("Remote Git Branch",
227                         self.format.get_format_description())
228
229    def test_get_network_name(self):
230        self.assertEqual(b"git", self.format.network_name())
231
232    def test_supports_tags(self):
233        self.assertTrue(self.format.supports_tags())
234
235
236class TestRemoteGitBranch(TestCaseWithTransport):
237
238    _test_needs_features = [ExecutableFeature('git')]
239
240    def setUp(self):
241        TestCaseWithTransport.setUp(self)
242        self.remote_real = GitRepo.init('remote', mkdir=True)
243        self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
244        self.permit_url(self.remote_url)
245
246    def test_set_last_revision_info(self):
247        c1 = self.remote_real.do_commit(
248            message=b'message 1',
249            committer=b'committer <committer@example.com>',
250            author=b'author <author@example.com>',
251            ref=b'refs/heads/newbranch')
252        c2 = self.remote_real.do_commit(
253            message=b'message 2',
254            committer=b'committer <committer@example.com>',
255            author=b'author <author@example.com>',
256            ref=b'refs/heads/newbranch')
257
258        remote = ControlDir.open(self.remote_url)
259        newbranch = remote.open_branch('newbranch')
260        self.assertEqual(newbranch.lookup_foreign_revision_id(c2),
261                         newbranch.last_revision())
262        newbranch.set_last_revision_info(
263            1, newbranch.lookup_foreign_revision_id(c1))
264        self.assertEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
265        self.assertEqual(newbranch.last_revision(),
266                         newbranch.lookup_foreign_revision_id(c1))
267
268
269class FetchFromRemoteTestBase(object):
270
271    _test_needs_features = [ExecutableFeature('git')]
272
273    _to_format = None
274
275    def setUp(self):
276        TestCaseWithTransport.setUp(self)
277        self.remote_real = GitRepo.init('remote', mkdir=True)
278        self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
279        self.permit_url(self.remote_url)
280
281    def test_sprout_simple(self):
282        self.remote_real.do_commit(
283            message=b'message',
284            committer=b'committer <committer@example.com>',
285            author=b'author <author@example.com>')
286
287        remote = ControlDir.open(self.remote_url)
288        self.make_controldir('local', format=self._to_format)
289        local = remote.sprout('local')
290        self.assertEqual(
291            default_mapping.revision_id_foreign_to_bzr(
292                self.remote_real.head()),
293            local.open_branch().last_revision())
294
295    def test_sprout_with_tags(self):
296        c1 = self.remote_real.do_commit(
297            message=b'message',
298            committer=b'committer <committer@example.com>',
299            author=b'author <author@example.com>')
300        c2 = self.remote_real.do_commit(
301            message=b'another commit',
302            committer=b'committer <committer@example.com>',
303            author=b'author <author@example.com>',
304            ref=b'refs/tags/another')
305        self.remote_real.refs[b'refs/tags/blah'] = self.remote_real.head()
306
307        remote = ControlDir.open(self.remote_url)
308        self.make_controldir('local', format=self._to_format)
309        local = remote.sprout('local')
310        local_branch = local.open_branch()
311        self.assertEqual(
312            default_mapping.revision_id_foreign_to_bzr(c1),
313            local_branch.last_revision())
314        self.assertEqual(
315            {'blah': local_branch.last_revision(),
316             'another': default_mapping.revision_id_foreign_to_bzr(c2)},
317            local_branch.tags.get_tag_dict())
318
319    def test_sprout_with_annotated_tag(self):
320        c1 = self.remote_real.do_commit(
321            message=b'message',
322            committer=b'committer <committer@example.com>',
323            author=b'author <author@example.com>')
324        c2 = self.remote_real.do_commit(
325            message=b'another commit',
326            committer=b'committer <committer@example.com>',
327            author=b'author <author@example.com>',
328            ref=b'refs/heads/another')
329        porcelain.tag_create(
330            self.remote_real,
331            tag=b"blah",
332            author=b'author <author@example.com>',
333            objectish=c2,
334            tag_time=int(time.time()),
335            tag_timezone=0,
336            annotated=True,
337            message=b"Annotated tag")
338
339        remote = ControlDir.open(self.remote_url)
340        self.make_controldir('local', format=self._to_format)
341        local = remote.sprout(
342            'local', revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
343        local_branch = local.open_branch()
344        self.assertEqual(
345            default_mapping.revision_id_foreign_to_bzr(c1),
346            local_branch.last_revision())
347        self.assertEqual(
348            {'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
349            local_branch.tags.get_tag_dict())
350
351    def test_sprout_with_annotated_tag_unreferenced(self):
352        c1 = self.remote_real.do_commit(
353            message=b'message',
354            committer=b'committer <committer@example.com>',
355            author=b'author <author@example.com>')
356        c2 = self.remote_real.do_commit(
357            message=b'another commit',
358            committer=b'committer <committer@example.com>',
359            author=b'author <author@example.com>')
360        porcelain.tag_create(
361            self.remote_real,
362            tag=b"blah",
363            author=b'author <author@example.com>',
364            objectish=c1,
365            tag_time=int(time.time()),
366            tag_timezone=0,
367            annotated=True,
368            message=b"Annotated tag")
369
370        remote = ControlDir.open(self.remote_url)
371        self.make_controldir('local', format=self._to_format)
372        local = remote.sprout(
373            'local',
374            revision_id=default_mapping.revision_id_foreign_to_bzr(c1))
375        local_branch = local.open_branch()
376        self.assertEqual(
377            default_mapping.revision_id_foreign_to_bzr(c1),
378            local_branch.last_revision())
379        self.assertEqual(
380            {'blah': default_mapping.revision_id_foreign_to_bzr(c1)},
381            local_branch.tags.get_tag_dict())
382
383
384class FetchFromRemoteToBzrTests(FetchFromRemoteTestBase, TestCaseWithTransport):
385
386    _to_format = '2a'
387
388
389class FetchFromRemoteToGitTests(FetchFromRemoteTestBase, TestCaseWithTransport):
390
391    _to_format = 'git'
392
393
394class PushToRemoteBase(object):
395
396    _test_needs_features = [ExecutableFeature('git')]
397
398    _from_format = None
399
400    def setUp(self):
401        TestCaseWithTransport.setUp(self)
402        self.remote_real = GitRepo.init('remote', mkdir=True)
403        self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
404        self.permit_url(self.remote_url)
405
406    def test_push_branch_new(self):
407        remote = ControlDir.open(self.remote_url)
408        wt = self.make_branch_and_tree('local', format=self._from_format)
409        self.build_tree(['local/blah'])
410        wt.add(['blah'])
411        revid = wt.commit('blah')
412
413        if self._from_format == 'git':
414            result = remote.push_branch(wt.branch, name='newbranch')
415        else:
416            result = remote.push_branch(
417                wt.branch, lossy=True, name='newbranch')
418
419        self.assertEqual(0, result.old_revno)
420        if self._from_format == 'git':
421            self.assertEqual(1, result.new_revno)
422        else:
423            self.assertIs(None, result.new_revno)
424
425        result.report(BytesIO())
426
427        self.assertEqual(
428            {b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
429             },
430            self.remote_real.get_refs())
431
432    def test_push_branch_symref(self):
433        cfg = self.remote_real.get_config()
434        cfg.set((b'core', ), b'bare', True)
435        cfg.write_to_path()
436        self.remote_real.refs.set_symbolic_ref(b'HEAD', b'refs/heads/master')
437        c1 = self.remote_real.do_commit(
438            message=b'message',
439            committer=b'committer <committer@example.com>',
440            author=b'author <author@example.com>',
441            ref=b'refs/heads/master')
442        remote = ControlDir.open(self.remote_url)
443        wt = self.make_branch_and_tree('local', format=self._from_format)
444        self.build_tree(['local/blah'])
445        wt.add(['blah'])
446        revid = wt.commit('blah')
447
448        if self._from_format == 'git':
449            result = remote.push_branch(wt.branch, overwrite=True)
450        else:
451            result = remote.push_branch(wt.branch, lossy=True, overwrite=True)
452
453        self.assertEqual(None, result.old_revno)
454        if self._from_format == 'git':
455            self.assertEqual(1, result.new_revno)
456        else:
457            self.assertIs(None, result.new_revno)
458
459        result.report(BytesIO())
460
461        self.assertEqual(
462            {
463                b'HEAD': self.remote_real.refs[b'refs/heads/master'],
464                b'refs/heads/master': self.remote_real.refs[b'refs/heads/master'],
465            },
466            self.remote_real.get_refs())
467
468    def test_push_branch_new_with_tags(self):
469        remote = ControlDir.open(self.remote_url)
470        builder = self.make_branch_builder('local', format=self._from_format)
471        builder.start_series()
472        rev_1 = builder.build_snapshot(None, [
473            ('add', ('', None, 'directory', '')),
474            ('add', ('filename', None, 'file', b'content'))])
475        rev_2 = builder.build_snapshot(
476            [rev_1], [('modify', ('filename', b'new-content\n'))])
477        rev_3 = builder.build_snapshot(
478            [rev_1], [('modify', ('filename', b'new-new-content\n'))])
479        builder.finish_series()
480        branch = builder.get_branch()
481        try:
482            branch.tags.set_tag('atag', rev_2)
483        except TagsNotSupported:
484            raise TestNotApplicable('source format does not support tags')
485
486        branch.get_config_stack().set('branch.fetch_tags', True)
487        if self._from_format == 'git':
488            result = remote.push_branch(branch, name='newbranch')
489        else:
490            result = remote.push_branch(
491                branch, lossy=True, name='newbranch')
492
493        self.assertEqual(0, result.old_revno)
494        if self._from_format == 'git':
495            self.assertEqual(2, result.new_revno)
496        else:
497            self.assertIs(None, result.new_revno)
498
499        result.report(BytesIO())
500
501        self.assertEqual(
502            {b'refs/heads/newbranch', b'refs/tags/atag'},
503            set(self.remote_real.get_refs().keys()))
504
505    def test_push(self):
506        c1 = self.remote_real.do_commit(
507            message=b'message',
508            committer=b'committer <committer@example.com>',
509            author=b'author <author@example.com>')
510
511        remote = ControlDir.open(self.remote_url)
512        self.make_controldir('local', format=self._from_format)
513        local = remote.sprout('local')
514        self.build_tree(['local/blah'])
515        wt = local.open_workingtree()
516        wt.add(['blah'])
517        revid = wt.commit('blah')
518        wt.branch.tags.set_tag('sometag', revid)
519        wt.branch.get_config_stack().set('branch.fetch_tags', True)
520
521        if self._from_format == 'git':
522            result = wt.branch.push(remote.create_branch('newbranch'))
523        else:
524            result = wt.branch.push(
525                remote.create_branch('newbranch'), lossy=True)
526
527        self.assertEqual(0, result.old_revno)
528        self.assertEqual(2, result.new_revno)
529
530        result.report(BytesIO())
531
532        self.assertEqual(
533            {b'refs/heads/master': self.remote_real.head(),
534             b'HEAD': self.remote_real.head(),
535             b'refs/heads/newbranch': self.remote_real.refs[b'refs/heads/newbranch'],
536             b'refs/tags/sometag': self.remote_real.refs[b'refs/heads/newbranch'],
537             },
538            self.remote_real.get_refs())
539
540    def test_push_diverged(self):
541        c1 = self.remote_real.do_commit(
542            message=b'message',
543            committer=b'committer <committer@example.com>',
544            author=b'author <author@example.com>',
545            ref=b'refs/heads/newbranch')
546
547        remote = ControlDir.open(self.remote_url)
548        wt = self.make_branch_and_tree('local', format=self._from_format)
549        self.build_tree(['local/blah'])
550        wt.add(['blah'])
551        revid = wt.commit('blah')
552
553        newbranch = remote.open_branch('newbranch')
554        if self._from_format == 'git':
555            self.assertRaises(DivergedBranches, wt.branch.push, newbranch)
556        else:
557            self.assertRaises(DivergedBranches, wt.branch.push,
558                              newbranch, lossy=True)
559
560        self.assertEqual(
561            {b'refs/heads/newbranch': c1},
562            self.remote_real.get_refs())
563
564        if self._from_format == 'git':
565            wt.branch.push(newbranch, overwrite=True)
566        else:
567            wt.branch.push(newbranch, lossy=True, overwrite=True)
568
569        self.assertNotEqual(c1, self.remote_real.refs[b'refs/heads/newbranch'])
570
571
572class PushToRemoteFromBzrTests(PushToRemoteBase, TestCaseWithTransport):
573
574    _from_format = '2a'
575
576
577class PushToRemoteFromGitTests(PushToRemoteBase, TestCaseWithTransport):
578
579    _from_format = 'git'
580
581
582class RemoteControlDirTests(TestCaseWithTransport):
583
584    _test_needs_features = [ExecutableFeature('git')]
585
586    def setUp(self):
587        TestCaseWithTransport.setUp(self)
588        self.remote_real = GitRepo.init('remote', mkdir=True)
589        self.remote_url = 'git://%s/' % os.path.abspath(self.remote_real.path)
590        self.permit_url(self.remote_url)
591
592    def test_remove_branch(self):
593        c1 = self.remote_real.do_commit(
594            message=b'message',
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>',
601            ref=b'refs/heads/blah')
602
603        remote = ControlDir.open(self.remote_url)
604        remote.destroy_branch(name='blah')
605        self.assertEqual(
606            self.remote_real.get_refs(),
607            {b'refs/heads/master': self.remote_real.head(),
608             b'HEAD': self.remote_real.head(),
609             })
610
611    def test_list_branches(self):
612        c1 = self.remote_real.do_commit(
613            message=b'message',
614            committer=b'committer <committer@example.com>',
615            author=b'author <author@example.com>')
616        c2 = self.remote_real.do_commit(
617            message=b'another commit',
618            committer=b'committer <committer@example.com>',
619            author=b'author <author@example.com>',
620            ref=b'refs/heads/blah')
621
622        remote = ControlDir.open(self.remote_url)
623        self.assertEqual(
624            set(['master', 'blah', 'master']),
625            set([b.name for b in remote.list_branches()]))
626
627    def test_get_branches(self):
628        c1 = self.remote_real.do_commit(
629            message=b'message',
630            committer=b'committer <committer@example.com>',
631            author=b'author <author@example.com>')
632        c2 = self.remote_real.do_commit(
633            message=b'another commit',
634            committer=b'committer <committer@example.com>',
635            author=b'author <author@example.com>',
636            ref=b'refs/heads/blah')
637
638        remote = ControlDir.open(self.remote_url)
639        self.assertEqual(
640            {'': 'master', 'blah': 'blah', 'master': 'master'},
641            {n: b.name for (n, b) in remote.get_branches().items()})
642        self.assertEqual(
643            set(['', 'blah', 'master']), set(remote.branch_names()))
644
645    def test_remove_tag(self):
646        c1 = self.remote_real.do_commit(
647            message=b'message',
648            committer=b'committer <committer@example.com>',
649            author=b'author <author@example.com>')
650        c2 = self.remote_real.do_commit(
651            message=b'another commit',
652            committer=b'committer <committer@example.com>',
653            author=b'author <author@example.com>',
654            ref=b'refs/tags/blah')
655
656        remote = ControlDir.open(self.remote_url)
657        remote_branch = remote.open_branch()
658        remote_branch.tags.delete_tag('blah')
659        self.assertRaises(NoSuchTag, remote_branch.tags.delete_tag, 'blah')
660        self.assertEqual(
661            self.remote_real.get_refs(),
662            {b'refs/heads/master': self.remote_real.head(),
663             b'HEAD': self.remote_real.head(),
664             })
665
666    def test_set_tag(self):
667        c1 = self.remote_real.do_commit(
668            message=b'message',
669            committer=b'committer <committer@example.com>',
670            author=b'author <author@example.com>')
671        c2 = self.remote_real.do_commit(
672            message=b'another commit',
673            committer=b'committer <committer@example.com>',
674            author=b'author <author@example.com>')
675
676        remote = ControlDir.open(self.remote_url)
677        remote.open_branch().tags.set_tag(
678            b'blah', default_mapping.revision_id_foreign_to_bzr(c1))
679        self.assertEqual(
680            self.remote_real.get_refs(),
681            {b'refs/heads/master': self.remote_real.head(),
682             b'refs/tags/blah': c1,
683             b'HEAD': self.remote_real.head(),
684             })
685
686    def test_annotated_tag(self):
687        c1 = self.remote_real.do_commit(
688            message=b'message',
689            committer=b'committer <committer@example.com>',
690            author=b'author <author@example.com>')
691        c2 = self.remote_real.do_commit(
692            message=b'another commit',
693            committer=b'committer <committer@example.com>',
694            author=b'author <author@example.com>')
695
696        porcelain.tag_create(
697            self.remote_real,
698            tag=b"blah",
699            author=b'author <author@example.com>',
700            objectish=c2,
701            tag_time=int(time.time()),
702            tag_timezone=0,
703            annotated=True,
704            message=b"Annotated tag")
705
706        remote = ControlDir.open(self.remote_url)
707        remote_branch = remote.open_branch()
708        self.assertEqual({
709            'blah': default_mapping.revision_id_foreign_to_bzr(c2)},
710            remote_branch.tags.get_tag_dict())
711
712    def test_get_branch_reference(self):
713        c1 = self.remote_real.do_commit(
714            message=b'message',
715            committer=b'committer <committer@example.com>',
716            author=b'author <author@example.com>')
717        c2 = self.remote_real.do_commit(
718            message=b'another commit',
719            committer=b'committer <committer@example.com>',
720            author=b'author <author@example.com>')
721
722        remote = ControlDir.open(self.remote_url)
723        self.assertEqual(b'refs/heads/master', remote.get_branch_reference(''))
724        self.assertEqual(None, remote.get_branch_reference('master'))
725
726    def test_get_branch_nick(self):
727        c1 = self.remote_real.do_commit(
728            message=b'message',
729            committer=b'committer <committer@example.com>',
730            author=b'author <author@example.com>')
731        remote = ControlDir.open(self.remote_url)
732        self.assertEqual('master', remote.open_branch().nick)
733
734
735class GitUrlAndPathFromTransportTests(TestCase):
736
737    def test_file(self):
738        split_url = _git_url_and_path_from_transport('file:///home/blah')
739        self.assertEqual(split_url.scheme, 'file')
740        self.assertEqual(split_url.path, '/home/blah')
741
742    def test_file_segment_params(self):
743        split_url = _git_url_and_path_from_transport('file:///home/blah,branch=master')
744        self.assertEqual(split_url.scheme, 'file')
745        self.assertEqual(split_url.path, '/home/blah')
746
747    def test_git_smart(self):
748        split_url = _git_url_and_path_from_transport(
749            'git://github.com/dulwich/dulwich,branch=master')
750        self.assertEqual(split_url.scheme, 'git')
751        self.assertEqual(split_url.path, '/dulwich/dulwich')
752
753    def test_https(self):
754        split_url = _git_url_and_path_from_transport(
755            'https://github.com/dulwich/dulwich')
756        self.assertEqual(split_url.scheme, 'https')
757        self.assertEqual(split_url.path, '/dulwich/dulwich')
758
759    def test_https_segment_params(self):
760        split_url = _git_url_and_path_from_transport(
761            'https://github.com/dulwich/dulwich,branch=master')
762        self.assertEqual(split_url.scheme, 'https')
763        self.assertEqual(split_url.path, '/dulwich/dulwich')
764