/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to breezy/git/dir.py

  • Committer: Jelmer Vernooij
  • Date: 2020-07-18 23:14:00 UTC
  • mfrom: (7490.40.62 work)
  • mto: This revision was merged to the branch mainline in revision 7519.
  • Revision ID: jelmer@jelmer.uk-20200718231400-jaes9qltn8oi8xss
Merge lp:brz/3.1.

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/git/dir.py
 
1
# Copyright (C) 2007 Canonical Ltd
 
2
# Copyright (C) 2010-2018 Jelmer Vernooij
 
3
#
 
4
# This program is free software; you can redistribute it and/or modify
 
5
# it under the terms of the GNU General Public License as published by
 
6
# the Free Software Foundation; either version 2 of the License, or
 
7
# (at your option) any later version.
 
8
#
 
9
# This program is distributed in the hope that it will be useful,
 
10
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
11
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
12
# GNU General Public License for more details.
 
13
#
 
14
# You should have received a copy of the GNU General Public License
 
15
# along with this program; if not, write to the Free Software
 
16
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
17
 
 
18
"""An adapter between a Git control dir and a Bazaar ControlDir."""
 
19
 
 
20
import contextlib
 
21
 
 
22
from .. import (
 
23
    branch as _mod_branch,
 
24
    errors as brz_errors,
 
25
    trace,
 
26
    osutils,
 
27
    urlutils,
 
28
    )
 
29
from ..transport import (
 
30
    do_catching_redirections,
 
31
    get_transport_from_path,
 
32
    )
 
33
 
 
34
from ..controldir import (
 
35
    BranchReferenceLoop,
 
36
    ControlDir,
 
37
    ControlDirFormat,
 
38
    format_registry,
 
39
    RepositoryAcquisitionPolicy,
 
40
    )
 
41
 
 
42
from .mapping import (
 
43
    decode_git_path,
 
44
    encode_git_path,
 
45
    )
 
46
from .push import (
 
47
    GitPushResult,
 
48
    )
 
49
from .transportgit import (
 
50
    OBJECTDIR,
 
51
    TransportObjectStore,
 
52
    )
 
53
 
 
54
 
 
55
class GitDirConfig(object):
 
56
 
 
57
    def get_default_stack_on(self):
 
58
        return None
 
59
 
 
60
    def set_default_stack_on(self, value):
 
61
        raise brz_errors.BzrError("Cannot set configuration")
 
62
 
 
63
 
 
64
class GitControlDirFormat(ControlDirFormat):
 
65
 
 
66
    colocated_branches = True
 
67
    fixed_components = True
 
68
 
 
69
    def __eq__(self, other):
 
70
        return type(self) == type(other)
 
71
 
 
72
    def is_supported(self):
 
73
        return True
 
74
 
 
75
    def network_name(self):
 
76
        return b"git"
 
77
 
 
78
 
 
79
class UseExistingRepository(RepositoryAcquisitionPolicy):
 
80
    """A policy of reusing an existing repository"""
 
81
 
 
82
    def __init__(self, repository, stack_on=None, stack_on_pwd=None,
 
83
                 require_stacking=False):
 
84
        """Constructor.
 
85
 
 
86
        :param repository: The repository to use.
 
87
        :param stack_on: A location to stack on
 
88
        :param stack_on_pwd: If stack_on is relative, the location it is
 
89
            relative to.
 
90
        """
 
91
        super(UseExistingRepository, self).__init__(
 
92
            stack_on, stack_on_pwd, require_stacking)
 
93
        self._repository = repository
 
94
 
 
95
    def acquire_repository(self, make_working_trees=None, shared=False,
 
96
                           possible_transports=None):
 
97
        """Implementation of RepositoryAcquisitionPolicy.acquire_repository
 
98
 
 
99
        Returns an existing repository to use.
 
100
        """
 
101
        return self._repository, False
 
102
 
 
103
 
 
104
class GitDir(ControlDir):
 
105
    """An adapter to the '.git' dir used by git."""
 
106
 
 
107
    def is_supported(self):
 
108
        return True
 
109
 
 
110
    def can_convert_format(self):
 
111
        return False
 
112
 
 
113
    def break_lock(self):
 
114
        # There are no global locks, so nothing to break.
 
