From c670c93d0fa96fbd24d34c7f3525cf6ce786e056 Mon Sep 17 00:00:00 2001 From: Daniel Mach Date: Mon, 7 Oct 2024 09:55:40 +0200 Subject: [PATCH] Refactor code handling _scmsync.obsinfo to obs_api.scmsync_obsinfo.ScmsyncObsinfo class --- osc/core.py | 14 +++--------- osc/obs_api/scmsync_obsinfo.py | 42 ++++++++++++++++++++++++++++++++++ 2 files changed, 45 insertions(+), 11 deletions(-) create mode 100644 osc/obs_api/scmsync_obsinfo.py diff --git a/osc/core.py b/osc/core.py index 2799939af..fcd31c831 100644 --- a/osc/core.py +++ b/osc/core.py @@ -3160,17 +3160,9 @@ def checkout_package( if revision is not None: # search for the git sha sum based on the OBS DISTURL package source revision # we need also take into account that the url was different at that point of time - url = shasum = None - u = makeurl(apiurl, ['source', project, package, '_scmsync.obsinfo'], {'rev': revision}) - f = http_GET(u) - for line in f.readlines(): - if line.startswith(b"revision: "): - shasum = line[10:].rstrip() - if line.startswith(b"url: "): - url = line[5:].rstrip() - if shasum is None: - raise oscerr.OscIOError(None, 'Unable to find git shasum for given revision') - scm_url = url + b'#' + shasum + from .obs_api.scmsync_obsinfo import ScmsyncObsinfo + scmsync_obsinfo = ScmsyncObsinfo.from_api(apiurl, project, package, rev=revision) + scm_url = f"{scmsync_obsinfo.url}#{scmsync_obsinfo.revision}" os.putenv("OSC_VERSION", get_osc_version()) run_external(['/usr/lib/obs/service/obs_scm_bridge', '--outdir', directory, '--url', scm_url]) diff --git a/osc/obs_api/scmsync_obsinfo.py b/osc/obs_api/scmsync_obsinfo.py new file mode 100644 index 000000000..0f32a59b5 --- /dev/null +++ b/osc/obs_api/scmsync_obsinfo.py @@ -0,0 +1,42 @@ +import typing + +from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import + + +class ScmsyncObsinfo(BaseModel): + """ + Class for handling _scmsync.obsinfo files + """ + + mtime: int = Field() + commit: str = Field() + url: str = Field() + revision: str = Field() + + @classmethod + def from_string(cls, data: str) -> "ScmsyncObsinfo": + kwargs = {} + for line in data.splitlines(): + line = line.strip() + if not line: + continue + key, value = line.split(": ", 1) + field = cls.__fields__.get(key, None) + if field and field.type is int: + value = int(value) + kwargs[key] = value + return cls(**kwargs) + + @classmethod + def from_file(cls, file: Union[str, typing.IO]) -> "ScmsyncObsinfo": + if isinstance(file, str): + with open(file, "r", encoding="utf-8") as f: + return cls.from_string(f.read()) + return cls.from_string(file.read()) + + @classmethod + def from_api(cls, apiurl: str, project: str, package: str, *, rev: str) -> "ScmsyncObsinfo": + url_path = ["source", project, package, "_scmsync.obsinfo"] + url_query = {"rev": rev} + response = cls.xml_request("GET", apiurl, url_path, url_query) + return cls.from_file(response, apiurl=apiurl)