Merge pull request openSUSE#1963 from dmach/git-obs-improve-meta

Improve handling meta in git-obs

Author: dmach

osc/commands_git/meta_info.py

@@ -0,0 +1,100 @@
+import osc.commandline_git
+
+
+class MetaInfoCommand(osc.commandline_git.GitObsCommand):
+    """
+    Resolve and print metadata about the current checkout
+    """
+
+    name = "info"
+    parent = "MetaCommand"
+
+    def init_arguments(self):
+        group = self.parser.add_mutually_exclusive_group()
+        group.add_argument(
+            "--type",
+            action="store_true",
+            help="Print type",
+        )
+        group.add_argument(
+            "--apiurl",
+            action="store_true",
+            help="Print apiurl",
+        )
+        group.add_argument(
+            "--project",
+            action="store_true",
+            help="Print project",
+        )
+        group.add_argument(
+            "--package",
+            action="store_true",
+            help="Print package",
+        )
+        group.add_argument(
+            "--branch",
+            action="store_true",
+            help="Print branch",
+        )
+        group.add_argument(
+            "--commit",
+            action="store_true",
+            help="Print commit",
+        )
+        group.add_argument(
+            "--remote",
+            action="store_true",
+            help="Print remote",
+        )
+        group.add_argument(
+            "--remote-url",
+            action="store_true",
+            help="Print remote_url",
+        )
+        group.add_argument(
+            "--export",
+            action="store_true",
+            help="Show json objects instead of human readable text",
+        )
+
+    def run(self, args):
+        from osc.git_scm.store import GitStore
+
+        store = GitStore(".")
+        result = {
+            "type": store.type,
+            "apiurl": store.apiurl,
+            "project": store.project,
+            "package": store.package,
+            "branch": store._git.current_branch,
+            "commit": store._git.get_branch_head(),
+            "remote": store._git.get_current_remote(),
+            "remote_url": store._git.get_remote_url(),
+        }
+        if args.export:
+            import json
+
+            print(json.dumps(result, indent=4))
+        elif args.type:
+            print(result["type"] or "")
+        elif args.apiurl:
+            print(result["apiurl"] or "")
+        elif args.project:
+            print(result["project"] or "")
+        elif args.package:
+            print(result["package"] or "")
+        elif args.branch:
+            print(result["branch"] or "")
+        elif args.commit:
+            print(result["commit"] or "")
+        elif args.remote:
+            print(result["remote"] or "")
+        elif args.remote_url:
+            print(result["remote_url"] or "")
+        else:
+            from osc.output import KeyValueTable
+
+            table = KeyValueTable(min_key_length=10)
+            for key, value in result.items():
+                table.add(key, value)
+            print(str(table))

osc/commands_git/meta_pull.py

@@ -13,64 +13,17 @@ def init_arguments(self):
         pass
 
     def run(self, args):
-        from osc import gitea_api
-        from osc.git_scm.configuration import Configuration
-        from osc.git_scm.manifest import Manifest
-        from osc.git_scm.store import LocalGitStore
+        from osc.git_scm.store import GitStore
         from osc.output import KeyValueTable
 
         self.print_gitea_settings()
 
-        store = LocalGitStore(".")
-
-        apiurl = None
-        project = None
-
-        # read apiurl and project from _manifest that lives in <owner>/_ObsPrj, matching <branch>
-        # XXX: when the target branch doesn't exist, file from the default branch is returned
-        if store.is_package:
-            try:
-                owner, _ = store._git.get_owner_repo()
-                repo = "_ObsPrj"
-                branch = store._git.current_branch
-
-                url = self.gitea_conn.makeurl("repos", owner, repo, "raw", "_manifest", query={"ref": branch})
-                response = self.gitea_conn.request("GET", url)
-                if response.data:
-                    manifest = Manifest.from_string(response.data.decode("utf-8"))
-                    if manifest.obs_apiurl:
-                        apiurl = manifest.obs_apiurl
-                    if manifest.obs_project:
-                        project = manifest.obs_project
-            except gitea_api.GiteaException as e:
-                if e.status != 404:
-                    raise
-
-        # read apiurl from the global configuration in obs/configuration. branch
-        if not apiurl:
-            try:
-                url = self.gitea_conn.makeurl("repos", "obs", "configuration", "raw", "configuration.yaml", query={"ref": "main"})
-                response = self.gitea_conn.request("GET", url)
-                if response.data:
-                    configuration = Configuration.from_string(response.data.decode("utf-8"))
-                    if configuration.obs_apiurl:
-                        apiurl = configuration.obs_apiurl
-            except gitea_api.GiteaException as e:
-                if e.status != 404:
-                    raise
-
-        if apiurl:
-            store.apiurl = apiurl
-
-        if project:
-            store.project = project
-
+        store = GitStore(".", check=False)
         branch = store._git.current_branch
-        meta = store._read_meta(branch=branch).dict()
-        meta.pop("header", None)
+        changed = store.pull(self.gitea_conn)
 
         table = KeyValueTable(min_key_length=10)
         table.add("Branch", branch, color="bold")
-        for key, value in meta.items():
+        for key, value in changed.items():
             table.add(key, value)
         print(str(table))

osc/commands_git/repo_clone.py

@@ -18,6 +18,12 @@ class RepoCloneCommand(osc.commandline_git.GitObsCommand):
     def init_arguments(self):
         self.add_argument_owner_repo(nargs="+")
 
