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")