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."""
20
from __future__ import absolute_import
23
branch as _mod_branch,
30
from ..sixish import (
34
from ..transport import (
35
do_catching_redirections,
36
get_transport_from_path,
39
from ..controldir import (
44
RepositoryAcquisitionPolicy,
50
from .transportgit import (
56
class GitDirConfig(object):
58
def get_default_stack_on(self):
61
def set_default_stack_on(self, value):
62
raise brz_errors.BzrError("Cannot set configuration")
65
class GitControlDirFormat(ControlDirFormat):
67
colocated_branches = True
68
fixed_components = True
70
def __eq__(self, other):
71
return type(self) == type(other)
73
def is_supported(self):
76
def network_name(self):
80
class UseExistingRepository(RepositoryAcquisitionPolicy):
81
"""A policy of reusing an existing repository"""
83
def __init__(self, repository, stack_on=None, stack_on_pwd=None,
84
require_stacking=False):
87
:param repository: The repository to use.
88
:param stack_on: A location to stack on
89
:param stack_on_pwd: If stack_on is relative, the location it is
92
super(UseExistingRepository, self).__init__(
93
stack_on, stack_on_pwd, require_stacking)
94
self._repository = repository
96
def acquire_repository(self, make_working_trees=None, shared=False,
97
possible_transports=None):
98
"""Implementation of RepositoryAcquisitionPolicy.acquire_repository
100
Returns an existing repository to use.
102
return self._repository, False
105
class GitDir(ControlDir):
106
"""An adapter to the '.git' dir used by git."""
108
def is_supported(self):
111
def can_convert_format(self):
114
def break_lock(self):
115
# There are no global locks, so nothing to break.
116
raise NotImplementedError(self.break_lock)
118
def cloning_metadir(self, stacked=False):
119
return format_registry.make_controldir("git")
121
def checkout_metadir(self, stacked=False):
122
return format_registry.make_controldir("git")
124
def _get_selected_ref(self, branch, ref=None):
125
if ref is not None and branch is not None:
126
raise brz_errors.BzrError("can't specify both ref and branch")
129
if branch is not None:
130
from .refs import branch_name_to_ref
131
return branch_name_to_ref(branch)
132
segment_parameters = getattr(
133
self.user_transport, "get_segment_parameters", lambda: {})()
134
ref = segment_parameters.get("ref")
136
return urlutils.unquote_to_bytes(ref)
137
if branch is None and getattr(self, "_get_selected_branch", False):
138
branch = self._get_selected_branch()
139
if branch is not None:
140
from .refs import branch_name_to_ref
141
return branch_name_to_ref(branch)
144
def get_config(self):
145
return GitDirConfig()
147
def _available_backup_name(self, base):
148
return osutils.available_backup_name(base, self.root_transport.has)
150
def sprout(self, url, revision_id=None, force_new_repo=False,
151
recurse='down', possible_transports=None,
152
accelerator_tree=None, hardlink=False, stacked=False,
153
source_branch=None, create_tree_if_local=True):
154
from ..repository import InterRepository
155
from ..transport.local import LocalTransport
156
from ..transport import get_transport
157
target_transport = get_transport(url, possible_transports)
158
target_transport.ensure_base()
159
cloning_format = self.cloning_metadir()
160
# Create/update the result branch
162
result = ControlDir.open_from_transport(target_transport)
163
except brz_errors.NotBranchError:
164
result = cloning_format.initialize_on_transport(target_transport)
165
source_branch = self.open_branch()
166
source_repository = self.find_repository()
168
result_repo = result.find_repository()
169
except brz_errors.NoRepositoryPresent:
170
result_repo = result.create_repository()
172
raise _mod_branch.UnstackableBranchFormat(
173
self._format, self.user_url)
174
interrepo = InterRepository.get(source_repository, result_repo)
176
if revision_id is not None:
177
determine_wants = interrepo.get_determine_wants_revids(
178
[revision_id], include_tags=True)
180
determine_wants = interrepo.determine_wants_all
181
interrepo.fetch_objects(determine_wants=determine_wants,
182
mapping=source_branch.mapping)
183
result_branch = source_branch.sprout(
184
result, revision_id=revision_id, repository=result_repo)
185
if (create_tree_if_local and
186
result.open_branch(name="").name == result_branch.name and
187
isinstance(target_transport, LocalTransport) and
188
(result_repo is None or result_repo.make_working_trees())):
189
wt = result.create_workingtree(
190
accelerator_tree=accelerator_tree,
191
hardlink=hardlink, from_branch=result_branch)
194
if recurse == 'down':
195
with cleanup.ExitStack() as stack:
198
basis = wt.basis_tree()
199
elif result_branch is not None:
200
basis = result_branch.basis_tree()
201
elif source_branch is not None:
202
basis = source_branch.basis_tree()
203
if basis is not None:
204
stack.enter_context(basis.lock_read())
205
subtrees = basis.iter_references()
208
for path in subtrees:
209
target = urlutils.join(url, urlutils.escape(path))
210
sublocation = wt.reference_parent(
211
path, possible_transports=possible_transports)
212
if sublocation is None:
214
'Ignoring nested tree %s, parent location unknown.',
217
sublocation.controldir.sprout(
218
target, basis.get_reference_revision(path),
219
force_new_repo=force_new_repo, recurse=recurse,
223
def clone_on_transport(self, transport, revision_id=None,
224
force_new_repo=False, preserve_stacking=False,
225
stacked_on=None, create_prefix=False,
226
use_existing_dir=True, no_tree=False,
228
"""See ControlDir.clone_on_transport."""
229
from ..repository import InterRepository
230
from .mapping import default_mapping
231
from ..transport.local import LocalTransport
232
if stacked_on is not None:
233
raise _mod_branch.UnstackableBranchFormat(
234
self._format, self.user_url)
236
format = BareLocalGitControlDirFormat()
238
format = LocalGitControlDirFormat()
239
(target_repo, target_controldir, stacking,
240
repo_policy) = format.initialize_on_transport_ex(
241
transport, use_existing_dir=use_existing_dir,
242
create_prefix=create_prefix,
243
force_new_repo=force_new_repo)
244
target_repo = target_controldir.find_repository()
245
target_git_repo = target_repo._git
246
source_repo = self.find_repository()
247
interrepo = InterRepository.get(source_repo, target_repo)
248
if revision_id is not None:
249
determine_wants = interrepo.get_determine_wants_revids(
250
[revision_id], include_tags=True, tag_selector=tag_selector)
252
determine_wants = interrepo.determine_wants_all
253
(pack_hint, _, refs) = interrepo.fetch_objects(determine_wants,
254
mapping=default_mapping)
255
for name, val in viewitems(refs):
256
target_git_repo.refs[name] = val
257
result_dir = LocalGitDir(transport, target_git_repo, format)
258
if revision_id is not None:
259
result_dir.open_branch().set_last_revision(revision_id)
260
if not no_tree and isinstance(result_dir.root_transport, LocalTransport):
261
if result_dir.open_repository().make_working_trees():
263
local_wt = self.open_workingtree()
264
except brz_errors.NotLocalUrl:
265
result_dir.create_workingtree(revision_id=revision_id)
267
local_wt.clone(result_dir, revision_id=revision_id)
271
def find_repository(self):
272
"""Find the repository that should be used.
274
This does not require a branch as we use it to find the repo for
275
new branches as well as to hook existing branches up to their
278
return self._gitrepository_class(self._find_commondir())
280
def get_refs_container(self):
281
"""Retrieve the refs container.
283
raise NotImplementedError(self.get_refs_container)
285
def determine_repository_policy(self, force_new_repo=False, stack_on=None,
286
stack_on_pwd=None, require_stacking=False):
287
"""Return an object representing a policy to use.
289
This controls whether a new repository is created, and the format of
290
that repository, or some existing shared repository used instead.
292
If stack_on is supplied, will not seek a containing shared repo.
294
:param force_new_repo: If True, require a new repository to be created.
295
:param stack_on: If supplied, the location to stack on. If not
296
supplied, a default_stack_on location may be used.
297
:param stack_on_pwd: If stack_on is relative, the location it is
300
return UseExistingRepository(self.find_repository())
302
def branch_names(self):
303
from .refs import ref_to_branch_name
305
for ref in self.get_refs_container().keys():
307
branch_name = ref_to_branch_name(ref)
308
except UnicodeDecodeError:
309
trace.warning("Ignoring branch %r with unicode error ref", ref)
313
ret.append(branch_name)
316
def get_branches(self):
317
from .refs import ref_to_branch_name
319
for ref in self.get_refs_container().keys():
321
branch_name = ref_to_branch_name(ref)
322
except UnicodeDecodeError:
323
trace.warning("Ignoring branch %r with unicode error ref", ref)
327
ret[branch_name] = self.open_branch(ref=ref)
330
def list_branches(self):
331
return list(self.get_branches().values())
333
def push_branch(self, source, revision_id=None, overwrite=False,
334
remember=False, create_prefix=False, lossy=False,
335
name=None, tag_selector=None):
336
"""Push the source branch into this ControlDir."""
337
push_result = GitPushResult()
338
push_result.workingtree_updated = None
339
push_result.master_branch = None
340
push_result.source_branch = source
341
push_result.stacked_on = None
342
from .branch import GitBranch
343
if isinstance(source, GitBranch) and lossy:
344
raise brz_errors.LossyPushToSameVCS(source.controldir, self)
345
target = self.open_branch(name, nascent_ok=True)
346
push_result.branch_push_result = source.push(
347
target, overwrite=overwrite, stop_revision=revision_id,
348
lossy=lossy, tag_selector=tag_selector)
349
push_result.new_revid = push_result.branch_push_result.new_revid
350
push_result.old_revid = push_result.branch_push_result.old_revid
352
wt = self.open_workingtree()
353
except brz_errors.NoWorkingTree:
354
push_result.workingtree_updated = None
356
if self.open_branch(name="").name == target.name:
358
old_revision=push_result.old_revid,
359
new_revision=push_result.new_revid)
360
push_result.workingtree_updated = True
362
push_result.workingtree_updated = False
363
push_result.target_branch = target
364
if source.get_push_location() is None or remember:
365
source.set_push_location(push_result.target_branch.base)
369
class LocalGitControlDirFormat(GitControlDirFormat):
370
"""The .git directory control format."""
375
def _known_formats(self):
376
return set([LocalGitControlDirFormat()])
379
def repository_format(self):
380
from .repository import GitRepositoryFormat
381
return GitRepositoryFormat()
384
def workingtree_format(self):
385
from .workingtree import GitWorkingTreeFormat
386
return GitWorkingTreeFormat()
388
def get_branch_format(self):
389
from .branch import LocalGitBranchFormat
390
return LocalGitBranchFormat()
392
def open(self, transport, _found=None):
393
"""Open this directory.
396
from .transportgit import TransportRepo
398
def _open(transport):
400
return TransportRepo(transport, self.bare,
401
refs_text=getattr(self, "_refs_text", None))
402
except ValueError as e:
403
if e.args == ('Expected file to start with \'gitdir: \'', ):
404
raise brz_errors.NotBranchError(path=transport.base)
407
def redirected(transport, e, redirection_notice):
408
trace.note(redirection_notice)
409
return transport._redirected_to(e.source, e.target)
410
gitrepo = do_catching_redirections(_open, transport, redirected)
411
if not _found and not gitrepo._controltransport.has('objects'):
412
raise brz_errors.NotBranchError(path=transport.base)
413
return LocalGitDir(transport, gitrepo, self)
415
def get_format_description(self):
416
return "Local Git Repository"
418
def initialize_on_transport(self, transport):
419
from .transportgit import TransportRepo
420
git_repo = TransportRepo.init(transport, bare=self.bare)
421
return LocalGitDir(transport, git_repo, self)
423
def initialize_on_transport_ex(self, transport, use_existing_dir=False,
424
create_prefix=False, force_new_repo=False,
426
stack_on_pwd=None, repo_format_name=None,
427
make_working_trees=None,
428
shared_repo=False, vfs_only=False):
430
raise brz_errors.SharedRepositoriesUnsupported(self)
432
def make_directory(transport):
436
def redirected(transport, e, redirection_notice):
437
trace.note(redirection_notice)
438
return transport._redirected_to(e.source, e.target)
440
transport = do_catching_redirections(
441
make_directory, transport, redirected)
442
except brz_errors.FileExists:
443
if not use_existing_dir:
445
except brz_errors.NoSuchFile:
446
if not create_prefix:
448
transport.create_prefix()
449
controldir = self.initialize_on_transport(transport)
451
result_repo = controldir.find_repository()
452
repository_policy = UseExistingRepository(result_repo)
453
result_repo.lock_write()
456
repository_policy = None
457
return (result_repo, controldir, False,
460
def is_supported(self):
463
def supports_transport(self, transport):
465
external_url = transport.external_url()
466
except brz_errors.InProcessTransport:
467
raise brz_errors.NotBranchError(path=transport.base)
468
return external_url.startswith("file:")
470
def is_control_filename(self, filename):
471
return (filename == '.git'
472
or filename.startswith('.git/')
473
or filename.startswith('.git\\'))
476
class BareLocalGitControlDirFormat(LocalGitControlDirFormat):
479
supports_workingtrees = False
481
def get_format_description(self):
482
return "Local Git Repository (bare)"
484
def is_control_filename(self, filename):
488
class LocalGitDir(GitDir):
489
"""An adapter to the '.git' dir used by git."""
491
def _get_gitrepository_class(self):
492
from .repository import LocalGitRepository
493
return LocalGitRepository
496
return "<%s at %r>" % (
497
self.__class__.__name__, self.root_transport.base)
499
_gitrepository_class = property(_get_gitrepository_class)
502
def user_transport(self):
503
return self.root_transport
506
def control_transport(self):
507
return self._git._controltransport
509
def __init__(self, transport, gitrepo, format):
510
self._format = format
511
self.root_transport = transport
512
self._mode_check_done = False
515
self.transport = transport
517
self.transport = transport.clone('.git')
518
self._mode_check_done = None
520
def _get_symref(self, ref):
521
ref_chain, unused_sha = self._git.refs.follow(ref)
522
if len(ref_chain) == 1:
526
def set_branch_reference(self, target_branch, name=None):
527
ref = self._get_selected_ref(name)
528
target_transport = target_branch.controldir.control_transport
529
if self.control_transport.base == target_transport.base:
530
if ref == target_branch.ref:
531
raise BranchReferenceLoop(target_branch)
532
self._git.refs.set_symbolic_ref(ref, target_branch.ref)
536
target_branch.controldir.control_transport.local_abspath(
538
except brz_errors.NotLocalUrl:
539
raise brz_errors.IncompatibleFormat(
540
target_branch._format, self._format)
541
# TODO(jelmer): Do some consistency checking across branches..
542
self.control_transport.put_bytes(
543
'commondir', target_path.encode('utf-8'))
544
# TODO(jelmer): Urgh, avoid mucking about with internals.
545
self._git._commontransport = (
546
target_branch.repository._git._commontransport.clone())
547
self._git.object_store = TransportObjectStore(
548
self._git._commontransport.clone(OBJECTDIR))
549
self._git.refs.transport = self._git._commontransport
550
target_ref_chain, unused_sha = (
551
target_branch.controldir._git.refs.follow(target_branch.ref))
552
for target_ref in target_ref_chain:
553
if target_ref == b'HEAD':
557
# Can't create a reference to something that is not a in a repository.
558
raise brz_errors.IncompatibleFormat(
559
self.set_branch_reference, self)
560
self._git.refs.set_symbolic_ref(ref, target_ref)
562
def get_branch_reference(self, name=None):
563
ref = self._get_selected_ref(name)
564
target_ref = self._get_symref(ref)
565
if target_ref is not None:
566
from .refs import ref_to_branch_name
568
branch_name = ref_to_branch_name(target_ref)
570
params = {'ref': urlutils.quote(
571
target_ref.decode('utf-8'), '')}
573
if branch_name != '':
574
params = {'branch': urlutils.quote(branch_name, '')}
578
commondir = self.control_transport.get_bytes('commondir')
579
except brz_errors.NoSuchFile:
580
base_url = self.user_url.rstrip('/')
582
base_url = urlutils.local_path_to_url(
583
commondir.decode(osutils._fs_enc)).rstrip('/.git/') + '/'
585
params = {k: v.encode('utf-8') for (k, v) in viewitems(params)}
586
return urlutils.join_segment_parameters(base_url, params)
589
def find_branch_format(self, name=None):
590
from .branch import (
591
LocalGitBranchFormat,
593
return LocalGitBranchFormat()
595
def get_branch_transport(self, branch_format, name=None):
596
if branch_format is None:
597
return self.transport
598
if isinstance(branch_format, LocalGitControlDirFormat):
599
return self.transport
600
raise brz_errors.IncompatibleFormat(branch_format, self._format)
602
def get_repository_transport(self, format):
604
return self.transport
605
if isinstance(format, LocalGitControlDirFormat):
606
return self.transport
607
raise brz_errors.IncompatibleFormat(format, self._format)
609
def get_workingtree_transport(self, format):
611
return self.transport
612
if isinstance(format, LocalGitControlDirFormat):
613
return self.transport
614
raise brz_errors.IncompatibleFormat(format, self._format)
616
def open_branch(self, name=None, unsupported=False, ignore_fallbacks=None,
617
ref=None, possible_transports=None, nascent_ok=False):
618
"""'create' a branch for this dir."""
619
repo = self.find_repository()
620
from .branch import LocalGitBranch
621
ref = self._get_selected_ref(name, ref)
622
if not nascent_ok and ref not in self._git.refs:
623
raise brz_errors.NotBranchError(
624
self.root_transport.base, controldir=self)
625
ref_chain, unused_sha = self._git.refs.follow(ref)
626
if ref_chain[-1] == b'HEAD':
629
controldir = self._find_commondir()
630
return LocalGitBranch(controldir, repo, ref_chain[-1])
632
def destroy_branch(self, name=None):
633
refname = self._get_selected_ref(name)
634
if refname == b'HEAD':
635
# HEAD can't be removed
636
raise brz_errors.UnsupportedOperation(
637
self.destroy_branch, self)
639
del self._git.refs[refname]
641
raise brz_errors.NotBranchError(
642
self.root_transport.base, controldir=self)
644
def destroy_repository(self):
645
raise brz_errors.UnsupportedOperation(self.destroy_repository, self)
647
def destroy_workingtree(self):
648
raise brz_errors.UnsupportedOperation(self.destroy_workingtree, self)
650
def destroy_workingtree_metadata(self):
651
raise brz_errors.UnsupportedOperation(
652
self.destroy_workingtree_metadata, self)
654
def needs_format_conversion(self, format=None):
655
return not isinstance(self._format, format.__class__)
657
def open_repository(self):
658
"""'open' a repository for this dir."""
659
if self.control_transport.has('commondir'):
660
raise brz_errors.NoRepositoryPresent(self)
661
return self._gitrepository_class(self)
663
def has_workingtree(self):
664
return not self._git.bare
666
def open_workingtree(self, recommend_upgrade=True, unsupported=False):
667
if not self._git.bare:
668
repo = self.find_repository()
669
from .workingtree import GitWorkingTree
670
branch = self.open_branch(ref=b'HEAD', nascent_ok=True)
671
return GitWorkingTree(self, repo, branch)
672
loc = urlutils.unescape_for_display(self.root_transport.base, 'ascii')
673
raise brz_errors.NoWorkingTree(loc)
675
def create_repository(self, shared=False):
676
from .repository import GitRepositoryFormat
678
raise brz_errors.IncompatibleFormat(
679
GitRepositoryFormat(), self._format)
680
return self.find_repository()
682
def create_branch(self, name=None, repository=None,
683
append_revisions_only=None, ref=None):
684
refname = self._get_selected_ref(name, ref)
685
if refname != b'HEAD' and refname in self._git.refs:
686
raise brz_errors.AlreadyBranchError(self.user_url)
687
repo = self.open_repository()
688
if refname in self._git.refs:
689
ref_chain, unused_sha = self._git.refs.follow(
690
self._get_selected_ref(None))
691
if ref_chain[0] == b'HEAD':
692
refname = ref_chain[1]
693
from .branch import LocalGitBranch
694
branch = LocalGitBranch(self, repo, refname)
695
if append_revisions_only:
696
branch.set_append_revisions_only(append_revisions_only)
699
def backup_bzrdir(self):
700
if not self._git.bare:
701
self.root_transport.copy_tree(".git", ".git.backup")
702
return (self.root_transport.abspath(".git"),
703
self.root_transport.abspath(".git.backup"))
705
basename = urlutils.basename(self.root_transport.base)
706
parent = self.root_transport.clone('..')
707
parent.copy_tree(basename, basename + ".backup")
709
def create_workingtree(self, revision_id=None, from_branch=None,
710
accelerator_tree=None, hardlink=False):
712
raise brz_errors.UnsupportedOperation(
713
self.create_workingtree, self)
714
if from_branch is None:
715
from_branch = self.open_branch(nascent_ok=True)
716
if revision_id is None:
717
revision_id = from_branch.last_revision()
718
repo = self.find_repository()
719
from .workingtree import GitWorkingTree
720
wt = GitWorkingTree(self, repo, from_branch)
721
wt.set_last_revision(revision_id)
722
wt._build_checkout_with_index()
725
def _find_or_create_repository(self, force_new_repo=None):
726
return self.create_repository(shared=False)
728
def _find_creation_modes(self):
729
"""Determine the appropriate modes for files and directories.
731
They're always set to be consistent with the base directory,
732
assuming that this transport allows setting modes.
734
# TODO: Do we need or want an option (maybe a config setting) to turn
735
# this off or override it for particular locations? -- mbp 20080512
736
if self._mode_check_done:
738
self._mode_check_done = True
740
st = self.transport.stat('.')
741
except brz_errors.TransportNotPossible:
742
self._dir_mode = None
743
self._file_mode = None
745
# Check the directory mode, but also make sure the created
746
# directories and files are read-write for this user. This is
747
# mostly a workaround for filesystems which lie about being able to
748
# write to a directory (cygwin & win32)
749
if (st.st_mode & 0o7777 == 0o0000):
750
# FTP allows stat but does not return dir/file modes
751
self._dir_mode = None
752
self._file_mode = None
754
self._dir_mode = (st.st_mode & 0o7777) | 0o0700
755
# Remove the sticky and execute bits for files
756
self._file_mode = self._dir_mode & ~0o7111
758
def _get_file_mode(self):
759
"""Return Unix mode for newly created files, or None.
761
if not self._mode_check_done:
762
self._find_creation_modes()
763
return self._file_mode
765
def _get_dir_mode(self):
766
"""Return Unix mode for newly created directories, or None.
768
if not self._mode_check_done:
769
self._find_creation_modes()
770
return self._dir_mode
772
def get_refs_container(self):
773
return self._git.refs
775
def get_peeled(self, ref):
776
return self._git.get_peeled(ref)
778
def _find_commondir(self):
780
commondir = self.control_transport.get_bytes('commondir')
781
except brz_errors.NoSuchFile:
784
commondir = commondir.rstrip(b'/.git/').decode(osutils._fs_enc)
785
return ControlDir.open_from_transport(
786
get_transport_from_path(commondir))
added
removed
breezy/git/dir.py
