From 887ab698f8eec8efc38082f343f0f34d1dd34fd5 Mon Sep 17 00:00:00 2001 From: lomnido Date: Thu, 14 Mar 2024 11:18:40 +0100 Subject: [PATCH] Deep Manifest support should be ready --- tsrc/cli/manifest.py | 52 ++++-- tsrc/status_endpoint.py | 402 ++++++++++++++++++++++++---------------- 2 files changed, 273 insertions(+), 181 deletions(-) diff --git a/tsrc/cli/manifest.py b/tsrc/cli/manifest.py index 6ba0401c..7afad112 100644 --- a/tsrc/cli/manifest.py +++ b/tsrc/cli/manifest.py @@ -18,8 +18,8 @@ from tsrc.status_endpoint import ( Status, StatusCollector, + WorkspaceReposSummary, get_l_and_r_sha1_of_branch, - workspace_repositories_summary, ) from tsrc.workspace_config import WorkspaceConfig @@ -40,7 +40,6 @@ def run(args: argparse.Namespace) -> None: workspace = get_workspace_with_repos(args) cfg_path = workspace.cfg_path - # 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) @@ -60,13 +59,12 @@ def run(args: argparse.Namespace) -> None: ) = is_manifest_in_workspace(repos, workspace.config.manifest_url) """current: Workspace's > Manifest_repo's > branch""" - current_workspace_manifest_repo_branch = None + current_workspace_manifest_repo = None if static_manifest_manifest_dest: ui.info_2("Current integration into Workspace:") - # statuses_items = statuses.items() - current_workspace_manifest_repo_branch = workspace_repositories_summary( + wrs = WorkspaceReposSummary( workspace.root_path, statuses, static_manifest_manifest_dest, @@ -75,6 +73,8 @@ def run(args: argparse.Namespace) -> None: only_manifest=True, ) + current_workspace_manifest_repo = wrs.summary() + mi = ManifestReport( workspace_config, cfg_path, @@ -83,7 +83,7 @@ def run(args: argparse.Namespace) -> None: args.manifest_branch, static_manifest_manifest_dest, static_manifest_manifest_branch, - current_workspace_manifest_repo_branch, + current_workspace_manifest_repo, ) mi.report() @@ -99,7 +99,7 @@ def __init__( set_manifest_branch: str, static_manifest_manifest_dest: Union[str, None], static_manifest_manifest_branch: Union[str, None], - current_workspace_manifest_repo_branch: Union[str, None], + current_workspace_manifest_repo: Union[Repo, None], ): self.w_c = workspace_config self.cfg_path = cfg_path @@ -108,7 +108,7 @@ def __init__( self.set_manifest_branch = set_manifest_branch self.s_m_m_dest = static_manifest_manifest_dest self.static_manifest_manifest_branch = static_manifest_manifest_branch - self.c_w_m_r_branch = current_workspace_manifest_repo_branch + self.c_w_m_repo = current_workspace_manifest_repo def report(self) -> None: if self.set_manifest_branch: @@ -141,7 +141,8 @@ def on_set_branch(self) -> None: self.w_c.save_to_file(self.cfg_path) """workspace is now updated""" self.uip_workspace_updated() - workspace_repositories_summary( + + wrs = WorkspaceReposSummary( self.workspace_root_path, self.statuses, self.s_m_m_dest, @@ -150,7 +151,9 @@ def on_set_branch(self) -> None: do_update=True, only_manifest=True, ) - self.report_what_wha_sync() + + wrs.summary() + self.report_what_after_sync() else: """branch is nowhere to be found""" if self.s_m_m_dest: @@ -162,9 +165,9 @@ def on_set_branch(self) -> None: def on_default_display(self) -> None: """just report final status of current state, do not update anything""" - self.report_what_wha_sync() + self.report_what_after_sync() - def report_what_wha_sync(self) -> None: + def report_what_after_sync(self) -> None: """report what will_happen_after sync with Manifest""" if self.s_m_m_dest: self.report_iro_m_branch_in_w() @@ -180,24 +183,31 @@ def report_what_wha_sync(self) -> None: 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: + c_w_m_repo_branch = None + if self.c_w_m_repo: + c_w_m_repo_branch = self.c_w_m_repo.branch + if c_w_m_repo_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: + deep_m_repo = self.get_w_d_m_repo() + if deep_m_repo: + if deep_m_repo.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() + # manifest repo should have only one remote, thus: [0] + if deep_m_repo.remotes[0].url != self.w_c.manifest_url: + self.uip_same_branch_different_url(deep_m_repo.remotes[0].url) + else: + self.uip_ok_after_sync_same_branch() - def get_w_d_m_branch(self) -> Union[str, None]: + def get_w_d_m_repo(self) -> Union[Repo, 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 + return deep_manifest.get_repo(self.s_m_m_dest) else: return None @@ -220,6 +230,10 @@ def uip_after_sync_branch_change(self) -> None: ui.reset, ) + def uip_same_branch_different_url(self, url: str) -> None: + ui.info_2("Deep manifest is using different remote URL") + ui.info_1("Deep Manifest's URL:", url) + 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, diff --git a/tsrc/status_endpoint.py b/tsrc/status_endpoint.py index 731a386d..28e8e0a2 100644 --- a/tsrc/status_endpoint.py +++ b/tsrc/status_endpoint.py @@ -96,161 +96,234 @@ 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, +class WorkspaceReposSummary: + def __init__( + self, + workspace_root_path: Path, + statuses: Dict[str, StatusOrError], + st_m_m_dest: Union[str, None], + st_m_m_branch: Union[str, None], + # TODO: check: should not be also here 'None'? + w_c_m_branch: str, + do_update: bool = False, + only_manifest: bool = False, + ): + self.w_r_path = workspace_root_path + self.statuses = statuses + self.st_m_m_dest = st_m_m_dest + self.st_m_m_branch = st_m_m_branch + self.w_c_m_branch = w_c_m_branch + self.do_update = do_update + self.only_manifest = only_manifest + + def summary(self) -> Union[Repo, None]: + + deep_manifest = None + max_m_branch, deep_manifest = self.max_len_manifest_branch( + self.w_r_path, + self.st_m_m_dest, + self.statuses, ) - 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 + max_dest = 0 + if self.only_manifest is False: + max_dest = max(len(x) for x in self.statuses.keys()) 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, + max_m_branch = 0 + + # this shoud ensure always sorted items by key + o_stats = collections.OrderedDict(sorted(self.statuses.items())) + has_d_m_d = collections.OrderedDict() + for dest in o_stats.keys(): + d_m_repo_found, _ = self.check_if_deep_manifest_repo_dest( + deep_manifest, dest, - st_m_m_dest, - max_m_branch, ) + has_d_m_d[dest] = d_m_repo_found + s_has_d_m_d = collections.OrderedDict() - message += describe_status(status) + # sort base on if there is a deep manifest destination + for key in sorted(has_d_m_d, key=has_d_m_d.__getitem__): + s_has_d_m_d[key] = has_d_m_d[key] - if dest == st_m_m_dest: - message += describe_on_manifest_repo_status(st_m_m_branch, w_c_m_branch) - - ui.info(*message) + # let us print the final list of repos + cur_w_m_repo = None + if deep_manifest: + # prepare to take care of letfovers from deep_manifest + d_m_repos = deep_manifest.get_repos() - return cur_w_m_r_branch + cur_w_m_repo = self.core_message_print( + deep_manifest, + s_has_d_m_d, + max_dest, + max_m_branch, + d_m_repos, + ) + if self.only_manifest is False: + # recollect leftovers only if there is full list + for leftover in d_m_repos: + # print("DEBUG repos: dest:", i.dest, " branch: ", i.branch) + message = [ui.reset, "*", ui.purple, dest.ljust(max_dest)] + message += [ui.brown, "[", ui.purple] + message += [leftover.branch.ljust(max_m_branch)] + message += [ui.brown, "]", ui.reset] + ui.info(*message) + + return cur_w_m_repo + + def core_message_print( + self, + deep_manifest: Manifest, + s_has_d_m_d: collections.OrderedDict, + max_dest: int, + max_m_branch: int, + d_m_repos: List[Repo], + ) -> Union[Repo, None]: + """Prints a summary of Workspace repository status. + + How to understand the output: + * repository path (default color) + * [ branch_of_Manifest_repository_from_manifest.yml ]= + (represents some kind of inside block as if Manifest expands) + * current branch (and description) from workspace point of view + * <—— MANIFEST: (purple) (for Manifest repo only) + (represents pointer) + * manifest branch from last 'sync' + * ~~> (default color) (represents transition) + * newly configured manifest branch (if there is such) + """ + cur_w_m_repo = None + for dest in s_has_d_m_d.keys(): + status = self.statuses[dest] + d_m_repo_found, d_m_repo = self.check_if_deep_manifest_repo_dest( + deep_manifest, + dest, + ) -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 == self.st_m_m_dest: + if self.do_update and self.only_manifest is True: + ui.info_2("New state after Workspace update:") + if isinstance(status, Status): + cur_w_m_repo = d_m_repo + else: + if self.only_manifest is True: + continue + + message = [ui.green, "*", ui.reset, dest.ljust(max_dest)] + + if deep_manifest: + d_m_repo_branch = None + if d_m_repo: + d_m_repo_branch = d_m_repo.branch + d_m_repos.pop(d_m_repos.index(d_m_repo)) + message += self.deep_manifest_describe( + d_m_repo_found, + d_m_repo_branch, + dest, + self.st_m_m_dest, + max_m_branch, + ) + + message += self.describe_status(status) + + if dest == self.st_m_m_dest: + message += self.describe_on_manifest_repo_status( + self.st_m_m_branch, self.w_c_m_branch + ) + + ui.info(*message) + + return cur_w_m_repo + + def check_if_deep_manifest_repo_dest( + self, + deep_manifest: Union[Manifest, None], + dest: str, + ) -> Tuple[bool, Union[Repo, None]]: + d_m_repo_found = True + d_m_repo = None + if not deep_manifest: + return False, None + try: + d_m_repo = deep_manifest.get_repo(dest) + except RepoNotFound: + d_m_repo_found = False + return d_m_repo_found, d_m_repo + + def deep_manifest_describe( + self, + 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: - 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): - return [ui.red, "error: missing repo"] - if isinstance(status, Exception): - return [ui.red, "error: ", status] - git_status = status.git.describe() - manifest_status = status.manifest.describe() - return git_status + manifest_status + 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( + self, + 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(): + """test if such repo exists first before checking it in the deep manifest""" + 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) + max_m_all = 0 + if d_m: + d_m_stats = d_m.get_repos() + max_m_all = max(len(x.branch) for x in d_m_stats) + max_m_branch = max(max_m_branch, max_m_all) + return max_m_branch, d_m + + def describe_on_manifest_repo_status( + self, 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(self, status: StatusOrError) -> List[ui.Token]: + """Return a list of tokens suitable for ui.info().""" + if isinstance(status, MissingRepo): + return [ui.red, "error: missing repo"] + if isinstance(status, Exception): + return [ui.red, "error: ", status] + git_status = status.git.describe() + manifest_status = status.manifest.describe() + return git_status + manifest_status def get_l_and_r_sha1_of_branch( @@ -261,7 +334,7 @@ def get_l_and_r_sha1_of_branch( """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( + rc, l_b_sha = run_git_captured( w_r_path / dest, "rev-parse", "--verify", @@ -271,30 +344,35 @@ def get_l_and_r_sha1_of_branch( 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, + _, l_ref = run_git_captured(w_r_path / dest, "symbolic-ref", "-q", "HEAD") + _, r_ref = run_git_captured( + w_r_path / dest, "for-each-ref", "--format='%(upstream)'", l_ref ) - r_m_sha = None + r_b_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( + tmp_r_ref = r_ref.split("/") + this_remote = tmp_r_ref[2] + _, r_b_sha = run_git_captured( w_r_path / dest, "ls-remote", "--exit-code", "--head", this_remote, - this_r_ref, + l_ref, check=True, ) - if r_m_sha: - return l_m_sha, r_m_sha.split()[0] + if r_b_sha: + return l_b_sha, r_b_sha.split()[0] else: - return l_m_sha, None + return l_b_sha, None + + +def describe_status(status: StatusOrError) -> List[ui.Token]: + """Return a list of tokens suitable for ui.info().""" + if isinstance(status, MissingRepo): + return [ui.red, "error: missing repo"] + if isinstance(status, Exception): + return [ui.red, "error: ", status] + git_status = status.git.describe() + manifest_status = status.manifest.describe() + return git_status + manifest_status