Skip to content

Commit

Permalink
Merge pull request #4 from alalkamys/fix/push-changes
Browse files Browse the repository at this point in the history 
Fix: Push Changes on Consecutive Runs to Achieve Idempotency
  • Loading branch information
@alalkamys
alalkamys authored Mar 22, 2024
2 parents 498f6c3 + 2315722 commit 96af9cb
Show file tree
Hide file tree
Showing 3 changed files with 165 additions and 50 deletions.
2 changes: 1 addition & 1 deletion README.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -292,7 +292,7 @@ Below is an explanation of each field in the configuration file:
| `AZURE_DEVOPS_PAT` | Azure DevOps Personal Access Token (PAT) | `None` |
| `GITHUB_TOKEN` | GitHub Personal Access Token (PAT) | `None` |
| `GITHUB_ENTERPRISE_TOKEN` | GitHub Enterprise Personal Access Token (PAT) | `None` |
| `CODE_MIGRATION_ASSISTANT_USER_AGENT` | User agent used for HTTP requests by Code Migration Assistant | `alalkamys`/code-migration-assistant |
| `CODE_MIGRATION_ASSISTANT_USER_AGENT` | User agent used for HTTP requests by Code Migration Assistant | `alalkamys/code-migration-assistant` |

<!--********************* R E F E R E N C E S *********************-->

Expand Down
201 changes: 152 additions & 49 deletions app/helpers/git.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,15 +3,17 @@

from git import Actor
from git import Repo
from git.exc import GitCommandError
from git.exc import NoSuchPathError
from git.refs.head import Head
from git.remote import PushInfo
from git.remote import PushInfoList
from git.exc import GitCommandError
from git.exc import NoSuchPathError
from typing import Any
from typing import Union
import json
import logging
import os
import re
import sys

_logger = logging.getLogger(app_config.APP_NAME)
Expand Down Expand Up @@ -133,7 +135,7 @@ def load_target_repos(repos: list[dict]) -> list[Repo]:
return result


def identity_setup(repo: Repo, actor_username: str, actor_email: str) -> None:
def identity_setup(repo: Repo, actor_username: str, actor_email: str) -> bool:
"""Set up identity configuration for a GitPython repository.
Args:
Expand All @@ -158,37 +160,91 @@ def identity_setup(repo: Repo, actor_username: str, actor_email: str) -> None:
return False


def checkout_branch(repo: Repo, branch_name: str, from_branch: str = None) -> bool:
"""Checkout or create a new branch in the local repository.
def configure_divergent_branches_reconciliation_method(repo: Repo, rebase: bool = False, fast_forward_only: bool = False) -> bool:
"""Configure the reconciliation method for handling divergent branches in a GitPython repository.
This function configures the reconciliation method to handle situations where the local and remote branches have diverged during pull operations.
Args:
repo (Repo): The GitPython repository object.
rebase (bool, optional): If True, set the reconciliation method to rebase. Defaults to False.
fast_forward_only (bool, optional): If True, set the reconciliation method to fast-forward only. Ignored if 'rebase' is True. Defaults to False.
Returns:
bool: True if the reconciliation method was configured successfully, False otherwise.
"""
try:
config_writer = repo.config_writer()
if fast_forward_only:
_logger.debug(
"Setting reconciliation method to fast-forward only..")
config_writer.set_value('pull', 'ff', 'only').release()
elif rebase:
_logger.debug("Setting reconciliation method to rebase..")
config_writer.set_value('pull', 'rebase', 'true').release()
else:
_logger.debug("Setting reconciliation method to merge..")
config_writer.set_value('pull', 'rebase', 'false').release()
del (config_writer)
return True
except Exception as e:
_logger.error(f"An error occurred while setting up reconciliation method: {
str(e).strip()}")
return False


