Commit f484eb84 authored by nimrod's avatar nimrod
Browse files

Substantial work on the Git infrastructure.

- Refactor of the git modules.
- Update git-namespace-backup accordingly.
- Implement mirror and archive commands in git-manage.

Left to do:
- GitHub empty commit.
- Create command in git-manage.
parent fc4ba82c
Loading
Loading
Loading
Loading
+206 −14
Original line number Diff line number Diff line
@@ -6,31 +6,227 @@ import argparse
import os
import os.path
import sys
import github3.exceptions  # pylint: disable=import-error
import gitlab.exceptions  # pylint: disable=import-error
import passhole.passhole  # pylint: disable=import-error

sys.path.append(os.path.expanduser("~/Documents/bin"))

# import rcfiles.git  # noqa: E402 pylint: disable=wrong-import-position
# import rcfiles.github  # noqa: E402 pylint: disable=wrong-import-position
# import rcfiles.gitlab  # noqa: E402 pylint: disable=wrong-import-position
import rcfiles.git  # noqa: E402 pylint: disable=wrong-import-position
import rcfiles.github  # noqa: E402 pylint: disable=wrong-import-position
import rcfiles.gitlab  # noqa: E402 pylint: disable=wrong-import-position


GH_MIRROR_PREFIX = "https://*****@github.com"


def github_mirrors(project, gh_conn):
    """Return a list of GitHub repositories that are mirrored from GitLab."""
    repos = []
    for mirror in project.remote_mirrors.list(all=True):
        if mirror.url.startswith(GH_MIRROR_PREFIX):
            name, owner = (
                mirror.url.removeprefix(GH_MIRROR_PREFIX)
                .removesuffix("/")
                .removesuffix(".git")
                .split("/")
            )
            try:
                repos.append(gh_conn.repository(name, owner))
            except github3.exceptions.NotFoundError:
                pass
    return repos


def mirror_project(project, gh_conn, token):
    """Mirror a project to GitHub."""
    gh_me = rcfiles.github.me(gh_conn)
    mirror_url = f"https://{token}@github.com/{gh_me}/{project.name}.git"

    try:
        gh_repo = gh_conn.repository(project.name, gh_me)
    except github3.exceptions.NotFoundError:
        gh_repo = gh_conn.create_repository(project.name)

    gh_repo.edit(
        project.name,
        homepage=project.web_url,
        description=f"Mirror of {project.web_url}",
        has_issues=False,
        has_wiki=False,
        default_branch=project.default_branch,
        has_projects=False,
    )

    for mirror in project.remote_mirrors.list(all=True):
        if mirror.url.startswith(f"{GH_MIRROR_PREFIX}/{gh_me}/{project.name}"):
            mirror.url = mirror_url
            break
    else:
        mirror = project.remote_mirrors.create({"url": mirror_url})

    mirror.enabled = True
    mirror.only_protected_branches = False
    mirror.keep_divergent_refs = False
    mirror.save()

    return gh_repo


def get_mirror_token():
    """Get the GitHub token for mirroring a GitLab repository.

    Reads it from a Keepass password database using Passhole.
    """
    ENTRY_PATH = "Web Sites/GitHub".split("/")  # noqa
    TOKEN_FIELD = "GitLab mirroring token"  # noqa nosec

    # The following line requires an interactive session for getting the
    # password.
    db = passhole.passhole.open_database()
    entry = db.find_entries(path=ENTRY_PATH, first=True)
    return passhole.passhole.get_field(entry, TOKEN_FIELD)


def guess_remote(remote_type="gitlab", gl_conn=None):
    # pylint: disable=too-many-return-statements
    """Try to guess the right remote of the Git repo we're in.

    return None if failed, or the repo dictionary from get_all_remotes().
    """
    if remote_type not in ["gitlab", "github"]:
        return None

    if remote_type == "gitlab" and gl_conn is None:
        return None

    remotes = rcfiles.git.get_all_remotes()
    if remotes is None:
        return None

    if "origin" in remotes:
        remote = remotes["origin"]
        if remote_type == "gitlab" and rcfiles.gitlab.is_gitlab_url(
            gl_conn, remote["url"]
        ):
            return remote
        if remote_type == "github" and rcfiles.github.is_github_url(
            remote["url"]
        ):
            return remote

    for remote in remotes.values():
        if remote_type == "gitlab" and rcfiles.gitlab.is_gitlab_url(
            gl_conn, remote["url"]
        ):
            return remote
        if remote_type == "github" and rcfiles.github.is_github_url(
            remote["url"]
        ):
            return remote

    return None


