Add failing test as xfail

backend/tests/fixtures/repos/conflicting-branches/initial__main/commit1/test.gql

@@ -0,0 +1,10 @@
+query tags {
+  BuiltinTag {
+    edges {
+      node {
+    id
+    hfid
+     }
+   }
+ }
+}

backend/tests/fixtures/repos/conflicting-branches/initial__main/commit2/test.gql

@@ -0,0 +1,10 @@
+ query tags_query {
+  BuiltinTag {
+    edges {
+      node {
+	id
+	hfid
+      }
+    }
+  }
+}

backend/tests/fixtures/repos/conflicting-branches/pr_0001__change1/commit1/base_commit

@@ -0,0 +1 @@
+main-step1

backend/tests/fixtures/repos/conflicting-branches/pr_0001__change1/commit1/test.gql

@@ -0,0 +1,10 @@
+query my_tags {
+ BuiltinTag {
+   edges {
+     node {
+   id
+   hfid
+     }
+   }
+ }
+}

backend/tests/helpers/file_repo.py

@@ -33,8 +33,10 @@ def repo(self) -> Repo:
     def _initial_directory(self, repo_base: Path) -> str:
         initial_candidates = list(repo_base.glob("initial__*"))
         assert len(initial_candidates) == 1
-        initial_directory = str(initial_candidates[0]).replace(f"{repo_base}/", "")
+
+        initial_directory = str(initial_candidates[0].relative_to(repo_base))
         _, branch = initial_directory.split("__")
+
         self._initial_branch = self._initial_branch or branch
         self._branches.append(self._initial_branch)
         return initial_directory
@@ -67,3 +69,67 @@ def __post_init__(self) -> None:
     @property
     def path(self) -> str:
         return str(self.sources_directory / self.name)
+
+
+@dataclass
+class MultipleStagesFileRepo(FileRepo):
+    """Redefines methods to support more complex repository workflow with multiple commits and branching with these."""
+
+    def _initial_directory(self, repo_base: Path) -> str:
+        initial_candidates = list(repo_base.glob("initial__*"))
+        assert len(initial_candidates) == 1
+
+        initial_directory = str(initial_candidates[0].relative_to(repo_base))
+        _, branch = initial_directory.split("__")
+
+        self._initial_branch = self._initial_branch or branch
+        self._branches.append(self._initial_branch)
+
+        return initial_directory
+
+    def _setup_initial_branch(self, directory: Path) -> None:
+        """Setup the initial branch with multiple commits."""
+        initial_commit_folders = sorted(directory.glob("commit*"))
+        for i, commit_folder in enumerate(initial_commit_folders, start=1):
+            shutil.copytree(commit_folder, self.sources_directory / self.name, dirs_exist_ok=True)
+            self.repo.git.add(".")
+            self.repo.index.commit(f"Step {i}")
+            # Tag commit for later reference
+            self.repo.create_tag(f"{self._initial_branch}-step{i}")
+
+    def _apply_pull_requests(self, repo_base: Path) -> None:
+        pull_requests = sorted(list(repo_base.glob("pr*")))
+        for pull_request in pull_requests:
+            branch = str(pull_request).split("__")[-1]
+            base_commit_path = pull_request / "base_commit"
+            base_commit: str | None = None
+
+            if base_commit_path.exists():
+                base_commit = base_commit_path.read_text().strip()
+
+            if branch in self._branches:
+                self.repo.git.checkout(branch)
+            else:
+                # Checkout the base commit or fallback to the initial branch
+                self.repo.git.checkout(base_commit or self._initial_branch)
+                self.repo.git.checkout("-b", branch)
+                self._branches.append(branch)
+
+            # Apply changes and create multiple commits if specified
+            commit_folders = sorted(pull_request.glob("commit*"))
+            for i, commit_folder in enumerate(commit_folders, start=1):
+                shutil.copytree(commit_folder, self.sources_directory / self.name, dirs_exist_ok=True)
+                self.repo.git.add(".")
+                self.repo.git.commit("-m", f"{pull_request} step {i}")
+
+    def __post_init__(self) -> None:
+        repo_base = self.local_repo_base_path / self.name
+        initial_directory = self._initial_directory(repo_base=repo_base)
+
+        shutil.copytree(repo_base / initial_directory, self.sources_directory / self.name)
+        self._repo = Repo.init(self.sources_directory / self.name, initial_branch=self._initial_branch)
+
+        self._setup_initial_branch(directory=repo_base / initial_directory)
+        self._apply_pull_requests(repo_base=repo_base)
+
+        self.repo.git.checkout(self._initial_branch)

backend/tests/unit/git/test_git_repository.py

@@ -30,6 +30,7 @@
 from infrahub.git.worktree import Worktree
 from infrahub.utils import find_first_file_in_directory
 from tests.conftest import TestHelper
+from tests.helpers.file_repo import MultipleStagesFileRepo
 from tests.helpers.test_client import dummy_async_request
 
 
@@ -297,6 +298,19 @@ async def test_create_branch_in_git_not_in_remote(git_repo_01: InfrahubRepositor
     assert len(worktrees) == 3
 
 
+@pytest.mark.xfail(reason="Failing at reproducing conflicts without remote branches to trigger the function to test")
+async def test_has_conflicting_changes(git_repos_source_dir_module_scope: Path):
+    test_repo = MultipleStagesFileRepo(name="conflicting-branches", sources_directory=git_repos_source_dir_module_scope)
+    repository = await InfrahubRepository.new(
+        id=UUIDT.new(),
+        name=test_repo.name,
+        location=test_repo.path,
+        client=InfrahubClient(config=Config(requester=dummy_async_request)),
+    )
+
+    assert repository.has_conflicting_changes(target_branch="main", source_branch="change1")
+
+
 async def test_pull_branch(git_repo_04: InfrahubRepository):
     repo = git_repo_04
     await repo.fetch()