def checkout_branch(repo: Repo, branch_name: str, from_branch: str = None, remote_name: str = "origin") -> bool:
"""Checkout an existing branch or create a new branch in the local repository.
This function checks out an existing branch or creates a new branch in the local repository.
If the specified branch already exists locally, it switches to that branch.
If the branch does not exist locally but exists in the remote repository, it creates a new local branch
tracking the remote branch and switches to it.
If the specified branch does not exist locally or remotely, it attempts to create a new branch
based on the provided base branch (or the current branch if not specified).
Args:
repo (Repo): The GitPython Repo object representing the local repository.
branch_name (str): The name of the branch to checkout or create.
from_branch (str, optional): The name of the base branch to create the new branch from.
If None, create the new branch from the current branch. Defaults to None.
remote_name (str, optional): The name of the remote repository. Defaults to "origin".
Returns:
bool: True if the branch was successfully checked out or created, False otherwise.
"""
try:
remote_branch_name = f"{remote_name}/{branch_name}"
if branch_name in repo.branches:
_logger.info(
f"'{branch_name}' branch already exists, checking out..")
f"'{branch_name}' branch already exists. Switching..")
branch = repo.branches[branch_name]
elif remote_branch_name in repo.refs:
_logger.info(f"'{remote_branch_name}' exists.")
branch = repo.create_head(branch_name, commit=remote_branch_name)
_logger.info(f"Branch '{branch_name}' set up to track '{
remote_branch_name}'")
branch.set_tracking_branch(repo.refs[remote_branch_name])
else:
_logger.info(f"'{branch_name}' doesn't exist, creating..")
from_branch = from_branch or repo.active_branch.name
from_branch = from_branch or get_default_branch_name(repo=repo, remote_name=remote_name)
remote_from_branch = f"{remote_name}/{from_branch}"
if from_branch in repo.branches:
branch = repo.create_head(branch_name, commit=from_branch)
_logger.info(f"Created new branch '{
branch_name}' based on '{from_branch}' branch")
branch_name}' based on '{from_branch}' branch. Switching..")
elif remote_from_branch in repo.refs:
branch = repo.create_head(
branch_name, commit=remote_from_branch)
_logger.info(f"Created new branch '{
branch_name}' based on '{remote_from_branch}' branch. Switching..")
else:
_logger.error(
f"Error: '{from_branch}' based on branch doesn't exist")
return False

branch.checkout()
_logger.info(f"Checked out branch '{branch_name}' successfully.")
_logger.info(f"Switched to branch '{branch_name}' successfully.")
return True