115
        raise NotImplementedError(self.break_lock)
 
116
 
 
117
    def cloning_metadir(self, stacked=False):
 
118
        return format_registry.make_controldir("git")
 
119
 
 
120
    def checkout_metadir(self, stacked=False):
 
121
        return format_registry.make_controldir("git")
 
122
 
 
123
    def _get_selected_ref(self, branch, ref=None):
 
124
        if ref is not None and branch is not None:
 
125
            raise brz_errors.BzrError("can't specify both ref and branch")
 
126
        if ref is not None:
 
127
            return ref
 
128
        if branch is not None:
 
129
            from .refs import branch_name_to_ref
 
130
            return branch_name_to_ref(branch)
 
131
        segment_parameters = getattr(
 
132
            self.user_transport, "get_segment_parameters", lambda: {})()
 
133
        ref = segment_parameters.get("ref")
 
134
        if ref is not None:
 
135
            return urlutils.unquote_to_bytes(ref)
 
136
        if branch is None and getattr(self, "_get_selected_branch", False):
 
137
            branch = self._get_selected_branch()
 
138
            if branch is not None:
 
139
                from .refs import branch_name_to_ref
 
140
                return branch_name_to_ref(branch)
 
141
        return b"HEAD"
 
142
 
 
143
    def get_config(self):
 
144
        return GitDirConfig()
 
145
 
 
146
    def _available_backup_name(self, base):
 
147
        return osutils.available_backup_name(base, self.root_transport.has)
 
148
 
 
149
    def sprout(self, url, revision_id=None, force_new_repo=False,
 
150
               recurse='down', possible_transports=None,
 
151
               accelerator_tree=None, hardlink=False, stacked=False,
 
152
               source_branch=None, create_tree_if_local=True):
 
153
        from ..repository import InterRepository
 
154
        from ..transport.local import LocalTransport
 
155
        from ..transport import get_transport
 
156
        target_transport = get_transport(url, possible_transports)
 
157
        target_transport.ensure_base()
 
158
        cloning_format = self.cloning_metadir()
 
159
        # Create/update the result branch
 
160
        try:
 
161
            result = ControlDir.open_from_transport(target_transport)
 
162
        except brz_errors.NotBranchError:
 
163
            result = cloning_format.initialize_on_transport(target_transport)
 
164
        source_branch = self.open_branch()
 
165
        source_repository = self.find_repository()
 
166
        try:
 
167
            result_repo = result.find_repository()
 
168
        except brz_errors.NoRepositoryPresent:
 
169
            result_repo = result.create_repository()
 
170
        if stacked:
 
171
            raise _mod_branch.UnstackableBranchFormat(
 
172
                self._format, self.user_url)
 
173
        interrepo = InterRepository.get(source_repository, result_repo)
 
174
 
 
175
        if revision_id is not None:
 
176
            determine_wants = interrepo.get_determine_wants_revids(
 
177
                [revision_id], include_tags=True)
 
178
        else:
 
179
            determine_wants = interrepo.determine_wants_all
 
180
        interrepo.fetch_objects(determine_wants=determine_wants,
 
181
                                mapping=source_branch.mapping)
 
182
        result_branch = source_branch.sprout(
 
183
            result, revision_id=revision_id, repository=result_repo)
 
184
        if (create_tree_if_local and
 
185
            result.open_branch(name="").name == result_branch.name and
 
186
            isinstance(target_transport, LocalTransport) and
 
187
                (result_repo is None or result_repo.make_working_trees())):
 
188
            wt = result.create_workingtree(
 
189
                accelerator_tree=accelerator_tree,
 
190
                hardlink=hardlink, from_branch=result_branch)
 
191
        else:
 
192
            wt = None
 
193
        if recurse == 'down':
 
194
            with contextlib.ExitStack() as stack:
 
195
                basis = None
 
196
                if wt is not None:
 
197
                    basis = wt.basis_tree()
 
198
                elif result_branch is not None:
 
199
                    basis = result_branch.basis_tree()
 
200
                elif source_branch is not None:
 
201
                    basis = source_branch.basis_tree()
 
202
                if basis is not None:
 
203
                    stack.enter_context(basis.lock_read())
 
204
                    subtrees = basis.iter_references()
 
205
                else:
 
206
                    subtrees = []
 
