Refactor code handling _scmsync.obsinfo to obs_api.scmsync_obsinfo.ScmsyncObsinfo class

Author: dmach

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

osc/obs_api/scmsync_obsinfo.py

@@ -0,0 +1,56 @@
+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())
+        data = file.read()
+        if isinstance(data, bytes):
+            data = data.decode("utf-8")
+        return cls.from_string(data)
+
+    @classmethod
+    def from_api(cls, apiurl: str, project: str, package: str, *, rev: str) -> "ScmsyncObsinfo":
+        import urllib.error
+        from .. import oscerr
+        from ..connection import http_request
+        from ..core import makeurl
+
+        url_path = ["source", project, package, "_scmsync.obsinfo"]
+        url_query = {"rev": rev}
+        url = makeurl(apiurl, url_path, url_query)
+        try:
+            response = http_request("GET", url)
+        except urllib.error.HTTPError as e:
+            if e.status == 404:
+                raise oscerr.NotFoundAPIError(f"File '_scmsync.obsinfo' was not found in {project}/{package}, rev={rev}")
+            raise
+        return cls.from_file(response)