Skip to content

Commit

Permalink
fix, refact: fix GitError exception instantiation, update find_first_…
Browse files Browse the repository at this point in the history 
…branch_matching_pattern return
  • Loading branch information
@renatav
renatav committed Feb 3, 2024
1 parent 8e10776 commit 8c781a2
Show file tree
Hide file tree
Showing 2 changed files with 64 additions and 33 deletions.
94 changes: 62 additions & 32 deletions taf/git.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -290,7 +290,8 @@ def all_commits_on_branch(
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not list commits. pygit repository could not be instantiated."
self,
message="Could not list commits. pygit repository could not be instantiated.",
)
if branch:
branch_obj = repo.branches.get(branch)
Expand Down Expand Up @@ -332,7 +333,8 @@ def all_commits_since_commit(
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not list commits. pygit repository could not be instantiated."
self,
message="Could not list commits. pygit repository could not be instantiated.",
)
if branch:
branch_obj = repo.branches.get(branch)
Expand Down Expand Up @@ -361,7 +363,9 @@ def add_remote(
try:
if self.remote_exists(upstream_name):
if raise_error_if_exists:
raise GitError(f"Remote {upstream_name} already exists")
raise GitError(
self, message=f"Remote {upstream_name} already exists"
)
return
self._git("remote add {} {}", upstream_name, upstream_url)
except GitError as e:
Expand All @@ -375,7 +379,8 @@ def branches(
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not list branches. pygit repository could not be instantiated."
self,
message="Could not list branches. pygit repository could not be instantiated.",
)
if all:
branches = set(repo.branches)
Expand Down Expand Up @@ -405,7 +410,8 @@ def branches_containing_commit(
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not list branches. pygit repository could not be instantiated."
self,
message="Could not list branches. pygit repository could not be instantiated.",
)
local_branches = remote_branches = []
try:
Expand Down Expand Up @@ -443,7 +449,8 @@ def branch_exists(
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not check if branch exists. pygit repository could not be instantiated"
self,
message="Could not check if branch exists. pygit repository could not be instantiated",
)
branch = repo.branches.get(branch_name)
# this git command should return the branch's name if it exists
Expand Down Expand Up @@ -471,7 +478,8 @@ def branch_off_commit(self, branch_name: str, commit_sha: str) -> None:
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not create a new branch. pygit repo could not be instantiated"
self,
message="Could not create a new branch. pygit repo could not be instantiated",
)
try:
commit = repo[commit_sha]
Expand Down Expand Up @@ -507,7 +515,8 @@ def checkout_branch(
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not checkout branch. pygit repository could not be instantiated"
self,
message="Could not checkout branch. pygit repository could not be instantiated",
)
try:
branch = repo.lookup_branch(branch_name)
Expand Down Expand Up @@ -552,7 +561,8 @@ def checkout_paths(self, commit_sha: str, *args) -> None:
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not checkout paths. pygit repository could not be instantiated."
self,
message="Could not checkout paths. pygit repository could not be instantiated.",
)
commit = repo.get(commit_sha)
repo.checkout_tree(commit, paths=list(args))
Expand Down Expand Up @@ -677,12 +687,13 @@ def create_and_checkout_branch(
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not create a new branch. pygit repo could not be instantiated"
self,
message="Could not create a new branch. pygit repo could not be instantiated",
)
try:
branch = repo.lookup_branch(branch_name)
if branch is not None and raise_error_if_exists:
raise GitError(f"Branch {branch_name} already exists")
raise GitError(self, message=f"Branch {branch_name} already exists")
try:
commit = repo.revparse_single("HEAD")
repo.branches.local.create(branch_name, commit)
Expand Down Expand Up @@ -711,7 +722,8 @@ def create_branch(self, branch_name: str, commit: Optional[str] = None) -> None:
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not create branch. pygit repository could not be instantiated."
self,
message="Could not create branch. pygit repository could not be instantiated.",
)
try:
if commit is not None:
Expand Down Expand Up @@ -780,7 +792,8 @@ def commit_before_commit(self, commit: str) -> Optional[str]:
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not find commit. pygit repository could not be instantiated."
self,
message="Could not find commit. pygit repository could not be instantiated.",
)
repo_commit_id = repo.get(commit).id
for comm in repo.walk(repo_commit_id):
Expand All @@ -795,7 +808,8 @@ def delete_local_branch(self, branch_name: str) -> None:
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not delete branch. pygit repository could not be instantiated."
self,
message="Could not delete branch. pygit repository could not be instantiated.",
)
repo.branches.delete(branch_name)
except KeyError:
Expand Down Expand Up @@ -824,7 +838,8 @@ def get_commit_date(self, commit_sha: str) -> str:
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not get commit message. pygit repository could not be instantiated."
self,
message="Could not get commit message. pygit repository could not be instantiated.",
)
commit = repo.get(commit_sha)
date = datetime.datetime.utcfromtimestamp(
Expand All @@ -838,7 +853,8 @@ def get_commit_message(self, commit_sha: str) -> str:
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not get commit message. pygit repository could not be instantiated."
self,
message="Could not get commit message. pygit repository could not be instantiated.",
)
commit = repo.get(commit_sha)
return commit.message
Expand Down Expand Up @@ -919,7 +935,8 @@ def head_commit_sha(self) -> Optional[str]:
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not find head commit sha. pygit repository could not be instantiated."
self,
message="Could not find head commit sha. pygit repository could not be instantiated.",
)
try:
return repo.revparse_single("HEAD").id.hex
Expand Down Expand Up @@ -953,13 +970,14 @@ def find_first_branch_matching_pattern(
pattern_func: Callable[[str], bool],
include_remotes: bool = False,
sort_key_func: Optional[Callable[[str], bool]] = None,
) -> Tuple[Optional[str], List[str]]:
) -> Optional[str]:

