diff --git a/osc/commands/fork.py b/osc/commands/fork.py index b703e168..affa49b3 100644 --- a/osc/commands/fork.py +++ b/osc/commands/fork.py @@ -1,3 +1,4 @@ +import re import sys import urllib.parse @@ -44,27 +45,58 @@ class ForkCommand(osc.commandline.OscCommand): self.add_argument_new_repo_name() + self.add_argument( + "--no-devel-project", + action="store_true", + help="Fork the specified package instead the package from the devel project (which is the place where the package is developed)", + ) + def run(self, args): from osc import conf as osc_conf from osc import gitea_api from osc import obs_api + from osc.git_scm import GitStore from osc.output import tty - is_package = args.package is not None + # make a copy of project, package; if we change them, the original values remain in args + project = args.project + package = args.package + + is_package = package is not None + use_devel_project = False if not is_package and args.target_package: self.parser.error("The '--target-package' option requires the 'package' argument to be set") + if not is_package and args.no_devel_project: + self.parser.error("The '--no-devel-project' option can be used only when forking a package") + if is_package: # 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 element must be set in the package meta)" - ) + pkg = obs_api.Package.from_api(args.apiurl, project, package) + + if not args.no_devel_project: + # devel project is not set in package meta as usual but we parse it from "OBS:RejectBranch" attribute + attributes = obs_api.Attributes.from_api(args.apiurl, project, package, attr="OBS:RejectBranch").attribute_list + if attributes: + attribute = attributes[0].value + # the pattern starts with a non-greedy match so we capture the first url + match = re.match(r".*?(https://[^ ]+).*", attribute) + if match: + devel_project_url = match.group(1) + build_project = GitStore.get_build_project(devel_project_url) + # override the package we're cloning with the one from the devel project + use_devel_project = True + project = build_project + pkg = obs_api.Package.from_api(args.apiurl, project, package) + + if not pkg.scmsync: + print(f"{tty.colorize('ERROR', 'red,bold')}: Forking is possible only with packages managed in Git (the element must be set in the package meta)") + sys.exit(1) + else: # get the project meta from the OBS API first - project = obs_api.Project.from_api(args.apiurl, args.project) + project = obs_api.Project.from_api(args.apiurl, project) if not project.scmsync: raise RuntimeError( "Forking is possible only with projects managed in Git (the element must be set in the project meta)" @@ -72,11 +104,16 @@ class ForkCommand(osc.commandline.OscCommand): # parse gitea url, owner, repo and branch from the scmsync url if is_package: - parsed_scmsync_url = urllib.parse.urlparse(package.scmsync, scheme="https") + parsed_scmsync_url = urllib.parse.urlparse(pkg.scmsync, scheme="https") else: parsed_scmsync_url = urllib.parse.urlparse(project.scmsync, scheme="https") url = urllib.parse.urlunparse((parsed_scmsync_url.scheme, parsed_scmsync_url.netloc, "", "", "", "")) owner, repo = parsed_scmsync_url.path.strip("/").split("/") + + # remove trailing ".git" from repo + if repo.endswith(".git"): + repo = repo[:-4] + # temporary hack to allow people using fork atm at all, when packages # are managed via git project. # fallback always to default branch for now, but we actually need to @@ -125,15 +162,17 @@ class ForkCommand(osc.commandline.OscCommand): print() if is_package: - print(f"Forking OBS package {args.project}/{args.package} ...") + print(f"Forking OBS package {project}/{package} ...") + if use_devel_project: + print(f" * {tty.colorize('NOTE', 'bold')}: Forking from the devel project instead of the specified {args.project}/{args.package}") else: - print(f"Forking OBS project {args.project} ...") + print(f"Forking OBS project {project} ...") print(f" * OBS apiurl: {args.apiurl}") # we use a single API endpoint for forking both projects and packages (project requires setting package to "_project") status = obs_api.Package.cmd_fork( args.apiurl, - args.project, - args.package if is_package else "_project", + project, + package if is_package else "_project", scmsync=fork_scmsync, target_project=args.target_project, target_package=args.target_package if is_package else None, diff --git a/osc/git_scm/store.py b/osc/git_scm/store.py index e3f4bc1d..c18866cc 100644 --- a/osc/git_scm/store.py +++ b/osc/git_scm/store.py @@ -28,6 +28,66 @@ class GitStore: return False return store.is_package + @staticmethod + def get_build_project(git_repo_url: str): + """ + Get the project we use for building from _ObsPrj git repo. + The _ObsPrj is located under the same owner as the repo with the package. + They share the same branch. + """ + import tempfile + + from osc import gitea_api + + # parse the git_repo_url (which usually corresponds with the url of the 'origin' remote of the local git repo) + scheme, netloc, path, params, query, fragment = gitea_api.Git.urlparse(git_repo_url) + + # scheme + host + gitea_host = urllib.parse.urlunparse((scheme, netloc, "", None, None, None)) + + # OBS and Gitea usernames are identical + # XXX: we're using the configured apiurl; it would be great to have a mapping from -G/--gitea-login to -A/--apiurl so we don't have to provide -A on the command-line + apiurl = osc_conf.config["apiurl"] + gitea_user = osc_conf.get_apiurl_usr(apiurl) + + # remove trailing ".git" from path + if path.endswith(".git"): + path = path[:-4] + + gitea_owner, gitea_repo = path.strip("/").split("/")[-2:] + + # replace gitea_repo with _ObsPrj + gitea_repo = "_ObsPrj" + + # XXX: we assume that the _ObsPrj project has the same branch as the package + gitea_branch = fragment + + gitea_conf = gitea_api.Config() + try: + gitea_login = gitea_conf.get_login_by_url_user(url=gitea_host, user=gitea_user) + except gitea_api.Login.DoesNotExist: + # matching login entry doesn't exist in git-obs config + return None + + gitea_conn = gitea_api.Connection(gitea_login) + + with tempfile.TemporaryDirectory(prefix="osc_devel_project_git") as tmp_dir: + try: + gitea_api.Repo.clone(gitea_conn, gitea_owner, gitea_repo, branch=gitea_branch, quiet=True, directory=tmp_dir) + project_build_path = os.path.join(tmp_dir, "project.build") + with open(project_build_path, "r", encoding="utf-8") as f: + project_build = f.readline().strip() + return project_build + except gitea_api.GiteaException: + # "_ObsPrj" repo doesn't exist + return None + except subprocess.CalledProcessError: + # branch doesn't exist + return None + except FileNotFoundError: + # "project.build" file doesn't exist + return None + @property def git_project_dir(self): if not hasattr(self, "_git_project_dir"): @@ -135,6 +195,10 @@ class GitStore: if self.is_package and self.project_obs_scm_store: # read project from parent directory that contains a project with .osc metadata self._project = self.project_obs_scm_store.project + if not self._project: + # read project from Gitea (identical owner, repo: _ObsPrj, file: project.build) + origin = self._run_git(["remote", "get-url", "origin"]) + self._project = self.get_build_project(origin) if not self._project: # HACK: assume openSUSE:Factory project if project metadata is missing self._project = "openSUSE:Factory" diff --git a/osc/gitea_api/conf.py b/osc/gitea_api/conf.py index c9e310fe..98d19ee9 100644 --- a/osc/gitea_api/conf.py +++ b/osc/gitea_api/conf.py @@ -121,9 +121,31 @@ class Config: """ Return ``Login`` object for the given ``url`` and ``user``. """ + from .git import Git + + # exact match for login in self.list_logins(): if (login.url, login.user) == (url, user): return login + + def url_to_hostname(value): + netloc = Git.urlparse(value).netloc + + # remove user from hostname + if "@" in netloc: + netloc = netloc.split("@")[-1] + + # remove port from hostname + if ":" in netloc: + netloc = netloc.split(":")[0] + + return netloc + + # match only hostname (netloc without 'user@' and ':port') + user + for login in self.list_logins(): + if (url_to_hostname(login.url), login.user) == (url_to_hostname(url), user): + return login + raise Login.DoesNotExist(url=url, user=user) def add_login(self, login: Login): diff --git a/osc/gitea_api/git.py b/osc/gitea_api/git.py index da39f0b8..665a0542 100644 --- a/osc/gitea_api/git.py +++ b/osc/gitea_api/git.py @@ -9,6 +9,38 @@ from typing import Tuple class Git: + @staticmethod + def urlparse(url) -> urllib.parse.ParseResult: + """ + Parse git url. + + Supported formats: + - https://example.com/owner/repo.git + - https://example.com:1234/owner/repo.git + - example.com/owner/repo.git + - user@example.com:owner/repo.git + - user@example.com:1234:owner/repo.git" + """ + # try ssh clone url first + pattern = r"(?P[^@:]+@[^@:]+(:[0-9]+)?):(?P.+)" + match = re.match(pattern, url) + if match: + scheme = "" + netloc = match.groupdict()["netloc"] + path = match.groupdict()["path"] + params = '' + query = '' + fragment = '' + result = urllib.parse.ParseResult(scheme, netloc, path, params, query, fragment) + return result + + result = urllib.parse.urlparse(url) + if not result.netloc: + # empty netloc is most likely an error, prepend and then discard scheme to trick urlparse() + result = urllib.parse.urlparse("https://" + url) + result = urllib.parse.ParseResult("", *list(result)[1:]) + return result + def __init__(self, workdir): self.abspath = os.path.abspath(workdir) diff --git a/osc/gitea_api/repo.py b/osc/gitea_api/repo.py index 659f743a..87849edb 100644 --- a/osc/gitea_api/repo.py +++ b/osc/gitea_api/repo.py @@ -44,6 +44,8 @@ class Repo: owner: str, repo: str, *, + branch: Optional[str] = None, + quiet: bool = False, directory: Optional[str] = None, cwd: Optional[str] = None, anonymous: bool = False, @@ -112,6 +114,13 @@ class Repo: # clone cmd = ["git", "clone", clone_url, directory] + + if branch: + cmd += ["--branch", branch] + + if quiet: + cmd += ["--quiet"] + subprocess.run(cmd, cwd=cwd, env=env, check=True) # setup remotes diff --git a/osc/obs_api/__init__.py b/osc/obs_api/__init__.py index 9010c732..14a4aa56 100644 --- a/osc/obs_api/__init__.py +++ b/osc/obs_api/__init__.py @@ -1,3 +1,4 @@ +from .attributes import Attributes from .keyinfo import Keyinfo from .package import Package from .package_sources import PackageSources diff --git a/osc/obs_api/attributes.py b/osc/obs_api/attributes.py new file mode 100644 index 00000000..0dc07b62 --- /dev/null +++ b/osc/obs_api/attributes.py @@ -0,0 +1,46 @@ +from typing import Any + +from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import + + +class Attribute(XmlModel): + name: str = Field( + xml_attribute=True, + ) + namespace: str = Field( + xml_attribute=True, + ) + value: str = Field( + xml_set_text=True, + xml_wrapped=True, + ) + + +class Attributes(XmlModel): + XML_TAG = "attributes" + + attribute_list: List[Attribute] = Field( + xml_name="attribute", + ) + + @classmethod + def from_api( + cls, apiurl: str, project: str, package: Optional[str] = None, *, attr: Optional[str] = None + ) -> "Attributes": + import urllib.error + from .. import oscerr + from ..connection import http_request + from ..core import makeurl + + if package: + url_path = ["source", project, package, "_attribute"] + else: + url_path = ["source", project, "_attribute"] + + if attr: + url_path.append(attr) + + url_query: Dict[str, Any] = {} + url = makeurl(apiurl, url_path, url_query) + response = http_request("GET", url) + return cls.from_file(response) diff --git a/tests/test_gitea_api_git.py b/tests/test_gitea_api_git.py new file mode 100644 index 00000000..7cf2369e --- /dev/null +++ b/tests/test_gitea_api_git.py @@ -0,0 +1,36 @@ +import unittest + +from osc.gitea_api import Git + + +class TestGiteaApiGit(unittest.TestCase): + def test_urlparse(self): + # https url without port + url = "https://example.com/owner/repo.git" + result = Git.urlparse(url) + self.assertEqual(list(result), ['https', 'example.com', '/owner/repo.git', '', '', '']) + + # https url with port + url = "https://example.com:1234/owner/repo.git" + result = Git.urlparse(url) + self.assertEqual(list(result), ['https', 'example.com:1234', '/owner/repo.git', '', '', '']) + + # url without scheme + # urllib.parse.urlparse() would normally return ['', '', 'example.com/owner/repo.git', '', '', ''] + url = "example.com/owner/repo.git" + result = Git.urlparse(url) + self.assertEqual(list(result), ['', 'example.com', '/owner/repo.git', '', '', '']) + + # ssh url + url = "user@example.com:owner/repo.git" + result = Git.urlparse(url) + self.assertEqual(list(result), ['', 'user@example.com', 'owner/repo.git', '', '', '']) + + # ssh url with port + url = "user@example.com:1234:owner/repo.git" + result = Git.urlparse(url) + self.assertEqual(list(result), ['', 'user@example.com:1234', 'owner/repo.git', '', '', '']) + + +if __name__ == "__main__": + unittest.main()