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
61 changes: 55 additions & 6 deletions src/sc/branching/branching.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@
import sys

from git import Repo
from git_flow_library import GitFlowLibrary
from repo_library import RepoLibrary

from .branch import Branch, BranchType
from .commands.checkout import Checkout
Expand All @@ -30,10 +32,9 @@
from .commands.push import Push
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 git_flow_library import GitFlowLibrary
from repo_library import RepoLibrary

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -81,7 +82,7 @@ def checkout(
)

@staticmethod
def sc_status(
def status(
run_dir: Path = Path.cwd(),
):
top_dir, project_type = detect_project(run_dir)
Expand All @@ -91,7 +92,7 @@ def sc_status(
)

@staticmethod
def sc_clean(
def clean(
run_dir: Path = Path.cwd(),
):
top_dir, project_type = detect_project(run_dir)
Expand All @@ -101,7 +102,7 @@ def sc_clean(
)

@staticmethod
def sc_reset(
def reset(
run_dir: Path = Path.cwd(),
):
top_dir, project_type = detect_project(run_dir)
Expand All @@ -110,7 +111,6 @@ def sc_reset(
project_type
)


@staticmethod
def push(
branch_type: BranchType,
Expand Down Expand Up @@ -150,6 +150,55 @@ def list(branch_type: BranchType, run_dir: Path = Path.cwd()):
project_type
)

@staticmethod
def tag_list(run_dir: Path = Path.cwd()):
top_dir, project_type = detect_project(run_dir)
run_command_by_project_type(
TagList(top_dir),
project_type
)

@staticmethod
def tag_show(tag: str, run_dir: Path = Path.cwd()):
top_dir, project_type = detect_project(run_dir)
run_command_by_project_type(
TagShow(top_dir, tag),
project_type
)

@staticmethod
def tag_create(tag: str, run_dir: Path = Path.cwd()):
top_dir, project_type = detect_project(run_dir)
run_command_by_project_type(
TagCreate(top_dir, tag),
project_type
)

@staticmethod
def tag_rm(tag: str, remote: bool, run_dir: Path = Path.cwd()):
top_dir, project_type = detect_project(run_dir)
run_command_by_project_type(
TagRm(top_dir, tag, remote),
project_type
)

@staticmethod
def tag_push(tag: str, run_dir: Path = Path.cwd()):
top_dir, project_type = detect_project(run_dir)
run_command_by_project_type(
TagPush(top_dir, tag),
project_type
)

@staticmethod
def tag_check(tag: str, run_dir: Path = Path.cwd()):
top_dir, project_type = detect_project(run_dir)
run_command_by_project_type(
TagCheck(top_dir, tag),
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
210 changes: 210 additions & 0 deletions src/sc/branching/commands/tag.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,210 @@
# 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.
"""Module for `sc tag` functionality."""

from dataclasses import dataclass
import logging
from pathlib import Path
import subprocess
import sys

from git import Repo
from sc_manifest_parser import ScManifest

from .command import Command

logger = logging.getLogger(__name__)

@dataclass
class TagShow(Command):
"""Show information about a particular tag."""
tag: str

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

def run_repo_command(self):
manifest = ScManifest.from_repo_root(self.top_dir / ".repo")
for proj in manifest.projects:
logger.info(f"Operating in: {self.top_dir / proj.path}")
if proj.lock_status:
logger.info(f"GIT_LOCK_STATUS: {proj.lock_status}")
self._git_show(self.top_dir / proj.path)

logger.info("-" * 100)

logger.info(f"Operating on manifest {self.top_dir / '.repo' / 'manifests'}")
self._git_show(self.top_dir / ".repo" / "manifests")

def _git_show(self, repo_path: Path):
try:
out = subprocess.run(
["git", "show", self.tag, "--color=always"],
cwd=repo_path,
check=True,
capture_output=True,
text=True
)
print(out.stdout)
except subprocess.CalledProcessError:
logger.warning(f"Tag {self.tag} not found!")

@dataclass
class TagList(Command):
"""List all tags."""
def run_git_command(self):
self._list_tags(self.top_dir)

def run_repo_command(self):
logger.info(f"Tags in mainfest: {self.top_dir / '.repo' / 'manifests'}")
self._list_tags(self.top_dir / '.repo' / 'manifests')

def _list_tags(self, repo_path: Path):
subprocess.run(["git", "tag", "-l"], cwd=repo_path, check=False)

@dataclass
class TagCreate(Command):
"""Create a tag."""
tag: str

def run_git_command(self):
subprocess.run(["git", "tag", self.tag], cwd=self.top_dir, check=False)

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

for proj in manifest.projects:
logger.info(f"Operating on: {self.top_dir / proj.path}")
if proj.lock_status == "READ_ONLY":
logger.info("READ_ONLY, skipping creating tag.")
continue

Repo(self.top_dir / proj.path).git.tag(self.tag)
logger.info(f"Tagged with {self.tag}")

logger.info(f"Operating on manifest: {self.top_dir / '.repo' / 'manifests'}")
Repo(self.top_dir / '.repo' / 'manifests').git.tag(self.tag)
logger.info(f"Tagged with {self.tag}")

def _error_if_tag_already_exists(self, manifest: ScManifest):
existing = [
self.top_dir / proj.path for proj in manifest.projects
if proj.lock_status != "READ_ONLY" # We aren't tagging READ_ONLY anyway
and self._tag_exists(self.top_dir / proj.path)
]

if self._tag_exists(self.top_dir / '.repo' / 'manifests'):
logger.error(f"Tag {self.tag} already exists in the manifest.")
existing.append(self.top_dir / '.repo' / 'manifests')

if existing:
logger.error(
"Tag already exists in the following projects:\n" +
"\n".join(str(p) for p in existing)
)
sys.exit(1)

def _tag_exists(self, repo_path: Path):
return any(t.name == self.tag for t in Repo(repo_path).tags)

@dataclass
class TagRm(Command):
"""Remove a tag."""
tag: str
remote: bool

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

def run_repo_command(self):
manifest = ScManifest.from_repo_root(self.top_dir / '.repo')
for proj in manifest.projects:
logger.info(f"Operating on: {self.top_dir / proj.path}")
if proj.lock_status == "READ_ONLY":
logger.info("READ_ONLY, skipping removing tag")
continue

self._delete_tag(self.top_dir / proj.path, proj.remote)

logger.info(f"Operating on manifest: {self.top_dir / '.repo' / 'manifests'}")
self._delete_tag(self.top_dir / '.repo' / 'manifests')

def _delete_tag(self, repo_path: Path, remote: str = "origin"):
subprocess.run(["git", "tag", "--delete", self.tag], cwd=repo_path, check=False)
if self.remote:
subprocess.run(
["git", "push", remote, f":refs/tags/{self.tag}"],
cwd=repo_path,
check=False
)

@dataclass
class TagPush(Command):
"""Push a tag."""
tag: str

def run_git_command(self):
remote = Repo(self.top_dir).remotes[0].name
self._push_tags(self.top_dir, remote)

def run_repo_command(self):
manifest = ScManifest.from_repo_root(self.top_dir / '.repo')
for proj in manifest.projects:
logger.info(f"Operating on: {self.top_dir / proj.path}")
if proj.lock_status == "READ_ONLY":
logger.info("READ_ONLY, skipping pushing tags")
continue

self._push_tags(self.top_dir / proj.path, proj.remote)

logger.info(f"Operating on manifest: {self.top_dir / '.repo' / 'manifests'}")
self._push_tags(self.top_dir / '.repo' / 'manifests')

logger.info("Push tags complete.")

def _push_tags(self, repo_path: Path, remote: str = "origin"):
subprocess.run(
["git", "push", remote, f"refs/tags/{self.tag}"],
cwd=repo_path,
check=False
)

@dataclass
class TagCheck(Command):
"""Check all repos for a specific tag."""
tag: str

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

def run_repo_command(self):
manifest = ScManifest.from_repo_root(self.top_dir / '.repo')
for proj in manifest.projects:
logger.info(f"Operating on: {self.top_dir / proj.path}")
if proj.lock_status == "READ_ONLY":
logger.info("READ_ONLY, skipping checking tags.")
continue

self._check_tag(self.top_dir / proj.path)

logger.info(f"Operating on manifest: {self.top_dir / '.repo' / 'manifests'}")
self._check_tag(self.top_dir / '.repo' / 'manifests')

def _check_tag(self, repo_path: Path):
subprocess.run(
["git", "show-ref", "--tags", "--verify", f"refs/tags/{self.tag}"],
cwd=repo_path,
check=False
)
50 changes: 43 additions & 7 deletions src/sc/branching_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,21 +33,17 @@ def init():
@cli.command()
def clean():
"""Clean all modules. (git clean -fdx)."""
SCBranching.sc_clean()
SCBranching.clean()

@cli.command()
def status():
"""Show the working tree status."""
SCBranching.sc_status()


SCBranching.status()

@cli.command()
def reset():
"""Clean and Reset all modules to remote manifest. (git reset --hard REMOTE)"""
SCBranching.sc_reset()


SCBranching.reset()

@cli.command()
def build():
Expand Down Expand Up @@ -299,5 +295,45 @@ def checkout(name, force, verify):
"""Checkout a support branch."""
SCBranching.checkout(BranchType.SUPPORT, name, force, verify)

@cli.group()
def tag():
pass

@tag.command()
def list():
"""List tags in manifest or git repo."""
SCBranching.tag_list()

@tag.command()
@click.argument("tag")
def show(tag):
"""Show information about a tag in all repos."""
SCBranching.tag_show(tag)

@tag.command()
@click.argument("tag")
def create(tag):
"""Create a tag in all non READ_ONLY repos."""
SCBranching.tag_create(tag)

@tag.command()
@click.argument("tag")
@click.option('-r', '--remote', is_flag=True, help="Remove in remotes as well as local.")
def rm(tag, remote):
"""Remove a tag in all non READ_ONLY repos."""
SCBranching.tag_rm(tag, remote)

@tag.command()
@click.argument("tag")
def push(tag):
"""Push given tag in all non READ_ONLY repos."""
SCBranching.tag_push(tag)

@tag.command()
@click.argument("tag")
def check(tag):
"""Check if a tag exists on all non READ_ONLY repos."""
SCBranching.tag_check(tag)

if __name__ == '__main__':
cli()