Implement osc-git command with several subcommands

osc-git.py

@@ -0,0 +1,49 @@
+#!/usr/bin/env python3
+
+import sys
+
+import osc.commandline
+import osc.commands_git
+from osc import oscerr
+from osc.output import print_msg
+
+
+class OscGitMainCommand(osc.commandline.MainCommand):
+    name = "osc-git"
+
+    MODULES = (
+        ("osc.commands_git", osc.commands_git.__path__[0]),
+    )
+
+    def init_arguments(self):
+        pass
+
+    def post_parse_args(self, args):
+        pass
+
+    @classmethod
+    def main(cls, argv=None, run=True):
+        """
+        Initialize OscMainCommand, load all commands and run the selected command.
+        """
+        cmd = cls()
+        cmd.load_commands()
+        if run:
+            args = cmd.parse_args(args=argv)
+            exit_code = cmd.run(args)
+            sys.exit(exit_code)
+        else:
+            args = None
+        return cmd, args
+
+
+def main():
+    try:
+        OscGitMainCommand.main()
+    except oscerr.OscBaseError as e:
+        print_msg(str(e), print_to="error")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

osc/commands_git/clone.py

@@ -0,0 +1,44 @@
+import osc.commandline
+
+from . import common
+
+
+class CloneCommand(osc.commandline.OscCommand):
+    """
+    Clone a project or a package
+    """
+
+    name = "clone"
+
+    def init_arguments(self):
+        common.cmd_add_login(self)
+        common.cmd_add_owner(self)
+        common.cmd_add_repo(self)
+
+        self.add_argument(
+            "-a",
+            "--anonymous",
+            action="store_true",
+            default=None,
+            help="Clone anonymously via the http protocol",
+        )
+
+        self.add_argument(
+            "--directory",
+            help="Clone into the given directory",
+        )
+
+    def run(self, args):
+        from osc import gitea_api
+
+        conf = gitea_api.Config()
+        login = conf.get_login(name=args.gitea_login_name)
+        conn = gitea_api.Connection(login)
+        gitea_api.clone_repo(
+            conn,
+            args.owner,
+            args.repo,
+            directory=args.directory,
+            anonymous=args.anonymous,
+            add_remotes=True,
+        )

osc/commands_git/common.py

@@ -0,0 +1,60 @@
+import osc.commandline
+
+
+# GIT / GITEA
+
+
+def cmd_add_login(cmd: osc.commandline.OscCommand):
+    # TODO: option name? make a global option?
+    cmd.add_argument(
+        "-G",
+        "--gitea-login-name",
+        help="Name of the login entry in the config file",
+    )
+
+
+def cmd_add_owner(cmd: osc.commandline.OscCommand):
+    cmd.add_argument(
+        "owner",
+        help="Name of the repository owner (login, org)",
+    )
+
+
+def cmd_add_repo(cmd: osc.commandline.OscCommand):
+    cmd.add_argument(
+        "repo",
+        help="Name of the repository",
+    )
+
+
+def cmd_add_new_repo_name(cmd: osc.commandline.OscCommand):
+    cmd.add_argument(
+        "--new-repo-name",
+        help="Name of the newly forked repo",
+    )
+
+
+# OBS
+
+
+def cmd_add_apiurl(cmd: osc.commandline.OscCommand):
+    cmd.add_argument(
+        "-A",
+        "--apiurl",
+        metavar="URL",
+        help="Open Build Service API URL or a configured alias",
+    )
+
+
+def cmd_add_project(cmd: osc.commandline.OscCommand):
+    cmd.add_argument(
+        "project",
+        help="Name of the OBS project",
+    )
+
+
+def cmd_add_package(cmd: osc.commandline.OscCommand):
+    cmd.add_argument(
+        "package",
+        help="Name of the OBS package",
+    )

osc/commands_git/fork.py

