From d0de9bc1322ca134d502cc30dd547eb843ad5f4a Mon Sep 17 00:00:00 2001 From: Yashwanth Nannapaneni Date: Mon, 18 May 2026 11:43:54 -0700 Subject: [PATCH] Adding stacky recreate command to recreate a stack from a passed in stacky info --- src/stacky/commands/recreate.py | 120 ++++++++++++++++++ src/stacky/main.py | 12 ++ .../tests/test_commands/test_recreate.py | 110 ++++++++++++++++ 3 files changed, 242 insertions(+) create mode 100644 src/stacky/commands/recreate.py create mode 100644 src/stacky/tests/test_commands/test_recreate.py diff --git a/src/stacky/commands/recreate.py b/src/stacky/commands/recreate.py new file mode 100644 index 0000000..8ea1fce --- /dev/null +++ b/src/stacky/commands/recreate.py @@ -0,0 +1,120 @@ +"""Recreate command - rebuild a stack from `stacky info` output.""" + +import re +import sys +from typing import Iterable, List, Optional, Tuple + +from stacky.git.branch import get_all_branches +from stacky.git.refs import get_merge_base, set_parent, set_parent_commit +from stacky.stack.models import StackBranchSet +from stacky.utils.logging import cout, die, info +from stacky.utils.types import BranchName, Commit, STACK_BOTTOMS +from stacky.utils.ui import confirm + + +_TREE_CHARS = "│┌├└─" # │ ┌ ├ └ ─ +_LINE_RE = re.compile( + r"^(?P[\s│┌├└─]*)" + r"(?:[!~*]+\s*)*" + r"(?P[A-Za-z0-9_][\w./-]*)" + r"(?:\s+\(.*\))?\s*$" +) + + +def _indent_of(line: str) -> int: + """Visual indent of the branch on a line, measured by the leftmost tree + character. Lines without any tree character (e.g. `* main`) are roots and + get -1 so they sort below everything else. + """ + for i, ch in enumerate(line): + if ch in "┌├└": # ┌ ├ └ + return i + return -1 + + +def parse_stack_info(text: str) -> List[Tuple[BranchName, Optional[BranchName]]]: + """Parse `stacky info` output into a list of (child, parent) edges. + + The bottom-most branch (e.g. `main`) is returned with parent=None. + Lines are processed top-to-bottom; each branch's parent is the next line + *below* it with a strictly smaller indent (since stacky prints the tree + upside-down, so visually-lower = closer to the root). + """ + parsed: List[Tuple[int, BranchName]] = [] + for raw in text.splitlines(): + if not raw.strip(): + continue + m = _LINE_RE.match(raw) + if not m: + continue + name = BranchName(m.group("name")) + parsed.append((_indent_of(raw), name)) + + edges: List[Tuple[BranchName, Optional[BranchName]]] = [] + for i, (indent, name) in enumerate(parsed): + parent: Optional[BranchName] = None + for j in range(i + 1, len(parsed)): + j_indent, j_name = parsed[j] + if j_indent < indent: + parent = j_name + break + edges.append((name, parent)) + return edges + + +def _validate_branches_exist(edges: Iterable[Tuple[BranchName, Optional[BranchName]]]): + existing = set(get_all_branches()) + missing: List[BranchName] = [] + for child, parent in edges: + if child not in existing: + missing.append(child) + if parent is not None and parent not in existing: + missing.append(parent) + if missing: + die("Branches not found in repo: {}", ", ".join(sorted(set(missing)))) + + +def cmd_recreate(stack: StackBranchSet, args): + """Rebuild a stack from `stacky info` output piped on stdin or via --file.""" + if args.file: + with open(args.file) as f: + text = f.read() + else: + if sys.stdin.isatty(): + die("No input provided. Pipe `stacky info` output or pass --file.") + text = sys.stdin.read() + + edges = parse_stack_info(text) + if not edges: + die("No branches parsed from input") + + _validate_branches_exist(edges) + + to_apply = [(c, p) for c, p in edges if p is not None] + roots = [c for c, p in edges if p is None] + + cout("Will recreate stack with {} edges:\n", len(to_apply), fg="green") + for child, parent in to_apply: + cout(" {} -> {}\n", child, parent) + if roots: + cout("Stack bottoms (unchanged): {}\n", ", ".join(roots), fg="cyan") + + for root in roots: + if root not in STACK_BOTTOMS: + info( + "Note: {} is treated as a root but is not a known stack bottom", + root, + ) + + if not args.force: + confirm() + + for child, parent in to_apply: + base = get_merge_base(child, parent) # type: ignore[arg-type] + if base is None: + die("Could not find merge-base between {} and {}", child, parent) + set_parent(child, parent, set_origin=True) + set_parent_commit(child, Commit(base)) + info("Set {} -> {} (parent commit {})", child, parent, base[:8]) + + cout("Done. Run `stacky info` to verify.\n", fg="green") diff --git a/src/stacky/main.py b/src/stacky/main.py index 8bf8399..7f8ff19 100644 --- a/src/stacky/main.py +++ b/src/stacky/main.py @@ -37,6 +37,7 @@ from stacky.commands.fold import ( cmd_fold, inner_do_fold, finish_merge_fold_operation ) +from stacky.commands.recreate import cmd_recreate def main(): @@ -115,6 +116,8 @@ def main(): current_branch = get_current_branch_name() if args.command == "continue": _handle_continue(stack, current_branch) + elif args.command == "recreate": + args.func(stack, args) else: if current_branch not in stack.stack: main_branch = get_real_stack_bottom() @@ -336,3 +339,12 @@ def _setup_other_commands(subparsers): fold_parser = subparsers.add_parser("fold", help="Fold current branch into parent branch and delete current branch") fold_parser.add_argument("--allow-empty", action="store_true", help="Allow empty commits during cherry-pick") fold_parser.set_defaults(func=cmd_fold) + + # recreate + recreate_parser = subparsers.add_parser( + "recreate", + help="Rebuild a stack from `stacky info` output (read from stdin or --file)", + ) + recreate_parser.add_argument("--file", help="Read stack info from a file instead of stdin") + recreate_parser.add_argument("--force", "-f", action="store_true", help="Bypass confirmation") + recreate_parser.set_defaults(func=cmd_recreate) diff --git a/src/stacky/tests/test_commands/test_recreate.py b/src/stacky/tests/test_commands/test_recreate.py new file mode 100644 index 0000000..fa7d1e3 --- /dev/null +++ b/src/stacky/tests/test_commands/test_recreate.py @@ -0,0 +1,110 @@ +"""Tests for stacky.commands.recreate.""" + +import unittest + +from stacky.commands.recreate import parse_stack_info + + +class TestParseStackInfo(unittest.TestCase): + def test_simple_linear_stack(self): + text = "\n".join([ + " ┌── feat-c", + " ├── feat-b", + " ├── feat-a", + "* main", + ]) + edges = parse_stack_info(text) + self.assertEqual(edges, [ + ("feat-c", "main"), + ("feat-b", "main"), + ("feat-a", "main"), + ("main", None), + ]) + + def test_branched_stack(self): + # Two children of feat-a, plus a sibling chain rooted at main. + text = "\n".join([ + " │ ┌── feat-a-c2", + " │ ┌── feat-a-c1", + " ├── feat-a", + " ├── feat-b", + "* main", + ]) + edges = parse_stack_info(text) + self.assertEqual(edges, [ + ("feat-a-c2", "feat-a"), + ("feat-a-c1", "feat-a"), + ("feat-a", "main"), + ("feat-b", "main"), + ("main", None), + ]) + + def test_strips_status_markers_and_pr_suffix(self): + text = "\n".join([ + " ┌── ! feat-x", + " ├── !~ feat-y (#42 some title)", + "* main", + ]) + edges = parse_stack_info(text) + self.assertEqual(edges, [ + ("feat-x", "main"), + ("feat-y", "main"), + ("main", None), + ]) + + def test_full_frostdb_example(self): + # The exact tree the user reconstructed by hand. + text = "\n".join([ + " ┌── ! ynannapaneni/adding-p-validation-for-hybrid", + " ├── ! ynannapaneni/FDBCORE-XX-simulator-for-log-system-v2", + " ├── !~ ynannapaneni/FDBCORE-XX-agentbox-git-ignore", + " │ ┌── ynannapaneni/FDBCORE-4496-dmi-uses-real-time-stream-manager", + " │ ┌── ynannapaneni/FDBCORE-45165-rsm-on-connection-closed", + " │ ┌── ynannapaneni/FDBCORE-45165-component-test-for-realtime-delivery-framework", + " ├── ! ynannapaneni/FDBCORE-44860-real-time-mutation-delivery-framework", + " ├── !~ ynannapaneni/FDBCORE-40432-fix-bw-lag-metric-spike-issues", + " │ ┌── ynannapaneni/FDBCORE-37707-adding-hybrid-worker-dd-queue", + " │ ┌── ynannapaneni/FDBCORE-33707-adding-dd-queue-unit-tests", + " ├── ! ynannapaneni/FDBCORE-33707-creating-relocate-data-for-shard", + " ├── !~ frostdb_4", + " ├── !~ frostdb_3", + " ├── !~ frostdb_2", + "* main", + ]) + edges = dict(parse_stack_info(text)) + self.assertEqual(edges["main"], None) + self.assertEqual(edges["frostdb_2"], "main") + self.assertEqual(edges["frostdb_3"], "main") + self.assertEqual(edges["frostdb_4"], "main") + self.assertEqual( + edges["ynannapaneni/FDBCORE-45165-component-test-for-realtime-delivery-framework"], + "ynannapaneni/FDBCORE-44860-real-time-mutation-delivery-framework", + ) + self.assertEqual( + edges["ynannapaneni/FDBCORE-45165-rsm-on-connection-closed"], + "ynannapaneni/FDBCORE-45165-component-test-for-realtime-delivery-framework", + ) + self.assertEqual( + edges["ynannapaneni/FDBCORE-4496-dmi-uses-real-time-stream-manager"], + "ynannapaneni/FDBCORE-45165-rsm-on-connection-closed", + ) + self.assertEqual( + edges["ynannapaneni/FDBCORE-37707-adding-hybrid-worker-dd-queue"], + "ynannapaneni/FDBCORE-33707-adding-dd-queue-unit-tests", + ) + self.assertEqual( + edges["ynannapaneni/FDBCORE-33707-adding-dd-queue-unit-tests"], + "ynannapaneni/FDBCORE-33707-creating-relocate-data-for-shard", + ) + self.assertEqual( + edges["ynannapaneni/FDBCORE-33707-creating-relocate-data-for-shard"], + "main", + ) + + def test_empty_input(self): + self.assertEqual(parse_stack_info(""), []) + self.assertEqual(parse_stack_info(" \n \n"), []) + + +if __name__ == "__main__": + unittest.main()