207
                for path in subtrees:
 
208
                    target = urlutils.join(url, urlutils.escape(path))
 
209
                    sublocation = wt.reference_parent(
 
210
                        path, possible_transports=possible_transports)
 
211
                    if sublocation is None:
 
212
                        trace.warning(
 
213
                            'Ignoring nested tree %s, parent location unknown.',
 
214
                            path)
 
215
                        continue
 
216
                    sublocation.controldir.sprout(
 
217
                        target, basis.get_reference_revision(path),
 
218
                        force_new_repo=force_new_repo, recurse=recurse,
 
219
                        stacked=stacked)
 
220
        return result
 
221
 
 
222
    def clone_on_transport(self, transport, revision_id=None,
 
223
                           force_new_repo=False, preserve_stacking=False,
 
224
                           stacked_on=None, create_prefix=False,
 
225
                           use_existing_dir=True, no_tree=False,
 
226
                           tag_selector=None):
 
227
        """See ControlDir.clone_on_transport."""
 
228
        from ..repository import InterRepository
 
229
        from .mapping import default_mapping
 
230
        from ..transport.local import LocalTransport
 
231
        if stacked_on is not None:
 
232
            raise _mod_branch.UnstackableBranchFormat(
 
233
                self._format, self.user_url)
 
234
        if no_tree:
 
235
            format = BareLocalGitControlDirFormat()
 
236
        else:
 
237
            format = LocalGitControlDirFormat()
 
238
        (target_repo, target_controldir, stacking,
 
239
         repo_policy) = format.initialize_on_transport_ex(
 
240
            transport, use_existing_dir=use_existing_dir,
 
241
            create_prefix=create_prefix,
 
242
            force_new_repo=force_new_repo)
 
243
        target_repo = target_controldir.find_repository()
 
244
        target_git_repo = target_repo._git
 
245
        source_repo = self.find_repository()
 
246
        interrepo = InterRepository.get(source_repo, target_repo)
 
247
        if revision_id is not None:
 
248
            determine_wants = interrepo.get_determine_wants_revids(
 
249
                [revision_id], include_tags=True, tag_selector=tag_selector)
 
250
        else:
 
251
            determine_wants = interrepo.determine_wants_all
 
252
        (pack_hint, _, refs) = interrepo.fetch_objects(determine_wants,
 
253
                                                       mapping=default_mapping)
 
254
        for name, val in refs.items():
 
255
            target_git_repo.refs[name] = val
 
256
        result_dir = LocalGitDir(transport, target_git_repo, format)
 
257
        if revision_id is not None:
 
258
            result_dir.open_branch().set_last_revision(revision_id)
 
259
        if not no_tree and isinstance(result_dir.root_transport, LocalTransport):
 
260
            if result_dir.open_repository().make_working_trees():
 
261
                try:
 
262
                    local_wt = self.open_workingtree()
 
263
                except brz_errors.NoWorkingTree:
 
264
                    pass
 
265
                except brz_errors.NotLocalUrl:
 
266
                    result_dir.create_workingtree(revision_id=revision_id)
 
267
                else:
 
268
                    local_wt.clone(result_dir, revision_id=revision_id)
 
269
 
 
270
        return result_dir
 
271
 
 
272
    def find_repository(self):
 
273
        """Find the repository that should be used.
 
274
 
 
275
        This does not require a branch as we use it to find the repo for
 
276
        new branches as well as to hook existing branches up to their
 
277
        repository.
 
278
        """
 
279
        return self._gitrepository_class(self._find_commondir())
 
280
 
 
281
    def get_refs_container(self):
 
282
        """Retrieve the refs container.
 
283
        """
 
284
        raise NotImplementedError(self.get_refs_container)
 
285
 
 
286
    def determine_repository_policy(self, force_new_repo=False, stack_on=None,
 
287
                                    stack_on_pwd=None, require_stacking=False):
 
288
        """Return an object representing a policy to use.
 
289
 
 
290
        This controls whether a new repository is created, and the format of
 
291
        that repository, or some existing shared repository used instead.
 
292
 
 
293
        If stack_on is supplied, will not seek a containing shared repo.
 
294
 
 
295
        :param force_new_repo: If True, require a new repository to be created.
 
296
        :param stack_on: If supplied, the location to stack on.  If not
 
297
            supplied, a default_stack_on location may be used.
 
298
        :param stack_on_pwd: If stack_on is relative, the location it is
 
299
            relative to.
 
300
        """
 
