refactoring: properly reuse code for 'cli/manifest.py'

tsrc/cli/status.py

@@ -70,96 +70,3 @@ def run(args: argparse.Namespace) -> None:
                 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
-        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
-
-
-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()