Daniel Garcia Moreno
4e91302fb6
This command is useful to add or remove reviewers to one or more pull reuqests.
405 lines
12 KiB
Python
405 lines
12 KiB
Python
#!/usr/bin/python3
|
|
|
|
import sys
|
|
import osc.commandline
|
|
import osc.commandline_git
|
|
|
|
|
|
class PackagingSubcommand:
|
|
|
|
name = "default"
|
|
aliases = []
|
|
|
|
def __init__(self, parent):
|
|
self._parent = parent
|
|
self._parser = None
|
|
self._init_arguments = False
|
|
|
|
def init_arguments(self):
|
|
if self._init_arguments:
|
|
return
|
|
|
|
self._parser = self._parent.subparsers.add_parser(
|
|
self.name, aliases=self.aliases, help=self.__class__.__doc__
|
|
)
|
|
|
|
self._init_arguments = True
|
|
|
|
def add_argument(self, *args, **kwargs):
|
|
return self._parser.add_argument(*args, **kwargs)
|
|
|
|
def run(self, args):
|
|
raise NotImplemented
|
|
|
|
@property
|
|
def output(self):
|
|
return self._parent.output
|
|
|
|
@property
|
|
def apiurl(self):
|
|
return self._parent.apiurl
|
|
|
|
@property
|
|
def args(self):
|
|
return self._parent.args
|
|
|
|
@property
|
|
def gitea_conn(self):
|
|
return self._parent.gitea_conn
|
|
|
|
@property
|
|
def user(self):
|
|
return self._parent.user
|
|
|
|
def fork(self, repo, base_branch, new_branch=None):
|
|
from osc import gitea_api
|
|
|
|
owner, repo = repo.owner, repo.repo
|
|
print(f"Forking git repo {owner}/{repo} ...", file=sys.stderr)
|
|
try:
|
|
repo_obj = gitea_api.Fork.create(self.gitea_conn, owner, repo)
|
|
fork_owner = repo_obj.owner
|
|
fork_repo = repo_obj.repo
|
|
print(f" * Fork created: {fork_owner}/{fork_repo}", file=sys.stderr)
|
|
except gitea_api.ForkExists as e:
|
|
fork_owner = e.fork_owner
|
|
fork_repo = e.fork_repo
|
|
print(f" * Fork already exists: {fork_owner}/{fork_repo}", file=sys.stderr)
|
|
except gitea_api.GiteaException as e:
|
|
if e.status == 404:
|
|
print(f" * ERROR: Repo doesn't exist: {owner}/{repo}", file=sys.stderr)
|
|
return None
|
|
raise
|
|
|
|
r = gitea_api.Repo.get(self.gitea_conn, fork_owner, fork_repo)
|
|
if new_branch:
|
|
try:
|
|
gitea_api.Branch.create(
|
|
self.gitea_conn, fork_owner, fork_repo, new_branch_name=new_branch, old_ref_name=base_branch
|
|
)
|
|
except gitea_api.BranchExists:
|
|
print(
|
|
f" * Warning: Branch already exists, not creating it: {fork_owner}/{fork_repo}#{new_branch}",
|
|
file=sys.stderr,
|
|
)
|
|
|
|
branch = new_branch or base_branch
|
|
return f"{r.clone_url}#{branch}"
|
|
|
|
def parse_repo(self, repo):
|
|
"""
|
|
Convert org/repo#branch into gitea repo
|
|
org and branch are optional, default org is "pool" and default
|
|
branch is the default branch configured in gitea.
|
|
|
|
returns [Repo, branch]
|
|
"""
|
|
from osc import gitea_api
|
|
|
|
org = "pool"
|
|
if "/" in repo:
|
|
org, repo = repo.split("/", maxsplit=1)
|
|
if "#" in repo:
|
|
repo, branch = repo.split("#", maxsplit=1)
|
|
else:
|
|
branch = None
|
|
|
|
r = gitea_api.Repo.get(self.gitea_conn, org, repo)
|
|
return r, branch or r.default_branch
|
|
|
|
|
|
class PackagingCommand(osc.commandline.Command):
|
|
"""
|
|
Packaging utilities to work with git <-> OBS
|
|
"""
|
|
|
|
name = "gitw"
|
|
|
|
# inject some git-obs methods to avoid duplicating code
|
|
post_parse_args = osc.commandline_git.GitObsMainCommand.post_parse_args
|
|
|
|
def __init__(self, full_name, parent=None):
|
|
self._subcommands = {}
|
|
|
|
# All declared subcommands
|
|
for k, v in globals().items():
|
|
if type(v) != type(PackagingSubcommand) or k == "PackagingSubcommand":
|
|
continue
|
|
if issubclass(v, PackagingSubcommand):
|
|
command = v(self)
|
|
self._subcommands[v.name] = command
|
|
for alias in v.aliases:
|
|
self._subcommands[alias] = command
|
|
|
|
super().__init__(full_name, parent)
|
|
|
|
def init_arguments(self):
|
|
self.subparsers = self.parser.add_subparsers(dest="command", title="commands")
|
|
|
|
# inherit global options from the main git-obs command
|
|
osc.commandline_git.GitObsMainCommand.init_arguments(self)
|
|
|
|
self.add_argument(
|
|
"-a",
|
|
"--api-url",
|
|
help="OBS api url, defaults to api.opensuse.org",
|
|
default="api.opensuse.org",
|
|
)
|
|
|
|
for command in self._subcommands.values():
|
|
if not command._init_arguments:
|
|
command.init_arguments()
|
|
|
|
def _setup_gitea(self, args):
|
|
from osc import gitea_api
|
|
|
|
self.gitea_conf = gitea_api.Config(args.gitea_config)
|
|
self.gitea_login = self.gitea_conf.get_login(args.gitea_login or self.user)
|
|
self.gitea_conn = gitea_api.Connection(self.gitea_login)
|
|
|
|
def run(self, args):
|
|
from osc import conf
|
|
|
|
self.output = []
|
|
self.apiurl = conf.sanitize_apiurl(args.api_url)
|
|
self.args = args
|
|
|
|
self.user = conf.get_apiurl_usr(self.apiurl)
|
|
self._setup_gitea(args)
|
|
|
|
if not self.args.command:
|
|
self.parser.print_help()
|
|
return
|
|
|
|
cmd = self._subcommands[self.args.command]
|
|
cmd.run(self.args)
|
|
|
|
print()
|
|
for line in self.output:
|
|
print(line, file=sys.stderr)
|
|
|
|
|
|
class TestProjectCommand(PackagingSubcommand):
|
|
"""Create a test project in OBS linking/forking packages from gitea"""
|
|
|
|
name = "testp"
|
|
aliases = ["p"]
|
|
|
|
def init_arguments(self):
|
|
super().init_arguments()
|
|
|
|
self.add_argument(
|
|
"-p",
|
|
"--prj",
|
|
help="OBS base project to copy config from, defaults to openSUSE:Factory",
|
|
)
|
|
self.add_argument(
|
|
"-l",
|
|
"--link",
|
|
action="store_true",
|
|
help="Do not create the project, just link the packages",
|
|
default=False,
|
|
)
|
|
self.add_argument(
|
|
"-b",
|
|
"--branch",
|
|
help="Git branch to use",
|
|
)
|
|
self.add_argument(
|
|
"-f",
|
|
"--fork",
|
|
action="store_true",
|
|
help="Fork gitea pkg repos in your user home, defaults to False",
|
|
default=False,
|
|
)
|
|
self.add_argument(
|
|
"prj_name",
|
|
help="test project name to create in home:USER:$prj_name",
|
|
)
|
|
self.add_argument(
|
|
"pkg_repo",
|
|
nargs="*",
|
|
help="pkg gitea ref, it should be [org/]repo[#branch]",
|
|
)
|
|
|
|
def run(self, args):
|
|
from osc import conf
|
|
|
|
project = f"home:{self.user}:{self.args.prj_name}"
|
|
self.weburl = self.apiurl.replace("api", "build", count=1)
|
|
|
|
if not self.args.link:
|
|
self.create_project(project)
|
|
self.output.append(f"OBS test project created: {project}")
|
|
|
|
self.output.append(f"{self.weburl}/project/show/{project}")
|
|
self.output.append("")
|
|
|
|
for repo in self.args.pkg_repo:
|
|
repo, branch = self.parse_repo(repo)
|
|
package, scm = self.create_package(project, repo, branch)
|
|
self.output.append(f" * Linked pkg: {package} <- {scm}")
|
|
|
|
def create_project(self, project):
|
|
from osc.util.xml import ET
|
|
from osc.core import edit_meta, xml_fromstring, show_project_meta
|
|
|
|
# Get base project meta
|
|
if self.args.prj:
|
|
meta_data = b"".join(show_project_meta(self.apiurl, self.args.prj))
|
|
root = xml_fromstring(meta_data)
|
|
repos = "\n".join(ET.tostring(i).decode() for i in root.findall("repository"))
|
|
else:
|
|
repos = """
|
|
<repository name="openSUSE_Tumbleweed">
|
|
<path project="openSUSE:Tumbleweed" repository="standard"/>
|
|
<arch>x86_64</arch>
|
|
</repository>
|
|
"""
|
|
|
|
data = f"""
|
|
<project name="{project}">
|
|
<title>Test project gitea</title>
|
|
<description/>
|
|
<person userid="{self.user}" role="maintainer"/>
|
|
{repos}
|
|
</project>
|
|
"""
|
|
|
|
# Create the project in OBS
|
|
edit_meta(metatype="prj", data=data, apiurl=self.apiurl, path_args=(project,))
|
|
|
|
def create_package(self, project, repo, branch):
|
|
from osc.core import edit_meta
|
|
|
|
package = repo.repo
|
|
if self.args.fork:
|
|
scm = self.fork(repo, branch, self.args.branch)
|
|
else:
|
|
scm = f"{repo.clone_url}#{branch}"
|
|
|
|
# Create the package in OBS
|
|
data = f"""
|
|
<package name="{package}" project="{project}">
|
|
<title/>
|
|
<description/>
|
|
<person userid="{self.user}" role="maintainer"/>
|
|
<scmsync>{scm}</scmsync>
|
|
</package>
|
|
"""
|
|
|
|
edit_meta(metatype="pkg", data=data, apiurl=self.apiurl, path_args=(project, package))
|
|
|
|
return package, scm
|
|
|
|
|
|
class AddRemotesCommand(PackagingSubcommand):
|
|
"""Add all obs/ibs remotes in the current git clone"""
|
|
|
|
name = "add-remotes"
|
|
aliases = ["ar"]
|
|
|
|
def init_arguments(self):
|
|
PackagingSubcommand.init_arguments(self)
|
|
|
|
self.add_argument(
|
|
"-f",
|
|
"--fork",
|
|
action="store_true",
|
|
help="Fork in your user home, and add the remote defaults to False",
|
|
default=False,
|
|
)
|
|
|
|
self.add_argument(
|
|
"-o",
|
|
"--owner",
|
|
nargs="?",
|
|
help="Add owner remote",
|
|
)
|
|
|
|
def run(self, args):
|
|
from osc.gitea_api import Repo
|
|
from osc.gitea_api.git import Git
|
|
|
|
base = self.gitea_conn.host
|
|
obs = "opensuse" in base
|
|
prefix = "" if obs else "i"
|
|
|
|
remotes = {
|
|
"pool": "gitea@{base}:pool/{package}",
|
|
"fork": "gitea@{base}:{org}/{package}",
|
|
}
|
|
|
|
git = Git(".")
|
|
owner, repo = git.get_owner_repo()
|
|
|
|
if self.args.fork:
|
|
r = Repo.get(self.gitea_conn, "pool", repo)
|
|
self.fork(r, None)
|
|
remote = remotes["fork"].format(base=base, org=self.user, package=repo)
|
|
try:
|
|
git.add_remote(f"{prefix}fork", remote)
|
|
except:
|
|
self.output.append(f"Can't add remote {remote}")
|
|
|
|
if self.args.owner:
|
|
remote = remotes["fork"].format(base=base, org=self.args.owner, package=repo)
|
|
try:
|
|
git.add_remote(self.args.owner, remote)
|
|
except:
|
|
self.output.append(f"Can't add remote {remote}")
|
|
|
|
self.output.append("")
|
|
self.output.extend(git._run_git(["remote", "-v"]).split("\n"))
|
|
|
|
|
|
class ReviewCommand(PackagingSubcommand):
|
|
"""Add/Remove reviewers to a PR"""
|
|
|
|
name = "rrev"
|
|
aliases = ["rr"]
|
|
|
|
def init_arguments(self):
|
|
PackagingSubcommand.init_arguments(self)
|
|
|
|
self.add_argument(
|
|
"-r",
|
|
"--remove",
|
|
action="store_true",
|
|
help="Remove reviewer from PR",
|
|
default=False,
|
|
)
|
|
|
|
self.add_argument(
|
|
"reviewers",
|
|
help="Reviewer gitea id, use a comma separate list to add more than one",
|
|
)
|
|
|
|
self.add_argument(
|
|
"pull_request",
|
|
nargs="+",
|
|
help="pull request gitea ref, it should be [org/]repo[#PR]",
|
|
)
|
|
|
|
def run(self, args):
|
|
method = "POST"
|
|
if args.remove:
|
|
method = "DELETE"
|
|
|
|
reviewers = [i.strip() for i in args.reviewers.split(",")]
|
|
json_data = {"reviewers": reviewers}
|
|
|
|
for pr in args.pull_request:
|
|
r, n = self.parse_repo(pr)
|
|
url = f"repos/{r.owner}/{r.repo}/pulls/{n}/requested_reviewers"
|
|
url = self.gitea_conn.makeurl(url)
|
|
|
|
response = self.gitea_conn.request(
|
|
method=method, url=url, json_data=json_data
|
|
)
|
|
|
|
if args.remove:
|
|
self.output.append(f"Reviewers removed {r.clone_url}/pulls/{n}: {reviewers}")
|
|
else:
|
|
self.output.append(f"Reviewers added {r.clone_url}/pulls/{n}: {reviewers}")
|