enhance output of 'status' by these features:

tsrc/cli/status.py

@@ -15,11 +15,12 @@
 )
 from tsrc.errors import MissingRepo
 from tsrc.executor import Outcome, Task, process_items
-from tsrc.git import GitStatus, get_git_status
+from tsrc.git import GitStatus, get_git_status, run_git_captured
 from tsrc.manifest import Manifest
 from tsrc.repo import Repo
 from tsrc.utils import erase_last_line
 from tsrc.workspace import Workspace
+from tsrc.status_endpoint import StatusCollector, ManifestStatus, describe_status
 
 
 def configure_parser(subparser: argparse._SubParsersAction) -> None:
@@ -43,98 +44,29 @@ def run(args: argparse.Namespace) -> None:
     erase_last_line()
     ui.info_2("Workspace status:")
     statuses = status_collector.statuses
+
+    """detect same Manifest's repository in the workspace repositories"""
+    """and also check if there is missing upstream"""
+    static_manifest_manifest_dest = None    # "static" as it cannot be changed, "manifest_manifest" = Manifest repo in Manifest.yml; this may not exist but if it does, it will be the 'dest'intion == "direcory_name"
+    static_manifest_manifest_branch = None  # "static" as it cannot be changed, "manifest_manifest" = Manifest repo in Manifest.yml; this may not exist, but if it does, it will be the (git) 'branch'
+    workspace_manifest_repo_branch = None   # Workspace's > Manifest_repo's > branch
+    for x in repos:
+        this_dest = x.dest
+        this_branch = x.branch
+        for y in x.remotes:
+            if y.url == workspace.config.manifest_url:
+                static_manifest_manifest_dest = this_dest
+                workspace_manifest_repo_branch = workspace.local_manifest.current_branch()
+                static_manifest_manifest_branch = this_branch
+
     max_dest = max(len(x) for x in statuses.keys())
     for dest, status in statuses.items():
         message = [ui.green, "*", ui.reset, dest.ljust(max_dest)]
         message += describe_status(status)
+        if dest == static_manifest_manifest_dest:
+            message += [ui.purple, "<---", "MANIFEST:"]
+            message += [ui.green, static_manifest_manifest_branch]
+            if workspace.config.manifest_branch != static_manifest_manifest_branch:
+                message += [ui.reset, "~~~>", ui.green, workspace.config.manifest_branch]
         ui.info(*message)
 
-
-class ManifestStatus:
-    """Represent the status of a repo w.r.t the manifest."""
-
-    def __init__(self, repo: Repo, *, manifest: Manifest):
-        self.repo = repo
-        self.manifest = manifest
-        self.incorrect_branch: Optional[Tuple[str, str]] = None
-
-    def update(self, git_status: GitStatus) -> None:
-        """Set self.incorrect_branch if the local git status
-        does not match the branch set in the manifest.
-        """
-        expected_branch = self.repo.branch
-        actual_branch = git_status.branch
-        if actual_branch and actual_branch != expected_branch:
-            self.incorrect_branch = (actual_branch, expected_branch)
-
-    def describe(self) -> List[ui.Token]:
-        """Return a list of tokens suitable for ui.info()`."""
-        res: List[ui.Token] = []
-        incorrect_branch = self.incorrect_branch
-        if incorrect_branch:
-            actual, expected = incorrect_branch
-            res += [ui.red, "(expected: " + expected + ")"]
-        return res
-
-
-class Status:
-    """Wrapper class for both ManifestStatus and GitStatus"""
-
-    def __init__(self, *, git: GitStatus, manifest: ManifestStatus):
-        self.git = git
-        self.manifest = manifest
-
-
-StatusOrError = Union[Status, Exception]
-CollectedStatuses = Dict[str, StatusOrError]
-
-
-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
-
-
-class StatusCollector(Task[Repo]):
-    """Implement a Task to collect local git status and
-    stats w.r.t the manifest for each repo.
-    """
-
-    def __init__(self, workspace: Workspace) -> None:
-        self.workspace = workspace
-        self.manifest = workspace.get_manifest()
-        self.statuses: CollectedStatuses = collections.OrderedDict()
-
-    def describe_item(self, item: Repo) -> str:
-        return item.dest
-
-    def describe_process_start(self, item: Repo) -> List[ui.Token]:
-        return [item.dest]
-
-    def describe_process_end(self, item: Repo) -> List[ui.Token]:
-        return []
-
-    def process(self, index: int, count: int, repo: Repo) -> Outcome:
-        # Note: Outcome is always empty here, because we
-        # use self.statuses in the main `run()` function instead
-        # of calling OutcomeCollection.print_summary()
-        full_path = self.workspace.root_path / repo.dest
-        self.info_count(index, count, repo.dest, end="\r")
-        if not full_path.exists():
-            self.statuses[repo.dest] = MissingRepo(repo.dest)
-        try:
-            git_status = get_git_status(full_path)
-            manifest_status = ManifestStatus(repo, manifest=self.manifest)
-            manifest_status.update(git_status)
-            status = Status(git=git_status, manifest=manifest_status)
-            self.statuses[repo.dest] = status
-        except Exception as e:
-            self.statuses[repo.dest] = e
-        if not self.parallel:
-            erase_last_line()
-        return Outcome.empty()

