From 8a1af93fd464f4ebd8e4fea83964754468df76b4 Mon Sep 17 00:00:00 2001 From: lomnido Date: Fri, 8 Mar 2024 11:59:14 +0100 Subject: [PATCH] 'cli/manifest.py' should be ready to make use of deep manifest Introducing consideration towards deep manifest: Deep Manifest is when we have Manifest repository in the Workspace, and there is 'manifest.yml'. And this Manifest also contains information that can tell us what will happen after 'tsrc sync'. While on 'tsrc manifest' this does little help, but the full power will be seen when it gets implemented into 'tsrc status' --- tsrc/cli/manifest.py | 143 +++++++++++++++-------------- tsrc/status_endpoint.py | 198 +++++++++++++++++++++++++++++++++++++++- 2 files changed, 272 insertions(+), 69 deletions(-) diff --git a/tsrc/cli/manifest.py b/tsrc/cli/manifest.py index aeadb0f7..6ba0401c 100644 --- a/tsrc/cli/manifest.py +++ b/tsrc/cli/manifest.py @@ -12,9 +12,15 @@ get_workspace_with_repos, ) from tsrc.executor import process_items -from tsrc.git import run_git_captured +from tsrc.git import get_git_status, run_git_captured +from tsrc.manifest import load_manifest from tsrc.repo import Repo -from tsrc.status_endpoint import Status, StatusCollector, describe_status +from tsrc.status_endpoint import ( + Status, + StatusCollector, + get_l_and_r_sha1_of_branch, + workspace_repositories_summary, +) from tsrc.workspace_config import WorkspaceConfig @@ -37,7 +43,7 @@ def run(args: argparse.Namespace) -> None: # manifest_branch = workspace.local_manifest.current_branch() workspace_config = workspace.config - ui.info_1("Manifest's URL: ", ui.purple, workspace_config.manifest_url, ui.reset) + ui.info_1("Manifest's URL:", ui.purple, workspace_config.manifest_url, ui.reset) status_collector = StatusCollector(workspace) repos = workspace.repos @@ -57,17 +63,19 @@ def run(args: argparse.Namespace) -> None: current_workspace_manifest_repo_branch = None if static_manifest_manifest_dest: - ui.info_2("Integrated into Workspace as another repository:") + ui.info_2("Current integration into Workspace:") # statuses_items = statuses.items() - current_workspace_manifest_repo_branch = workspace_integration_summary( + current_workspace_manifest_repo_branch = workspace_repositories_summary( + workspace.root_path, statuses, static_manifest_manifest_dest, static_manifest_manifest_branch, workspace.config.manifest_branch, + only_manifest=True, ) - mi = ManifestInfo( + mi = ManifestReport( workspace_config, cfg_path, workspace.root_path, @@ -78,10 +86,10 @@ def run(args: argparse.Namespace) -> None: current_workspace_manifest_repo_branch, ) - mi.info() + mi.report() -class ManifestInfo: +class ManifestReport: def __init__( self, workspace_config: WorkspaceConfig, @@ -102,7 +110,7 @@ def __init__( self.static_manifest_manifest_branch = static_manifest_manifest_branch self.c_w_m_r_branch = current_workspace_manifest_repo_branch - def info(self) -> None: + def report(self) -> None: if self.set_manifest_branch: self.on_set_branch() else: @@ -133,33 +141,16 @@ def on_set_branch(self) -> None: self.w_c.save_to_file(self.cfg_path) """workspace is now updated""" self.uip_workspace_updated() - workspace_integration_summary( + workspace_repositories_summary( + self.workspace_root_path, self.statuses, self.s_m_m_dest, self.static_manifest_manifest_branch, self.w_c.manifest_branch, - True, + do_update=True, + only_manifest=True, ) - if self.s_m_m_dest: - """when there is Manifest repository in the Workspace""" - if rc_is_on_remote == 0: - if self.c_w_m_r_branch != self.w_c.manifest_branch: - self.uip_after_sync_branch_change() - else: - self.uip_ok_after_sync_same_branch() - else: - self.uip_push_first_for_sync_to_work() - else: - """use 'manifest_branch_0' to determine if brach will change""" - if self.w_c.manifest_branch != self.w_c.manifest_branch_0: - self.uip_branch_will_change_after_sync( - self.w_c.manifest_branch, - self.w_c.manifest_branch_0, - ) - else: - self.uip_branch_will_stay_the_same_after_sync( - self.w_c.manifest_branch - ) + self.report_what_wha_sync() else: """branch is nowhere to be found""" if self.s_m_m_dest: @@ -171,8 +162,12 @@ def on_set_branch(self) -> None: def on_default_display(self) -> None: """just report final status of current state, do not update anything""" - if self.s_m_m_dest and self.c_w_m_r_branch != self.w_c.manifest_branch: - self.uip_branch_will_change_after_sync(self.w_c.manifest_branch) + self.report_what_wha_sync() + + def report_what_wha_sync(self) -> None: + """report what will_happen_after sync with Manifest""" + if self.s_m_m_dest: + self.report_iro_m_branch_in_w() else: """use 'manifest_branch_0' to determine if brach will change""" if self.w_c.manifest_branch != self.w_c.manifest_branch_0: @@ -183,11 +178,36 @@ def on_default_display(self) -> None: else: self.uip_branch_will_stay_the_same_after_sync(self.w_c.manifest_branch) + def report_iro_m_branch_in_w(self) -> None: + """report in regards of Manifest branch in Workspace (has change or not)""" + if self.c_w_m_r_branch != self.w_c.manifest_branch: + self.uip_branch_will_change_after_sync(self.w_c.manifest_branch) + else: + deep_m_branch = self.get_w_d_m_branch() + if deep_m_branch: + if deep_m_branch != self.w_c.manifest_branch: + self.uip_branch_will_change_after_sync(self.w_c.manifest_branch) + else: + self.uip_ok_after_sync_same_branch() + + def get_w_d_m_branch(self) -> Union[str, None]: + """get Workspace-deep manifest branch. This means: + Workspace:Manifest repository:Manifest file:Manifest repository:branch""" + if isinstance(self.s_m_m_dest, str): + deep_manifest = load_manifest( + self.workspace_root_path / self.s_m_m_dest / "manifest.yml" + ) + return deep_manifest.get_repo(self.s_m_m_dest).branch + else: + return None + + """ui prints|errors segment follows:""" + def uip_skip_set_branch(self) -> None: ui.info_1("Skipping configuring the same branch") def uip_using_new_branch(self, branch: str) -> None: - ui.info_2("Using new branch: ", ui.green, branch, ui.reset) + ui.info_2("Using new branch:", ui.green, branch, ui.reset) def uip_workspace_updated(self) -> None: ui.info_1("Workspace updated") @@ -201,9 +221,30 @@ def uip_after_sync_branch_change(self) -> None: ) def uip_ok_after_sync_same_branch(self) -> None: + """check if repository is clean, + and also if remote commit SHA1 is same as local commit SHA1, + as only then we can say for sure, it will stays the same""" + if self.s_m_m_dest: + m_g_status = get_git_status(self.workspace_root_path / self.s_m_m_dest) + if not ( + m_g_status.dirty is False # noqa: W503 + and m_g_status.ahead == 0 # noqa: W503 + and m_g_status.behind == 0 # noqa: W503 + and m_g_status.upstreamed is True # noqa: W503 + ): + ui.info_2("Clean Manifest repository before calling 'sync'") + return + l_m_sha, r_m_sha = get_l_and_r_sha1_of_branch( + self.workspace_root_path, + self.s_m_m_dest, + self.w_c.manifest_branch, + ) + if r_m_sha and l_m_sha != r_m_sha: + ui.info_2("Remote branch does not have same HEAD") + return ui.info_2( ui.blue, - "OK: After 'sync' the repository will stays on the same branch", + "OK: After 'sync', Manifest repository will stays on the same branch", ) def uip_push_first_for_sync_to_work(self) -> None: @@ -281,38 +322,6 @@ def is_manifest_in_workspace( return static_manifest_manifest_dest, static_manifest_manifest_branch -def workspace_integration_summary( - statuses: Dict[str, StatusOrError], - st_m_m_dest: Union[str, None], - st_m_m_branch: Union[str, None], - w_c_m_branch: str, - do_update: bool = False, -) -> Union[str, None]: - """prints a summary of Manifest repository status. - the same output should be used when using 'tsrc status' - but other repositories included""" - cur_w_m_r_branch = None - for dest, status in statuses.items(): - if dest == st_m_m_dest: - if do_update: - ui.info_2("Updating configured Manifest branch. See new overall state:") - if isinstance(status, Status): - cur_w_m_r_branch = status.git.branch - message = [ui.green, "*", ui.reset, dest] - message += describe_status(status) - message += [ui.purple, "<---", "MANIFEST:"] - message += [ui.green, st_m_m_branch] - if w_c_m_branch != st_m_m_branch: - message += [ - ui.reset, - "~~~>", - ui.green, - w_c_m_branch, - ] - ui.info(*message) - return cur_w_m_r_branch - - def manifest_remote_branch_exist(url: str, branch: str) -> int: """check for manifest remote branch, as only if it exist, we should allow to set it. as you can see, there is diff --git a/tsrc/status_endpoint.py b/tsrc/status_endpoint.py index 70abdf02..731a386d 100644 --- a/tsrc/status_endpoint.py +++ b/tsrc/status_endpoint.py @@ -1,12 +1,13 @@ import collections +from pathlib import Path from typing import Dict, List, Optional, Tuple, Union import cli_ui as ui from tsrc.errors import MissingRepo from tsrc.executor import Outcome, Task -from tsrc.git import GitStatus, get_git_status -from tsrc.manifest import Manifest +from tsrc.git import GitStatus, get_git_status, run_git_captured +from tsrc.manifest import Manifest, RepoNotFound, load_manifest from tsrc.repo import Repo from tsrc.utils import erase_last_line from tsrc.workspace import Workspace @@ -95,6 +96,152 @@ def process(self, index: int, count: int, repo: Repo) -> Outcome: CollectedStatuses = Dict[str, StatusOrError] +def workspace_repositories_summary( + workspace_root_path: Path, + statuses: Dict[str, StatusOrError], + st_m_m_dest: Union[str, None], + st_m_m_branch: Union[str, None], + w_c_m_branch: str, + do_update: bool = False, + only_manifest: bool = False, +) -> Union[str, None]: + """prints a summary of Manifest repository status. + the same output should be used when using 'tsrc status' + but with other repositories included + + Few points to output: + * repository path (default color) + * [ effective_branch_of_Manifest_repository_from_manifest.yml ]= + (should represent some kind of inside block as if Manifest expands) + * current branch from workspace point of view + * status descriptions + * <—— MANIFEST: (purple) (should represent pointer) + * manifest branch from last 'sync' + * ~~> (default color) (should represent transition) + * newly configured manifest branch (if there is such) + """ + deep_manifest = None + max_m_branch, deep_manifest = max_len_manifest_branch( + workspace_root_path, + st_m_m_dest, + statuses, + ) + + cur_w_m_r_branch = None + max_dest = 0 + if only_manifest is False: + max_dest = max(len(x) for x in statuses.keys()) + else: + max_m_branch = 0 + + for dest, status in statuses.items(): + d_m_repo_found, d_m_branch = check_if_deep_manifest_repo_dest( + deep_manifest, + dest, + ) + + if dest == st_m_m_dest: + if do_update: + ui.info_2("New state after Workspace update:") + if isinstance(status, Status): + cur_w_m_r_branch = status.git.branch + else: + if only_manifest is True: + continue + + message = [ui.green, "*", ui.reset, dest.ljust(max_dest)] + + if deep_manifest: + message += deep_manifest_describe( + d_m_repo_found, + d_m_branch, + dest, + st_m_m_dest, + max_m_branch, + ) + + message += describe_status(status) + + if dest == st_m_m_dest: + message += describe_on_manifest_repo_status(st_m_m_branch, w_c_m_branch) + + ui.info(*message) + + return cur_w_m_r_branch + + +def check_if_deep_manifest_repo_dest( + deep_manifest: Union[Manifest, None], + dest: str, +) -> Tuple[bool, Union[str, None]]: + d_m_repo_found = True + d_m_branch = None + if not deep_manifest: + return False, None + try: + d_m_branch = deep_manifest.get_repo(dest).branch + except RepoNotFound: + d_m_repo_found = False + return d_m_repo_found, d_m_branch + + +def deep_manifest_describe( + d_m_r_found: bool, + d_m_branch: Union[str, None], + dest: str, + st_m_m_dest: Union[str, None], + max_m_branch: int, +) -> List[ui.Token]: + message = [] + if d_m_r_found is True and isinstance(d_m_branch, str): + message += [ui.brown, "[", ui.green] + message += [d_m_branch.ljust(max_m_branch)] + if dest == st_m_m_dest: + message += [ui.brown, "]=", ui.reset] + else: + message += [ui.brown, "] ", ui.reset] + else: + message += [" ".ljust(max_m_branch + 2 + 2 + 1)] + return message + + +def max_len_manifest_branch( + w_r_path: Path, st_m_m_dest: Union[str, None], statuses: Dict[str, StatusOrError] +) -> Tuple[int, Union[Manifest, None]]: + """calculate maximum lenght for deep manifest branch (if present)""" + max_m_branch = 0 + d_m = None + if st_m_m_dest: + d_m = load_manifest(w_r_path / st_m_m_dest / "manifest.yml") + all_m_branch_len = [] + for dest, _status in statuses.items(): + try: + this_len = len(d_m.get_repo(dest).branch) + except RepoNotFound: + continue + if dest == st_m_m_dest: + this_len = this_len + 1 + all_m_branch_len += [this_len] + max_m_branch = max(all_m_branch_len) + return max_m_branch, d_m + + +def describe_on_manifest_repo_status( + s_branch: Union[str, None], c_branch: str +) -> List[ui.Token]: + """When exactly on Manifest repository integrated into Workspace""" + message = [ui.purple, "<——", "MANIFEST:"] + message += [ui.green, s_branch] + if c_branch != s_branch: + message += [ + ui.reset, + "~~>", + ui.green, + c_branch, + ] + return message + + def describe_status(status: StatusOrError) -> List[ui.Token]: """Return a list of tokens suitable for ui.info().""" if isinstance(status, MissingRepo): @@ -104,3 +251,50 @@ def describe_status(status: StatusOrError) -> List[ui.Token]: git_status = status.git.describe() manifest_status = status.manifest.describe() return git_status + manifest_status + + +def get_l_and_r_sha1_of_branch( + w_r_path: Path, + dest: str, + branch: str, +) -> Tuple[Union[str, None], Union[str, None]]: + """obtain local and remote SHA1 of given branch. + This is useful when we need to check if we are exactly + updated with remote down to the commit""" + rc, l_m_sha = run_git_captured( + w_r_path / dest, + "rev-parse", + "--verify", + "HEAD", + check=False, + ) + if rc != 0: + return None, None + + tmp_ref = "{}@{{upstream}}" + rc, this_ref = run_git_captured( + w_r_path / dest, + "rev-parse", + "--symbolic-full-name", + "--abbrev-ref", + tmp_ref.format(branch), + check=False, + ) + r_m_sha = None + if rc == 0: + tmp_r_ref = this_ref.split("/") + this_remote = tmp_r_ref[0] + this_r_ref = "refs/heads/" + tmp_r_ref[1] + _, r_m_sha = run_git_captured( + w_r_path / dest, + "ls-remote", + "--exit-code", + "--head", + this_remote, + this_r_ref, + check=True, + ) + if r_m_sha: + return l_m_sha, r_m_sha.split()[0] + else: + return l_m_sha, None