branch_tips = {}
repo: pygit2.Repository = self.pygit_repo
if repo is None:
raise GitError(
"Could not find first branch matching pattern. pygit repository could not be instantiated."
self,
message="Could not find first branch matching pattern. pygit repository could not be instantiated.",
)

# Obtain the branch reference
Expand All @@ -969,7 +987,9 @@ def find_first_branch_matching_pattern(
# Get the target commit of the branch
branch_target = branch_ref.target
else:
raise GitError(f"Branch {traverse_branch_name} does not exist")
raise GitError(
self, message=f"Branch {traverse_branch_name} does not exist"
)

branches = self.branches(all=include_remotes)
all_branch_names = []
Expand Down Expand Up @@ -998,15 +1018,16 @@ def find_first_branch_matching_pattern(
if tip_hex is not None and (
commit.hex == tip_hex or repo.descendant_of(tip_hex, commit.hex)
):
return branch_name, []
return None, all_branch_names
return branch_name
return None

def get_current_branch(self, full_name: Optional[bool] = False) -> str:
"""Return current branch."""
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not find current branch. pygit repository could not be instantiated."
self,
message="Could not find current branch. pygit repository could not be instantiated.",
)
branch = repo.lookup_reference("HEAD").resolve()
if full_name:
Expand Down Expand Up @@ -1036,7 +1057,8 @@ def get_merge_base(self, branch1: str, branch2: str) -> str:
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not find merge base. pygit repository could not be instantiated."
self,
message="Could not find merge base. pygit repository could not be instantiated.",
)
commit1 = self.top_commit_of_branch(branch1)
commit2 = self.top_commit_of_branch(branch2)
Expand Down Expand Up @@ -1073,7 +1095,8 @@ def is_remote_branch(self, branch_name: str) -> bool:
def is_bare_repository(self) -> bool:
if self.pygit_repo is None:
raise GitError(
"Could determine if bare repository. pygit repository could not be instantiated."
self,
message="Could determine if bare repository. pygit repository could not be instantiated.",
)
return self.pygit_repo.is_bare

Expand Down Expand Up @@ -1107,7 +1130,8 @@ def list_changed_files_at_revision(self, commit: str) -> List[str]:
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not list changes. pygit repository could not be instantiated."
self,
message="Could not list changes. pygit repository could not be instantiated.",
)
commit1 = repo.get(commit)
commit2 = self.commit_before_commit(commit)
Expand All @@ -1127,7 +1151,8 @@ def list_commits(self, branch: Optional[str] = "") -> List[pygit2.Commit]:
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not list commits. pygit repository could not be instantiated."
self,
message="Could not list commits. pygit repository could not be instantiated.",
)
if branch:
branch_obj = repo.branches.get(branch)
Expand All @@ -1154,7 +1179,8 @@ def list_n_commits(
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not list commits. pygit repository could not be instantiated."
self,
message="Could not list commits. pygit repository could not be instantiated.",
)
if start_commit_sha is not None:
start_commit_id = repo.get(start_commit_sha).id
Expand Down Expand Up @@ -1237,7 +1263,8 @@ def merge_branch(
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could merge branch. pygit repository could not be instantiated."
self,
message="Could merge branch. pygit repository could not be instantiated.",
)
branch = repo.lookup_branch(branch_name)
oid = branch.target
Expand Down Expand Up @@ -1342,14 +1369,16 @@ def synced_with_remote(
remote_url = self.get_remote_url()
if remote_url is None:
raise GitError(
"URL not specified and could not be automatically determined. Cannot check if synced"
self,
message="URL not specified and could not be automatically determined. Cannot check if synced",
)
urls = [remote_url]

branch_name = branch or self.default_branch
if branch_name is None:
raise GitError(
"Branch not specified and default branch could not be determined"
self,
message="Branch not specified and default branch could not be determined",
)
for url in urls:
tracking_branch = self.get_tracking_branch(branch_name, strip_remote=True)
Expand All @@ -1370,7 +1399,8 @@ def top_commit_of_branch(self, branch_name: str) -> Optional[str]:
repo = self.pygit_repo
if repo is None:
raise GitError(
"Could not find top commit. pygit repository could not be instantiated."
self,
message="Could not find top commit. pygit repository could not be instantiated.",
)
branch = repo.branches.get(branch_name)
if branch is not None:
Expand Down
3 changes: 2 additions & 1 deletion taf/repositoriesdb.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,8 @@ def check_if_repositories_json_exists(
commit = auth_repo.head_commit_sha()
if commit is None:
raise GitError(
"Could not check if repositroies.json exists. Commit is not specified and head commit could not be determined"
auth_repo,
message="Could not check if repositroies.json exists. Commit is not specified and head commit could not be determined",
)
try:
auth_repo.get_json(commit, REPOSITORIES_JSON_PATH)
Expand Down

0 comments on commit 8c781a2

Please sign in to comment.