tsrc/git.py

@@ -77,6 +77,7 @@ def __init__(self, working_path: Path) -> None:
         self.tag: Optional[str] = None
         self.branch: Optional[str] = None
         self.sha1: Optional[str] = None
+        self.upstreamed = False
 
     def update(self) -> None:
         # Try and gather as many information about the git repository as
@@ -90,6 +91,7 @@ def update(self) -> None:
         self.update_tag()
         self.update_remote_status()
         self.update_worktree_status()
+        self.update_upstreamed()
 
     def update_sha1(self) -> None:
         self.sha1 = get_sha1(self.working_path, short=True)
@@ -136,6 +138,11 @@ def update_worktree_status(self) -> None:
                 self.added += 1
                 self.dirty = True
 
+    def update_upstreamed(self) -> None:
+        rc, _ = run_git_captured(self.working_path, "config", "--get", f"branch.{self.branch}.remote", check=False)
+        if rc == 0:
+            self.upstreamed = True
+
     def describe(self) -> List[ui.Token]:
         """Return a list of tokens suitable for ui.info."""
         res: List[ui.Token] = []

tsrc/status_endpoint.py

@@ -0,0 +1,105 @@
+import collections
+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.repo import Repo
+from tsrc.utils import erase_last_line
+from tsrc.workspace import Workspace
+
+class ManifestStatus:
+    """Represent the status of a repo w.r.t the manifest."""
+
+    def __init__(self, repo: Repo, *, manifest: Manifest):
+        self.repo = repo
+        self.manifest = manifest
+        self.incorrect_branch: Optional[Tuple[str, str]] = None
+        self.missing_upstream = None
+
+    def update(self, git_status: GitStatus) -> None:
+        """Set self.incorrect_branch if the local git status
+        does not match the branch set in the manifest.
+        """
+        expected_branch = self.repo.branch
+        actual_branch = git_status.branch
+        if actual_branch and actual_branch != expected_branch:
+            self.incorrect_branch = (actual_branch, expected_branch)
+        self.missing_upstream = not git_status.upstreamed
+
+    def describe(self) -> List[ui.Token]:
+        """Return a list of tokens suitable for ui.info()`."""
+        res: List[ui.Token] = []
+        incorrect_branch = self.incorrect_branch
+        if incorrect_branch:
+            actual, expected = incorrect_branch
+            res += [ui.red, "(expected: " + expected + ")"]
+        if self.missing_upstream:
+            res += [ui.red, "(missing upstream)"]
+        return res
+
+
+class Status:
+    """Wrapper class for both ManifestStatus and GitStatus"""
+
+    def __init__(self, *, git: GitStatus, manifest: ManifestStatus):
+        self.git = git
+        self.manifest = manifest
+
+class StatusCollector(Task[Repo]):
+    """Implement a Task to collect local git status and
+    stats w.r.t the manifest for each repo.
+    """
+
+    def __init__(self, workspace: Workspace) -> None:
+        self.workspace = workspace
+        self.manifest = workspace.get_manifest()
+        self.statuses: CollectedStatuses = collections.OrderedDict()
+
+    def describe_item(self, item: Repo) -> str:
+        return item.dest
+
+    def describe_process_start(self, item: Repo) -> List[ui.Token]:
+        return [item.dest]
+
+    def describe_process_end(self, item: Repo) -> List[ui.Token]:
+        return []
+
+    def process(self, index: int, count: int, repo: Repo) -> Outcome:
+        # Note: Outcome is always empty here, because we
+        # use self.statuses in the main `run()` function instead
+        # of calling OutcomeCollection.print_summary()
+        full_path = self.workspace.root_path / repo.dest
+        self.info_count(index, count, repo.dest, end="\r")
+        if not full_path.exists():
+            self.statuses[repo.dest] = MissingRepo(repo.dest)
+        try:
+            git_status = get_git_status(full_path)
+            manifest_status = ManifestStatus(repo, manifest=self.manifest)
+            manifest_status.update(git_status)
+            status = Status(git=git_status, manifest=manifest_status)
+            self.statuses[repo.dest] = status
+        except Exception as e:
+            self.statuses[repo.dest] = e
+        if not self.parallel:
+            erase_last_line()
+        return Outcome.empty()
+
+
+
+StatusOrError = Union[Status, Exception]
+# CollectedStatuses = Dict[str, StatusOrError]
+
+
+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