def guess_name(args, gh_conn=None, gl_conn=None):
    """Try to guess the name from the arguments and the repository remotes."""
    if args.name is None:
        if not rcfiles.git.in_repo():
            parser.error("Name not provided and not in a Git repo.")

        remote = guess_remote(
            remote_type="github"
            if "github" in args and args.github
            else "gitlab",
            gl_conn=gl_conn,
        )
        if remote is None:
            if "github" in args and args.github:
                parser.error(
                    "Name not provided and could not find a GitHub remote."
                )
            else:
                parser.error(
                    "Name not provided and could not find a GitLab remote."
                )
        else:
            if "github" in args and args.github:
                name = rcfiles.github.url_to_name(remote["url"])
            else:
                name = rcfiles.gitlab.url_to_name(gl_conn, remote["url"])
    else:
        if "github" in args and args.github:
            if "/" in args.name:
                name = args.name
            else:
                name = f"{rcfiles.github.me(gh_conn)}/{args.name}"
        else:
            if "/" in args.name:
                name = args.name
            else:
                name = f"{rcfiles.gitlab.me(gl_conn)}/{args.name}"
    return name


def create_repo(args):
    """Create a new repository."""
    print(args)
    if args.mirror and args.github:
        parser.error("Can't mirror from GitHub to GitLab.")


def mirror_repo(args):
    """Mirror a GitLab repository to GitHub."""
    print(args)
    try:
        gh_conn = rcfiles.github.connect()
    except Exception as e:  # pylint: disable=broad-except
        parser.error(f"Failed to connect to GitHub: {e}")
    try:
        gl_conn = rcfiles.gitlab.connect()
    except Exception as e:  # pylint: disable=broad-except
        parser.error(f"Failed to connect to GitLab: {e}")

    name = guess_name(args, gh_conn, gl_conn)

    try:
        project = gl_conn.projects.get(name)
    except gitlab.exceptions.GitlabGetError:
        parser.error(f"Could not find GitLab project {name}.")

    gh_repo = mirror_project(project, gh_conn, get_mirror_token())
    if args.name is None and rcfiles.git.in_repo():
        rcfiles.git.add_remote(".", "github", gh_repo.ssh_url)


def archive_repo(args):
    """Archive a repository."""
    print(args)
    try:
        gh_conn = rcfiles.github.connect()
    except Exception as e:  # pylint: disable=broad-except
        parser.error(f"Failed to connect to GitHub: {e}")
    if not args.github:
        try:
            gl_conn = rcfiles.gitlab.connect()
        except Exception as e:  # pylint: disable=broad-except
            parser.error(f"Failed to connect to GitLab: {e}")

    if args.github:
        owner, name = guess_name(args, gh_conn=gh_conn, gl_conn=None).split(
            "/"
        )
    else:
        name = guess_name(args, gh_conn=None, gl_conn=gl_conn)

    if args.github:
        repo = gh_conn.repository(owner, name)
        repo.edit(name, archived=True)
    else:
        project = gl_conn.projects.get(name)
        project.archive()
        for mirror in github_mirrors(project, gh_conn):
            mirror.edit(mirror.name, archived=True)

def main():
    """The main function."""

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    subparsers = parser.add_subparsers(
        title="Commands", required=True, dest="command"
@@ -71,9 +267,5 @@ def main():
        "--github", help="The repository is in GitHub.", action="store_true"
    )

    args = parser.parse_args()
    args.func(args)


if __name__ == "__main__":
    main()
    _args = parser.parse_args()
    _args.func(_args)
+22 −15
Original line number Diff line number Diff line
@@ -14,6 +14,7 @@ import pathlib
import sys
import os
import os.path
import gitlab.exception  # pylint: disable=import-error
import sh  # pylint: disable=import-error
from sh.contrib import git  # pylint: disable=import-error