301
        return UseExistingRepository(self.find_repository())
 
302
 
 
303
    def branch_names(self):
 
304
        from .refs import ref_to_branch_name
 
305
        ret = []
 
306
        for ref in self.get_refs_container().keys():
 
307
            try:
 
308
                branch_name = ref_to_branch_name(ref)
 
309
            except UnicodeDecodeError:
 
310
                trace.warning("Ignoring branch %r with unicode error ref", ref)
 
311
                continue
 
312
            except ValueError:
 
313
                continue
 
314
            ret.append(branch_name)
 
315
        return ret
 
316
 
 
317
    def get_branches(self):
 
318
        from .refs import ref_to_branch_name
 
319
        ret = {}
 
320
        for ref in self.get_refs_container().keys():
 
321
            try:
 
322
                branch_name = ref_to_branch_name(ref)
 
323
            except UnicodeDecodeError:
 
324
                trace.warning("Ignoring branch %r with unicode error ref", ref)
 
325
                continue
 
326
            except ValueError:
 
327
                continue
 
328
            ret[branch_name] = self.open_branch(ref=ref)
 
329
        return ret
 
330
 
 
331
    def list_branches(self):
 
332
        return list(self.get_branches().values())
 
333
 
 
334
    def push_branch(self, source, revision_id=None, overwrite=False,
 
335
                    remember=False, create_prefix=False, lossy=False,
 
336
                    name=None, tag_selector=None):
 
337
        """Push the source branch into this ControlDir."""
 
338
        push_result = GitPushResult()
 
339
        push_result.workingtree_updated = None
 
340
        push_result.master_branch = None
 
341
        push_result.source_branch = source
 
342
        push_result.stacked_on = None
 
343
        from .branch import GitBranch
 
344
        if isinstance(source, GitBranch) and lossy:
 
345
            raise brz_errors.LossyPushToSameVCS(source.controldir, self)
 
346
        target = self.open_branch(name, nascent_ok=True)
 
347
        push_result.branch_push_result = source.push(
 
348
            target, overwrite=overwrite, stop_revision=revision_id,
 
349
            lossy=lossy, tag_selector=tag_selector)
 
350
        push_result.new_revid = push_result.branch_push_result.new_revid
 
351
        push_result.old_revid = push_result.branch_push_result.old_revid
 
352
        try:
 
353
            wt = self.open_workingtree()
 
354
        except brz_errors.NoWorkingTree:
 
355
            push_result.workingtree_updated = None
 
356
        else:
 
357
            if self.open_branch(name="").name == target.name:
 
358
                wt._update_git_tree(
 
359
                    old_revision=push_result.old_revid,
 
360
                    new_revision=push_result.new_revid)
 
361
                push_result.workingtree_updated = True
 
362
            else:
 
363
                push_result.workingtree_updated = False
 
364
        push_result.target_branch = target
 
365
        if source.get_push_location() is None or remember:
 
366
            source.set_push_location(push_result.target_branch.base)
 
367
        return push_result
 
368
 
 
369
 
 
370
class LocalGitControlDirFormat(GitControlDirFormat):
 
371
    """The .git directory control format."""
 
372
 
 
373
    bare = False
 
374
 
 
375
    @classmethod
 
376
    def _known_formats(self):
 
377
        return set([LocalGitControlDirFormat()])
 
378
 
 
379
    @property
 
380
    def repository_format(self):
 
381
        from .repository import GitRepositoryFormat
 
382
        return GitRepositoryFormat()
 
383
 
 
384
    @property
 
385
    def workingtree_format(self):
 
386
        from .workingtree import GitWorkingTreeFormat
 
387
        return GitWorkingTreeFormat()
 
388
 
 
389
    def get_branch_format(self):
 
390
        from .branch import LocalGitBranchFormat
 
391
        return LocalGitBranchFormat()
 
392
 
 
393
    def open(self, transport, _found=None):
 
394
        """Open this directory.
 
395
 
 
396
        """
 
397
        from .transportgit import TransportRepo
 
398
 
 
399
        def _open(transport):
 
400
            try:
 
