Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 12 additions & 10 deletions pr_split/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -244,7 +244,7 @@ def _handle_loc_bound_warnings(warnings: list[str], *, strict_loc_bounds: bool)
if strict_loc_bounds and warnings:
console.print(f"[red]{ErrorMsg.LOC_BOUNDS_STRICT_FAILED()}[/red]")
for warning in warnings:
console.print(f"[red]- {warning}[/red]")
console.print(f"[red]- {escape(warning)}[/red]")
raise typer.Exit(1)

for warning in warnings:
Expand Down Expand Up @@ -731,7 +731,7 @@ def _move_assignment(
if from_id == to_id:
console.print(
f"[yellow]Source and destination are the same"
f" ('{from_id}'). No move performed.[/yellow]"
f" ('{escape(from_id)}'). No move performed.[/yellow]"
)
return False

Expand Down Expand Up @@ -1393,7 +1393,7 @@ def split(
),
)
except (ValidationError, ValueError) as exc:
console.print(f"[red]{exc}[/red]")
console.print(f"[red]{escape(str(exc))}[/red]")
raise typer.Exit(1) from exc
try:
validate_no_binary_files(parsed_diff)
Expand Down Expand Up @@ -1445,7 +1445,7 @@ def split(
logger.success("Edited plan validation passed")
_report_oversized_groups(groups, settings.max_loc, hunk_counts)
except PRSplitError as exc:
console.print(f"[red]Edited plan is invalid: {exc}[/red]")
console.print(f"[red]Edited plan is invalid: {escape(str(exc))}[/red]")
raise typer.Exit(1) from exc

merge_base_ref = merge_base(diff_base, dev_branch)
Expand Down Expand Up @@ -1571,7 +1571,9 @@ def status() -> None:
# so showing it would claim OPEN for a PR that may be gone.
pr_state = "UNKNOWN"
unverified.append(pr_record.pr_number)
table.add_row(group.id, group.title, branch_name, pr_info, pr_state, review)
table.add_row(
escape(group.id), escape(group.title), escape(branch_name), pr_info, pr_state, review
)

console.print(table)
for gid, branch, parent in stale_layers(plan_file):
Expand Down Expand Up @@ -1981,7 +1983,7 @@ def execute(
validate_coverage(plan.groups, parsed_diff)
validate_new_file_pieces(plan.groups, parsed_diff, dag)
except PlanValidationError as exc:
console.print(f"[red]{exc}[/red]")
console.print(f"[red]{escape(str(exc))}[/red]")
raise typer.Exit(1) from exc

# Any plan with dependency edges is laid out along its DAG (stacked or not),
Expand Down Expand Up @@ -2234,15 +2236,15 @@ def merge_all(

console.print()
if merged:
console.print(f"[green]Merged ({len(merged)}): {', '.join(merged)}[/green]")
console.print(f"[green]Merged ({len(merged)}): {escape(', '.join(merged))}[/green]")
if skipped:
console.print(f"[yellow]Skipped ({len(skipped)}): {', '.join(skipped)}[/yellow]")
console.print(f"[yellow]Skipped ({len(skipped)}): {escape(', '.join(skipped))}[/yellow]")
if failed:
console.print(f"[red]Failed ({len(failed)}): {', '.join(failed)}[/red]")
console.print(f"[red]Failed ({len(failed)}): {escape(', '.join(failed))}[/red]")
if blocked:
console.print(
f"[yellow]Blocked by unmerged dependencies ({len(blocked)}): "
f"{', '.join(blocked)}. Re-run once those PRs are merged.[/yellow]"
f"{escape(', '.join(blocked))}. Re-run once those PRs are merged.[/yellow]"
)
if notify:
exit_reason = (
Expand Down
92 changes: 92 additions & 0 deletions tests/test_cli_coverage.py
Original file line number Diff line number Diff line change
Expand Up @@ -1038,6 +1038,98 @@ def test_move_messages_keep_bracketed_paths(self) -> None:
assert "Group '[pr-9]' not found." in out


class TestRichEscapingOfStatusAndSummaries:
@patch("pr_split.cli.get_pr_state", return_value={"state": "OPEN", "reviewDecision": None})
@patch("pr_split.cli.load_plan")
@patch("pr_split.cli.plan_exists", return_value=True)
def test_status_table_keeps_bracketed_title(
self, mock_pe: MagicMock, mock_load: MagicMock, mock_state: MagicMock
) -> None:
from pr_split.constants import Priority
from pr_split.schemas import BranchRecord, GitState, PlanFile, PRRecord, SplitPlan

plan = SplitPlan(
dev_branch="feature",
base_branch="main",
max_loc=400,
priority=Priority.ORTHOGONAL,
groups=[_group("pr-1", "Add [core] typing")],
)
git_state = GitState(
branches=[BranchRecord(group_id="pr-1", branch_name="b[1]", base_branch="main")],
prs=[PRRecord(group_id="pr-1", pr_number=7, pr_url="u")],
)
mock_load.return_value = PlanFile(plan=plan, git_state=git_state)
result = runner.invoke(app, ["status"])
assert result.exit_code == 0
assert "Add [core] typing" in result.output
assert "b[1]" in result.output

def test_error_echo_keeps_bracketed_path(self) -> None:
from pr_split.cli import _handle_loc_bound_warnings, console

# The strict path prints each warning in red (the non-strict path only logs).
with console.capture() as capture, pytest.raises(typer.Exit):
_handle_loc_bound_warnings(
["Group 'pr-1' touches src/[id].py"], strict_loc_bounds=True
)
assert "src/[id].py" in capture.get()

def test_editor_same_group_message_keeps_bracketed_id(self) -> None:
from pr_split.cli import console

with console.capture() as capture:
assert not _move_assignment([], MagicMock(), "a.py", 0, "[pr-1]", "[pr-1]")
assert "Source and destination are the same ('[pr-1]')" in capture.get()


class TestRichEscapingOfMergeSummary:
@patch("pr_split.cli.merge_pr")
@patch("pr_split.cli.get_pr_state")
@patch("pr_split.cli.load_plan")
@patch("pr_split.cli.plan_exists", return_value=True)
def test_summary_lines_keep_bracketed_group_ids(
self,
mock_pe: MagicMock,
mock_load: MagicMock,
mock_state: MagicMock,
mock_merge: MagicMock,
) -> None:
from pr_split.constants import Priority
from pr_split.exceptions import GitOperationError
from pr_split.schemas import GitState, PlanFile, PRRecord, SplitPlan

groups = [_group("[pr-1]", "a"), _group("[pr-2]", "b"), _group("[pr-3]", "c")]
plan = SplitPlan(
dev_branch="feature",
base_branch="main",
max_loc=400,
priority=Priority.ORTHOGONAL,
groups=groups,
)
prs = [PRRecord(group_id=g.id, pr_number=i + 1, pr_url="u") for i, g in enumerate(groups)]
mock_load.return_value = PlanFile(plan=plan, git_state=GitState(prs=prs))
states = {
1: {"state": "OPEN", "isDraft": False, "reviewDecision": None},
2: {"state": "OPEN", "isDraft": True, "reviewDecision": None},
3: {"state": "OPEN", "isDraft": False, "reviewDecision": None},
}
mock_state.side_effect = lambda n: states[n]

def merge(n: int, *, auto: bool = False) -> None:
if n == 3:
raise GitOperationError("conflict")

mock_merge.side_effect = merge

result = runner.invoke(app, ["merge"])

flat = " ".join(result.output.split())
assert "Merged (1): [pr-1]" in flat
assert "Skipped (1): [pr-2] (draft)" in flat
assert "Failed (1): [pr-3]" in flat


class TestExecuteRetriesAfterFailedPush:
"""Branches without PRs mean a failed push; execute must retry, not refuse.

Expand Down
Loading