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,
29
from ..sixish import (
33
from ..transport import (
34
do_catching_redirections,
35
get_transport_from_path,
38
from ..controldir import (
43
RepositoryAcquisitionPolicy,
49
from .transportgit import (
55
class GitDirConfig(object):
57
def get_default_stack_on(self):
60
def set_default_stack_on(self, value):
61
raise brz_errors.BzrError("Cannot set configuration")
64
class GitControlDirFormat(ControlDirFormat):
66
colocated_branches = True
67
fixed_components = True
69
def __eq__(self, other):
70
return type(self) == type(other)
72
def is_supported(self):
75
def network_name(self):
79
class UseExistingRepository(RepositoryAcquisitionPolicy):
80
"""A policy of reusing an existing repository"""
82
def __init__(self, repository, stack_on=None, stack_on_pwd=None,
83
require_stacking=False):
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
91
super(UseExistingRepository, self).__init__(
92
stack_on, stack_on_pwd, require_stacking)
93
self._repository = repository
95
def acquire_repository(self, make_working_trees=None, shared=False,
96
possible_transports=None):
97
"""Implementation of RepositoryAcquisitionPolicy.acquire_repository
99
Returns an existing repository to use.
101
return self._repository, False
104
class GitDir(ControlDir):
105
"""An adapter to the '.git' dir used by git."""
107
def is_supported(self):
110
def can_convert_format(self):
113
def break_lock(self):
114
# There are no global locks, so nothing to break.
115
raise NotImplementedError(self.break_lock)
117
def cloning_metadir(self, stacked=False):
118
return format_registry.make_controldir("git")
120
def checkout_metadir(self, stacked=False):
121
return format_registry.make_controldir("git")
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")
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")
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)
143
def get_config(self):
144
return GitDirConfig()
146
def _available_backup_name(self, base):
147
return osutils.available_backup_name(base, self.root_transport.has)
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
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()
167
result_repo = result.find_repository()
168
except brz_errors.NoRepositoryPresent:
169
result_repo = result.create_repository()
171
raise _mod_branch.UnstackableBranchFormat(
172
self._format, self.user_url)
173
interrepo = InterRepository.get(source_repository, result_repo)
175
if revision_id is not None:
176
determine_wants = interrepo.get_determine_wants_revids(
177
[revision_id], include_tags=True)
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
result.create_workingtree(
189
accelerator_tree=accelerator_tree,
190
hardlink=hardlink, from_branch=result_branch)
193
def clone_on_transport(self, transport, revision_id=None,
194
force_new_repo=False, preserve_stacking=False,
195
stacked_on=None, create_prefix=False,
196
use_existing_dir=True, no_tree=False):
197
"""See ControlDir.clone_on_transport."""
198
from ..repository import InterRepository
199
from .mapping import default_mapping
200
if stacked_on is not None:
201
raise _mod_branch.UnstackableBranchFormat(
202
self._format, self.user_url)
204
format = BareLocalGitControlDirFormat()
206
format = LocalGitControlDirFormat()
207
(target_repo, target_controldir, stacking,
208
repo_policy) = format.initialize_on_transport_ex(
209
transport, use_existing_dir=use_existing_dir,
210
create_prefix=create_prefix,
211
force_new_repo=force_new_repo)
212
target_repo = target_controldir.find_repository()
213
target_git_repo = target_repo._git
214
source_repo = self.find_repository()
215
interrepo = InterRepository.get(source_repo, target_repo)
216
if revision_id is not None:
217
determine_wants = interrepo.get_determine_wants_revids(
218
[revision_id], include_tags=True)
220
determine_wants = interrepo.determine_wants_all
221
(pack_hint, _, refs) = interrepo.fetch_objects(determine_wants,
222
mapping=default_mapping)
223
for name, val in viewitems(refs):
224
target_git_repo.refs[name] = val
225
result_dir = self.__class__(transport, target_git_repo, format)
226
if revision_id is not None:
227
result_dir.open_branch().set_last_revision(revision_id)
229
# Cheaper to check if the target is not local, than to try making
231
result_dir.root_transport.local_abspath('.')
232
if result_dir.open_repository().make_working_trees():
233
self.open_workingtree().clone(
234
result_dir, revision_id=revision_id)
235
except (brz_errors.NoWorkingTree, brz_errors.NotLocalUrl):
240
def find_repository(self):
241
"""Find the repository that should be used.
243
This does not require a branch as we use it to find the repo for
244
new branches as well as to hook existing branches up to their
247
return self._gitrepository_class(self._find_commondir())
249
def get_refs_container(self):
250
"""Retrieve the refs container.
252
raise NotImplementedError(self.get_refs_container)
254
def determine_repository_policy(self, force_new_repo=False, stack_on=None,
255
stack_on_pwd=None, require_stacking=False):
256
"""Return an object representing a policy to use.
258
This controls whether a new repository is created, and the format of
259
that repository, or some existing shared repository used instead.
261
If stack_on is supplied, will not seek a containing shared repo.
263
:param force_new_repo: If True, require a new repository to be created.
264
:param stack_on: If supplied, the location to stack on. If not
265
supplied, a default_stack_on location may be used.
266
:param stack_on_pwd: If stack_on is relative, the location it is
269
return UseExistingRepository(self.find_repository())
271
def get_branches(self):
272
from .refs import ref_to_branch_name
274
for ref in self.get_refs_container().keys():
276
branch_name = ref_to_branch_name(ref)
277
except UnicodeDecodeError:
278
trace.warning("Ignoring branch %r with unicode error ref", ref)
282
ret[branch_name] = self.open_branch(ref=ref)
285
def list_branches(self):
286
return list(self.get_branches().values())
288
def push_branch(self, source, revision_id=None, overwrite=False,
289
remember=False, create_prefix=False, lossy=False,
291
"""Push the source branch into this ControlDir."""
292
push_result = GitPushResult()
293
push_result.workingtree_updated = None
294
push_result.master_branch = None
295
push_result.source_branch = source
296
push_result.stacked_on = None
297
from .branch import GitBranch
298
if isinstance(source, GitBranch) and lossy:
299
raise brz_errors.LossyPushToSameVCS(source.controldir, self)
300
target = self.open_branch(name, nascent_ok=True)
301
push_result.branch_push_result = source.push(
302
target, overwrite=overwrite, stop_revision=revision_id,
304
push_result.new_revid = push_result.branch_push_result.new_revid
305
push_result.old_revid = push_result.branch_push_result.old_revid
307
wt = self.open_workingtree()
308
except brz_errors.NoWorkingTree:
309
push_result.workingtree_updated = None
311
if self.open_branch(name="").name == target.name:
313
old_revision=push_result.old_revid,
314
new_revision=push_result.new_revid)
315
push_result.workingtree_updated = True
317
push_result.workingtree_updated = False
318
push_result.target_branch = target
319
if source.get_push_location() is None or remember:
320
source.set_push_location(push_result.target_branch.base)
324
class LocalGitControlDirFormat(GitControlDirFormat):
325
"""The .git directory control format."""
330
def _known_formats(self):
331
return set([LocalGitControlDirFormat()])
334
def repository_format(self):
335
from .repository import GitRepositoryFormat
336
return GitRepositoryFormat()
339
def workingtree_format(self):
340
from .workingtree import GitWorkingTreeFormat
341
return GitWorkingTreeFormat()
343
def get_branch_format(self):
344
from .branch import LocalGitBranchFormat
345
return LocalGitBranchFormat()
347
def open(self, transport, _found=None):
348
"""Open this directory.
351
from .transportgit import TransportRepo
353
def _open(transport):
355
return TransportRepo(transport, self.bare,
356
refs_text=getattr(self, "_refs_text", None))
357
except ValueError as e:
358
if e.args == ('Expected file to start with \'gitdir: \'', ):
359
raise brz_errors.NotBranchError(path=transport.base)
362
def redirected(transport, e, redirection_notice):
363
trace.note(redirection_notice)
364
return transport._redirected_to(e.source, e.target)
365
gitrepo = do_catching_redirections(_open, transport, redirected)
366
if not gitrepo._controltransport.has('HEAD'):
367
raise brz_errors.NotBranchError(path=transport.base)
368
return LocalGitDir(transport, gitrepo, self)
370
def get_format_description(self):
371
return "Local Git Repository"
373
def initialize_on_transport(self, transport):
374
from .transportgit import TransportRepo
375
git_repo = TransportRepo.init(transport, bare=self.bare)
376
return LocalGitDir(transport, git_repo, self)
378
def initialize_on_transport_ex(self, transport, use_existing_dir=False,
379
create_prefix=False, force_new_repo=False,
381
stack_on_pwd=None, repo_format_name=None,
382
make_working_trees=None,
383
shared_repo=False, vfs_only=False):
384
def make_directory(transport):
388
def redirected(transport, e, redirection_notice):
389
trace.note(redirection_notice)
390
return transport._redirected_to(e.source, e.target)
392
transport = do_catching_redirections(
393
make_directory, transport, redirected)
394
except brz_errors.FileExists:
395
if not use_existing_dir:
397
except brz_errors.NoSuchFile:
398
if not create_prefix:
400
transport.create_prefix()
401
controldir = self.initialize_on_transport(transport)
403
result_repo = controldir.find_repository()
404
repository_policy = UseExistingRepository(result_repo)
405
result_repo.lock_write()
408
repository_policy = None
409
return (result_repo, controldir, False,
412
def is_supported(self):
415
def supports_transport(self, transport):
417
external_url = transport.external_url()
418
except brz_errors.InProcessTransport:
419
raise brz_errors.NotBranchError(path=transport.base)
420
return external_url.startswith("file:")
422
def is_control_filename(self, filename):
423
return (filename == '.git'
424
or filename.startswith('.git/')
425
or filename.startswith('.git\\'))
428
class BareLocalGitControlDirFormat(LocalGitControlDirFormat):
431
supports_workingtrees = False
433
def get_format_description(self):
434
return "Local Git Repository (bare)"
436
def is_control_filename(self, filename):
440
class LocalGitDir(GitDir):
441
"""An adapter to the '.git' dir used by git."""
443
def _get_gitrepository_class(self):
444
from .repository import LocalGitRepository
445
return LocalGitRepository
448
return "<%s at %r>" % (
449
self.__class__.__name__, self.root_transport.base)
451
_gitrepository_class = property(_get_gitrepository_class)
454
def user_transport(self):
455
return self.root_transport
458
def control_transport(self):
459
return self._git._controltransport
461
def __init__(self, transport, gitrepo, format):
462
self._format = format
463
self.root_transport = transport
464
self._mode_check_done = False
467
self.transport = transport
469
self.transport = transport.clone('.git')
470
self._mode_check_done = None
472
def _get_symref(self, ref):
473
ref_chain, unused_sha = self._git.refs.follow(ref)
474
if len(ref_chain) == 1:
478
def set_branch_reference(self, target_branch, name=None):
479
ref = self._get_selected_ref(name)
480
target_transport = target_branch.controldir.control_transport
481
if self.control_transport.base == target_transport.base:
482
if ref == target_branch.ref:
483
raise BranchReferenceLoop(target_branch)
484
self._git.refs.set_symbolic_ref(ref, target_branch.ref)
488
target_branch.controldir.control_transport.local_abspath(
490
except brz_errors.NotLocalUrl:
491
raise brz_errors.IncompatibleFormat(
492
target_branch._format, self._format)
493
# TODO(jelmer): Do some consistency checking across branches..
494
self.control_transport.put_bytes(
495
'commondir', target_path.encode('utf-8'))
496
# TODO(jelmer): Urgh, avoid mucking about with internals.
497
self._git._commontransport = (
498
target_branch.repository._git._commontransport.clone())
499
self._git.object_store = TransportObjectStore(
500
self._git._commontransport.clone(OBJECTDIR))
501
self._git.refs.transport = self._git._commontransport
502
target_ref_chain, unused_sha = (
503
target_branch.controldir._git.refs.follow(target_branch.ref))
504
for target_ref in target_ref_chain:
505
if target_ref == b'HEAD':
509
# Can't create a reference to something that is not a in a repository.
510
raise brz_errors.IncompatibleFormat(
511
self.set_branch_reference, self)
512
self._git.refs.set_symbolic_ref(ref, target_ref)
514
def get_branch_reference(self, name=None):
515
ref = self._get_selected_ref(name)
516
target_ref = self._get_symref(ref)
517
if target_ref is not None:
518
from .refs import ref_to_branch_name
520
branch_name = ref_to_branch_name(target_ref)
522
params = {'ref': urlutils.quote(
523
target_ref.decode('utf-8'), '')}
525
if branch_name != '':
526
params = {'branch': urlutils.quote(branch_name, '')}
530
commondir = self.control_transport.get_bytes('commondir')
531
except brz_errors.NoSuchFile:
532
base_url = self.user_url.rstrip('/')
534
base_url = urlutils.local_path_to_url(
535
commondir.decode(osutils._fs_enc)).rstrip('/.git/') + '/'
537
params = {k: v.encode('utf-8') for (k, v) in viewitems(params)}
538
return urlutils.join_segment_parameters(base_url, params)
541
def find_branch_format(self, name=None):
542
from .branch import (
543
LocalGitBranchFormat,
545
return LocalGitBranchFormat()
547
def get_branch_transport(self, branch_format, name=None):
548
if branch_format is None:
549
return self.transport
550
if isinstance(branch_format, LocalGitControlDirFormat):
551
return self.transport
552
raise brz_errors.IncompatibleFormat(branch_format, self._format)
554
def get_repository_transport(self, format):
556
return self.transport
557
if isinstance(format, LocalGitControlDirFormat):
558
return self.transport
559
raise brz_errors.IncompatibleFormat(format, self._format)
561
def get_workingtree_transport(self, format):
563
return self.transport
564
if isinstance(format, LocalGitControlDirFormat):
565
return self.transport
566
raise brz_errors.IncompatibleFormat(format, self._format)
568
def open_branch(self, name=None, unsupported=False, ignore_fallbacks=None,
569
ref=None, possible_transports=None, nascent_ok=False):
570
"""'create' a branch for this dir."""
571
repo = self.find_repository()
572
from .branch import LocalGitBranch
573
ref = self._get_selected_ref(name, ref)
574
if not nascent_ok and ref not in self._git.refs:
575
raise brz_errors.NotBranchError(
576
self.root_transport.base, controldir=self)
577
ref_chain, unused_sha = self._git.refs.follow(ref)
578
if ref_chain[-1] == b'HEAD':
581
controldir = self._find_commondir()
582
return LocalGitBranch(controldir, repo, ref_chain[-1])
584
def destroy_branch(self, name=None):
585
refname = self._get_selected_ref(name)
586
if refname == b'HEAD':
587
# HEAD can't be removed
588
raise brz_errors.UnsupportedOperation(
589
self.destroy_branch, self)
591
del self._git.refs[refname]
593
raise brz_errors.NotBranchError(
594
self.root_transport.base, controldir=self)
596
def destroy_repository(self):
597
raise brz_errors.UnsupportedOperation(self.destroy_repository, self)
599
def destroy_workingtree(self):
600
raise brz_errors.UnsupportedOperation(self.destroy_workingtree, self)
602
def destroy_workingtree_metadata(self):
603
raise brz_errors.UnsupportedOperation(
604
self.destroy_workingtree_metadata, self)
606
def needs_format_conversion(self, format=None):
607
return not isinstance(self._format, format.__class__)
609
def open_repository(self):
610
"""'open' a repository for this dir."""
611
if self.control_transport.has('commondir'):
612
raise brz_errors.NoRepositoryPresent(self)
613
return self._gitrepository_class(self)
615
def has_workingtree(self):
616
return not self._git.bare
618
def open_workingtree(self, recommend_upgrade=True, unsupported=False):
619
if not self._git.bare:
620
repo = self.find_repository()
621
from .workingtree import GitWorkingTree
622
branch = self.open_branch(ref=b'HEAD', nascent_ok=True)
623
return GitWorkingTree(self, repo, branch)
624
loc = urlutils.unescape_for_display(self.root_transport.base, 'ascii')
625
raise brz_errors.NoWorkingTree(loc)
627
def create_repository(self, shared=False):
628
from .repository import GitRepositoryFormat
630
raise brz_errors.IncompatibleFormat(
631
GitRepositoryFormat(), self._format)
632
return self.find_repository()
634
def create_branch(self, name=None, repository=None,
635
append_revisions_only=None, ref=None):
636
refname = self._get_selected_ref(name, ref)
637
if refname != b'HEAD' and refname in self._git.refs:
638
raise brz_errors.AlreadyBranchError(self.user_url)
639
repo = self.open_repository()
640
if refname in self._git.refs:
641
ref_chain, unused_sha = self._git.refs.follow(
642
self._get_selected_ref(None))
643
if ref_chain[0] == b'HEAD':
644
refname = ref_chain[1]
645
from .branch import LocalGitBranch
646
branch = LocalGitBranch(self, repo, refname)
647
if append_revisions_only:
648
branch.set_append_revisions_only(append_revisions_only)
651
def backup_bzrdir(self):
652
if not self._git.bare:
653
self.root_transport.copy_tree(".git", ".git.backup")
654
return (self.root_transport.abspath(".git"),
655
self.root_transport.abspath(".git.backup"))
657
basename = urlutils.basename(self.root_transport.base)
658
parent = self.root_transport.clone('..')
659
parent.copy_tree(basename, basename + ".backup")
661
def create_workingtree(self, revision_id=None, from_branch=None,
662
accelerator_tree=None, hardlink=False):
664
raise brz_errors.UnsupportedOperation(
665
self.create_workingtree, self)
666
if from_branch is None:
667
from_branch = self.open_branch(nascent_ok=True)
668
if revision_id is None:
669
revision_id = from_branch.last_revision()
670
repo = self.find_repository()
671
from .workingtree import GitWorkingTree
672
wt = GitWorkingTree(self, repo, from_branch)
673
wt.set_last_revision(revision_id)
674
wt._build_checkout_with_index()
677
def _find_or_create_repository(self, force_new_repo=None):
678
return self.create_repository(shared=False)
680
def _find_creation_modes(self):
681
"""Determine the appropriate modes for files and directories.
683
They're always set to be consistent with the base directory,
684
assuming that this transport allows setting modes.
686
# TODO: Do we need or want an option (maybe a config setting) to turn
687
# this off or override it for particular locations? -- mbp 20080512
688
if self._mode_check_done:
690
self._mode_check_done = True
692
st = self.transport.stat('.')
693
except brz_errors.TransportNotPossible:
694
self._dir_mode = None
695
self._file_mode = None
697
# Check the directory mode, but also make sure the created
698
# directories and files are read-write for this user. This is
699
# mostly a workaround for filesystems which lie about being able to
700
# write to a directory (cygwin & win32)
701
if (st.st_mode & 0o7777 == 0o0000):
702
# FTP allows stat but does not return dir/file modes
703
self._dir_mode = None
704
self._file_mode = None
706
self._dir_mode = (st.st_mode & 0o7777) | 0o0700
707
# Remove the sticky and execute bits for files
708
self._file_mode = self._dir_mode & ~0o7111
710
def _get_file_mode(self):
711
"""Return Unix mode for newly created files, or None.
713
if not self._mode_check_done:
714
self._find_creation_modes()
715
return self._file_mode
717
def _get_dir_mode(self):
718
"""Return Unix mode for newly created directories, or None.
720
if not self._mode_check_done:
721
self._find_creation_modes()
722
return self._dir_mode
724
def get_refs_container(self):
725
return self._git.refs
727
def get_peeled(self, ref):
728
return self._git.get_peeled(ref)
730
def _find_commondir(self):
732
commondir = self.control_transport.get_bytes('commondir')
733
except brz_errors.NoSuchFile:
736
commondir = commondir.rstrip(b'/.git/').decode(osutils._fs_enc)
737
return ControlDir.open_from_transport(
738
get_transport_from_path(commondir))
added
removed
breezy/git/dir.py