401
                return TransportRepo(transport, self.bare,
 
402
                                     refs_text=getattr(self, "_refs_text", None))
 
403
            except ValueError as e:
 
404
                if e.args == ('Expected file to start with \'gitdir: \'', ):
 
405
                    raise brz_errors.NotBranchError(path=transport.base)
 
406
                raise
 
407
 
 
408
        def redirected(transport, e, redirection_notice):
 
409
            trace.note(redirection_notice)
 
410
            return transport._redirected_to(e.source, e.target)
 
411
        gitrepo = do_catching_redirections(_open, transport, redirected)
 
412
        if not _found and not gitrepo._controltransport.has('objects'):
 
413
            raise brz_errors.NotBranchError(path=transport.base)
 
414
        return LocalGitDir(transport, gitrepo, self)
 
415
 
 
416
    def get_format_description(self):
 
417
        return "Local Git Repository"
 
418
 
 
419
    def initialize_on_transport(self, transport):
 
420
        from .transportgit import TransportRepo
 
421
        git_repo = TransportRepo.init(transport, bare=self.bare)
 
422
        return LocalGitDir(transport, git_repo, self)
 
423
 
 
424
    def initialize_on_transport_ex(self, transport, use_existing_dir=False,
 
425
                                   create_prefix=False, force_new_repo=False,
 
426
                                   stacked_on=None,
 
427
                                   stack_on_pwd=None, repo_format_name=None,
 
428
                                   make_working_trees=None,
 
429
                                   shared_repo=False, vfs_only=False):
 
430
        if shared_repo:
 
431
            raise brz_errors.SharedRepositoriesUnsupported(self)
 
432
 
 
433
        def make_directory(transport):
 
434
            transport.mkdir('.')
 
435
            return transport
 
436
 
 
437
        def redirected(transport, e, redirection_notice):
 
438
            trace.note(redirection_notice)
 
439
            return transport._redirected_to(e.source, e.target)
 
440
        try:
 
441
            transport = do_catching_redirections(
 
442
                make_directory, transport, redirected)
 
443
        except brz_errors.FileExists:
 
444
            if not use_existing_dir:
 
445
                raise
 
446
        except brz_errors.NoSuchFile:
 
447
            if not create_prefix:
 
448
                raise
 
449
            transport.create_prefix()
 
450
        controldir = self.initialize_on_transport(transport)
 
451
        if repo_format_name:
 
452
            result_repo = controldir.find_repository()
 
453
            repository_policy = UseExistingRepository(result_repo)
 
454
            result_repo.lock_write()
 
455
        else:
 
456
            result_repo = None
 
457
            repository_policy = None
 
458
        return (result_repo, controldir, False,
 
459
                repository_policy)
 
460
 
 
461
    def is_supported(self):
 
462
        return True
 
463
 
 
464
    def supports_transport(self, transport):
 
465
        try:
 
466
            external_url = transport.external_url()
 
467
        except brz_errors.InProcessTransport:
 
468
            raise brz_errors.NotBranchError(path=transport.base)
 
469
        return external_url.startswith("file:")
 
470
 
 
471
    def is_control_filename(self, filename):
 
472
        return (filename == '.git'
 
473
                or filename.startswith('.git/')
 
474
                or filename.startswith('.git\\'))
 
475
 
 
476
 
 
477
class BareLocalGitControlDirFormat(LocalGitControlDirFormat):
 
478
 
 
479
    bare = True
 
480
    supports_workingtrees = False
 
481
 
 
482
    def get_format_description(self):
 
483
        return "Local Git Repository (bare)"
 
484
 
 
485
    def is_control_filename(self, filename):
 
486
        return False
 
487
 
 
488
 
 
489
class LocalGitDir(GitDir):
 
490
    """An adapter to the '.git' dir used by git."""
 
491
 
 
492
    def _get_gitrepository_class(self):
 
493
        from .repository import LocalGitRepository
 
494
        return LocalGitRepository
 
495
 
 
496
    def __repr__(self):
 
497
        return "<%s at %r>" % (
 
498
            self.__class__.__name__, self.root_transport.base)
 
499
 
 
500
    _gitrepository_class = property(_get_gitrepository_class)
 
501
 
 
502
    @property
 
503
    def user_transport(self):
 
504
        return self.root_transport
 