@@ -0,0 +1,46 @@
+import osc.commandline
+
+from . import common
+
+
+# TODO: move 'fork' and 'clone' commands under 'repo' command?
+
+
+class ForkCommand(osc.commandline.OscCommand):
+    """
+    Fork a package that is managed in Git
+    """
+
+    name = "fork"
+
+    def init_arguments(self):
+        common.cmd_add_login(self)
+        common.cmd_add_owner(self)
+        common.cmd_add_repo(self)
+        common.cmd_add_new_repo_name(self)
+
+    def run(self, args):
+        import urllib.parse
+        from osc import conf as osc_conf
+        from osc import gitea_api
+        from osc.output import print_msg
+
+        conf = gitea_api.Config()
+        login = conf.get_login(args.gitea_login_name)
+
+        print_msg(f"Forking git repo {args.owner}/{args.repo} ...", print_to="stderr")
+        print_msg(f" * URL: {login.url}", print_to="stderr")
+        print_msg(f" * User: {login.user}", print_to="stderr")
+
+        conn = gitea_api.Connection(login)
+
+        try:
+            response = gitea_api.fork_repo(conn, args.owner, args.repo, new_repo_name=args.new_repo_name)
+            repo = response.json()
+            fork_owner = repo["owner"]["login"]
+            fork_repo = repo["name"]
+            print_msg(f" * Fork created: {fork_owner}/{fork_repo}", print_to="stderr")
+        except gitea_api.ForkExists as e:
+            fork_owner = e.fork_owner
+            fork_repo = e.fork_repo
+            print_msg(f" * Fork already exists: {fork_owner}/{fork_repo}", print_to="stderr")

osc/commands_git/fork_obs_package.py

@@ -0,0 +1,99 @@
+import osc.commandline
+
+from . import common
+
+
+class ForkObsPackageCommand(osc.commandline.OscCommand):
+    """
+    Fork an OBS package that is managed in Git
+    """
+
+    name = "fork-obs-package"
+
+    def init_arguments(self):
+        common.cmd_add_apiurl(self)
+        common.cmd_add_project(self)
+        common.cmd_add_package(self)
+        common.cmd_add_new_repo_name(self)
+
+    def run(self, args):
+        import sys
+        import urllib.parse
+        from osc import conf as osc_conf
+        from osc import gitea_api
+        from osc import obs_api
+        from osc.output import print_msg
+
+        osc_conf.get_config(override_apiurl=args.apiurl)
+        args.apiurl = osc_conf.config.apiurl
+
+        # get the package meta from the OBS API first
+        package = obs_api.Package.from_api(args.apiurl, args.project, args.package)
+        if not package.scmsync:
+            raise RuntimeError(
+                "Forking is possible only with packages managed in Git (the <scmsync> element must be set in the package meta)"
+            )
+
+        # parse gitea url, owner, repo and branch from the scmsync url
+        parsed_scmsync_url = urllib.parse.urlparse(package.scmsync, scheme="https")
+        url = urllib.parse.urlunparse((parsed_scmsync_url.scheme, parsed_scmsync_url.netloc, "", "", "", ""))
+        owner, repo = parsed_scmsync_url.path.strip("/").split("/")
+        branch = parsed_scmsync_url.fragment or None
+
+        conf = gitea_api.Config()
+        # find a credentials entry for url and OBS user (there can be multiple users configured for a single URL in the config file)
+        login = conf.get_login_by_url_user(url=url, user=osc_conf.get_apiurl_usr(args.apiurl))
+        conn = gitea_api.Connection(login)
+
+        print_msg(f"Forking git repo {owner}/{repo} ...", print_to="stderr")
+        print_msg(f" * URL: {login.url}", print_to="stderr")
+        print_msg(f" * User: {login.user}", print_to="stderr")
+
+        # the branch was not specified, fetch the default branch from the repo
+        if branch:
+            fork_branch = branch
+        else:
+            response = gitea_api.get_repo(conn, owner, repo)
+            repo = response.json()
+            branch = repo["default_branch"]
+            fork_branch = branch
+
+        # check if the scmsync branch exists in the source repo
+        parent_branch_data = gitea_api.get_branch(conn, owner, repo, fork_branch).json()
+
+        try:
+            response = gitea_api.fork_repo(conn, owner, repo, new_repo_name=args.new_repo_name)
+            repo = response.json()
+            fork_owner = repo["owner"]["login"]
+            fork_repo = repo["name"]
+            print_msg(f" * Fork created: {fork_owner}/{fork_repo}", print_to="stderr")
+        except gitea_api.ForkExists as e:
+            fork_owner = e.fork_owner
+            fork_repo = e.fork_repo
+            print_msg(f" * Fork already exists: {fork_owner}/{fork_repo}", print_to="stderr")
+
+        # XXX: implicit branch name should be forbidden; assumptions are bad
+        fork_scmsync = urllib.parse.urlunparse(
+            (parsed_scmsync_url.scheme, parsed_scmsync_url.netloc, f"{fork_owner}/{fork_repo}", "", "", fork_branch)
+        )
+
+        print_msg(f"Forking OBS package {args.project}/{args.package} ...", print_to="stderr")
+        print_msg(f" * OBS apiurl: {args.apiurl}", print_to="stderr")
+        status = obs_api.Package.cmd_fork(args.apiurl, args.project, args.package, scmsync=fork_scmsync)
+        target_project = status.data["targetproject"]
+        target_package = status.data["targetpackage"]
+        # XXX: the current OBS API is not ideal; we don't get any info whether the new package exists already; 404 would be probably nicer
+        print_msg(f" * Fork created: {target_project}/{target_package}", print_to="stderr")
+        print_msg(f" * scmsync URL: {fork_scmsync}", print_to="stderr")
+
+        # check if the scmsync branch exists in the forked repo
+        fork_branch_data = gitea_api.get_branch(conn, fork_owner, fork_repo, fork_branch).json()
+
+        parent_commit = parent_branch_data["commit"]["id"]
+        fork_commit = fork_branch_data["commit"]["id"]
+        if parent_commit != fork_commit:
+            print_msg(f"The branch in the forked repo is out of sync with the parent", print_to="error")
+            print_msg(f" * Fork: {fork_owner}/{fork_repo}#{fork_branch}, commit: {fork_commit}", print_to="error")
+            print_msg(f" * Parent: {owner}/{repo}#{fork_branch}, commit: {parent_commit}", print_to="error")
+            print_msg(" * If this is not intentional, please clone the fork and fix the branch manually", print_to="error")
+            sys.exit(1)

