1
# Copyright (C) 2007 Canonical Ltd
2
# Copyright (C) 2010-2018 Jelmer Vernooij
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.
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.
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
18
"""An adapter between a Git control dir and a Bazaar ControlDir."""
23
branch as _mod_branch,
29
from ..transport import (
30
do_catching_redirections,
31
get_transport_from_path,
34
from ..controldir import (
39
RepositoryAcquisitionPolicy,
45
from .transportgit import (
51
class GitDirConfig(object):
53
def get_default_stack_on(self):
56
def set_default_stack_on(self, value):
57
raise brz_errors.BzrError("Cannot set configuration")
60
class GitControlDirFormat(ControlDirFormat):
62
colocated_branches = True
63
fixed_components = True
65
def __eq__(self, other):
66
return type(self) == type(other)
68
def is_supported(self):
71
def network_name(self):
75
class UseExistingRepository(RepositoryAcquisitionPolicy):
76
"""A policy of reusing an existing repository"""
78
def __init__(self, repository, stack_on=None, stack_on_pwd=None,
79
require_stacking=False):
82
:param repository: The repository to use.
83
:param stack_on: A location to stack on
84
:param stack_on_pwd: If stack_on is relative, the location it is
87
super(UseExistingRepository, self).__init__(
88
stack_on, stack_on_pwd, require_stacking)
89
self._repository = repository
91
def acquire_repository(self, make_working_trees=None, shared=False,
92
possible_transports=None):
93
"""Implementation of RepositoryAcquisitionPolicy.acquire_repository
95
Returns an existing repository to use.
97
return self._repository, False
100
class GitDir(ControlDir):
101
"""An adapter to the '.git' dir used by git."""
103
def is_supported(self):
106
def can_convert_format(self):
109
def break_lock(self):
110
# There are no global locks, so nothing to break.
111
raise NotImplementedError(self.break_lock)
113
def cloning_metadir(self, stacked=False):
114
return format_registry.make_controldir("git")
116
def checkout_metadir(self, stacked=False):
117
return format_registry.make_controldir("git")
119
def _get_selected_ref(self, branch, ref=None):
120
if ref is not None and branch is not None:
121
raise brz_errors.BzrError("can't specify both ref and branch")
124
if branch is not None:
125
from .refs import branch_name_to_ref
126
return branch_name_to_ref(branch)
127
segment_parameters = getattr(
128
self.user_transport, "get_segment_parameters", lambda: {})()
129
ref = segment_parameters.get("ref")
131
return urlutils.unquote_to_bytes(ref)
132
if branch is None and getattr(self, "_get_selected_branch", False):
133
branch = self._get_selected_branch()
134
if branch is not None:
135
from .refs import branch_name_to_ref
136
return branch_name_to_ref(branch)
139
def get_config(self):
140
return GitDirConfig()
142
def _available_backup_name(self, base):
143
return osutils.available_backup_name(base, self.root_transport.has)
145
def sprout(self, url, revision_id=None, force_new_repo=False,
146
recurse='down', possible_transports=None,
147
accelerator_tree=None, hardlink=False, stacked=False,
148
source_branch=None, create_tree_if_local=True):
149
from ..repository import InterRepository
150
from ..transport.local import LocalTransport
151
from ..transport import get_transport
152
target_transport = get_transport(url, possible_transports)
153
target_transport.ensure_base()
154
cloning_format = self.cloning_metadir()
155
# Create/update the result branch
157
result = ControlDir.open_from_transport(target_transport)
158
except brz_errors.NotBranchError:
159
result = cloning_format.initialize_on_transport(target_transport)
160
source_branch = self.open_branch()
161
source_repository = self.find_repository()
163
result_repo = result.find_repository()
164
except brz_errors.NoRepositoryPresent:
165
result_repo = result.create_repository()
167
raise _mod_branch.UnstackableBranchFormat(
168
self._format, self.user_url)
169
interrepo = InterRepository.get(source_repository, result_repo)
171
if revision_id is not None:
172
determine_wants = interrepo.get_determine_wants_revids(
173
[revision_id], include_tags=True)
175
determine_wants = interrepo.determine_wants_all
176
interrepo.fetch_objects(determine_wants=determine_wants,
177
mapping=source_branch.mapping)
178
result_branch = source_branch.sprout(
179
result, revision_id=revision_id, repository=result_repo)
180
if (create_tree_if_local and
181
result.open_branch(name="").name == result_branch.name and
182
isinstance(target_transport, LocalTransport) and
183
(result_repo is None or result_repo.make_working_trees())):
184
wt = result.create_workingtree(
185
accelerator_tree=accelerator_tree,
186
hardlink=hardlink, from_branch=result_branch)
189
if recurse == 'down':
190
with contextlib.ExitStack() as stack:
193
basis = wt.basis_tree()
194
elif result_branch is not None:
195
basis = result_branch.basis_tree()
196
elif source_branch is not None:
197
basis = source_branch.basis_tree()
198
if basis is not None:
199
stack.enter_context(basis.lock_read())
200
subtrees = basis.iter_references()
203
for path in subtrees:
204
target = urlutils.join(url, urlutils.escape(path))
205
sublocation = wt.reference_parent(
206
path, possible_transports=possible_transports)
207
if sublocation is None:
209
'Ignoring nested tree %s, parent location unknown.',
212
sublocation.controldir.sprout(
213
target, basis.get_reference_revision(path),
214
force_new_repo=force_new_repo, recurse=recurse,
218
def clone_on_transport(self, transport, revision_id=None,
219
force_new_repo=False, preserve_stacking=False,
220
stacked_on=None, create_prefix=False,
221
use_existing_dir=True, no_tree=False):
222
"""See ControlDir.clone_on_transport."""
223
from ..repository import InterRepository
224
from .mapping import default_mapping
225
if stacked_on is not None:
226
raise _mod_branch.UnstackableBranchFormat(
227
self._format, self.user_url)
229
format = BareLocalGitControlDirFormat()
231
format = LocalGitControlDirFormat()
232
(target_repo, target_controldir, stacking,
233
repo_policy) = format.initialize_on_transport_ex(
234
transport, use_existing_dir=use_existing_dir,
235
create_prefix=create_prefix,
236
force_new_repo=force_new_repo)
237
target_repo = target_controldir.find_repository()
238
target_git_repo = target_repo._git
239
source_repo = self.find_repository()
240
interrepo = InterRepository.get(source_repo, target_repo)
241
if revision_id is not None:
242
determine_wants = interrepo.get_determine_wants_revids(
243
[revision_id], include_tags=True)
245
determine_wants = interrepo.determine_wants_all
246
(pack_hint, _, refs) = interrepo.fetch_objects(determine_wants,
247
mapping=default_mapping)
248
for name, val in refs.items():
249
target_git_repo.refs[name] = val
250
result_dir = self.__class__(transport, target_git_repo, format)
251
if revision_id is not None:
252
result_dir.open_branch().set_last_revision(revision_id)
254
# Cheaper to check if the target is not local, than to try making
256
result_dir.root_transport.local_abspath('.')
257
if result_dir.open_repository().make_working_trees():
258
self.open_workingtree().clone(
259
result_dir, revision_id=revision_id)
260
except (brz_errors.NoWorkingTree, brz_errors.NotLocalUrl):
265
def find_repository(self):
266
"""Find the repository that should be used.
268
This does not require a branch as we use it to find the repo for
269
new branches as well as to hook existing branches up to their
272
return self._gitrepository_class(self._find_commondir())
274
def get_refs_container(self):
275
"""Retrieve the refs container.
277
raise NotImplementedError(self.get_refs_container)
279
def determine_repository_policy(self, force_new_repo=False, stack_on=None,
280
stack_on_pwd=None, require_stacking=False):
281
"""Return an object representing a policy to use.
283
This controls whether a new repository is created, and the format of
284
that repository, or some existing shared repository used instead.
286
If stack_on is supplied, will not seek a containing shared repo.
288
:param force_new_repo: If True, require a new repository to be created.
289
:param stack_on: If supplied, the location to stack on. If not
290
supplied, a default_stack_on location may be used.
291
:param stack_on_pwd: If stack_on is relative, the location it is
294
return UseExistingRepository(self.find_repository())
296
def get_branches(self):
297
from .refs import ref_to_branch_name
299
for ref in self.get_refs_container().keys():
301
branch_name = ref_to_branch_name(ref)
302
except UnicodeDecodeError:
303
trace.warning("Ignoring branch %r with unicode error ref", ref)
307
ret[branch_name] = self.open_branch(ref=ref)
310
def list_branches(self):
311
return list(self.get_branches().values())
313
def push_branch(self, source, revision_id=None, overwrite=False,
314
remember=False, create_prefix=False, lossy=False,
316
"""Push the source branch into this ControlDir."""
317
push_result = GitPushResult()
318
push_result.workingtree_updated = None
319
push_result.master_branch = None
320
push_result.source_branch = source
321
push_result.stacked_on = None
322
from .branch import GitBranch
323
if isinstance(source, GitBranch) and lossy:
324
raise brz_errors.LossyPushToSameVCS(source.controldir, self)
325
target = self.open_branch(name, nascent_ok=True)
326
push_result.branch_push_result = source.push(
327
target, overwrite=overwrite, stop_revision=revision_id,
329
push_result.new_revid = push_result.branch_push_result.new_revid
330
push_result.old_revid = push_result.branch_push_result.old_revid
332
wt = self.open_workingtree()
333
except brz_errors.NoWorkingTree:
334
push_result.workingtree_updated = None
336
if self.open_branch(name="").name == target.name:
338
old_revision=push_result.old_revid,
339
new_revision=push_result.new_revid)
340
push_result.workingtree_updated = True
342
push_result.workingtree_updated = False
343
push_result.target_branch = target
344
if source.get_push_location() is None or remember:
345
source.set_push_location(push_result.target_branch.base)
349
class LocalGitControlDirFormat(GitControlDirFormat):
350
"""The .git directory control format."""
355
def _known_formats(self):
356
return set([LocalGitControlDirFormat()])
359
def repository_format(self):
360
from .repository import GitRepositoryFormat
361
return GitRepositoryFormat()
364
def workingtree_format(self):
365
from .workingtree import GitWorkingTreeFormat
366
return GitWorkingTreeFormat()
368
def get_branch_format(self):
369
from .branch import LocalGitBranchFormat
370
return LocalGitBranchFormat()
372
def open(self, transport, _found=None):
373
"""Open this directory.
376
from .transportgit import TransportRepo
378
def _open(transport):
380
return TransportRepo(transport, self.bare,
381
refs_text=getattr(self, "_refs_text", None))
382
except ValueError as e:
383
if e.args == ('Expected file to start with \'gitdir: \'', ):
384
raise brz_errors.NotBranchError(path=transport.base)
387
def redirected(transport, e, redirection_notice):
388
trace.note(redirection_notice)
389
return transport._redirected_to(e.source, e.target)
390
gitrepo = do_catching_redirections(_open, transport, redirected)
391
if not _found and not gitrepo._controltransport.has('objects'):
392
raise brz_errors.NotBranchError(path=transport.base)
393
return LocalGitDir(transport, gitrepo, self)
395
def get_format_description(self):
396
return "Local Git Repository"
398
def initialize_on_transport(self, transport):
399
from .transportgit import TransportRepo
400
git_repo = TransportRepo.init(transport, bare=self.bare)
401
return LocalGitDir(transport, git_repo, self)
403
def initialize_on_transport_ex(self, transport, use_existing_dir=False,
404
create_prefix=False, force_new_repo=False,
406
stack_on_pwd=None, repo_format_name=None,
407
make_working_trees=None,
408
shared_repo=False, vfs_only=False):
410
raise brz_errors.SharedRepositoriesUnsupported(self)
412
def make_directory(transport):
416
def redirected(transport, e, redirection_notice):
417
trace.note(redirection_notice)
418
return transport._redirected_to(e.source, e.target)
420
transport = do_catching_redirections(
421
make_directory, transport, redirected)
422
except brz_errors.FileExists:
423
if not use_existing_dir:
425
except brz_errors.NoSuchFile:
426
if not create_prefix:
428
transport.create_prefix()
429
controldir = self.initialize_on_transport(transport)
431
result_repo = controldir.find_repository()
432
repository_policy = UseExistingRepository(result_repo)
433
result_repo.lock_write()
436
repository_policy = None
437
return (result_repo, controldir, False,
440
def is_supported(self):
443
def supports_transport(self, transport):
445
external_url = transport.external_url()
446
except brz_errors.InProcessTransport:
447
raise brz_errors.NotBranchError(path=transport.base)
448
return external_url.startswith("file:")
450
def is_control_filename(self, filename):
451
return (filename == '.git'
452
or filename.startswith('.git/')
453
or filename.startswith('.git\\'))
456
class BareLocalGitControlDirFormat(LocalGitControlDirFormat):
459
supports_workingtrees = False
461
def get_format_description(self):
462
return "Local Git Repository (bare)"
464
def is_control_filename(self, filename):
468
class LocalGitDir(GitDir):
469
"""An adapter to the '.git' dir used by git."""
471
def _get_gitrepository_class(self):
472
from .repository import LocalGitRepository
473
return LocalGitRepository
476
return "<%s at %r>" % (
477
self.__class__.__name__, self.root_transport.base)
479
_gitrepository_class = property(_get_gitrepository_class)
482
def user_transport(self):
483
return self.root_transport
486
def control_transport(self):
487
return self._git._controltransport
489
def __init__(self, transport, gitrepo, format):
490
self._format = format
491
self.root_transport = transport
492
self._mode_check_done = False
495
self.transport = transport
497
self.transport = transport.clone('.git')
498
self._mode_check_done = None
500
def _get_symref(self, ref):
501
ref_chain, unused_sha = self._git.refs.follow(ref)
502
if len(ref_chain) == 1:
506
def set_branch_reference(self, target_branch, name=None):
507
ref = self._get_selected_ref(name)
508
target_transport = target_branch.controldir.control_transport
509
if self.control_transport.base == target_transport.base:
510
if ref == target_branch.ref:
511
raise BranchReferenceLoop(target_branch)
512
self._git.refs.set_symbolic_ref(ref, target_branch.ref)
516
target_branch.controldir.control_transport.local_abspath(
518
except brz_errors.NotLocalUrl:
519
raise brz_errors.IncompatibleFormat(
520
target_branch._format, self._format)
521
# TODO(jelmer): Do some consistency checking across branches..
522
self.control_transport.put_bytes(
523
'commondir', target_path.encode('utf-8'))
524
# TODO(jelmer): Urgh, avoid mucking about with internals.
525
self._git._commontransport = (
526
target_branch.repository._git._commontransport.clone())
527
self._git.object_store = TransportObjectStore(
528
self._git._commontransport.clone(OBJECTDIR))
529
self._git.refs.transport = self._git._commontransport
530
target_ref_chain, unused_sha = (
531
target_branch.controldir._git.refs.follow(target_branch.ref))
532
for target_ref in target_ref_chain:
533
if target_ref == b'HEAD':
537
# Can't create a reference to something that is not a in a repository.
538
raise brz_errors.IncompatibleFormat(
539
self.set_branch_reference, self)
540
self._git.refs.set_symbolic_ref(ref, target_ref)
542
def get_branch_reference(self, name=None):
543
ref = self._get_selected_ref(name)
544
target_ref = self._get_symref(ref)
545
if target_ref is not None:
546
from .refs import ref_to_branch_name
548
branch_name = ref_to_branch_name(target_ref)
550
params = {'ref': urlutils.quote(
551
target_ref.decode('utf-8'), '')}
553
if branch_name != '':
554
params = {'branch': urlutils.quote(branch_name, '')}
558
commondir = self.control_transport.get_bytes('commondir')
559
except brz_errors.NoSuchFile:
560
base_url = self.user_url.rstrip('/')
562
base_url = urlutils.local_path_to_url(
563
commondir.decode(osutils._fs_enc)).rstrip('/.git/') + '/'
564
return urlutils.join_segment_parameters(base_url, params)
567
def find_branch_format(self, name=None):
568
from .branch import (
569
LocalGitBranchFormat,
571
return LocalGitBranchFormat()
573
def get_branch_transport(self, branch_format, name=None):
574
if branch_format is None:
575
return self.transport
576
if isinstance(branch_format, LocalGitControlDirFormat):
577
return self.transport
578
raise brz_errors.IncompatibleFormat(branch_format, self._format)
580
def get_repository_transport(self, format):
582
return self.transport
583
if isinstance(format, LocalGitControlDirFormat):
584
return self.transport
585
raise brz_errors.IncompatibleFormat(format, self._format)
587
def get_workingtree_transport(self, format):
589
return self.transport
590
if isinstance(format, LocalGitControlDirFormat):
591
return self.transport
592
raise brz_errors.IncompatibleFormat(format, self._format)
594
def open_branch(self, name=None, unsupported=False, ignore_fallbacks=None,
595
ref=None, possible_transports=None, nascent_ok=False):
596
"""'create' a branch for this dir."""
597
repo = self.find_repository()
598
from .branch import LocalGitBranch
599
ref = self._get_selected_ref(name, ref)
600
if not nascent_ok and ref not in self._git.refs:
601
raise brz_errors.NotBranchError(
602
self.root_transport.base, controldir=self)
603
ref_chain, unused_sha = self._git.refs.follow(ref)
604
if ref_chain[-1] == b'HEAD':
607
controldir = self._find_commondir()
608
return LocalGitBranch(controldir, repo, ref_chain[-1])
610
def destroy_branch(self, name=None):
611
refname = self._get_selected_ref(name)
612
if refname == b'HEAD':
613
# HEAD can't be removed
614
raise brz_errors.UnsupportedOperation(
615
self.destroy_branch, self)
617
del self._git.refs[refname]
619
raise brz_errors.NotBranchError(
620
self.root_transport.base, controldir=self)
622
def destroy_repository(self):
623
raise brz_errors.UnsupportedOperation(self.destroy_repository, self)
625
def destroy_workingtree(self):
626
raise brz_errors.UnsupportedOperation(self.destroy_workingtree, self)
628
def destroy_workingtree_metadata(self):
629
raise brz_errors.UnsupportedOperation(
630
self.destroy_workingtree_metadata, self)
632
def needs_format_conversion(self, format=None):
633
return not isinstance(self._format, format.__class__)
635
def open_repository(self):
636
"""'open' a repository for this dir."""
637
if self.control_transport.has('commondir'):
638
raise brz_errors.NoRepositoryPresent(self)
639
return self._gitrepository_class(self)
641
def has_workingtree(self):
642
return not self._git.bare
644
def open_workingtree(self, recommend_upgrade=True, unsupported=False):
645
if not self._git.bare:
646
repo = self.find_repository()
647
from .workingtree import GitWorkingTree
648
branch = self.open_branch(ref=b'HEAD', nascent_ok=True)
649
return GitWorkingTree(self, repo, branch)
650
loc = urlutils.unescape_for_display(self.root_transport.base, 'ascii')
651
raise brz_errors.NoWorkingTree(loc)
653
def create_repository(self, shared=False):
654
from .repository import GitRepositoryFormat
656
raise brz_errors.IncompatibleFormat(
657
GitRepositoryFormat(), self._format)
658
return self.find_repository()
660
def create_branch(self, name=None, repository=None,
661
append_revisions_only=None, ref=None):
662
refname = self._get_selected_ref(name, ref)
663
if refname != b'HEAD' and refname in self._git.refs:
664
raise brz_errors.AlreadyBranchError(self.user_url)
665
repo = self.open_repository()
666
if refname in self._git.refs:
667
ref_chain, unused_sha = self._git.refs.follow(
668
self._get_selected_ref(None))
669
if ref_chain[0] == b'HEAD':
670
refname = ref_chain[1]
671
from .branch import LocalGitBranch
672
branch = LocalGitBranch(self, repo, refname)
673
if append_revisions_only:
674
branch.set_append_revisions_only(append_revisions_only)
677
def backup_bzrdir(self):
678
if not self._git.bare:
679
self.root_transport.copy_tree(".git", ".git.backup")
680
return (self.root_transport.abspath(".git"),
681
self.root_transport.abspath(".git.backup"))
683
basename = urlutils.basename(self.root_transport.base)
684
parent = self.root_transport.clone('..')
685
parent.copy_tree(basename, basename + ".backup")
687
def create_workingtree(self, revision_id=None, from_branch=None,
688
accelerator_tree=None, hardlink=False):
690
raise brz_errors.UnsupportedOperation(
691
self.create_workingtree, self)
692
if from_branch is None:
693
from_branch = self.open_branch(nascent_ok=True)
694
if revision_id is None:
695
revision_id = from_branch.last_revision()
696
repo = self.find_repository()
697
from .workingtree import GitWorkingTree
698
wt = GitWorkingTree(self, repo, from_branch)
699
wt.set_last_revision(revision_id)
700
wt._build_checkout_with_index()
703
def _find_or_create_repository(self, force_new_repo=None):
704
return self.create_repository(shared=False)
706
def _find_creation_modes(self):
707
"""Determine the appropriate modes for files and directories.
709
They're always set to be consistent with the base directory,
710
assuming that this transport allows setting modes.
712
# TODO: Do we need or want an option (maybe a config setting) to turn
713
# this off or override it for particular locations? -- mbp 20080512
714
if self._mode_check_done:
716
self._mode_check_done = True
718
st = self.transport.stat('.')
719
except brz_errors.TransportNotPossible:
720
self._dir_mode = None
721
self._file_mode = None
723
# Check the directory mode, but also make sure the created
724
# directories and files are read-write for this user. This is
725
# mostly a workaround for filesystems which lie about being able to
726
# write to a directory (cygwin & win32)
727
if (st.st_mode & 0o7777 == 0o0000):
728
# FTP allows stat but does not return dir/file modes
729
self._dir_mode = None
730
self._file_mode = None
732
self._dir_mode = (st.st_mode & 0o7777) | 0o0700
733
# Remove the sticky and execute bits for files
734
self._file_mode = self._dir_mode & ~0o7111
736
def _get_file_mode(self):
737
"""Return Unix mode for newly created files, or None.
739
if not self._mode_check_done:
740
self._find_creation_modes()
741
return self._file_mode
743
def _get_dir_mode(self):
744
"""Return Unix mode for newly created directories, or None.
746
if not self._mode_check_done:
747
self._find_creation_modes()
748
return self._dir_mode
750
def get_refs_container(self):
751
return self._git.refs
753
def get_peeled(self, ref):
754
return self._git.get_peeled(ref)
756
def _find_commondir(self):
758
commondir = self.control_transport.get_bytes('commondir')
759
except brz_errors.NoSuchFile:
762
commondir = commondir.rstrip(b'/.git/').decode(osutils._fs_enc)
763
return ControlDir.open_from_transport(
764
get_transport_from_path(commondir))
added
removed
breezy/git/dir.py