505
 
 
506
    @property
 
507
    def control_transport(self):
 
508
        return self._git._controltransport
 
509
 
 
510
    def __init__(self, transport, gitrepo, format):
 
511
        self._format = format
 
512
        self.root_transport = transport
 
513
        self._mode_check_done = False
 
514
        self._git = gitrepo
 
515
        if gitrepo.bare:
 
516
            self.transport = transport
 
517
        else:
 
518
            self.transport = transport.clone('.git')
 
519
        self._mode_check_done = None
 
520
 
 
521
    def _get_symref(self, ref):
 
522
        ref_chain, unused_sha = self._git.refs.follow(ref)
 
523
        if len(ref_chain) == 1:
 
524
            return None
 
525
        return ref_chain[1]
 
526
 
 
527
    def set_branch_reference(self, target_branch, name=None):
 
528
        ref = self._get_selected_ref(name)
 
529
        target_transport = target_branch.controldir.control_transport
 
530
        if self.control_transport.base == target_transport.base:
 
531
            if ref == target_branch.ref:
 
532
                raise BranchReferenceLoop(target_branch)
 
533
            self._git.refs.set_symbolic_ref(ref, target_branch.ref)
 
534
        else:
 
535
            try:
 
536
                target_path = (
 
537
                    target_branch.controldir.control_transport.local_abspath(
 
538
                        '.'))
 
539
            except brz_errors.NotLocalUrl:
 
540
                raise brz_errors.IncompatibleFormat(
 
541
                    target_branch._format, self._format)
 
542
            # TODO(jelmer): Do some consistency checking across branches..
 
543
            self.control_transport.put_bytes(
 
544
                'commondir', encode_git_path(target_path))
 
545
            # TODO(jelmer): Urgh, avoid mucking about with internals.
 
546
            self._git._commontransport = (
 
547
                target_branch.repository._git._commontransport.clone())
 
548
            self._git.object_store = TransportObjectStore(
 
549
                self._git._commontransport.clone(OBJECTDIR))
 
550
            self._git.refs.transport = self._git._commontransport
 
551
            target_ref_chain, unused_sha = (
 
552
                target_branch.controldir._git.refs.follow(target_branch.ref))
 
553
            for target_ref in target_ref_chain:
 
554
                if target_ref == b'HEAD':
 
555
                    continue
 
556
                break
 
557
            else:
 
558
                # Can't create a reference to something that is not a in a repository.
 
559
                raise brz_errors.IncompatibleFormat(
 
560
                    self.set_branch_reference, self)
 
561
            self._git.refs.set_symbolic_ref(ref, target_ref)
 
562
 
 
563
    def get_branch_reference(self, name=None):
 
564
        ref = self._get_selected_ref(name)
 
565
        target_ref = self._get_symref(ref)
 
566
        if target_ref is not None:
 
567
            from .refs import ref_to_branch_name
 
568
            try:
 
569
                branch_name = ref_to_branch_name(target_ref)
 
570
            except ValueError:
 
571
                params = {'ref': urlutils.quote(
 
572
                    target_ref.decode('utf-8'), '')}
 
573
            else:
 
574
                if branch_name != '':
 
575
                    params = {'branch': urlutils.quote(branch_name, '')}
 
576
                else:
 
577
                    params = {}
 
578
            try:
 
579
                commondir = self.control_transport.get_bytes('commondir')
 
580
            except brz_errors.NoSuchFile:
 
581
                base_url = self.user_url.rstrip('/')
 
582
            else:
 
583
                base_url = urlutils.local_path_to_url(
 
584
                    decode_git_path(commondir)).rstrip('/.git/') + '/'
 
585
            return urlutils.join_segment_parameters(base_url, params)
 
586
        return None
 
587
 
 
588
    def find_branch_format(self, name=None):
 
589
        from .branch import (
 
590
            LocalGitBranchFormat,
 
591
            )
 
592
        return LocalGitBranchFormat()
 
593
 
 
594
    def get_branch_transport(self, branch_format, name=None):
 
595
        if branch_format is None:
 
596
            return self.transport
 
597
        if isinstance(branch_format, LocalGitControlDirFormat):
 
598
            return self.transport
 
599
        raise brz_errors.IncompatibleFormat(branch_format, self._format)
 
