1#@!#!123s

: / kunden / usr / lib / python3 / dist-packages / breezy / git /
Filename : git_remote_helper.py
back
# vim: expandtab

# Copyright (C) 2011-2018 Jelmer Vernooij <jelmer@jelmer.uk>

# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA


"""Remote helper for git for accessing bzr repositories."""

CAPABILITIES = ["fetch", "option", "push"]

import os

from ..controldir import ControlDir
from ..errors import NoRepositoryPresent, NotBranchError
from ..plugins.fastimport import exporter as fastexporter
from ..repository import InterRepository
from ..transport import get_transport_from_path
from . import LocalGitProber
from .dir import BareLocalGitControlDirFormat, LocalGitControlDirFormat
from .object_store import get_object_store
from .refs import get_refs_container, ref_to_branch_name
from .repository import GitRepository

try:
    import fastimport  # noqa: F401
except ModuleNotFoundError:
    pass
else:
    CAPABILITIES.append("import")
    CAPABILITIES.append("refspec *:*")


def open_remote_dir(url):
    try:
        return ControlDir.open(url)
    except NotBranchError:
        return ControlDir.create(url)


def fetch(outf, wants, shortname, remote_dir, local_dir):
    remote_repo = remote_dir.find_repository()
    local_repo = local_dir.find_repository()
    inter = InterRepository.get(remote_repo, local_repo)
    revs = []
    for sha1, _ref in wants:
        revs.append((sha1, None))
    if isinstance(remote_repo, GitRepository) and isinstance(local_repo, GitRepository):
        lossy = False
    else:
        lossy = True
    inter.fetch_revs(revs, lossy=lossy)
    outf.write(b"\n")


def push(outf, wants, shortname, remote_dir, local_dir):
    for src_ref, dest_ref in wants:
        local_branch = local_dir.open_branch(ref=src_ref)
        dest_branch_name = ref_to_branch_name(dest_ref)
        if dest_branch_name == "master":
            dest_branch_name = None
        try:
            remote_branch = remote_dir.open_branch(name=dest_branch_name)
        except NotBranchError:
            remote_branch = remote_dir.create_branch(name=dest_branch_name)
        local_branch.push(remote_branch)
        outf.write(b"ok %s\n" % dest_ref)
    outf.write(b"\n")


class RemoteHelper:
    """Git remote helper."""

    def __init__(self, local_dir, shortname, remote_dir):
        self.local_dir = local_dir
        self.shortname = shortname
        self.remote_dir = remote_dir
        self.batchcmd = None
        self.wants = []

    def cmd_capabilities(self, outf, argv):
        outf.write(b"\n".join([c.encode() for c in CAPABILITIES]) + b"\n\n")

    def cmd_list(self, outf, argv):
        try:
            repo = self.remote_dir.find_repository()
        except NoRepositoryPresent:
            repo = self.remote_dir.create_repository()
        object_store = get_object_store(repo)
        with object_store.lock_read():
            refs = get_refs_container(self.remote_dir, object_store)
            for ref, git_sha1 in refs.as_dict().items():
                ref = ref.replace(b"~", b"_")
                outf.write(b"%s %s\n" % (git_sha1, ref))
            outf.write(b"\n")

    def cmd_option(self, outf, argv):
        outf.write(b"unsupported\n")

    def cmd_fetch(self, outf, argv):
        if self.batchcmd not in (None, "fetch"):
            raise Exception("fetch command inside other batch command")
        self.wants.append(tuple(argv[1:]))
        self.batchcmd = "fetch"

    def cmd_push(self, outf, argv):
        if self.batchcmd not in (None, "push"):
            raise Exception("push command inside other batch command")
        self.wants.append(tuple(argv[1].split(":", 1)))
        self.batchcmd = "push"

    def cmd_import(self, outf, argv):
        if "fastimport" in CAPABILITIES:
            raise Exception("install fastimport for 'import' command support")
        ref = argv[1].encode("utf-8")
        dest_branch_name = ref_to_branch_name(ref)
        if dest_branch_name == "master":
            dest_branch_name = None
        remote_branch = self.remote_dir.open_branch(name=dest_branch_name)
        exporter = fastexporter.BzrFastExporter(
            remote_branch,
            outf=outf,
            ref=ref,
            checkpoint=None,
            import_marks_file=None,
            export_marks_file=None,
            revision=None,
            verbose=None,
            plain_format=True,
            rewrite_tags=False,
        )
        exporter.run()

    commands = {
        "capabilities": cmd_capabilities,
        "list": cmd_list,
        "option": cmd_option,
        "fetch": cmd_fetch,
        "push": cmd_push,
        "import": cmd_import,
    }

    def process(self, inf, outf):
        while True:
            line = inf.readline()
            if not line:
                break
            self.process_line(line, outf)

    def process_line(self, l, outf):
        argv = l.strip().split()
        if argv == []:
            if self.batchcmd == "fetch":
                fetch(outf, self.wants, self.shortname, self.remote_dir, self.local_dir)
            elif self.batchcmd == "push":
                push(outf, self.wants, self.shortname, self.remote_dir, self.local_dir)
            elif self.batchcmd is None:
                return
            else:
                raise AssertionError("invalid batch {!r}".format(self.batchcmd))
            self.batchcmd = None
        else:
            try:
                self.commands[argv[0].decode()](self, outf, argv)
            except KeyError:
                raise Exception("Unknown remote command {!r}".format(argv))
        outf.flush()


def open_local_dir():
    try:
        git_path = os.environ["GIT_DIR"]
    except KeyError:
        git_transport = get_transport_from_path(".")
        git_format = LocalGitProber().probe_transport(git_transport)
    else:
        if git_path.endswith("/.git"):
            git_format = LocalGitControlDirFormat()
            git_path = git_path[:-4]
        else:
            git_format = BareLocalGitControlDirFormat()
        git_transport = get_transport_from_path(git_path)

    return git_format.open(git_transport)