From 8447840a3d40cf30b8dacc579434ce22b3c1a2d4 Mon Sep 17 00:00:00 2001
From: Dev-iL <6509619+Dev-iL@users.noreply.github.com>
Date: Sat, 25 Apr 2026 23:36:10 +0300
Subject: [PATCH] [v3-2-test] Add tests for git bundle force-push resilience,
 error recovery, etc. (#65750)

Expand GitDagBundle test coverage with 12 new tests addressing gaps
identified during investigation of repeated recloning after tag force-push.

Force-push resilience:
- Branch force-pushed to unrelated commit follows the new ref
- Tag force-pushed to unrelated commit follows the new ref
- Tag moved forward then backward across multiple refreshes
- Refresh after force-push never triggers Repo.clone_from
- Repeated refreshes after force-push remain stable
- Re-initialization reuses existing repos (no reclone)

Error handling:
- Upstream tag deletion does not break refresh (local copy persists)
- Failed bare-repo fetch preserves previous working tree
- Refresh on versioned bundle raises AirflowException
- Corrupted working repo triggers cleanup and retry

Edge cases:
- Real submodule fixture with ref change and submodule sync
- Ambiguous ref (branch+tag same name) documents branch preference
(cherry picked from commit 19ae9ee0c9f1fb9dbbd4013c150b6c07b93ecb66)

Co-authored-by: Dev-iL <6509619+Dev-iL@users.noreply.github.com>
---
 .../git/tests/unit/git/bundles/test_git.py    | 395 ++++++++++++++++++
 scripts/ci/prek/known_airflow_exceptions.txt  |   1 +
 2 files changed, 396 insertions(+)

diff --git a/providers/git/tests/unit/git/bundles/test_git.py b/providers/git/tests/unit/git/bundles/test_git.py
index ade0c30675558..dc4556d3f377e 100644
--- a/providers/git/tests/unit/git/bundles/test_git.py
+++ b/providers/git/tests/unit/git/bundles/test_git.py
@@ -380,6 +380,198 @@ def test_refresh_tag(self, mock_githook, git_repo):
         files_in_repo = {f.name for f in bundle.path.iterdir() if f.is_file()}
         assert {"test_dag.py", "new_test.py"} == files_in_repo
 
+    @mock.patch("airflow.providers.git.bundles.git.GitHook")
+    def test_refresh_tag_force_pushed_to_unrelated_commit(self, mock_githook, git_repo):
+        """Ensure that refresh follows a tag moved to an unrelated commit."""
+        repo_path, repo = git_repo
+        mock_githook.return_value.repo_url = repo_path
+
+        repo.create_tag("release")
+        initial_commit = repo.head.commit
+
+        bundle = GitDagBundle(name="test", git_conn_id=CONN_HTTPS, tracking_ref="release")
+        bundle.initialize()
+        assert bundle.get_current_version() == initial_commit.hexsha
+
+        repo.git.checkout("--orphan", "new-branch")
+        repo.git.rm("-rf", ".")
+
+        new_file = repo_path / "new_file.py"
+        new_file.write_text("unrelated content")
+        repo.index.add([new_file])
+        unrelated_commit = repo.index.commit("Unrelated commit")
+
+        repo.create_tag("release", force=True)
+
+        bundle.refresh()
+
+        assert bundle.get_current_version() == unrelated_commit.hexsha
+
+        files_in_repo = {f.name for f in bundle.path.iterdir() if f.is_file()}
+        assert {"new_file.py"} == files_in_repo
+
+    @mock.patch("airflow.providers.git.bundles.git.GitHook")
+    def test_refresh_branch_force_pushed_to_unrelated_commit(self, mock_githook, git_repo):
+        """Ensure that refresh follows a branch force-pushed to an unrelated commit."""
+        repo_path, repo = git_repo
+        mock_githook.return_value.repo_url = repo_path
+
+        # Create a second branch to track
+        repo.create_head("release")
+        initial_commit = repo.head.commit
+
+        bundle = GitDagBundle(name="test", git_conn_id=CONN_HTTPS, tracking_ref="release")
+        bundle.initialize()
+        assert bundle.get_current_version() == initial_commit.hexsha
+
+        # Force-push "release" to an orphan commit
+        repo.git.checkout("--orphan", "temp-orphan")
+        repo.git.rm("-rf", ".")
+
+        new_file = repo_path / "branch_new.py"
+        new_file.write_text("unrelated branch content")
+        repo.index.add([new_file])
+        unrelated_commit = repo.index.commit("Unrelated branch commit")
+
+        # Point the release branch to this orphan commit
+        repo.git.branch("-f", "release", unrelated_commit.hexsha)
+
+        bundle.refresh()
+
+        assert bundle.get_current_version() == unrelated_commit.hexsha
+
+        files_in_repo = {f.name for f in bundle.path.iterdir() if f.is_file()}
+        assert {"branch_new.py"} == files_in_repo
+
+    @mock.patch("airflow.providers.git.bundles.git.GitHook")
+    def test_refresh_tag_moved_forward_and_backward(self, mock_githook, git_repo):
+        """Ensure refresh follows a tag moved forward then backward across refreshes."""
+        repo_path, repo = git_repo
+        mock_githook.return_value.repo_url = repo_path
+
+        commit_a = repo.head.commit
+        repo.create_tag("moving")
+
+        bundle = GitDagBundle(name="test", git_conn_id=CONN_HTTPS, tracking_ref="moving")
+        bundle.initialize()
+        assert bundle.get_current_version() == commit_a.hexsha
+
+        # Move tag forward to commit B
+        file_b = repo_path / "file_b.py"
+        file_b.write_text("commit b content")
+        repo.index.add([file_b])
+        commit_b = repo.index.commit("Commit B")
+        repo.create_tag("moving", force=True)
+
+        bundle.refresh()
+        assert bundle.get_current_version() == commit_b.hexsha
+        files_in_repo = {f.name for f in bundle.path.iterdir() if f.is_file()}
+        assert {"test_dag.py", "file_b.py"} == files_in_repo
+
+        # Move tag backward to commit A
+        repo.create_tag("moving", ref=commit_a, force=True)
+
+        bundle.refresh()
+        assert bundle.get_current_version() == commit_a.hexsha
+        files_in_repo = {f.name for f in bundle.path.iterdir() if f.is_file()}
+        assert {"test_dag.py"} == files_in_repo
+
+    @mock.patch("airflow.providers.git.bundles.git.GitHook")
+    def test_refresh_after_force_push_does_not_reclone(self, mock_githook, git_repo):
+        """Refresh after force-push must fetch+reset, never clone."""
+        repo_path, repo = git_repo
+        mock_githook.return_value.repo_url = repo_path
+
+        repo.create_tag("release")
+        bundle = GitDagBundle(name="test", git_conn_id=CONN_HTTPS, tracking_ref="release")
+        bundle.initialize()
+
+        # Force-push tag to orphan commit
+        repo.git.checkout("--orphan", "orphan-branch")
+        repo.git.rm("-rf", ".")
+        new_file = repo_path / "orphan_file.py"
+        new_file.write_text("orphan content")
+        repo.index.add([new_file])
+        unrelated_commit = repo.index.commit("Orphan commit")
+        repo.create_tag("release", force=True)
+
+        bare_repo_path = bundle.bare_repo_path
+        working_repo_path = bundle.repo_path
+
+        with mock.patch("airflow.providers.git.bundles.git.Repo.clone_from") as mock_clone:
+            bundle.refresh()
+            mock_clone.assert_not_called()
+
+        # Repos were reused, not recreated
+        assert bare_repo_path.exists()
+        assert working_repo_path.exists()
+        assert bundle.get_current_version() == unrelated_commit.hexsha
+
+    @mock.patch("airflow.providers.git.bundles.git.GitHook")
+    def test_repeated_refreshes_after_force_push_stable(self, mock_githook, git_repo):
+        """Repeated refreshes after a force-push remain stable."""
+        repo_path, repo = git_repo
+        mock_githook.return_value.repo_url = repo_path
+
+        repo.create_tag("release")
+        bundle = GitDagBundle(name="test", git_conn_id=CONN_HTTPS, tracking_ref="release")
+        bundle.initialize()
+
+        # Force-push tag to orphan commit
+        repo.git.checkout("--orphan", "orphan-branch")
+        repo.git.rm("-rf", ".")
+        new_file = repo_path / "stable_file.py"
+        new_file.write_text("stable content")
+        repo.index.add([new_file])
+        new_commit = repo.index.commit("Stable orphan commit")
+        repo.create_tag("release", force=True)
+
+        # First refresh
+        bundle.refresh()
+        assert bundle.get_current_version() == new_commit.hexsha
+        files_after_first = {f.name for f in bundle.path.iterdir() if f.is_file()}
+        assert {"stable_file.py"} == files_after_first
+
+        # Second refresh (no upstream changes)
+        bundle.refresh()
+        assert bundle.get_current_version() == new_commit.hexsha
+        files_after_second = {f.name for f in bundle.path.iterdir() if f.is_file()}
+        assert {"stable_file.py"} == files_after_second
+
+    @mock.patch("airflow.providers.git.bundles.git.GitHook")
+    def test_reinitialize_reuses_repos_after_force_push(self, mock_githook, git_repo):
+        """Re-initialization with a new bundle object reuses existing repos after force-push."""
+        repo_path, repo = git_repo
+        mock_githook.return_value.repo_url = repo_path
+
+        repo.create_tag("release")
+        bundle1 = GitDagBundle(name="test", git_conn_id=CONN_HTTPS, tracking_ref="release")
+        bundle1.initialize()
+
+        # Force-push tag to orphan commit
+        repo.git.checkout("--orphan", "orphan-branch")
+        repo.git.rm("-rf", ".")
+        new_file = repo_path / "reinit_file.py"
+        new_file.write_text("reinit content")
+        repo.index.add([new_file])
+        new_commit = repo.index.commit("Reinit orphan commit")
+        repo.create_tag("release", force=True)
+
+        bundle1.refresh()
+        assert bundle1.get_current_version() == new_commit.hexsha
+
+        # Simulate DAG processor restart: new bundle object, same name
+        bundle2 = GitDagBundle(name="test", git_conn_id=CONN_HTTPS, tracking_ref="release")
+        with mock.patch("airflow.providers.git.bundles.git.Repo.clone_from") as mock_clone:
+            bundle2.initialize()
+            mock_clone.assert_not_called()
+
+        assert bundle2.get_current_version() == new_commit.hexsha
+        files_in_repo = {f.name for f in bundle2.path.iterdir() if f.is_file()}
+        assert {"reinit_file.py"} == files_in_repo
+
+        assert_repo_is_closed(bundle2)
+
     @mock.patch("airflow.providers.git.bundles.git.GitHook")
     def test_head(self, mock_githook, git_repo):
         repo_path, repo = git_repo
@@ -920,6 +1112,106 @@ def test_clone_bare_repo_invalid_repository_error_retry_fails(
             # Verify Repo was called twice (failed attempt + failed retry)
             assert mock_repo_class.call_count == 2
 
+    @mock.patch("airflow.providers.git.bundles.git.GitHook")
+    def test_refresh_survives_upstream_tag_deletion(self, mock_githook, git_repo):
+        """Refresh succeeds when tracked tag is deleted upstream; local copy persists."""
+        repo_path, repo = git_repo
+        mock_githook.return_value.repo_url = repo_path
+        initial_commit = repo.head.commit
+
+        repo.create_tag("ephemeral")
+        bundle = GitDagBundle(name="test", git_conn_id=CONN_HTTPS, tracking_ref="ephemeral")
+        bundle.initialize()
+        assert bundle.get_current_version() == initial_commit.hexsha
+
+        # Delete the tag from the upstream repo
+        repo.delete_tag("ephemeral")
+
+        # Refresh still succeeds because git fetch refspecs don't prune deleted tags
+        bundle.refresh()
+        assert bundle.get_current_version() == initial_commit.hexsha
+
+        files_in_repo = {f.name for f in bundle.path.iterdir() if f.is_file()}
+        assert {"test_dag.py"} == files_in_repo
+
+    @mock.patch("airflow.providers.git.bundles.git.GitHook")
+    def test_refresh_failure_preserves_previous_checkout(self, mock_githook, git_repo):
+        """A failed refresh must not corrupt the previous working tree."""
+        repo_path, repo = git_repo
+        mock_githook.return_value.repo_url = repo_path
+
+        bundle = GitDagBundle(name="test", git_conn_id=CONN_HTTPS, tracking_ref=GIT_DEFAULT_BRANCH)
+        bundle.initialize()
+
+        files_before = {f.name for f in bundle.path.iterdir() if f.is_file()}
+        version_before = bundle.get_current_version()
+
+        # Add a new commit upstream
+        new_file = repo_path / "new_file.py"
+        new_file.write_text("new content")
+        repo.index.add([new_file])
+        repo.index.commit("New commit")
+
+        # Make the bare repo fetch fail — this is the first step in refresh()
+        with mock.patch.object(bundle, "_fetch_bare_repo", side_effect=GitCommandError("fetch", "simulated")):
+            with pytest.raises(GitCommandError):
+                bundle.refresh()
+
+        # Working tree should be unchanged from before the failed refresh
+        files_after = {f.name for f in bundle.path.iterdir() if f.is_file()}
+        assert files_before == files_after
+        assert bundle.get_current_version() == version_before
+
+    @mock.patch("airflow.providers.git.bundles.git.GitHook")
+    def test_refresh_versioned_bundle_raises(self, mock_githook, git_repo):
+        """Calling refresh() on a versioned bundle must raise AirflowException."""
+        repo_path, repo = git_repo
+        mock_githook.return_value.repo_url = repo_path
+        version = repo.head.commit.hexsha
+
+        bundle = GitDagBundle(
+            name="test",
+            git_conn_id=CONN_HTTPS,
+            version=version,
+            tracking_ref=GIT_DEFAULT_BRANCH,
+            prune_dotgit_folder=False,
+        )
+        bundle.initialize()
+
+        with pytest.raises(AirflowException, match="Refreshing a specific version is not supported"):
+            bundle.refresh()
+
+        assert_repo_is_closed(bundle)
+
+    @mock.patch("airflow.providers.git.bundles.git.GitHook")
+    @mock.patch("airflow.providers.git.bundles.git.shutil.rmtree")
+    @mock.patch("airflow.providers.git.bundles.git.os.path.exists")
+    def test_clone_repo_invalid_repository_error_retry(self, mock_exists, mock_rmtree, mock_githook):
+        """Test that InvalidGitRepositoryError on working repo triggers cleanup and retry."""
+        mock_githook.return_value.repo_url = "git@github.com:apache/airflow.git"
+        mock_githook.return_value.env = {}
+
+        mock_exists.return_value = True
+
+        with mock.patch("airflow.providers.git.bundles.git.Repo") as mock_repo_class:
+            # First call to Repo() raises InvalidGitRepositoryError, second succeeds
+            mock_repo_class.side_effect = [
+                InvalidGitRepositoryError("Invalid git repository"),
+                mock.MagicMock(),  # Second attempt succeeds
+            ]
+            mock_repo_class.clone_from = mock.MagicMock()
+
+            bundle = GitDagBundle(name="test", git_conn_id=CONN_HTTPS, tracking_ref="main")
+            bundle.bare_repo_path = "/fake/bare"
+
+            bundle._clone_repo_if_required()
+
+            # Verify cleanup was called for the working repo path
+            mock_rmtree.assert_called_once_with(bundle.repo_path)
+
+            # Verify Repo was called twice (failed attempt + retry)
+            assert mock_repo_class.call_count == 2
+
     @mock.patch("airflow.providers.git.bundles.git.shutil.rmtree")
     @mock.patch("airflow.providers.git.bundles.git.os.path.exists")
     @mock.patch("airflow.providers.git.bundles.git.GitHook")
@@ -1035,3 +1327,106 @@ def test_submodule_fetch_error_raises_runtime_error(
             bundle.initialize()
 
         mock_rmtree.assert_not_called()
+
+    @patch.dict(
+        os.environ,
+        {
+            "GIT_CONFIG_COUNT": "1",
+            "GIT_CONFIG_KEY_0": "protocol.file.allow",
+            "GIT_CONFIG_VALUE_0": "always",
+        },
+    )
+    @mock.patch("airflow.providers.git.bundles.git.GitHook")
+    def test_refresh_with_real_submodules_after_ref_change(self, mock_githook, tmp_path_factory):
+        """Refresh with real submodules works when the main repo changes but submodule ref stays the same."""
+        # Create submodule repo
+        sub_dir = tmp_path_factory.mktemp("subrepo")
+        sub_repo = Repo.init(sub_dir)
+        sub_repo.git.symbolic_ref("HEAD", f"refs/heads/{GIT_DEFAULT_BRANCH}")
+        sub_file = sub_dir / "sub_file.py"
+        sub_file.write_text("sub content")
+        sub_repo.index.add([sub_file])
+        sub_repo.index.commit("Sub initial commit")
+
+        # Create main repo with submodule
+        main_dir = tmp_path_factory.mktemp("mainrepo")
+        main_repo = Repo.init(main_dir)
+        main_repo.git.symbolic_ref("HEAD", f"refs/heads/{GIT_DEFAULT_BRANCH}")
+        main_file = main_dir / "main_file.py"
+        main_file.write_text("main content")
+        main_repo.index.add([main_file])
+        main_repo.index.commit("Main initial commit")
+
+        # Add submodule
+        main_repo.git.submodule("add", str(sub_dir), "mysub")
+        main_repo.index.commit("Add submodule")
+        initial_commit = main_repo.head.commit
+        main_repo.create_tag("v1")
+
+        mock_githook.return_value.repo_url = str(main_dir)
+        mock_githook.return_value.env = {}
+        bundle = GitDagBundle(
+            name="test_submod",
+            git_conn_id=CONN_HTTPS,
+            tracking_ref="v1",
+            submodules=True,
+        )
+        bundle.initialize()
+        assert bundle.get_current_version() == initial_commit.hexsha
+
+        # Verify submodule content is checked out
+        sub_content = (bundle.repo_path / "mysub" / "sub_file.py").read_text()
+        assert sub_content == "sub content"
+
+        # Add a new file to main repo (submodule ref unchanged) and move the tag
+        new_main_file = main_dir / "extra.py"
+        new_main_file.write_text("extra content")
+        main_repo.index.add([new_main_file])
+        new_commit = main_repo.index.commit("Extra file")
+        main_repo.create_tag("v1", force=True)
+
+        bundle.refresh()
+
+        assert bundle.get_current_version() == new_commit.hexsha
+        files_in_repo = {f.name for f in bundle.repo_path.iterdir() if f.is_file()}
+        assert "extra.py" in files_in_repo
+
+        # Submodule content remains intact
+        sub_content = (bundle.repo_path / "mysub" / "sub_file.py").read_text()
+        assert sub_content == "sub content"
+
+    @mock.patch("airflow.providers.git.bundles.git.GitHook")
+    def test_refresh_ambiguous_ref_prefers_branch_over_tag(self, mock_githook, git_repo):
+        """When tracking_ref matches both a branch and tag, refresh follows the branch."""
+        repo_path, repo = git_repo
+        mock_githook.return_value.repo_url = repo_path
+
+        # Create a branch named "ambiguous" at the initial commit
+        repo.create_head("ambiguous")
+
+        # Create a new commit and point a tag named "ambiguous" at it
+        tag_file = repo_path / "tag_file.py"
+        tag_file.write_text("tag content")
+        repo.index.add([tag_file])
+        repo.index.commit("Tag commit")
+        repo.create_tag("ambiguous")
+
+        # Move the "ambiguous" branch to another new commit
+        repo.heads.ambiguous.checkout()
+        branch_file = repo_path / "branch_file.py"
+        branch_file.write_text("branch content")
+        repo.index.add([branch_file])
+        branch_commit = repo.index.commit("Branch commit")
+
+        # Switch back to main so the upstream repo has a clean HEAD
+        repo.heads[GIT_DEFAULT_BRANCH].checkout()
+
+        bundle = GitDagBundle(name="test", git_conn_id=CONN_HTTPS, tracking_ref="ambiguous")
+        bundle.initialize()
+
+        # refresh() prefers origin/ambiguous (branch) over the ambiguous tag
+        bundle.refresh()
+        assert bundle.get_current_version() == branch_commit.hexsha
+
+        files_in_repo = {f.name for f in bundle.path.iterdir() if f.is_file()}
+        assert "branch_file.py" in files_in_repo
diff --git a/scripts/ci/prek/known_airflow_exceptions.txt b/scripts/ci/prek/known_airflow_exceptions.txt
index da07a6302a3a6..815cfb903a1fb 100644
--- a/scripts/ci/prek/known_airflow_exceptions.txt
+++ b/scripts/ci/prek/known_airflow_exceptions.txt
@@ -204,6 +204,7 @@ providers/fab/src/airflow/providers/fab/www/extensions/init_security.py::1
 providers/facebook/src/airflow/providers/facebook/ads/hooks/ads.py::2
 providers/git/src/airflow/providers/git/bundles/git.py::5
 providers/git/src/airflow/providers/git/hooks/git.py::1
+providers/git/tests/unit/git/bundles/test_git.py::1
 providers/github/src/airflow/providers/github/operators/github.py::2
 providers/github/src/airflow/providers/github/sensors/github.py::3
 providers/github/tests/system/github/example_github.py::2
