diff --git a/pr_split/cli.py b/pr_split/cli.py index 61d99e9..ab6dc27 100644 --- a/pr_split/cli.py +++ b/pr_split/cli.py @@ -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: @@ -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 @@ -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) @@ -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) @@ -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): @@ -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), @@ -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 = ( diff --git a/tests/test_cli_coverage.py b/tests/test_cli_coverage.py index 4a9010d..e324f92 100644 --- a/tests/test_cli_coverage.py +++ b/tests/test_cli_coverage.py @@ -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.