600
 
 
601
    def get_repository_transport(self, format):
 
602
        if format is None:
 
603
            return self.transport
 
604
        if isinstance(format, LocalGitControlDirFormat):
 
605
            return self.transport
 
606
        raise brz_errors.IncompatibleFormat(format, self._format)
 
607
 
 
608
    def get_workingtree_transport(self, format):
 
609
        if format is None:
 
610
            return self.transport
 
611
        if isinstance(format, LocalGitControlDirFormat):
 
612
            return self.transport
 
613
        raise brz_errors.IncompatibleFormat(format, self._format)
 
614
 
 
615
    def open_branch(self, name=None, unsupported=False, ignore_fallbacks=None,
 
616
                    ref=None, possible_transports=None, nascent_ok=False):
 
617
        """'create' a branch for this dir."""
 
618
        repo = self.find_repository()
 
619
        from .branch import LocalGitBranch
 
620
        ref = self._get_selected_ref(name, ref)
 
621
        if not nascent_ok and ref not in self._git.refs:
 
622
            raise brz_errors.NotBranchError(
 
623
                self.root_transport.base, controldir=self)
 
624
        ref_chain, unused_sha = self._git.refs.follow(ref)
 
625
        if ref_chain[-1] == b'HEAD':
 
626
            controldir = self
 
627
        else:
 
628
            controldir = self._find_commondir()
 
629
        return LocalGitBranch(controldir, repo, ref_chain[-1])
 
630
 
 
631
    def destroy_branch(self, name=None):
 
632
        refname = self._get_selected_ref(name)
 
633
        if refname == b'HEAD':
 
634
            # HEAD can't be removed
 
635
            raise brz_errors.UnsupportedOperation(
 
636
                self.destroy_branch, self)
 
637
        try:
 
638
            del self._git.refs[refname]
 
639
        except KeyError:
 
640
            raise brz_errors.NotBranchError(
 
641
                self.root_transport.base, controldir=self)
 
642
 
 
643
    def destroy_repository(self):
 
644
        raise brz_errors.UnsupportedOperation(self.destroy_repository, self)
 
645
 
 
646
    def destroy_workingtree(self):
 
647
        raise brz_errors.UnsupportedOperation(self.destroy_workingtree, self)
 
648
 
 
649
    def destroy_workingtree_metadata(self):
 
650
        raise brz_errors.UnsupportedOperation(
 
651
            self.destroy_workingtree_metadata, self)
 
652
 
 
653
    def needs_format_conversion(self, format=None):
 
654
        return not isinstance(self._format, format.__class__)
 
655
 
 
656
    def open_repository(self):
 
657
        """'open' a repository for this dir."""
 
658
        if self.control_transport.has('commondir'):
 
659
            raise brz_errors.NoRepositoryPresent(self)
 
660
        return self._gitrepository_class(self)
 
661
 
 
662
    def has_workingtree(self):
 
663
        return not self._git.bare
 
664
 
 
665
    def open_workingtree(self, recommend_upgrade=True, unsupported=False):
 
666
        if not self._git.bare:
 
667
            repo = self.find_repository()
 
668
            from .workingtree import GitWorkingTree
 
669
            branch = self.open_branch(ref=b'HEAD', nascent_ok=True)
 
670
            return GitWorkingTree(self, repo, branch)
 
671
        loc = urlutils.unescape_for_display(self.root_transport.base, 'ascii')
 
672
        raise brz_errors.NoWorkingTree(loc)
 
673
 
 
674
    def create_repository(self, shared=False):
 
675
        from .repository import GitRepositoryFormat
 
676
        if shared:
 
677
            raise brz_errors.IncompatibleFormat(
 
678
                GitRepositoryFormat(), self._format)
 
679
        return self.find_repository()
 
680
 
 
681
    def create_branch(self, name=None, repository=None,
 
682
                      append_revisions_only=None, ref=None):
 
683
        refname = self._get_selected_ref(name, ref)
 
684
        if refname != b'HEAD' and refname in self._git.refs:
 
685
            raise brz_errors.AlreadyBranchError(self.user_url)
 
686
        repo = self.open_repository()
 
687
        if refname in self._git.refs:
 
688
            ref_chain, unused_sha = self._git.refs.follow(
 
689
                self._get_selected_ref(None))
 
