Skip to content
Merged
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
68 changes: 56 additions & 12 deletions src/sc/branching/branching.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@
from repo_library import RepoLibrary

from .branch import Branch, BranchType
from .commands.branch_rename import BranchRename
from .commands.branch_rm_merged import BranchRmMerged
from .commands.checkout import Checkout
from .commands.clean import Clean
from .commands.command import Command
Expand All @@ -33,13 +35,13 @@
from .commands.list import List
from .commands.pull import Pull
from .commands.push import Push
from .commands.show import ShowBranch, ShowLog, ShowRepoFlowConfig
from .commands.show import ShowBranch, ShowLog, ShowRepoFlowConfig, ShowMergedRelease
from .commands.start import Start
from .commands.status import Status
from .commands.tag import (TagCheck, TagCreate, TagList, TagPush,
TagRm, TagShow)
from .commands.reset import Reset
from .exceptions import ScInitError
from sc.exceptions import ScError

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -86,7 +88,7 @@ def checkout(
Checkout(top_dir, branch, force=force, verify=verify),
project_type
)

@staticmethod
def delete(
branch_type: BranchType,
Expand Down Expand Up @@ -238,6 +240,19 @@ def show_repo_flow_config(run_dir: Path = Path.cwd()):
project_type
)

@staticmethod
def show_merged_release(
previous_release: str | None,
current_release: str | None,
wiki: bool,
run_dir: Path = Path.cwd()
):
top_dir, project_type = detect_project(run_dir)
run_command_by_project_type(
ShowMergedRelease(top_dir, previous_release, current_release, wiki),
project_type
)

@staticmethod
def group_checkout(group: str, branch: str, run_dir: Path = Path.cwd()):
top_dir, project_type = detect_project(run_dir)
Expand Down Expand Up @@ -300,6 +315,35 @@ def group_tag(
project_type
)

@staticmethod
def branch_rename(
old_branch: str,
new_branch: str,
local_only: bool,
git_only: bool,
run_dir: Path = Path.cwd()
):
top_dir, project_type = detect_project(run_dir)
run_command_by_project_type(
BranchRename(top_dir, old_branch, new_branch, local_only, git_only),
project_type
)

@staticmethod
def branch_rm_merged(
not_merged: bool,
all: bool,
yes: bool,
git_only: bool,
dry_run: bool,
run_dir: Path = Path.cwd()
):
top_dir, project_type = detect_project(run_dir)
run_command_by_project_type(
BranchRmMerged(top_dir, not_merged, all, yes, git_only, dry_run),
project_type
)