osc/commands_git/login.py

@@ -0,0 +1,12 @@
+import osc.commandline
+
+
+class LoginCommand(osc.commandline.OscCommand):
+    """
+    Manage credentials to Gitea servers
+    """
+
+    name = "login"
+
+    def init_arguments(self):
+        pass

osc/commands_git/login_add.py

@@ -0,0 +1,30 @@
+import osc.commandline
+
+
+class LoginAddCommand(osc.commandline.OscCommand):
+    """
+    Add a Gitea credentials entry
+    """
+
+    name = "add"
+    parent = "LoginCommand"
+
+    def init_arguments(self):
+        self.parser.add_argument("name")
+        self.parser.add_argument("--url", required=True)
+        self.parser.add_argument("--user", required=True)
+        self.parser.add_argument("--token", required=True)
+        self.parser.add_argument("--set-as-default", action="store_true")
+
+    def run(self, args):
+        from osc import gitea_api
+
+        print_msg(f"Adding a Gitea credentials entry with name '{args.name}' ...", print_to="stderr")
+
+        conf = gitea_api.GiteaConfig()
+        print_msg(f" * Config path: {conf.path}", print_to="stderr")
+
+        login = gitea_api.Login(name=args.name, url=args.url, user=args.user, token=args.token)
+        conf.add_login(login)
+
+        print_msg(" * Entry added", print_to="stderr")

osc/commands_git/login_list.py

@@ -0,0 +1,21 @@
+import osc.commandline
+
+
+class LoginListCommand(osc.commandline.OscCommand):
+    """
+    List Gitea credentials entries
+    """
+
+    name = "list"
+    parent = "LoginCommand"
+
+    def init_arguments(self):
+        self.parser.add_argument("--show-tokens", action="store_true", help="Show tokens in the output")
+
+    def run(self, args):
+        from osc import gitea_api
+
+        conf = gitea_api.Config()
+        for login in conf.list_logins():
+            print(login.to_human_readable_string(show_token=args.show_tokens))
+            print()

osc/commands_git/login_remove.py

@@ -0,0 +1,26 @@
+import osc.commandline
+
+
+class LoginRemoveCommand(osc.commandline.OscCommand):
+    """
+    Remove a Gitea credentials entry
+    """
+
+    name = "remove"
+    parent = "LoginCommand"
+
+    def init_arguments(self):
+        self.parser.add_argument("name")
+
+    def run(self, args):
+        from osc import gitea_api
+        from osc.output import print_msg
+
+        print_msg(f"Removing a Gitea credentials entry with name '{args.name}' ...", print_to="stderr")
+
+        conf = gitea_api.Config()
+        print_msg(f" * Config path: {conf.path}", print_to="stderr")
+
+        conf.remove_login(args.name)
+
+        print_msg(f" * Entry removed", print_to="stderr")