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