def detect_project(run_dir: Path) -> tuple[Path | ProjectType]:
if root := RepoLibrary.get_repo_root_dir(run_dir):
return root.parent, ProjectType.REPO
Expand Down Expand Up @@ -334,13 +378,13 @@ def create_branch(
sys.exit(1)

def run_command_by_project_type(command: Command, project_type: ProjectType):
if project_type == ProjectType.GIT:
command.run_git_command()
elif project_type == ProjectType.REPO:
try:
try:
if project_type == ProjectType.GIT:
command.run_git_command()
elif project_type == ProjectType.REPO:
command.run_repo_command()
except ScInitError as e:
logger.error(e)
sys.exit(1)
else:
raise RuntimeError("Should not get here.")
else:
raise RuntimeError("Should not get here.")
except ScError as e:
logger.error(e)
sys.exit(1)
114 changes: 114 additions & 0 deletions src/sc/branching/commands/branch_rename.py
Comment thread
TB-1993 marked this conversation as resolved.
Original file line number Diff line number Diff line change
@@ -0,0 +1,114 @@
# Copyright 2025 RDK Management
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from dataclasses import dataclass
from enum import Enum, auto
import logging
from pathlib import Path
Comment thread
BenjiMilan marked this conversation as resolved.

import git
from git import Repo

from .command import Command
from sc.exceptions import ScError
from sc_manifest_parser import ScManifest

logger = logging.getLogger(__name__)

@dataclass
class BranchRename(Command):
old_branch: str
new_branch: str
local_only: bool
git_only: bool

def run_git_command(self):
self._rename_repo(self.top_dir)

def run_repo_command(self):
manifest = ScManifest.from_repo_root(self.top_dir / ".repo")

if self.git_only:
self._rename_repo(Path.cwd())
return

for proj in manifest.projects:
if proj.lock_status is None:
logger.info(f"Renaming local branch in repo: {self.top_dir / proj.path}")
self._rename_repo(self.top_dir / proj.path)

logger.info(f"Renaming local manifest branch: {self.top_dir / '.repo' / 'manifests'}")
self._rename_repo(self.top_dir / ".repo" / "manifests")

def _rename_repo(self, directory: Path):
try:
repo = Repo(directory)
except git.InvalidGitRepositoryError as e:
# We should hopefully never get here. Only if the user is missing repositories
# that should be present from their manifest.
logger.warning(f"Skipping renaming for {directory}: Not a valid git repository.")
return
Comment thread
BenjiMilan marked this conversation as resolved.

try:
self._rename_local(repo)
except ScError as e:
logger.warning(f"Skipping renaming for {directory}: {e}")
return

if not self.local_only:
try:
self._rename_remote(repo)
except ScError as e:
logger.warning(f"Skipping remote renaming for {directory}: {e}")

def _rename_local(self, repo: Repo):
try:
repo.git.branch("-m", self.old_branch, self.new_branch)
logger.info("Renamed locally.")
except git.GitCommandError as e:
raise ScError(
f"Unable to rename branch in repo {repo.working_dir}: {e.stderr}"
) from e

def _rename_remote(self, repo: Repo):
try:
remote = repo.remotes[0].name
except IndexError as e:
raise ScError(f"No remote found for repo {repo.working_dir}.") from e

try:
# Push/create the new branch before deleting the old branch.
repo.git.push("-u", remote, f"{self.new_branch}:refs/heads/{self.new_branch}")
except git.GitCommandError as e:
raise ScError(
f"Failed to push new branch to remote {self.new_branch}: {e.stderr}") from e

if self._has_remote_branch(repo, remote, self.old_branch) is not None:
try:
repo.git.push(remote, "--delete", self.old_branch)
except git.GitCommandError as e:
raise ScError(
f"Failed to delete old branch on remote {self.old_branch}: {e.stderr}") from e

logger.info("Renamed remotely.")

def _has_remote_branch(self, repo: Repo, remote_name: str, branch_name: str) -> bool:
out = repo.git.ls_remote("--heads", remote_name, f"refs/heads/{branch_name}")

for line in out.splitlines():
commit, ref = line.split()
if ref == f"refs/heads/{branch_name}":
return True

return False
109 changes: 109 additions & 0 deletions src/sc/branching/commands/branch_rm_merged.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,109 @@
# Copyright 2025 RDK Management
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from dataclasses import dataclass
from pathlib import Path

import git
from git import Repo
from git_flow_library import GitFlowLibrary

from ..branch import Branch
from .command import Command
from .delete import Delete
from sc.exceptions import ScError
from sc.prompter import Prompter
from sc.services.tickets.ticket_service import TicketService

@dataclass
class BranchRmMerged(Command):
not_merged: bool = False
all: bool = False
no_prompt: bool = False
git_only: bool = False
dry_run: bool = False
Comment thread
BenjiMilan marked this conversation as resolved.

def run_git_command(self):
self._verify_options()
self._rm_merged(self.top_dir, git_only=True)

def run_repo_command(self):
self._error_on_sc_uninitialised()
self._verify_options()

if self.git_only:
root = GitFlowLibrary.get_git_root(Path.cwd())
if not root:
raise ScError(f"{Path.cwd()} not a valid git repository!")
self._rm_merged(root, self.git_only)
else:
self._rm_merged(self.top_dir)

def _verify_options(self):
if self.not_merged and self.all:
raise ScError("Cannot pass both --all and --no-merged.")

def _rm_merged(self, path: Path, git_only: bool = False):
filtered_branches = self._get_feature_branches(path)

ticket_service = TicketService()

for branch in filtered_branches:
ticket = ticket_service.get_ticket_from_branch(branch.name)
print(ticket.to_terminal(one_line=True))
print(branch.name)
Comment thread
BenjiMilan marked this conversation as resolved.
Comment thread
BenjiMilan marked this conversation as resolved.

if self.dry_run:
continue

if self.no_prompt:
self._delete_branch(path, branch, git_only)
elif Prompter.yn("Delete branch?"):
self._delete_branch(path, branch, git_only)
Comment thread
BenjiMilan marked this conversation as resolved.

def _get_feature_branches(self, path: Path) -> list[Branch]:
try:
repo = Repo(path)
except git.InvalidGitRepositoryError as e:
raise ScError(f"Invalid git repo: {path}") from e

develop = GitFlowLibrary.get_develop_branch(path)
master = GitFlowLibrary.get_master_branch(path)
hotfix = GitFlowLibrary.get_config_value("prefix.hotfix", path)
release = GitFlowLibrary.get_config_value("prefix.release", path)
support = GitFlowLibrary.get_config_value("prefix.support", path)

branch_filters = [develop, master, hotfix, release, support, "m/feature"]

feature = GitFlowLibrary.get_config_value("prefix.feature", path)

merge_type = "--merged" if not self.not_merged else "--no-merged"
merge_type = "--all" if self.all else merge_type
branches = repo.git.branch("-r", merge_type, develop).splitlines()

feature_branches = [
branch.strip().split("/", 1)[1]
for branch in branches
if not any(f in branch for f in branch_filters)
and feature in branch
]

return [Branch(*b.split("/", 1)) for b in feature_branches]

def _delete_branch(self, path: Path, branch: Branch, git_only: bool):
if git_only:
Delete(path, branch, remote=True).run_git_command()
else:
Delete(path, branch, remote=True).run_repo_command()

Loading