diff --git a/src/sc/branching/branching.py b/src/sc/branching/branching.py index 2f4a0b9..7320428 100644 --- a/src/sc/branching/branching.py +++ b/src/sc/branching/branching.py @@ -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 @@ -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__) @@ -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) @@ -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) @@ -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) @@ -110,7 +111,6 @@ def sc_reset( project_type ) - @staticmethod def push( branch_type: BranchType, @@ -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 diff --git a/src/sc/branching/commands/tag.py b/src/sc/branching/commands/tag.py new file mode 100644 index 0000000..8838ed1 --- /dev/null +++ b/src/sc/branching/commands/tag.py @@ -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 + ) \ No newline at end of file diff --git a/src/sc/branching_cli.py b/src/sc/branching_cli.py index b29aadb..5a85195 100644 --- a/src/sc/branching_cli.py +++ b/src/sc/branching_cli.py @@ -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(): @@ -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()