690
            if ref_chain[0] == b'HEAD':
 
691
                refname = ref_chain[1]
 
692
        from .branch import LocalGitBranch
 
693
        branch = LocalGitBranch(self, repo, refname)
 
694
        if append_revisions_only:
 
695
            branch.set_append_revisions_only(append_revisions_only)
 
696
        return branch
 
697
 
 
698
    def backup_bzrdir(self):
 
699
        if not self._git.bare:
 
700
            self.root_transport.copy_tree(".git", ".git.backup")
 
701
            return (self.root_transport.abspath(".git"),
 
702
                    self.root_transport.abspath(".git.backup"))
 
703
        else:
 
704
            basename = urlutils.basename(self.root_transport.base)
 
705
            parent = self.root_transport.clone('..')
 
706
            parent.copy_tree(basename, basename + ".backup")
 
707
 
 
708
    def create_workingtree(self, revision_id=None, from_branch=None,
 
709
                           accelerator_tree=None, hardlink=False):
 
710
        if self._git.bare:
 
711
            raise brz_errors.UnsupportedOperation(
 
712
                self.create_workingtree, self)
 
713
        if from_branch is None:
 
714
            from_branch = self.open_branch(nascent_ok=True)
 
715
        if revision_id is None:
 
716
            revision_id = from_branch.last_revision()
 
717
        repo = self.find_repository()
 
718
        from .workingtree import GitWorkingTree
 
719
        wt = GitWorkingTree(self, repo, from_branch)
 
720
        wt.set_last_revision(revision_id)
 
721
        wt._build_checkout_with_index()
 
722
        return wt
 
723
 
 
724
    def _find_or_create_repository(self, force_new_repo=None):
 
725
        return self.create_repository(shared=False)
 
726
 
 
727
    def _find_creation_modes(self):
 
728
        """Determine the appropriate modes for files and directories.
 
729
 
 
730
        They're always set to be consistent with the base directory,
 
731
        assuming that this transport allows setting modes.
 
732
        """
 
733
        # TODO: Do we need or want an option (maybe a config setting) to turn
 
734
        # this off or override it for particular locations? -- mbp 20080512
 
735
        if self._mode_check_done:
 
736
            return
 
737
        self._mode_check_done = True
 
738
        try:
 
739
            st = self.transport.stat('.')
 
740
        except brz_errors.TransportNotPossible:
 
741
            self._dir_mode = None
 
742
            self._file_mode = None
 
743
        else:
 
744
            # Check the directory mode, but also make sure the created
 
745
            # directories and files are read-write for this user. This is
 
746
            # mostly a workaround for filesystems which lie about being able to
 
747
            # write to a directory (cygwin & win32)
 
748
            if (st.st_mode & 0o7777 == 0o0000):
 
749
                # FTP allows stat but does not return dir/file modes
 
750
                self._dir_mode = None
 
751
                self._file_mode = None
 
752
            else:
 
753
                self._dir_mode = (st.st_mode & 0o7777) | 0o0700
 
754
                # Remove the sticky and execute bits for files
 
755
                self._file_mode = self._dir_mode & ~0o7111
 
756
 
 
757
    def _get_file_mode(self):
 
758
        """Return Unix mode for newly created files, or None.
 
759
        """
 
760
        if not self._mode_check_done:
 
761
            self._find_creation_modes()
 
762
        return self._file_mode
 
763
 
 
764
    def _get_dir_mode(self):
 
765
        """Return Unix mode for newly created directories, or None.
 
766
        """
 
767
        if not self._mode_check_done:
 
768
            self._find_creation_modes()
 
769
        return self._dir_mode
 
770
 
 
771
    def get_refs_container(self):
 
772
        return self._git.refs
 
773
 
 
774
    def get_peeled(self, ref):
 
775
        return self._git.get_peeled(ref)
 
776
 
 
777
    def _find_commondir(self):
 
778
        try:
 
779
            commondir = self.control_transport.get_bytes('commondir')
 
780
        except brz_errors.NoSuchFile:
 
781
            return self
 
782
        else:
 
783
            commondir = commondir.rstrip(b'/.git/').decode(osutils._fs_enc)
 
784
            return ControlDir.open_from_transport(
 
785
                get_transport_from_path(commondir))