except GitCommandError as e:
Expand Down Expand Up @@ -244,7 +300,7 @@ def commit_changes(repo: Repo, title: str, description: str = None, author: Acto


def push_changes(repo: Repo, remote_name: str = 'origin', remote_branch_name: str | None = None, timeout: int | None = 180) -> bool:
"""Push changes to the remote repository.
"""Push changes to the remote repository, pulling changes from the remote branch if it exists.
Args:
repo (Repo): The GitPython Repo object representing the local repository.
Expand All @@ -264,7 +320,8 @@ def push_changes(repo: Repo, remote_name: str = 'origin', remote_branch_name: st
This function pushes changes from the active local branch to the specified remote branch.
It handles various scenarios such as existing and non-existing remotes, and provides detailed logging
information during the push operation. The timeout parameter allows customization of the maximum time
allowed for the push operation.
allowed for the push operation. Before pushing, if the specified remote branch exists, it pulls changes
from that branch to ensure synchronization.
Example:
# Push changes from the active branch to the 'main' branch of the remote repository 'origin'
Expand All @@ -276,44 +333,69 @@ def push_changes(repo: Repo, remote_name: str = 'origin', remote_branch_name: st
remote = repo.remotes[remote_name]
branch_name = repo.active_branch.name
remote_branch_name = remote_branch_name if remote_branch_name else branch_name
_logger.info(f"Pushing changes to '{
remote_branch_name}' branch of remote '{remote_name}'...")
result: PushInfoList = remote.push(
refspec=f"{branch_name}:{remote_branch_name}", progress=RemoteProgressReporter(_logger), kill_after_timeout=timeout)
try:
assert len(result) != 0
VALID_PUSH_INFO_FLAGS: list[int] = [PushInfo.FAST_FORWARD, PushInfo.NEW_HEAD,
PushInfo.UP_TO_DATE, PushInfo.FORCED_UPDATE, PushInfo.NEW_TAG]
for push_info in result:
_logger.debug("+------------+")
_logger.debug("| Push Info: |")
_logger.debug("+------------+")
_logger.debug(f"Flag: {push_info.flags}")
_logger.debug(f"Local ref: {push_info.local_ref}")
_logger.debug(f"Remote Ref: {push_info.remote_ref}")
_logger.debug(f"Remote ref string: {
push_info.remote_ref_string}")
_logger.debug(f"Old Commit: {push_info.old_commit}")
_logger.debug(f"Summary: {push_info.summary.strip()}")
if push_info.flags not in VALID_PUSH_INFO_FLAGS:
if push_info.flags == PushInfo.ERROR:
_logger.error(
f"Incomplete push error: Push contains rejected heads. Check your internet connection and run in 'debug' mode to see more details.")
else:
_logger.error(
"Unexpected push error, maybe the remote rejected heads. Check your internet connection and run in 'debug' mode to see more details.")
return False
except AssertionError:
_logger.error(f"Pushing changes to remote '{
remote_name}' completely failed. Check your internet connection and run in 'debug' mode to see the remote push progress.")
return False
_logger.info(f"Changes pushed successfully to '{
branch_name}' branch of remote '{remote_name}'.")

_logger.debug(f"Setting '{branch_name}' upstream branch to '{remote_name}/{
remote_branch_name}'..")
repo.active_branch.set_tracking_branch(
repo.refs[f"{remote_name}/{remote_branch_name}"])

push_is_needed = True

remote_refs = remote.refs
if remote_branch_name in remote_refs:
_logger.debug(
f"'{remote_name}/{remote_branch_name}' remote branch exists.")
if not has_tracking_branch(repo.active_branch):
_logger.debug(
f"'{branch_name}' has no tracking branch. Setting..")
repo.active_branch.set_tracking_branch(
repo.refs[f"{remote_name}/{remote_branch_name}"])
_logger.debug(f"Pulling changes from '{
remote_branch_name}' branch of remote '{remote_name}' to '{branch_name}'...")
remote.pull(
refspec=remote_branch_name, kill_after_timeout=timeout)

push_is_needed = needs_push(repo=repo, branch_name=branch_name)

if push_is_needed:
_logger.info(f"Pushing changes to '{
remote_branch_name}' branch of remote '{remote_name}'...")
result: PushInfoList = remote.push(
refspec=f"{branch_name}:{remote_branch_name}", progress=RemoteProgressReporter(_logger), kill_after_timeout=timeout)

try:
assert len(result) != 0
VALID_PUSH_INFO_FLAGS: list[int] = [PushInfo.FAST_FORWARD, PushInfo.NEW_HEAD,
PushInfo.UP_TO_DATE, PushInfo.FORCED_UPDATE, PushInfo.NEW_TAG]
for push_info in result:
_logger.debug("+------------+")
_logger.debug("| Push Info: |")
_logger.debug("+------------+")
_logger.debug(f"Flag: {push_info.flags}")
_logger.debug(f"Local ref: {push_info.local_ref}")
_logger.debug(f"Remote Ref: {push_info.remote_ref}")
_logger.debug(f"Remote ref string: {
push_info.remote_ref_string}")
_logger.debug(f"Old Commit: {push_info.old_commit}")
_logger.debug(f"Summary: {push_info.summary.strip()}")
if push_info.flags not in VALID_PUSH_INFO_FLAGS:
if push_info.flags == PushInfo.ERROR:
_logger.error(
"Incomplete push error: Push contains rejected heads. Check your internet connection and run in 'debug' mode to see more details.")
else:
_logger.error(
"Unexpected push error, maybe the remote rejected heads. Check your internet connection and run in 'debug' mode to see more details.")
return False
except AssertionError:
_logger.error(f"Pushing changes to remote '{
remote_name}' completely failed. Check your internet connection and run in 'debug' mode to see the remote push progress.")
return False

_logger.info(f"Changes pushed successfully to '{
branch_name}' branch of remote '{remote_name}'.")
if not has_tracking_branch(repo.active_branch):
_logger.debug(f"Setting '{branch_name}' upstream branch to '{
remote_name}/{remote_branch_name}'..")
repo.active_branch.set_tracking_branch(
repo.refs[f"{remote_name}/{remote_branch_name}"])

else:
_logger.info("Already up-to-date. Skipping..")
return True
except IndexError:
_logger.error(f"Error accessing remote '{
Expand Down Expand Up @@ -377,3 +459,24 @@ def needs_push(repo: Repo, branch_name: str | None = None) -> bool:
if tracking_branch:
return any(repo.iter_commits(f"{tracking_branch.name}..{branch.name}"))
return False


def get_default_branch_name(repo: Repo, remote_name: str = "origin") -> Union[str, None]:
"""Get the default branch name of a Git repository.
Args:
repo (Repo): The GitPython Repo object representing the local repository.
remote_name (str, optional): The name of the remote repository. Defaults to "origin".
Returns:
Union[str, None]: The name of the default branch, or None if not found.
"""
try:
show_result = repo.git.remote("show", remote_name)
matches = re.search(r"\s*HEAD branch:\s*(.*)", show_result)
if matches:
return matches.group(1)
except Exception as e:
_logger.error(f"Error while querying the default branch: {e}")

return None
12 changes: 12 additions & 0 deletions main.py
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from app.config import app_config
from app.helpers.git import checkout_branch
from app.helpers.git import commit_changes
from app.helpers.git import configure_divergent_branches_reconciliation_method
from app.helpers.git import get_files_count
from app.helpers.git import has_tracking_branch
from app.helpers.git import identity_setup
Expand Down Expand Up @@ -82,6 +83,17 @@
_logger.info("Exiting..")
sys.exit(2)

reconciliation_method_configured = configure_divergent_branches_reconciliation_method(
repo=repo, rebase=True)
if not reconciliation_method_configured:
_logger.error(
f"Failed to configure the reconciliation method for '{repo_name}'. Review the logs for more details")
if repo != TARGET_REPOS[-1]:
_logger.info("Skipping to the next migration..")
continue
_logger.info("Exiting..")
sys.exit(7)

if check_branch:
branch_checked = checkout_branch(
repo=repo, branch_name=TARGET_BRANCH['name'], from_branch=TARGET_BRANCH.get('from', None))
Expand Down

0 comments on commit 96af9cb

Please sign in to comment.