+        self.add_argument(
+            "-b",
+            "--branch",
+            help="Checkout the specified branch",
+        )
+
         self.add_argument(
             "-a",
             "--anonymous",
@@ -46,7 +52,7 @@ def init_arguments(self):
 
     def run(self, args):
         from osc import gitea_api
-
+        from osc.git_scm.store import GitStore
         from osc.output import tty
 
         self.print_gitea_settings()
@@ -59,16 +65,19 @@ def run(self, args):
         for owner, repo in args.owner_repo:
             print(f"Cloning git repo {owner}/{repo} ...", file=sys.stderr)
             try:
-                gitea_api.Repo.clone(
+                repo_path = gitea_api.Repo.clone(
                     self.gitea_conn,
                     owner,
                     repo,
+                    branch=args.branch,
                     directory=args.directory,
                     use_http=args.anonymous or self.gitea_login.git_uses_http,
                     add_remotes=True,
                     ssh_private_key_path=args.ssh_key or self.gitea_login.ssh_key,
                     ssh_strict_host_key_checking=not(args.no_ssh_strict_host_key_checking),
                 )
+                store = GitStore(repo_path, check=False)
+                store.pull(self.gitea_conn)
                 num_entries += 1
             except gitea_api.GiteaException as e:
                 if e.status == 404:

osc/git_scm/store.py

@@ -5,6 +5,7 @@
 from pathlib import Path
 import typing
 import urllib.parse
+from typing import Dict
 from typing import List
 from typing import Optional
 from typing import Tuple
@@ -16,7 +17,8 @@
 
 
 if typing.TYPE_CHECKING:
-    from .core import Repo
+    from ..core import Repo
+    from ..obs_api import GiteaConnection
 
 
 class Header(BaseModel):
@@ -539,6 +541,10 @@ def _resolve_meta(self, field_name: str, *, allow_none: bool = False):
                 with open(path, "r", encoding="utf-8") as f:
                     result = f.read().strip() or None
 
+        # package = directory name; package must be part of a project checkout
+        if result is None and self.is_package and self.project_store is not None and field_name == "package":
+            result = os.path.basename(self.topdir)
+
         # package = repo name from the current remote url
         if result is None and self.is_package and field_name == "package":
             remote_url = self._git.get_remote_url()
@@ -570,3 +576,69 @@ def package(self) -> Optional[str]:
     @property
     def scmurl(self) -> Optional[str]:
         return self._git.get_remote_url()
+
+    def pull(self, gitea_conn) -> Dict[str, Optional[str]]:
+        from osc.git_scm.configuration import Configuration
+        from osc.git_scm.manifest import Manifest
+        from .. import gitea_api
+
+        apiurl = None
+        project = None
+
+        # read apiurl and project from _manifest that lives in <owner>/_ObsPrj, matching <branch>
+        # XXX: when the target branch doesn't exist, file from the default branch is returned
+        if self.is_package:
+            owner, _ = self._git.get_owner_repo()
+            repo = "_ObsPrj"
+            branch = self._git.current_branch
+
+            try:
+                url = gitea_conn.makeurl("repos", owner, repo, "raw", "_manifest", query={"ref": branch})
+                response = gitea_conn.request("GET", url)
+                if response.data:
+                    manifest = Manifest.from_string(response.data.decode("utf-8"))
+                    if manifest.obs_apiurl:
+                        apiurl = manifest.obs_apiurl
+                    if manifest.obs_project:
+                        project = manifest.obs_project
+            except gitea_api.GiteaException as e:
+                if e.status != 404:
+                    raise
+
+            if not project:
+                try:
+                    url = gitea_conn.makeurl("repos", owner, repo, "raw", "project.build", query={"ref": branch})
+                    response = gitea_conn.request("GET", url)
+                    if response.data:
+                        value = response.data.decode("utf-8").strip()
+                        if value:
+                            project = value
+                except gitea_api.GiteaException as e:
+                    if e.status != 404:
+                        raise
+
+        # read apiurl from the global configuration in obs/configuration, 'main' branch, 'configuration.yaml' file
+        if not apiurl:
+            try:
+                url = gitea_conn.makeurl("repos", "obs", "configuration", "raw", "configuration.yaml", query={"ref": "main"})
+                response = gitea_conn.request("GET", url)
+                if response.data:
+                    configuration = Configuration.from_string(response.data.decode("utf-8"))
+                    if configuration.obs_apiurl:
+                        apiurl = configuration.obs_apiurl
+            except gitea_api.GiteaException as e:
+                if e.status != 404:
+                    raise
+
+        if apiurl:
+            self.set_apiurl(apiurl)
+
+        if project:
+            self.set_project(project)
+
+        # return the values we've set
+        result = {
+            "apiurl": apiurl,
+            "project": project,
+        }
+        return result

osc/gitea_api/git.py

@@ -8,6 +8,8 @@
 from typing import Optional
 from typing import Tuple
 
+from . import exceptions
+
 
 class SshParseResult(urllib.parse.ParseResult):
     """
@@ -154,7 +156,10 @@ def get_branch_head(self, branch: Optional[str] = None) -> str:
         if not branch:
             branch = self.current_branch
 
-        return self._run_git(["rev-parse", f"refs/heads/{branch}"])
+        try:
+            return self._run_git(["rev-parse", f"refs/heads/{branch}"], mute_stderr=True)
+        except subprocess.CalledProcessError:
+            raise exceptions.GitObsRuntimeError(f"Unable to retrieve HEAD from branch '{branch}'. Does the branch exist?")
 
     def branch_exists(self, branch: str) -> bool:
         try:
@@ -262,6 +267,8 @@ def fetch(self, name: Optional[str] = None):
 
     def get_owner_repo(self, remote: Optional[str] = None) -> Tuple[str, str]:
         remote_url = self.get_remote_url(name=remote)
+        if not remote_url:
+            raise exceptions.GitObsRuntimeError("Couldn't determine owner and repo due to a missing remote")
         return self.get_owner_repo_from_url(remote_url)
 
     @staticmethod