Handle some git exceptions when initializing GitDagBundle (apache#46253)

* Handle some git exceptions when initializing GitDagBundle

Permission error raises GitCommandError, and if the cloned path is manually removed,
it will result in a path not found error when cloning from the bare repo path

* Apply suggestions from code review

Co-authored-by: Jed Cunningham <66968678+jedcunningham@users.noreply.github.com>

* fixup! Apply suggestions from code review

* Fix comment

* Update airflow/dag_processing/bundles/git.py

Co-authored-by: Jed Cunningham <66968678+jedcunningham@users.noreply.github.com>

* fix test

---------

Co-authored-by: Jed Cunningham <66968678+jedcunningham@users.noreply.github.com>

Author: ephraimbuddy

airflow/dag_processing/bundles/git.py

@@ -23,7 +23,7 @@
 from urllib.parse import urlparse
 
 from git import Repo
-from git.exc import BadName
+from git.exc import BadName, GitCommandError, NoSuchPathError
 
 from airflow.dag_processing.bundles.base import BaseDagBundle
 from airflow.exceptions import AirflowException
@@ -157,10 +157,14 @@ def initialize(self) -> None:
     def _clone_repo_if_required(self) -> None:
         if not os.path.exists(self.repo_path):
             self.log.info("Cloning repository to %s from %s", self.repo_path, self.bare_repo_path)
-            Repo.clone_from(
-                url=self.bare_repo_path,
-                to_path=self.repo_path,
-            )
+            try:
+                Repo.clone_from(
+                    url=self.bare_repo_path,
+                    to_path=self.repo_path,
+                )
+            except NoSuchPathError as e:
+                # Protection should the bare repo be removed manually
+                raise AirflowException("Repository path: %s not found", self.bare_repo_path) from e
 
         self.repo = Repo(self.repo_path)
 
@@ -169,12 +173,15 @@ def _clone_bare_repo_if_required(self) -> None:
             raise AirflowException(f"Connection {self.git_conn_id} doesn't have a host url")
         if not os.path.exists(self.bare_repo_path):
             self.log.info("Cloning bare repository to %s", self.bare_repo_path)
-            Repo.clone_from(
-                url=self.repo_url,
-                to_path=self.bare_repo_path,
-                bare=True,
-                env=self.hook.env,
-            )
+            try:
+                Repo.clone_from(
+                    url=self.repo_url,
+                    to_path=self.bare_repo_path,
+                    bare=True,
+                    env=self.hook.env,
+                )
+            except GitCommandError as e:
+                raise AirflowException("Error cloning repository") from e
         self.bare_repo = Repo(self.bare_repo_path)
 
     def _ensure_version_in_bare_repo(self) -> None:

tests/dag_processing/test_dag_bundles.py

@@ -17,12 +17,14 @@
 
 from __future__ import annotations
 
+import re
 import tempfile
 from pathlib import Path
 from unittest import mock
 
 import pytest
 from git import Repo
+from git.exc import GitCommandError, NoSuchPathError
 
 from airflow.dag_processing.bundles.base import BaseDagBundle
 from airflow.dag_processing.bundles.git import GitDagBundle, GitHook
@@ -477,3 +479,30 @@ def test_view_url_returns_none_when_no_version_in_view_url(self, mock_gitrepo):
         )
         view_url = bundle.view_url(None)
         assert view_url is None
+
+    @mock.patch("airflow.dag_processing.bundles.git.GitHook")
+    def test_clone_bare_repo_git_command_error(self, mock_githook):
+        mock_githook.return_value.repo_url = "git@github.com:apache/airflow.git"
+        mock_githook.return_value.env = {}
+
+        with mock.patch("airflow.dag_processing.bundles.git.Repo.clone_from") as mock_clone:
+            mock_clone.side_effect = GitCommandError("clone", "Simulated error")
+            bundle = GitDagBundle(name="test", tracking_ref="main")
+            with pytest.raises(
+                AirflowException,
+                match=re.escape("Error cloning repository"),
+            ):
+                bundle.initialize()
+
+    @mock.patch("airflow.dag_processing.bundles.git.GitHook")
+    def test_clone_repo_no_such_path_error(self, mock_githook):
+        mock_githook.return_value.repo_url = "git@github.com:apache/airflow.git"
+
+        with mock.patch("airflow.dag_processing.bundles.git.os.path.exists", return_value=False):
+            with mock.patch("airflow.dag_processing.bundles.git.Repo.clone_from") as mock_clone:
+                mock_clone.side_effect = NoSuchPathError("Path not found")
+                bundle = GitDagBundle(name="test", tracking_ref="main")
+                with pytest.raises(AirflowException) as exc_info:
+                    bundle._clone_repo_if_required()
+
+                assert "Repository path: %s not found" in str(exc_info.value)