@@ -50,39 +51,45 @@ if __name__ == "__main__":
    args = parser.parse_args()

    with rcfiles.gitlab.connect() as conn:
        group = rcfiles.gitlab.get_group(conn, args.namespace.name)
        if group is None:
            group = rcfiles.gitlab.create_group(
                conn, args.namespace.name, visibility="internal"
            )
        try:
            group = conn.groups.get(args.namespace.name)
            print(
                "Created new group id: {}, name: {}, path: {}.".format(
                "Using existing group id: {}, name: {}, path: {}.".format(
                    group.id, group.name, group.path
                ),
                file=sys.stderr,
            )
        else:
        except gitlab.exceptions.GitlabGetError:
            group = conn.groups.create(
                {
                    "name": args.namespace.name,
                    "path": rcfiles.gitlab.name_to_path(args.namespace.name),
                    "visibility": "internal",
                }
            )
            print(
                "Using existing group id: {}, name: {}, path: {}.".format(
                "Created new group id: {}, name: {}, path: {}.".format(
                    group.id, group.name, group.path
                ),
                file=sys.stderr,
            )
        for repo in list_repositories(args.namespace):
            project = rcfiles.gitlab.get_project(conn, group.name, repo.name)
            if project is None:
                project = rcfiles.gitlab.create_project(
                    conn, repo.name, group.name
            try:
                project = conn.projects.get_project(
                    f"{group.path}/{repo.name}"
                )
                print(
                    "Created new project id: {}, name: {}, path: {}.".format(
                    "Using existing project id: {}, name: {}, path: {}.".format(  # noqa: E501
                        project.id, project.name, project.path
                    ),
                    file=sys.stderr,
                )
            else:
            except gitlab.exceptions.GitlabGetError:
                project = conn.projects.create(
                    {"name": repo.name, "namespace_id": group.id}
                )
                print(
                    "Using existing project id: {}, name: {}, path: {}.".format(  # noqa: E501
                    "Created new project id: {}, name: {}, path: {}.".format(
                        project.id, project.name, project.path
                    ),
                    file=sys.stderr,
+19 −44
Original line number Diff line number Diff line
@@ -3,8 +3,8 @@
import configparser
import os.path
import pathlib
import sh  # pylint: disable=import-error
from sh.contrib import git  # pylint: disable=import-error
from . import gitlab


def is_repo(path):
@@ -23,23 +23,14 @@ def in_repo():
    return is_repo(".")


def get_remotes():
def get_all_remotes():
    """Return a dictionary of remotes and their URL.
    Also, deduce the remote type ("gitlab", "github" or None if couldn't figure
    it out) and get the namespace and repository name.

    If not in a Git repository, return None.
    """
    if not in_repo:
        return None

    gitlab_http_url = gitlab.get_url()
    gitlab_ssh_url = (
        f'git@{gitlab_http_url.removeprefix("https://").removesuffix("/")}:'
    )
    github_http_url = "https://github.com/"
    github_ssh_url = "git@github.com:"

    config = configparser.ConfigParser()
    config.read(".git/config")

@@ -52,40 +43,24 @@ def get_remotes():
        if x.startswith("remote ")
    }

    for name, remote in remotes.items():
        if remote["url"].startswith(gitlab_http_url) or remote[
            "url"
        ].startswith(gitlab_ssh_url):
            remotes[name]["type"] = "gitlab"
            parts = (
                remote["url"]
                .removeprefix(gitlab_http_url)
                .removeprefix(gitlab_ssh_url)
                .removesuffix(".git")
                .split("/")
            )
            if len(parts) == 2:
                remotes[name]["namespace"] = parts[0]
                remotes[name]["name"] = parts[1]
        elif remote["url"].startswith(github_http_url) or remote[
            "url"
        ].startswith(github_ssh_url):
            remotes[name]["type"] = "github"
            parts = (
                remote["url"]
                .removeprefix(github_http_url)
                .removeprefix(github_ssh_url)
                .removesuffix(".git")
                .split("/")
            )
            if len(parts) == 2:
                remotes[name]["name"] = parts[1]
        else:
            remotes[remote]["type"] = None

    return remotes


def add_remote(name, url):
def add_remote(repo, name, url):
    """Add a remote to the Git repository."""
    with sh.pushd(repo):
        git.remote("add", name, url)


def author_name():
    """Get the author name."""
    if "GIT_AUTHOR_NAME" in os.environ:
        return os.environ["GIT_AUTHOR_NAME"]
    return git.config("--get", "user.name")


def author_email():
    """Get the author email."""
    if "GIT_AUTHOR_EMAil" in os.environ:
        return os.environ["GIT_AUTHOR_EMAIL"]
    return git.config("--get", "user.email")
+26 −23
Original line number Diff line number Diff line
@@ -3,40 +3,43 @@
import os
import github3  # pylint: disable=import-error

# from . import git

HTTP_URL = "https://github.com/"
SSH_URL = "git@github.com:"


def url_to_name(url):
    """Get the full name from the GitLab URL."""
    return (
        url.removeprefix(HTTP_URL)
        .removeprefix(SSH_URL)
        .removesuffix("/")
        .removesuffix(".git")
    )


def is_github_url(url):
    """Return is the URL for a GitHub repository."""
    return url.startswith(HTTP_URL) or url.startswith(SSH_URL)


def connect():
    """Return a GitHub session."""
    try:
        token = os.environ["GITHUB_TOKEN"]
    except KeyError:
        # pylint: disable-next=raise-missing-from
        raise Exception("GITHUB_TOKEN environment variable not set.")
    return github3.login(token=token)


def github_me(conn):
# pylint: disable=invalid-name
def me(conn):
    """Return my GitHub account name."""
    return conn.me().login


def create_repo(  # pylint: disable=too-many-arguments
    conn, name, description="", homepage="", has_issues=False, has_wiki=False
):
    """Create a new GitHub repository under the login namespace."""
    return conn.create_repository(
        name, description, homepage, has_issues=has_issues, has_wiki=has_wiki
    )


def get_repo(conn, name):
    """Return the GitHub repo object with the given name."""
    me = github_me(conn)  # pylint: disable=invalid-name
    for i in conn.repositries_by(me, type="owner"):
        if i.name == name:
            return i
    raise Exception(f"Could not find the {name} GitHub repository.")


def read_only_github(conn, name):
    """Make a GitHub repository read-only."""
    repo = get_repo(conn, name)
    repo.edit(name, archived=True)
def empty_commit(repository):
    """Commit an empty commit."""
    raise NotImplementedError
+73 −55
Original line number Diff line number Diff line
@@ -2,7 +2,34 @@

import os
import re
import gitlab  # pylint: disable=import-error
import gitlab  # pylint: disable=import-error,useless-suppression
import gitlab.exceptions  # pylint: disable=import-error
from . import git


def http_url(conn):
    """Return the HTTP url to the GitLab instance."""
    return conn.get_url()


def ssh_url(conn):
    """Return the SSH url to the GitLab instance."""
    return f'git@{http_url(conn).removeprefix("https://").removesuffix("/")}:'


def url_to_name(conn, url):
    """Get the full name from the GitLab URL."""
    return (
        url.removeprefix(http_url(conn))
        .removeprefix(ssh_url(conn.git))
        .removesuffix("/")
        .removesuffix(".git")
    )


def is_gitlab_url(conn, url):
    """Return is the URL for a GitLab repository."""
    return url.startswith(http_url(conn)) or url.startswith(ssh_url(conn))


def name_to_path(name):
@@ -17,11 +44,38 @@ def get_url():
    ).removesuffix("api/v4")


def get_remotes(conn):
    """Returns a list of all the GitLab remotes.

    Very similar to the get_all_remotes function from the general git module,
    but just the GitLab remotes and a bit more information.
    """
    remotes = git.get_all_remotes()
    if remotes is None:
        return None

    gl_remotes = {
        name: remote
        for name, remote in remotes.items()
        if is_gitlab_url(conn, remote["url"])
    }
    for name, remote in gl_remotes.items():
        try:
            gl_remotes[name]["project"] = conn.projects.get(
                url_to_name(conn, remote["url"])
            )
        except gitlab.exceptions.GitlabGetError:
            pass

    return gl_remotes


def connect():
    """Return the GitLab object."""
    try:
        token = os.environ["GITLAB_TOKEN"]
    except KeyError:
        # pylint: disable-next=raise-missing-from
        raise Exception("GITLAB_TOKEN environment variable not set.")
    url = get_url()
    conn = gitlab.Gitlab(url=url, private_token=token)
@@ -29,57 +83,21 @@ def connect():
    return conn


def get_group(conn, name):
    """Return the GitLab group object with the given name."""
    for group in conn.groups.list(all=True):
        if group.name == name:
            return group
    return None


def create_group(conn, name, visibility=None, description=None):
    """Create a new GitLab group and return that object."""
    data = {
        "name": name,
        "path": name_to_path(name),
        "visibility": "public" if visibility is None else visibility,
    }
    if description is not None:
        data["description"] = description
    return conn.groups.create(data)


def get_project(conn, group, name):
    """Returns a GitLab project."""
# pylint: disable=invalid-name
    g = get_group(conn, group)
    if g is None:
        return None
    for p in g.projects.list(all=True):
        if p.name == name:
            return p
    return None


def create_project(conn, name, group=None, description=None, visibility=None):
    """Create a new GitLab project and return that object."""
    # pylint: disable=invalid-name
    data = {
        "name": name,
def me(conn):
    """Return my GitLab account name."""
    return conn.user.username


def empty_commit(project):
    """Commit an empty commit."""
    return project.commit.create(
        {
            "id": project.id,
            "branch": project.default_branch,
            "commit_message": "Initial empty commit.",
            "actions": [],
            "author_email": git.author_email(),
            "author_name": git.author_name(),
        }
    if group is not None:
        g = get_group(conn, group)
        if g is None:
            return None
        data["namespace_id"] = g.id
    if description is not None:
        data["description"] = description
    if visibility is not None:
        data["visibility"] = visibility
    return conn.projects.create(data)


def read_only_project(conn, group, name):
    """Make a GitLab project read-only."""
    project = get_project(conn, group, name)
    project.archive()
    )