Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 8 additions & 6 deletions lib/crewai/src/crewai/cli/git.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,10 @@
from functools import lru_cache
import subprocess


class Repository:
def __init__(self, path: str = ".") -> None:
self.path = path
self._is_git_repo_cache: bool | None = None

if not self.is_git_installed():
raise ValueError("Git is not installed or not found in your PATH.")
Expand Down Expand Up @@ -40,22 +40,24 @@ def status(self) -> str:
encoding="utf-8",
).strip()

@lru_cache(maxsize=None) # noqa: B019
def is_git_repo(self) -> bool:
"""Check if the current directory is a git repository.

Notes:
- TODO: This method is cached to avoid redundant checks, but using lru_cache on methods can lead to memory leaks
Uses an instance-level cache to avoid redundant subprocess calls
without the memory leak caused by @lru_cache on instance methods.
"""
if self._is_git_repo_cache is not None:
return self._is_git_repo_cache
try:
subprocess.check_output(
["git", "rev-parse", "--is-inside-work-tree"], # noqa: S607
cwd=self.path,
encoding="utf-8",
)
return True
self._is_git_repo_cache = True
except subprocess.CalledProcessError:
return False
self._is_git_repo_cache = False
return self._is_git_repo_cache or False

def has_uncommitted_changes(self) -> bool:
"""Check if the repository has uncommitted changes."""
Expand Down
32 changes: 32 additions & 0 deletions lib/crewai/tests/cli/test_git.py
Original file line number Diff line number Diff line change
Expand Up @@ -99,3 +99,35 @@ def test_origin_url(fp, repository):
stdout="https://github.com/user/repo.git\n",
)
assert repository.origin_url() == "https://github.com/user/repo.git"


def test_is_git_repo_caches_result(fp):
"""Calling is_git_repo multiple times should only invoke subprocess once."""
fp.register(["git", "--version"], stdout="git version 2.30.0\n")
fp.register(["git", "rev-parse", "--is-inside-work-tree"], stdout="true\n")
fp.register(["git", "fetch"], stdout="")

repo = Repository(path=".")
# __init__ already called is_git_repo once via constructor
assert repo.is_git_repo() is True

# Second call should return cached value without a new subprocess.
# If caching were broken a second subprocess would be spawned, but
# only one "rev-parse" was registered so it would raise.
assert repo.is_git_repo() is True


def test_is_git_repo_no_global_cache_leak(fp):
"""Repository instances must be garbage-collectable (no global lru_cache holding refs)."""
import gc
import weakref

fp.register(["git", "--version"], stdout="git version 2.30.0\n")
fp.register(["git", "rev-parse", "--is-inside-work-tree"], stdout="true\n")
fp.register(["git", "fetch"], stdout="")

repo = Repository(path=".")
weak = weakref.ref(repo)
del repo
gc.collect()
assert weak() is None, "Repository instance was not garbage collected"
Loading