From ebec6aa8c74e965ce39ffb6165cc0cab9d8ded66 Mon Sep 17 00:00:00 2001 From: Ruari Phipps Date: Tue, 21 Jul 2026 16:40:21 +0100 Subject: [PATCH 1/9] Branch history --- src/poly/cli_commands/branch.py | 61 ++++++++++++++++++++++++++++++++ src/poly/handlers/interface.py | 14 ++++++++ src/poly/handlers/sdk.py | 27 ++++++++++++++ src/poly/handlers/sync_client.py | 14 ++++++++ src/poly/output/console.py | 23 ++++++++++++ src/poly/project.py | 11 ++++++ 6 files changed, 150 insertions(+) diff --git a/src/poly/cli_commands/branch.py b/src/poly/cli_commands/branch.py index 3c977c02..4b73b1f9 100644 --- a/src/poly/cli_commands/branch.py +++ b/src/poly/cli_commands/branch.py @@ -230,6 +230,22 @@ def add_arguments(cls, subparsers: _SubParsersAction[ArgumentParser], parents: P ) branch_merge_parser.set_defaults(branch_subcommand="merge") + branch_history_parser = branch_subparsers.add_parser( + "history", + parents=[parents.path, parents.verbose, parents.json, parents.debug], + help="Show the history of a branch.", + ) + + branch_history_parser.add_argument( + "--branch-name", + "-b", + type=str, + default=None, + help="Name of the branch to show history for. Defaults to the current branch.", + ) + + branch_history_parser.set_defaults(branch_subcommand="history") + @classmethod def run(cls, args: Namespace) -> None: """Dispatch to the matching branch sub-handler.""" @@ -265,6 +281,9 @@ def run(cls, args: Namespace) -> None: elif args.branch_subcommand == "merge": cls.branch_merge(args.path, args.message, args.json, args.interactive, args.resolutions) + elif args.branch_subcommand == "history": + cls.branch_history(args.path, args.branch_name, args.json) + @classmethod def branch_list(cls, base_path: str, output_json: bool = False) -> None: """List branches in the Agent Studio project.""" @@ -906,3 +925,45 @@ def branch_merge( show_type=True, panel_title="Remaining merge conflicts", ) + + @classmethod + def branch_history( + cls, base_path: str, branch_name: Optional[str] = None, output_json: bool = False + ) -> None: + """Show the history of a branch in the Agent Studio project.""" + from poly.output.console import plain, print_branch_history, warning + + project = load_project(base_path, output_json=output_json) + + current_branch, branches = project.get_branches() + if not branch_name: + branch_name = current_branch + + if not branch_name: + if output_json: + json_print( + { + "success": False, + "error": "No current branch found. Please specify a branch name.", + } + ) + else: + warning("No current branch found. Please specify a branch name.") + + branch_id = branches.get(branch_name) + if not branch_id: + warning(f"Branch '{branch_name}' does not exist.") + return + + history = project.get_branch_history(branch_id) + + if output_json: + json_print({"branch_name": branch_name, "branch_id": branch_id, "history": history}) + return + + if not history: + plain(f"[muted]No history found for branch '{branch_name}'.[/muted]") + return + + plain(f"History for branch '{branch_name}':") + print_branch_history(history) diff --git a/src/poly/handlers/interface.py b/src/poly/handlers/interface.py index 5bfa2f6b..a045e66e 100644 --- a/src/poly/handlers/interface.py +++ b/src/poly/handlers/interface.py @@ -1097,3 +1097,17 @@ def trigger_test_run( dict: The created test run response. """ return PlatformAPIHandler.trigger_test_run(region, project_id, test_case_ids, branch_id) + + def get_branch_history(self, branch_name: str) -> list[dict[str, Any]]: + """Get the history of a specific branch. + + Args: + branch_name (str): The name of the branch + + Returns: + list[dict[str, Any]]: A list of commit history entries for the branch + """ + try: + return self.sync_client.get_branch_history(branch_name) + except (requests.HTTPError, SourcererAPIError) as e: + self._handle_api_error(e) diff --git a/src/poly/handlers/sdk.py b/src/poly/handlers/sdk.py index 04d84c7a..0a874e2e 100644 --- a/src/poly/handlers/sdk.py +++ b/src/poly/handlers/sdk.py @@ -681,3 +681,30 @@ def get_full_projection_response(self, force_refresh: bool = False) -> dict[str, "projection": self._projection_cache, "last_known_sequence": self._last_known_sequence, } + + def get_branch_history(self, branch_id: str) -> list[dict[str, Any]]: + """Get the history of commands for a specific branch. + + Args: + branch_id: The branch ID to fetch history for. + + Returns: + A list of dictionaries containing commit information for the branch. + Raises: + SourcererAPIError: If the API request fails. + """ + try: + url = f"{self._get_branches_url()}/{branch_id}/merge-history" + response = self.session.get(url) + response.raise_for_status() + return response.json().get("merges", []) + except requests.exceptions.RequestException as e: + if hasattr(e, "response") and e.response is not None: + try: + error_detail = e.response.json() + error_msg = f"API Error {e.response.status_code}: {error_detail}" + except (ValueError, KeyError): + error_msg = f"API request failed: {e}" + else: + error_msg = f"Request failed: {e}" + raise SourcererAPIError(error_msg) from e diff --git a/src/poly/handlers/sync_client.py b/src/poly/handlers/sync_client.py index 4238e59f..41d33f30 100644 --- a/src/poly/handlers/sync_client.py +++ b/src/poly/handlers/sync_client.py @@ -310,3 +310,17 @@ def get_branch_chat_info(self, branch_id: str) -> dict[str, Any]: """Get deployment info needed to start a draft chat on a branch.""" self.assert_branch_exists() return self.sdk.get_branch_chat_info(branch_id) + + def get_branch_history(self, branch_id: str) -> list[dict[str, Any]]: + """Get the history of a specific branch. + + Args: + branch_id (str): The ID of the branch to retrieve history for. + + Returns: + list[dict[str, Any]]: A list of dictionaries containing commit information for the branch. + """ + logger.info(f"Fetching history for branch ID:'{branch_id}'") + history = self.sdk.get_branch_history(branch_id) + logger.info(f"Fetched {len(history)} commits for branch ID:'{branch_id}'") + return history diff --git a/src/poly/output/console.py b/src/poly/output/console.py index 86cfdf89..5b68f531 100644 --- a/src/poly/output/console.py +++ b/src/poly/output/console.py @@ -147,6 +147,29 @@ def print_branches(branches: dict[str, str] | list[str], current_branch: str | N console.print(f" {name}") +# {'merges': [{'branchId': 'BRANCH-RLX9GX78', 'branchName': 'Ruari Phipps / Release', 'mergedBy': 'ruari@poly-ai.com', 'mergedAt': '2026-07-21T14:07:06.909Z', 'source': None}], 'count': 1} +def print_branch_history(commits: list[dict[str, Any]]) -> None: + """Print a table of branch history commits.""" + if not commits: + console.print("[muted]No commits found for this branch.[/muted]") + return + + table = Table(box=None, show_header=False, header_style="bold", padding=(0, 1)) + table.add_column("Merged At", no_wrap=True) + table.add_column("Branch", no_wrap=True) + table.add_column("Merged By", no_wrap=True) + + for commit in commits: + merged_at = _format_iso_timestamp(commit.get("mergedAt", "")) + table.add_row( + merged_at, + commit.get("branchName", "—"), + commit.get("mergedBy", "—"), + ) + + console.print(table) + + def print_validation_errors(errors: list[str]) -> None: """Print validation errors in a styled list.""" console.print("[error]Project configuration is invalid.[/error]") diff --git a/src/poly/project.py b/src/poly/project.py index 7582e8e0..3e4605c7 100644 --- a/src/poly/project.py +++ b/src/poly/project.py @@ -2912,3 +2912,14 @@ def update_ab_test(self, ab_test_id: str, traffic_percentage: int) -> dict: ab_test_id=ab_test_id, traffic_percentage=traffic_percentage, ) + + def get_branch_history(self, branch_id: str) -> list[dict[str, Any]]: + """Get the history of a branch. + + Args: + branch_id (str): The ID of the branch to get history for. + + Returns: + list[dict[str, Any]]: A list of commit history entries for the branch. + """ + return self.api_handler.get_branch_history(branch_id) From 2359f593a464d2fef63e68a783dfa296030ad14b Mon Sep 17 00:00:00 2001 From: Ruari Phipps Date: Wed, 22 Jul 2026 10:53:23 +0100 Subject: [PATCH 2/9] Rename branch --- src/poly/cli_commands/branch.py | 78 ++++++++++++++++++++++++++++++++ src/poly/handlers/interface.py | 14 ++++++ src/poly/handlers/sdk.py | 32 +++++++++++++ src/poly/handlers/sync_client.py | 26 +++++++++++ src/poly/project.py | 23 ++++++++++ 5 files changed, 173 insertions(+) diff --git a/src/poly/cli_commands/branch.py b/src/poly/cli_commands/branch.py index 4b73b1f9..44663d34 100644 --- a/src/poly/cli_commands/branch.py +++ b/src/poly/cli_commands/branch.py @@ -246,6 +246,20 @@ def add_arguments(cls, subparsers: _SubParsersAction[ArgumentParser], parents: P branch_history_parser.set_defaults(branch_subcommand="history") + branch_rename_parser = branch_subparsers.add_parser( + "rename", + parents=[parents.path, parents.verbose, parents.json, parents.debug], + help="Rename a current branch.", + ) + branch_rename_parser.add_argument( + "new_branch_name", + type=str, + nargs="?", + default=None, + help="New name for the current branch.", + ) + branch_rename_parser.set_defaults(branch_subcommand="rename") + @classmethod def run(cls, args: Namespace) -> None: """Dispatch to the matching branch sub-handler.""" @@ -284,6 +298,9 @@ def run(cls, args: Namespace) -> None: elif args.branch_subcommand == "history": cls.branch_history(args.path, args.branch_name, args.json) + elif args.branch_subcommand == "rename": + cls.branch_rename(args.path, args.new_branch_name, args.json) + @classmethod def branch_list(cls, base_path: str, output_json: bool = False) -> None: """List branches in the Agent Studio project.""" @@ -967,3 +984,64 @@ def branch_history( plain(f"History for branch '{branch_name}':") print_branch_history(history) + + @classmethod + def branch_rename( + cls, base_path: str, new_branch_name: Optional[str] = None, output_json: bool = False + ) -> None: + """Rename the current branch in the Agent Studio project.""" + from poly.output.console import error, success, warning + + project = load_project(base_path, output_json=output_json) + + current_branch = project.get_current_branch() + if not current_branch: + if output_json: + json_print( + { + "success": False, + "error": "Current branch doesn't exist. Create a new branch before renaming.", + } + ) + else: + warning("Current branch doesn't exist. Create a new branch before renaming.") + return + + if current_branch == "main": + if output_json: + json_print({"success": False, "error": "Cannot rename the main branch."}) + else: + error("Cannot rename the main branch.") + return + + if not new_branch_name: + if output_json: + json_print({"success": False, "error": "No new branch name provided."}) + else: + new_branch_name = input("Enter the new name for the current branch: ").strip() + if not new_branch_name: + warning("No new branch name provided. Exiting.") + return + + try: + renamed = project.rename_branch(new_branch_name) + except (ValueError, Exception) as e: + if output_json: + json_print({"success": False, "error": str(e)}) + else: + error(str(e)) + return + + if output_json: + json_print( + { + "success": renamed, + "old_branch_name": current_branch, + "new_branch_name": new_branch_name, + } + ) + else: + if renamed: + success(f"Renamed branch '{current_branch}' to '{new_branch_name}'.") + else: + error(f"Failed to rename branch '{current_branch}' to '{new_branch_name}'.") diff --git a/src/poly/handlers/interface.py b/src/poly/handlers/interface.py index a045e66e..7acd7573 100644 --- a/src/poly/handlers/interface.py +++ b/src/poly/handlers/interface.py @@ -1111,3 +1111,17 @@ def get_branch_history(self, branch_name: str) -> list[dict[str, Any]]: return self.sync_client.get_branch_history(branch_name) except (requests.HTTPError, SourcererAPIError) as e: self._handle_api_error(e) + + def rename_branch(self, new_branch_name: str) -> bool: + """Rename the current branch to a new name. + + Args: + new_branch_name (str): The new name for the current branch + + Returns: + bool: True if the branch was renamed successfully, False otherwise + """ + try: + return self.sync_client.rename_branch(new_branch_name) + except (requests.HTTPError, SourcererAPIError) as e: + self._handle_api_error(e) diff --git a/src/poly/handlers/sdk.py b/src/poly/handlers/sdk.py index 0a874e2e..0351ca9b 100644 --- a/src/poly/handlers/sdk.py +++ b/src/poly/handlers/sdk.py @@ -708,3 +708,35 @@ def get_branch_history(self, branch_id: str) -> list[dict[str, Any]]: else: error_msg = f"Request failed: {e}" raise SourcererAPIError(error_msg) from e + + def rename_branch(self, new_branch_name: str) -> dict[str, Any]: + """Rename the current branch. + + Args: + new_branch_name: The new name for the current branch. + + Returns: + A dictionary containing the updated branch information. + + Raises: + SourcererAPIError: If the API request fails or if there is no current branch. + """ + if not self.branch_id: + raise SourcererAPIError("No current branch found. Cannot rename.") + + try: + url = f"{self._get_branches_url()}/{self.branch_id}" + payload = {"name": new_branch_name} + response = self.session.patch(url, json=payload) + response.raise_for_status() + return response.json() + except requests.exceptions.RequestException as e: + if hasattr(e, "response") and e.response is not None: + try: + error_detail = e.response.json() + error_msg = f"API Error {e.response.status_code}: {error_detail}" + except (ValueError, KeyError): + error_msg = f"API request failed: {e}" + else: + error_msg = f"Request failed: {e}" + raise SourcererAPIError(error_msg) from e diff --git a/src/poly/handlers/sync_client.py b/src/poly/handlers/sync_client.py index 41d33f30..87f5ac0d 100644 --- a/src/poly/handlers/sync_client.py +++ b/src/poly/handlers/sync_client.py @@ -324,3 +324,29 @@ def get_branch_history(self, branch_id: str) -> list[dict[str, Any]]: history = self.sdk.get_branch_history(branch_id) logger.info(f"Fetched {len(history)} commits for branch ID:'{branch_id}'") return history + + def rename_branch(self, new_branch_name: str) -> bool: + """Rename the current branch. + + Args: + new_branch_name (str): The new name for the current branch. + + Returns: + bool: True if the rename was successful, False otherwise. + """ + self.assert_branch_exists() + + if self.sdk.branch_id == "main": + logger.error("Cannot rename 'main' branch.") + return False + + logger.info(f"Renaming branch ID:'{self.sdk.branch_id}' to '{new_branch_name}'") + + try: + self.sdk.rename_branch(new_branch_name=new_branch_name) + except SourcererAPIError as e: + logger.error(f"Failed to rename branch ID:'{self.sdk.branch_id}': {e}") + return False + + logger.info(f"Successfully renamed branch ID:'{self.sdk.branch_id}' to '{new_branch_name}'") + return True diff --git a/src/poly/project.py b/src/poly/project.py index 3e4605c7..343a9da4 100644 --- a/src/poly/project.py +++ b/src/poly/project.py @@ -2923,3 +2923,26 @@ def get_branch_history(self, branch_id: str) -> list[dict[str, Any]]: list[dict[str, Any]]: A list of commit history entries for the branch. """ return self.api_handler.get_branch_history(branch_id) + + def rename_branch(self, new_branch_name: str) -> bool: + """Rename the current branch. + + Args: + new_branch_name (str): The new name for the current branch. + + Returns: + bool: True if the rename was successful, False otherwise. + """ + if not new_branch_name: + raise ValueError("New branch name must be provided.") + + if self.branch_id == "main": + raise ValueError("Renaming 'main' branch is not supported.") + + branches = self.api_handler.get_branches() + + if new_branch_name in branches: + raise ValueError(f"Branch {new_branch_name} already exists.") + + success = self.api_handler.rename_branch(new_branch_name=new_branch_name) + return success From f15350d2c36f579295495fd3f567bebb913bbd87 Mon Sep 17 00:00:00 2001 From: Ruari Phipps Date: Wed, 22 Jul 2026 11:24:17 +0100 Subject: [PATCH 3/9] Tests --- src/poly/tests/cli_test.py | 197 +++++++++++++++++++++++++++++++++ src/poly/tests/project_test.py | 75 ++++++++++++- 2 files changed, 271 insertions(+), 1 deletion(-) diff --git a/src/poly/tests/cli_test.py b/src/poly/tests/cli_test.py index 14ff4716..cbe003f5 100644 --- a/src/poly/tests/cli_test.py +++ b/src/poly/tests/cli_test.py @@ -2761,3 +2761,200 @@ def test_get_audio_writes_file(self, mock_success, mock_api, mock_open): ) mock_success.assert_called_once() self.assertIn("2.0 MB", mock_success.call_args[0][0]) + + +class BranchHistoryTest(unittest.TestCase): + """Tests for BranchCommand.branch_history CLI handler.""" + + SAMPLE_BRANCHES = {"main": "main-id", "feature-a": "branch-a-id"} + + def setUp(self): + self.mock_load_patcher = patch("poly.cli_commands.branch.load_project") + self.mock_load = self.mock_load_patcher.start() + self.proj = MagicMock() + self.proj.get_branches.return_value = ("feature-a", dict(self.SAMPLE_BRANCHES)) + self.mock_load.return_value = self.proj + + def tearDown(self): + patch.stopall() + + @patch("poly.output.console.print_branch_history") + @patch("poly.output.console.plain") + def test_defaults_to_current_branch(self, mock_plain, mock_print_history): + """When no branch_name is given, history uses the current branch.""" + self.proj.get_branch_history.return_value = [{"mergedAt": "2026-07-01", "branchName": "x"}] + + BranchCommand.branch_history(TEST_DIR) + + self.proj.get_branch_history.assert_called_once_with("branch-a-id") + mock_print_history.assert_called_once() + + @patch("poly.output.console.print_branch_history") + @patch("poly.output.console.plain") + def test_explicit_branch_name(self, mock_plain, mock_print_history): + """An explicit branch_name looks up its ID and fetches history.""" + self.proj.get_branch_history.return_value = [{"mergedAt": "2026-07-01"}] + + BranchCommand.branch_history(TEST_DIR, branch_name="main") + + self.proj.get_branch_history.assert_called_once_with("main-id") + + @patch("poly.cli_commands.branch.json_print") + def test_json_output(self, mock_json): + """JSON mode outputs branch_name, branch_id, and history.""" + history = [{"mergedAt": "2026-07-01", "branchName": "feat"}] + self.proj.get_branch_history.return_value = history + + BranchCommand.branch_history(TEST_DIR, branch_name="feature-a", output_json=True) + + mock_json.assert_called_once() + payload = mock_json.call_args[0][0] + self.assertEqual(payload["branch_name"], "feature-a") + self.assertEqual(payload["branch_id"], "branch-a-id") + self.assertEqual(payload["history"], history) + + @patch("poly.output.console.plain") + def test_empty_history_shows_message(self, mock_plain): + """When history is empty, a 'no history found' message is shown.""" + self.proj.get_branch_history.return_value = [] + + BranchCommand.branch_history(TEST_DIR, branch_name="feature-a") + + mock_plain.assert_called_once() + self.assertIn("No history found", mock_plain.call_args[0][0]) + + @patch("poly.output.console.warning") + def test_nonexistent_branch_shows_warning(self, mock_warning): + """A branch name not in the branches dict shows a warning.""" + BranchCommand.branch_history(TEST_DIR, branch_name="no-such-branch") + + self.proj.get_branch_history.assert_not_called() + mock_warning.assert_called_once() + self.assertIn("does not exist", mock_warning.call_args[0][0]) + + +class BranchRenameTest(unittest.TestCase): + """Tests for BranchCommand.branch_rename CLI handler.""" + + def setUp(self): + self.mock_load_patcher = patch("poly.cli_commands.branch.load_project") + self.mock_load = self.mock_load_patcher.start() + self.proj = MagicMock() + self.proj.get_current_branch.return_value = "feature-a" + self.proj.rename_branch.return_value = True + self.mock_load.return_value = self.proj + + def tearDown(self): + patch.stopall() + + @patch("poly.output.console.success") + def test_successful_rename(self, mock_success): + """A successful rename prints a success message.""" + BranchCommand.branch_rename(TEST_DIR, new_branch_name="new-name") + + self.proj.rename_branch.assert_called_once_with("new-name") + mock_success.assert_called_once() + self.assertIn("feature-a", mock_success.call_args[0][0]) + self.assertIn("new-name", mock_success.call_args[0][0]) + + @patch("poly.cli_commands.branch.json_print") + def test_successful_rename_json(self, mock_json): + """JSON mode outputs old_branch_name, new_branch_name, and success.""" + BranchCommand.branch_rename(TEST_DIR, new_branch_name="new-name", output_json=True) + + mock_json.assert_called_once() + payload = mock_json.call_args[0][0] + self.assertTrue(payload["success"]) + self.assertEqual(payload["old_branch_name"], "feature-a") + self.assertEqual(payload["new_branch_name"], "new-name") + + @patch("poly.output.console.error") + def test_rename_failure_shows_error(self, mock_error): + """When rename_branch returns False, a failure message is shown.""" + self.proj.rename_branch.return_value = False + + BranchCommand.branch_rename(TEST_DIR, new_branch_name="new-name") + + mock_error.assert_called_once() + self.assertIn("Failed to rename", mock_error.call_args[0][0]) + + @patch("poly.output.console.warning") + def test_no_current_branch_shows_warning(self, mock_warning): + """When current branch is None, a warning is shown.""" + self.proj.get_current_branch.return_value = None + + BranchCommand.branch_rename(TEST_DIR, new_branch_name="new-name") + + self.proj.rename_branch.assert_not_called() + mock_warning.assert_called_once() + self.assertIn("doesn't exist", mock_warning.call_args[0][0]) + + @patch("poly.cli_commands.branch.json_print") + def test_no_current_branch_json(self, mock_json): + """JSON mode outputs error when no current branch exists.""" + self.proj.get_current_branch.return_value = None + + BranchCommand.branch_rename(TEST_DIR, new_branch_name="new-name", output_json=True) + + mock_json.assert_called_once() + payload = mock_json.call_args[0][0] + self.assertFalse(payload["success"]) + self.assertIn("doesn't exist", payload["error"]) + + @patch("poly.output.console.error") + def test_main_branch_shows_error(self, mock_error): + """Renaming main branch shows an error.""" + self.proj.get_current_branch.return_value = "main" + + BranchCommand.branch_rename(TEST_DIR, new_branch_name="new-name") + + self.proj.rename_branch.assert_not_called() + mock_error.assert_called_once() + self.assertIn("Cannot rename the main branch", mock_error.call_args[0][0]) + + @patch("poly.cli_commands.branch.json_print") + def test_main_branch_json(self, mock_json): + """JSON mode outputs error when trying to rename main.""" + self.proj.get_current_branch.return_value = "main" + + BranchCommand.branch_rename(TEST_DIR, new_branch_name="new-name", output_json=True) + + mock_json.assert_called_once() + payload = mock_json.call_args[0][0] + self.assertFalse(payload["success"]) + self.assertIn("Cannot rename the main branch", payload["error"]) + + @patch("poly.cli_commands.branch.json_print") + def test_no_name_json_mode(self, mock_json): + """JSON mode outputs error when no name is provided.""" + self.proj.rename_branch.side_effect = ValueError("New branch name must be provided.") + + BranchCommand.branch_rename(TEST_DIR, new_branch_name=None, output_json=True) + + calls = mock_json.call_args_list + self.assertEqual(len(calls), 2) + self.assertFalse(calls[0][0][0]["success"]) + self.assertIn("No new branch name provided", calls[0][0][0]["error"]) + self.assertFalse(calls[1][0][0]["success"]) + + @patch("poly.output.console.error") + def test_rename_exception_shows_error(self, mock_error): + """When rename_branch raises, the error message is shown.""" + self.proj.rename_branch.side_effect = ValueError("Branch already exists.") + + BranchCommand.branch_rename(TEST_DIR, new_branch_name="existing") + + mock_error.assert_called_once() + self.assertIn("Branch already exists", mock_error.call_args[0][0]) + + @patch("poly.cli_commands.branch.json_print") + def test_rename_exception_json(self, mock_json): + """JSON mode outputs the error when rename_branch raises.""" + self.proj.rename_branch.side_effect = ValueError("Branch already exists.") + + BranchCommand.branch_rename(TEST_DIR, new_branch_name="existing", output_json=True) + + mock_json.assert_called_once() + payload = mock_json.call_args[0][0] + self.assertFalse(payload["success"]) + self.assertIn("Branch already exists", payload["error"]) diff --git a/src/poly/tests/project_test.py b/src/poly/tests/project_test.py index e8b91b67..1cfc1ddb 100644 --- a/src/poly/tests/project_test.py +++ b/src/poly/tests/project_test.py @@ -2386,7 +2386,9 @@ def test_validate_project_invalid_multiple(self): errors = project.validate_project() self.assertEqual(len(errors), 2) error_texts = "\n".join(errors) - self.assertIn("Invalid references: ['global_functions: FUNCTION-missing_function']", error_texts) + self.assertIn( + "Invalid references: ['global_functions: FUNCTION-missing_function']", error_texts + ) self.assertIn("Start step 'missing_step' not found.", error_texts) @@ -3708,5 +3710,76 @@ def test_already_migrated_entries_are_unchanged(self): self.assertEqual(flow_steps["FLOW-abc_step-1"]["resource_id"], "FLOW-abc_step-1") +class GetBranchHistoryProject(unittest.TestCase): + """Tests for AgentStudioProject.get_branch_history.""" + + def setUp(self): + self.project = AgentStudioProject.from_dict(PROJECT_DATA, TEST_DIR) + + def test_delegates_to_api_handler(self): + """get_branch_history passes through to the api_handler and returns its result.""" + expected = [{"commit_id": "c1"}, {"commit_id": "c2"}] + with patch.object(AgentStudioProject, "api_handler", new_callable=MagicMock) as mock_api: + mock_api.get_branch_history.return_value = expected + + result = self.project.get_branch_history("branch-1") + + self.assertEqual(result, expected) + mock_api.get_branch_history.assert_called_once_with("branch-1") + + +class RenameBranchProject(unittest.TestCase): + """Tests for AgentStudioProject.rename_branch.""" + + def setUp(self): + self.project = AgentStudioProject.from_dict(PROJECT_DATA, TEST_DIR) + + def test_empty_name_raises_value_error(self): + """An empty branch name raises ValueError.""" + with self.assertRaises(ValueError) as ctx: + self.project.rename_branch("") + + self.assertIn("New branch name must be provided", str(ctx.exception)) + + def test_none_name_raises_value_error(self): + """A None branch name raises ValueError.""" + with self.assertRaises(ValueError) as ctx: + self.project.rename_branch(None) + + self.assertIn("New branch name must be provided", str(ctx.exception)) + + def test_main_branch_raises_value_error(self): + """Renaming the main branch raises ValueError.""" + self.project.branch_id = "main" + + with self.assertRaises(ValueError) as ctx: + self.project.rename_branch("new-name") + + self.assertIn("main", str(ctx.exception)) + + def test_duplicate_name_raises_value_error(self): + """A name that already exists raises ValueError.""" + self.project.branch_id = "branch-1" + with patch.object(AgentStudioProject, "api_handler", new_callable=MagicMock) as mock_api: + mock_api.get_branches.return_value = {"new-name": "branch-id-123"} + + with self.assertRaises(ValueError) as ctx: + self.project.rename_branch("new-name") + + self.assertIn("already exists", str(ctx.exception)) + + def test_successful_rename_returns_true(self): + """A valid rename delegates to api_handler and returns its result.""" + self.project.branch_id = "branch-1" + with patch.object(AgentStudioProject, "api_handler", new_callable=MagicMock) as mock_api: + mock_api.get_branches.return_value = {"other-branch": "id-456"} + mock_api.rename_branch.return_value = True + + result = self.project.rename_branch("new-name") + + self.assertTrue(result) + mock_api.rename_branch.assert_called_once_with(new_branch_name="new-name") + + if __name__ == "__main__": unittest.main() From 33ff6e611d72494774fab103ab42740223355b1f Mon Sep 17 00:00:00 2001 From: Ruari Phipps Date: Wed, 22 Jul 2026 15:59:17 +0100 Subject: [PATCH 4/9] Archive restore --- src/poly/cli_commands/branch.py | 117 ++++++++++++- src/poly/handlers/interface.py | 25 +++ src/poly/handlers/sdk.py | 53 ++++++ src/poly/handlers/sync_client.py | 31 ++++ src/poly/output/console.py | 22 +++ src/poly/project.py | 37 ++++ src/poly/tests/api/interface_test.py | 134 +++++++++++++++ src/poly/tests/api/sync_client_test.py | 109 +++++++++++- src/poly/tests/cli_test.py | 226 ++++++++++++++++++++++++- src/poly/tests/project_test.py | 73 ++++++++ 10 files changed, 822 insertions(+), 5 deletions(-) diff --git a/src/poly/cli_commands/branch.py b/src/poly/cli_commands/branch.py index 44663d34..029287b9 100644 --- a/src/poly/cli_commands/branch.py +++ b/src/poly/cli_commands/branch.py @@ -118,6 +118,11 @@ def add_arguments(cls, subparsers: _SubParsersAction[ArgumentParser], parents: P parents=[parents.path, parents.verbose, parents.json, parents.debug], help="List all branches in the project.", ) + branch_list_parser.add_argument( + "--archived", + action="store_true", + help="Show soft-deleted (archived) branches instead of active ones.", + ) branch_list_parser.set_defaults(branch_subcommand="list") branch_create_parser = branch_subparsers.add_parser( @@ -260,11 +265,25 @@ def add_arguments(cls, subparsers: _SubParsersAction[ArgumentParser], parents: P ) branch_rename_parser.set_defaults(branch_subcommand="rename") + branch_restore_parser = branch_subparsers.add_parser( + "restore", + parents=[parents.path, parents.verbose, parents.json, parents.debug], + help="Restore a soft-deleted branch from the archive.", + ) + branch_restore_parser.add_argument( + "branch_name", + type=str, + nargs="?", + default=None, + help="Name of the archived branch to restore.", + ) + branch_restore_parser.set_defaults(branch_subcommand="restore") + @classmethod def run(cls, args: Namespace) -> None: """Dispatch to the matching branch sub-handler.""" if args.branch_subcommand == "list": - cls.branch_list(args.path, args.json) + cls.branch_list(args.path, args.json, getattr(args, "archived", False)) elif args.branch_subcommand == "create": cls.branch_create( @@ -301,13 +320,27 @@ def run(cls, args: Namespace) -> None: elif args.branch_subcommand == "rename": cls.branch_rename(args.path, args.new_branch_name, args.json) + elif args.branch_subcommand == "restore": + cls.branch_restore(args.path, args.branch_name, args.json) + @classmethod - def branch_list(cls, base_path: str, output_json: bool = False) -> None: + def branch_list(cls, base_path: str, output_json: bool = False, archived: bool = False) -> None: """List branches in the Agent Studio project.""" - from poly.output.console import plain, print_branches, warning + from poly.output.console import plain, print_archived_branches, print_branches, warning project = load_project(base_path, output_json=output_json) + if archived: + branches = project.list_archived_branches() + if output_json: + json_print({"archived_branches": branches}) + return + if not branches: + plain("[muted]No archived branches found.[/muted]") + return + print_archived_branches(branches) + return + current_branch, branches = project.get_branches() if output_json: @@ -1045,3 +1078,81 @@ def branch_rename( success(f"Renamed branch '{current_branch}' to '{new_branch_name}'.") else: error(f"Failed to rename branch '{current_branch}' to '{new_branch_name}'.") + + @classmethod + def branch_restore( + cls, + base_path: str, + branch_name: Optional[str] = None, + output_json: bool = False, + ) -> None: + """Restore a soft-deleted branch from the archive.""" + import questionary + + from poly.output.console import error, plain, success, warning + + project = load_project(base_path, output_json=output_json) + + if not branch_name: + if output_json: + json_print( + { + "success": False, + "error": "branch restore with --json requires a branch name argument.", + } + ) + sys.exit(1) + + archived = project.list_archived_branches() + if not archived: + plain("[muted]No archived branches to restore.[/muted]") + return + + choices = [] + branch_id_map: dict[str, str] = {} + for b in archived: + name = b.get("name", "—") + branch_id = b.get("branchId", "") + label = f"{name} ({branch_id})" + choices.append(label) + branch_id_map[label] = branch_id + + selected = questionary.select( + "Select branch to restore", + choices=choices, + use_search_filter=True, + use_jk_keys=False, + ).ask() + if not selected: + warning("No branch selected. Exiting.") + return + + selected_branch_id = branch_id_map[selected] + try: + restored = project.api_handler.restore_branch(selected_branch_id) + except (ValueError, Exception) as e: + error(str(e)) + return + if restored: + branch_name = selected.split(" (")[0] + success(f"Branch '{branch_name}' restored.") + else: + error("Failed to restore selected branch.") + return + + try: + restored = project.restore_branch(branch_name) + except (ValueError, Exception) as e: + if output_json: + json_print({"success": False, "error": str(e)}) + else: + error(str(e)) + return + + if output_json: + json_print({"success": restored, "branch_name": branch_name}) + else: + if restored: + success(f"Branch '{branch_name}' restored.") + else: + error(f"Failed to restore branch '{branch_name}'.") diff --git a/src/poly/handlers/interface.py b/src/poly/handlers/interface.py index 7acd7573..c614e61d 100644 --- a/src/poly/handlers/interface.py +++ b/src/poly/handlers/interface.py @@ -1125,3 +1125,28 @@ def rename_branch(self, new_branch_name: str) -> bool: return self.sync_client.rename_branch(new_branch_name) except (requests.HTTPError, SourcererAPIError) as e: self._handle_api_error(e) + + def list_archived_branches(self) -> list[dict[str, Any]]: + """List soft-deleted (archived) branches for the project. + + Returns: + list[dict[str, Any]]: A list of archived branch entries. + """ + try: + return self.sync_client.list_archived_branches() + except (requests.HTTPError, SourcererAPIError) as e: + self._handle_api_error(e) + + def restore_branch(self, branch_id: str) -> bool: + """Restore a soft-deleted branch from the archive. + + Args: + branch_id (str): The ID of the branch to restore. + + Returns: + bool: True if the branch was restored successfully, False otherwise. + """ + try: + return self.sync_client.restore_branch(branch_id) + except (requests.HTTPError, SourcererAPIError) as e: + self._handle_api_error(e) diff --git a/src/poly/handlers/sdk.py b/src/poly/handlers/sdk.py index 0351ca9b..a7c588de 100644 --- a/src/poly/handlers/sdk.py +++ b/src/poly/handlers/sdk.py @@ -709,6 +709,59 @@ def get_branch_history(self, branch_id: str) -> list[dict[str, Any]]: error_msg = f"Request failed: {e}" raise SourcererAPIError(error_msg) from e + def list_archived_branches(self) -> list[dict[str, Any]]: + """List soft-deleted (archived) branches for the project. + + Returns: + A list of dictionaries containing archived branch information. + + Raises: + SourcererAPIError: If the API request fails. + """ + try: + url = f"{self._get_branches_url()}/archive" + response = self.session.get(url) + response.raise_for_status() + return response.json().get("branches", []) + except requests.exceptions.RequestException as e: + if hasattr(e, "response") and e.response is not None: + try: + error_detail = e.response.json() + error_msg = f"API Error {e.response.status_code}: {error_detail}" + except (ValueError, KeyError): + error_msg = f"API request failed: {e}" + else: + error_msg = f"Request failed: {e}" + raise SourcererAPIError(error_msg) from e + + def restore_branch(self, branch_id: str) -> dict[str, Any]: + """Restore a soft-deleted branch from the archive. + + Args: + branch_id: The ID of the branch to restore. + + Returns: + A dictionary containing the restored branch information. + + Raises: + SourcererAPIError: If the API request fails. + """ + try: + url = f"{self._get_branches_url()}/{branch_id}/restore" + response = self.session.post(url) + response.raise_for_status() + return response.json() + except requests.exceptions.RequestException as e: + if hasattr(e, "response") and e.response is not None: + try: + error_detail = e.response.json() + error_msg = f"API Error {e.response.status_code}: {error_detail}" + except (ValueError, KeyError): + error_msg = f"API request failed: {e}" + else: + error_msg = f"Request failed: {e}" + raise SourcererAPIError(error_msg) from e + def rename_branch(self, new_branch_name: str) -> dict[str, Any]: """Rename the current branch. diff --git a/src/poly/handlers/sync_client.py b/src/poly/handlers/sync_client.py index 87f5ac0d..43ab9fec 100644 --- a/src/poly/handlers/sync_client.py +++ b/src/poly/handlers/sync_client.py @@ -350,3 +350,34 @@ def rename_branch(self, new_branch_name: str) -> bool: logger.info(f"Successfully renamed branch ID:'{self.sdk.branch_id}' to '{new_branch_name}'") return True + + def list_archived_branches(self) -> list[dict[str, Any]]: + """List soft-deleted (archived) branches for the project. + + Returns: + list[dict[str, Any]]: A list of dictionaries containing archived branch information. + """ + logger.info("Fetching archived branches") + branches = self.sdk.list_archived_branches() + logger.info(f"Fetched {len(branches)} archived branches") + return branches + + def restore_branch(self, branch_id: str) -> bool: + """Restore a soft-deleted branch from the archive. + + Args: + branch_id (str): The ID of the branch to restore. + + Returns: + bool: True if the restore was successful, False otherwise. + """ + logger.info(f"Restoring branch ID:'{branch_id}'") + + try: + self.sdk.restore_branch(branch_id) + except SourcererAPIError as e: + logger.error(f"Failed to restore branch ID:'{branch_id}': {e}") + return False + + logger.info(f"Successfully restored branch ID:'{branch_id}'") + return True diff --git a/src/poly/output/console.py b/src/poly/output/console.py index 5b68f531..841c4a7c 100644 --- a/src/poly/output/console.py +++ b/src/poly/output/console.py @@ -148,6 +148,28 @@ def print_branches(branches: dict[str, str] | list[str], current_branch: str | N # {'merges': [{'branchId': 'BRANCH-RLX9GX78', 'branchName': 'Ruari Phipps / Release', 'mergedBy': 'ruari@poly-ai.com', 'mergedAt': '2026-07-21T14:07:06.909Z', 'source': None}], 'count': 1} +def print_archived_branches(branches: list[dict[str, Any]]) -> None: + """Print a table of archived (soft-deleted) branches.""" + table = Table(box=None, show_header=True, header_style="bold", padding=(0, 1)) + table.add_column("Branch", no_wrap=True) + table.add_column("ID", style="muted", no_wrap=True) + table.add_column("Archived", no_wrap=True) + table.add_column("Expires", no_wrap=True) + + for branch in branches: + archived_at = _format_iso_timestamp(branch.get("archivedAt", "")) + days_left = branch.get("daysLeft") + expires = f"{days_left} days left" if days_left is not None else "—" + table.add_row( + branch.get("name", "—"), + branch.get("branchId", "—"), + archived_at, + expires, + ) + + console.print(table) + + def print_branch_history(commits: list[dict[str, Any]]) -> None: """Print a table of branch history commits.""" if not commits: diff --git a/src/poly/project.py b/src/poly/project.py index 343a9da4..d2d80adc 100644 --- a/src/poly/project.py +++ b/src/poly/project.py @@ -2946,3 +2946,40 @@ def rename_branch(self, new_branch_name: str) -> bool: success = self.api_handler.rename_branch(new_branch_name=new_branch_name) return success + + def list_archived_branches(self) -> list[dict[str, Any]]: + """List soft-deleted (archived) branches for the project. + + Returns: + list[dict[str, Any]]: A list of archived branch entries. + """ + return self.api_handler.list_archived_branches() + + def restore_branch(self, branch_name: str) -> bool: + """Restore a soft-deleted branch from the archive. + + Args: + branch_name (str): The name of the branch to restore. + + Returns: + bool: True if the branch was restored successfully, False otherwise. + """ + if not branch_name: + raise ValueError("Branch name must be provided.") + + archived = self.api_handler.list_archived_branches() + matches = [b for b in archived if b.get("name") == branch_name] + if not matches: + raise ValueError( + f"Branch '{branch_name}' not found in archive. " + "Use 'poly branch list --archived' to see available branches." + ) + + if len(matches) > 1: + branch_ids = ", ".join(b["branchId"] for b in matches) + raise ValueError( + f"Multiple archived branches named '{branch_name}' found ({branch_ids}). " + "Use the interactive picker (poly branch restore) to select the correct one." + ) + + return self.api_handler.restore_branch(matches[0]["branchId"]) diff --git a/src/poly/tests/api/interface_test.py b/src/poly/tests/api/interface_test.py index df822bac..14b9b048 100644 --- a/src/poly/tests/api/interface_test.py +++ b/src/poly/tests/api/interface_test.py @@ -201,5 +201,139 @@ def test_priority_delete_types_are_queued_first(self): self.assertLess(types.index("variable_delete"), types.index("delete_topic")) +class GetBranchHistoryInterface(unittest.TestCase): + """Tests for AgentStudioInterface.get_branch_history.""" + + def setUp(self): + self.interface = AgentStudioInterface() + self.interface.sync_client = MagicMock() + + def test_returns_history_from_sync_client(self): + """The interface delegates to sync_client and returns the result.""" + expected = [{"commit_id": "c1", "message": "first commit"}] + self.interface.sync_client.get_branch_history.return_value = expected + + result = self.interface.get_branch_history("branch-1") + + self.assertEqual(result, expected) + self.interface.sync_client.get_branch_history.assert_called_once_with("branch-1") + + def test_translates_http_error(self): + """An HTTPError from the sync client is translated into a ValueError.""" + self.interface.sync_client.get_branch_history.side_effect = requests.HTTPError("boom") + + with self.assertRaises(ValueError): + self.interface.get_branch_history("branch-1") + + def test_translates_sourcerer_api_error(self): + """A SourcererAPIError from the sync client is translated into a ValueError.""" + self.interface.sync_client.get_branch_history.side_effect = SourcererAPIError("boom") + + with self.assertRaises(ValueError): + self.interface.get_branch_history("branch-1") + + +class RenameBranchInterface(unittest.TestCase): + """Tests for AgentStudioInterface.rename_branch.""" + + def setUp(self): + self.interface = AgentStudioInterface() + self.interface.sync_client = MagicMock() + + def test_returns_result_from_sync_client(self): + """The interface delegates to sync_client and returns True on success.""" + self.interface.sync_client.rename_branch.return_value = True + + result = self.interface.rename_branch("new-name") + + self.assertTrue(result) + self.interface.sync_client.rename_branch.assert_called_once_with("new-name") + + def test_translates_http_error(self): + """An HTTPError from the sync client is translated into a ValueError.""" + self.interface.sync_client.rename_branch.side_effect = requests.HTTPError("boom") + + with self.assertRaises(ValueError): + self.interface.rename_branch("new-name") + + def test_translates_sourcerer_api_error(self): + """A SourcererAPIError from the sync client is translated into a ValueError.""" + self.interface.sync_client.rename_branch.side_effect = SourcererAPIError("boom") + + with self.assertRaises(ValueError): + self.interface.rename_branch("new-name") + + +class ListArchivedBranchesInterface(unittest.TestCase): + """Tests for AgentStudioInterface.list_archived_branches.""" + + def setUp(self): + self.interface = AgentStudioInterface() + self.interface.sync_client = MagicMock() + + def test_returns_result_from_sync_client(self): + """The interface delegates to sync_client and returns the result.""" + expected = [{"branchId": "b-1", "name": "old", "archivedAt": "2026-07-01"}] + self.interface.sync_client.list_archived_branches.return_value = expected + + result = self.interface.list_archived_branches() + + self.assertEqual(result, expected) + self.interface.sync_client.list_archived_branches.assert_called_once() + + def test_translates_http_error(self): + """An HTTPError from the sync client is translated into a ValueError.""" + self.interface.sync_client.list_archived_branches.side_effect = requests.HTTPError("boom") + + with self.assertRaises(ValueError): + self.interface.list_archived_branches() + + def test_translates_sourcerer_api_error(self): + """A SourcererAPIError is translated into a ValueError.""" + self.interface.sync_client.list_archived_branches.side_effect = SourcererAPIError("boom") + + with self.assertRaises(ValueError): + self.interface.list_archived_branches() + + +class RestoreBranchInterface(unittest.TestCase): + """Tests for AgentStudioInterface.restore_branch.""" + + def setUp(self): + self.interface = AgentStudioInterface() + self.interface.sync_client = MagicMock() + + def test_returns_result_from_sync_client(self): + """The interface delegates to sync_client and returns True on success.""" + self.interface.sync_client.restore_branch.return_value = True + + result = self.interface.restore_branch("branch-1") + + self.assertTrue(result) + self.interface.sync_client.restore_branch.assert_called_once_with("branch-1") + + def test_returns_false_from_sync_client(self): + """The interface returns False when sync_client returns False.""" + self.interface.sync_client.restore_branch.return_value = False + + result = self.interface.restore_branch("branch-1") + + self.assertFalse(result) + + def test_translates_http_error(self): + """An HTTPError from the sync client is translated into a ValueError.""" + self.interface.sync_client.restore_branch.side_effect = requests.HTTPError("boom") + + with self.assertRaises(ValueError): + self.interface.restore_branch("branch-1") + + def test_translates_sourcerer_api_error(self): + """A SourcererAPIError is translated into a ValueError.""" + self.interface.sync_client.restore_branch.side_effect = SourcererAPIError("boom") + + with self.assertRaises(ValueError): + self.interface.restore_branch("branch-1") + + if __name__ == "__main__": unittest.main() diff --git a/src/poly/tests/api/sync_client_test.py b/src/poly/tests/api/sync_client_test.py index baa3936f..3f7555b8 100644 --- a/src/poly/tests/api/sync_client_test.py +++ b/src/poly/tests/api/sync_client_test.py @@ -7,7 +7,7 @@ from unittest.mock import MagicMock, patch from poly.handlers.protobuf.commands_pb2 import Command -from poly.handlers.sdk import SourcererSDK +from poly.handlers.sdk import SourcererAPIError, SourcererSDK from poly.handlers.sync_client import SyncClientHandler @@ -64,5 +64,112 @@ def test_successful_send_returns_true(self): handler._sdk.send_command_batch.assert_called_once() +class GetBranchHistory(unittest.TestCase): + """Tests for SyncClientHandler.get_branch_history.""" + + def test_returns_history_from_sdk(self): + """The handler delegates to sdk.get_branch_history and returns its result.""" + handler = build_handler() + expected = [{"commit_id": "c1", "message": "initial"}] + handler._sdk.get_branch_history.return_value = expected + + result = handler.get_branch_history("branch-1") + + self.assertEqual(result, expected) + handler._sdk.get_branch_history.assert_called_once_with("branch-1") + + def test_returns_empty_list_when_no_history(self): + """An empty history list is returned as-is.""" + handler = build_handler() + handler._sdk.get_branch_history.return_value = [] + + result = handler.get_branch_history("branch-1") + + self.assertEqual(result, []) + + +class RenameBranch(unittest.TestCase): + """Tests for SyncClientHandler.rename_branch.""" + + def test_successful_rename_returns_true(self): + """A successful SDK rename call returns True.""" + handler = build_handler() + handler._sdk.branch_id = "branch-1" + handler._sdk.fetch_branches.return_value = {"branches": [{"branchId": "branch-1"}]} + + result = handler.rename_branch("new-name") + + self.assertTrue(result) + handler._sdk.rename_branch.assert_called_once_with(new_branch_name="new-name") + + def test_main_branch_returns_false(self): + """Renaming the main branch returns False without calling the SDK.""" + handler = build_handler() + handler._sdk.branch_id = "main" + handler._sdk.fetch_branches.return_value = {"branches": [{"branchId": "main"}]} + + result = handler.rename_branch("new-name") + + self.assertFalse(result) + handler._sdk.rename_branch.assert_not_called() + + def test_api_error_returns_false(self): + """A SourcererAPIError during rename returns False.""" + handler = build_handler() + handler._sdk.branch_id = "branch-1" + handler._sdk.fetch_branches.return_value = {"branches": [{"branchId": "branch-1"}]} + handler._sdk.rename_branch.side_effect = SourcererAPIError("rename failed") + + result = handler.rename_branch("new-name") + + self.assertFalse(result) + + +class ListArchivedBranches(unittest.TestCase): + """Tests for SyncClientHandler.list_archived_branches.""" + + def test_returns_archived_branches_from_sdk(self): + """The handler delegates to sdk.list_archived_branches and returns its result.""" + handler = build_handler() + expected = [{"branchId": "b-1", "name": "old-branch", "archivedAt": "2026-07-01"}] + handler._sdk.list_archived_branches.return_value = expected + + result = handler.list_archived_branches() + + self.assertEqual(result, expected) + handler._sdk.list_archived_branches.assert_called_once() + + def test_returns_empty_list_when_no_archived_branches(self): + """An empty list is returned as-is.""" + handler = build_handler() + handler._sdk.list_archived_branches.return_value = [] + + result = handler.list_archived_branches() + + self.assertEqual(result, []) + + +class RestoreBranch(unittest.TestCase): + """Tests for SyncClientHandler.restore_branch.""" + + def test_successful_restore_returns_true(self): + """A successful SDK restore call returns True.""" + handler = build_handler() + + result = handler.restore_branch("branch-1") + + self.assertTrue(result) + handler._sdk.restore_branch.assert_called_once_with("branch-1") + + def test_api_error_returns_false(self): + """A SourcererAPIError during restore returns False.""" + handler = build_handler() + handler._sdk.restore_branch.side_effect = SourcererAPIError("restore failed") + + result = handler.restore_branch("branch-1") + + self.assertFalse(result) + + if __name__ == "__main__": unittest.main() diff --git a/src/poly/tests/cli_test.py b/src/poly/tests/cli_test.py index cbe003f5..ee6d8d38 100644 --- a/src/poly/tests/cli_test.py +++ b/src/poly/tests/cli_test.py @@ -2282,7 +2282,9 @@ def test_user_cancels_account_selection(self, mock_select, mock_iface_cls, mock_ @patch("poly.cli_commands.project.AgentStudioInterface") @patch("questionary.text") @patch("questionary.select") - def test_user_cancels_project_name_entry(self, mock_select, mock_text, mock_iface_cls, mock_init): + def test_user_cancels_project_name_entry( + self, mock_select, mock_text, mock_iface_cls, mock_init + ): """create project returns early when user enters empty project name.""" mock_iface = mock_iface_cls.return_value mock_iface.get_accessible_regions.return_value = ["us-1", "euw-1"] @@ -2958,3 +2960,225 @@ def test_rename_exception_json(self, mock_json): payload = mock_json.call_args[0][0] self.assertFalse(payload["success"]) self.assertIn("Branch already exists", payload["error"]) + + +class BranchListArchivedTest(unittest.TestCase): + """Tests for BranchCommand.branch_list with --archived flag.""" + + def setUp(self): + self.mock_load_patcher = patch("poly.cli_commands.branch.load_project") + self.mock_load = self.mock_load_patcher.start() + self.proj = MagicMock() + self.mock_load.return_value = self.proj + + def tearDown(self): + patch.stopall() + + @patch("poly.output.console.print_archived_branches") + def test_archived_flag_calls_list_archived(self, mock_print): + """--archived delegates to list_archived_branches and prints the table.""" + archived = [ + { + "branchId": "BRANCH-1", + "name": "old-prompts", + "archivedAt": "2026-07-05", + "daysLeft": 15, + }, + ] + self.proj.list_archived_branches.return_value = archived + + BranchCommand.branch_list(TEST_DIR, archived=True) + + self.proj.list_archived_branches.assert_called_once() + mock_print.assert_called_once_with(archived) + + @patch("poly.output.console.plain") + def test_archived_empty_shows_message(self, mock_plain): + """When no archived branches exist, a 'no archived' message is shown.""" + self.proj.list_archived_branches.return_value = [] + + BranchCommand.branch_list(TEST_DIR, archived=True) + + mock_plain.assert_called_once() + self.assertIn("No archived branches", mock_plain.call_args[0][0]) + + @patch("poly.cli_commands.branch.json_print") + def test_archived_json_output(self, mock_json): + """JSON mode with --archived outputs archived_branches.""" + archived = [{"branchId": "BRANCH-1", "name": "old", "daysLeft": 30}] + self.proj.list_archived_branches.return_value = archived + + BranchCommand.branch_list(TEST_DIR, output_json=True, archived=True) + + mock_json.assert_called_once() + payload = mock_json.call_args[0][0] + self.assertEqual(payload["archived_branches"], archived) + + @patch("poly.output.console.print_branches") + def test_no_archived_flag_uses_normal_list(self, mock_print): + """Without --archived, branch_list uses the normal get_branches flow.""" + self.proj.get_branches.return_value = ("main", {"main": "main-id"}) + + BranchCommand.branch_list(TEST_DIR, archived=False) + + self.proj.list_archived_branches.assert_not_called() + self.proj.get_branches.assert_called_once() + + +class BranchRestoreTest(unittest.TestCase): + """Tests for BranchCommand.branch_restore CLI handler.""" + + def setUp(self): + self.mock_load_patcher = patch("poly.cli_commands.branch.load_project") + self.mock_load = self.mock_load_patcher.start() + self.proj = MagicMock() + self.proj.restore_branch.return_value = True + self.mock_load.return_value = self.proj + + def tearDown(self): + patch.stopall() + + @patch("poly.output.console.success") + def test_successful_restore(self, mock_success): + """A successful restore prints a success message.""" + BranchCommand.branch_restore(TEST_DIR, branch_name="old-branch") + + self.proj.restore_branch.assert_called_once_with("old-branch") + mock_success.assert_called_once() + self.assertIn("old-branch", mock_success.call_args[0][0]) + + @patch("poly.cli_commands.branch.json_print") + def test_successful_restore_json(self, mock_json): + """JSON mode outputs success and the branch name.""" + BranchCommand.branch_restore(TEST_DIR, branch_name="old-branch", output_json=True) + + mock_json.assert_called_once() + payload = mock_json.call_args[0][0] + self.assertTrue(payload["success"]) + self.assertEqual(payload["branch_name"], "old-branch") + + @patch("poly.output.console.error") + def test_restore_failure(self, mock_error): + """When restore_branch returns False, a failure message is shown.""" + self.proj.restore_branch.return_value = False + + BranchCommand.branch_restore(TEST_DIR, branch_name="old-branch") + + mock_error.assert_called_once() + self.assertIn("Failed to restore", mock_error.call_args[0][0]) + + @patch("poly.output.console.error") + def test_restore_not_found_shows_error(self, mock_error): + """When the branch isn't in the archive, the ValueError is shown.""" + self.proj.restore_branch.side_effect = ValueError("not found in archive") + + BranchCommand.branch_restore(TEST_DIR, branch_name="no-such-branch") + + mock_error.assert_called_once() + self.assertIn("not found in archive", mock_error.call_args[0][0]) + + @patch("poly.cli_commands.branch.json_print") + def test_restore_not_found_json(self, mock_json): + """JSON mode outputs the error when restore_branch raises.""" + self.proj.restore_branch.side_effect = ValueError("not found in archive") + + BranchCommand.branch_restore(TEST_DIR, branch_name="no-such-branch", output_json=True) + + mock_json.assert_called_once() + payload = mock_json.call_args[0][0] + self.assertFalse(payload["success"]) + self.assertIn("not found in archive", payload["error"]) + + @patch("poly.cli_commands.branch.json_print") + def test_duplicate_name_json(self, mock_json): + """JSON mode outputs the error when restore_branch raises for duplicate names.""" + self.proj.restore_branch.side_effect = ValueError( + "Multiple archived branches named 'release' found" + ) + + BranchCommand.branch_restore(TEST_DIR, branch_name="release", output_json=True) + + mock_json.assert_called_once() + payload = mock_json.call_args[0][0] + self.assertFalse(payload["success"]) + self.assertIn("Multiple archived branches", payload["error"]) + + @patch("poly.output.console.error") + def test_duplicate_name_shows_error(self, mock_error): + """When multiple archived branches share a name, the error is shown.""" + self.proj.restore_branch.side_effect = ValueError( + "Multiple archived branches named 'release' found" + ) + + BranchCommand.branch_restore(TEST_DIR, branch_name="release") + + mock_error.assert_called_once() + self.assertIn("Multiple archived branches", mock_error.call_args[0][0]) + + @patch("poly.cli_commands.branch.json_print") + def test_no_name_json_mode_exits(self, mock_json): + """JSON mode without a branch name prints error and exits.""" + with self.assertRaises(SystemExit): + BranchCommand.branch_restore(TEST_DIR, branch_name=None, output_json=True) + + mock_json.assert_called_once() + payload = mock_json.call_args[0][0] + self.assertFalse(payload["success"]) + self.assertIn("requires a branch name", payload["error"]) + + @patch("poly.output.console.plain") + def test_no_name_empty_archive_shows_message(self, mock_plain): + """Interactive mode with no archived branches shows a message.""" + self.proj.list_archived_branches.return_value = [] + + BranchCommand.branch_restore(TEST_DIR, branch_name=None) + + mock_plain.assert_called_once() + self.assertIn("No archived branches", mock_plain.call_args[0][0]) + self.proj.restore_branch.assert_not_called() + + @patch("questionary.select") + @patch("poly.output.console.warning") + def test_no_name_user_cancels_shows_warning(self, mock_warning, mock_select): + """Interactive mode where user cancels shows a warning.""" + self.proj.list_archived_branches.return_value = [ + {"branchId": "BRANCH-1", "name": "old-branch"} + ] + mock_select.return_value.ask.return_value = None + + BranchCommand.branch_restore(TEST_DIR, branch_name=None) + + mock_warning.assert_called_once() + self.assertIn("No branch selected", mock_warning.call_args[0][0]) + + @patch("questionary.select") + @patch("poly.output.console.success") + def test_no_name_interactive_success(self, mock_success, mock_select): + """Interactive mode selects a branch and restores it.""" + self.proj.list_archived_branches.return_value = [ + {"branchId": "BRANCH-1", "name": "old-branch"}, + {"branchId": "BRANCH-2", "name": "old-branch"}, + ] + mock_select.return_value.ask.return_value = "old-branch (BRANCH-2)" + self.proj.api_handler.restore_branch.return_value = True + + BranchCommand.branch_restore(TEST_DIR, branch_name=None) + + self.proj.api_handler.restore_branch.assert_called_once_with("BRANCH-2") + mock_success.assert_called_once() + self.assertIn("old-branch", mock_success.call_args[0][0]) + + @patch("questionary.select") + @patch("poly.output.console.error") + def test_no_name_interactive_restore_fails(self, mock_error, mock_select): + """Interactive mode shows error when restore returns False.""" + self.proj.list_archived_branches.return_value = [ + {"branchId": "BRANCH-1", "name": "old-branch"}, + ] + mock_select.return_value.ask.return_value = "old-branch (BRANCH-1)" + self.proj.api_handler.restore_branch.return_value = False + + BranchCommand.branch_restore(TEST_DIR, branch_name=None) + + mock_error.assert_called_once() + self.assertIn("Failed to restore", mock_error.call_args[0][0]) diff --git a/src/poly/tests/project_test.py b/src/poly/tests/project_test.py index 1cfc1ddb..2f36f725 100644 --- a/src/poly/tests/project_test.py +++ b/src/poly/tests/project_test.py @@ -3781,5 +3781,78 @@ def test_successful_rename_returns_true(self): mock_api.rename_branch.assert_called_once_with(new_branch_name="new-name") +class ListArchivedBranchesProject(unittest.TestCase): + """Tests for AgentStudioProject.list_archived_branches.""" + + def setUp(self): + self.project = AgentStudioProject.from_dict(PROJECT_DATA, TEST_DIR) + + def test_delegates_to_api_handler(self): + """list_archived_branches passes through to the api_handler.""" + expected = [{"branchId": "b-1", "name": "old", "archivedAt": "2026-07-01"}] + with patch.object(AgentStudioProject, "api_handler", new_callable=MagicMock) as mock_api: + mock_api.list_archived_branches.return_value = expected + + result = self.project.list_archived_branches() + + self.assertEqual(result, expected) + mock_api.list_archived_branches.assert_called_once() + + +class RestoreBranchProject(unittest.TestCase): + """Tests for AgentStudioProject.restore_branch.""" + + def setUp(self): + self.project = AgentStudioProject.from_dict(PROJECT_DATA, TEST_DIR) + + def test_empty_name_raises_value_error(self): + """An empty branch name raises ValueError.""" + with self.assertRaises(ValueError) as ctx: + self.project.restore_branch("") + + self.assertIn("Branch name must be provided", str(ctx.exception)) + + def test_branch_not_in_archive_raises_value_error(self): + """A name not found in the archive raises ValueError.""" + with patch.object(AgentStudioProject, "api_handler", new_callable=MagicMock) as mock_api: + mock_api.list_archived_branches.return_value = [ + {"branchId": "b-1", "name": "other-branch"} + ] + + with self.assertRaises(ValueError) as ctx: + self.project.restore_branch("no-such-branch") + + self.assertIn("not found in archive", str(ctx.exception)) + + def test_successful_restore_returns_true(self): + """A valid restore looks up the branch ID and delegates to api_handler.""" + with patch.object(AgentStudioProject, "api_handler", new_callable=MagicMock) as mock_api: + mock_api.list_archived_branches.return_value = [ + {"branchId": "b-1", "name": "old-branch", "archivedAt": "2026-07-01"}, + ] + mock_api.restore_branch.return_value = True + + result = self.project.restore_branch("old-branch") + + self.assertTrue(result) + mock_api.restore_branch.assert_called_once_with("b-1") + + def test_duplicate_name_raises_value_error(self): + """Multiple archived branches with the same name raises ValueError.""" + with patch.object(AgentStudioProject, "api_handler", new_callable=MagicMock) as mock_api: + mock_api.list_archived_branches.return_value = [ + {"branchId": "BRANCH-1", "name": "release"}, + {"branchId": "BRANCH-2", "name": "release"}, + ] + + with self.assertRaises(ValueError) as ctx: + self.project.restore_branch("release") + + self.assertIn("Multiple archived branches", str(ctx.exception)) + self.assertIn("BRANCH-1", str(ctx.exception)) + self.assertIn("BRANCH-2", str(ctx.exception)) + mock_api.restore_branch.assert_not_called() + + if __name__ == "__main__": unittest.main() From 6c30475571d4e89cac502b9ffd958e0429d5a9ff Mon Sep 17 00:00:00 2001 From: Ruari Phipps Date: Thu, 23 Jul 2026 10:42:51 +0100 Subject: [PATCH 5/9] list --- src/poly/cli_commands/branch.py | 15 ++++++++++-- src/poly/tests/cli_test.py | 42 +++++++++++++++++++++++++++++++++ 2 files changed, 55 insertions(+), 2 deletions(-) diff --git a/src/poly/cli_commands/branch.py b/src/poly/cli_commands/branch.py index 029287b9..acafa27a 100644 --- a/src/poly/cli_commands/branch.py +++ b/src/poly/cli_commands/branch.py @@ -248,6 +248,12 @@ def add_arguments(cls, subparsers: _SubParsersAction[ArgumentParser], parents: P default=None, help="Name of the branch to show history for. Defaults to the current branch.", ) + branch_history_parser.add_argument( + "--limit", + type=int, + default=10, + help="Number of history entries to show. Defaults to 10.", + ) branch_history_parser.set_defaults(branch_subcommand="history") @@ -315,7 +321,7 @@ def run(cls, args: Namespace) -> None: cls.branch_merge(args.path, args.message, args.json, args.interactive, args.resolutions) elif args.branch_subcommand == "history": - cls.branch_history(args.path, args.branch_name, args.json) + cls.branch_history(args.path, args.branch_name, args.json, args.limit) elif args.branch_subcommand == "rename": cls.branch_rename(args.path, args.new_branch_name, args.json) @@ -978,7 +984,11 @@ def branch_merge( @classmethod def branch_history( - cls, base_path: str, branch_name: Optional[str] = None, output_json: bool = False + cls, + base_path: str, + branch_name: Optional[str] = None, + output_json: bool = False, + limit: int = 10, ) -> None: """Show the history of a branch in the Agent Studio project.""" from poly.output.console import plain, print_branch_history, warning @@ -1006,6 +1016,7 @@ def branch_history( return history = project.get_branch_history(branch_id) + history = history[:limit] if output_json: json_print({"branch_name": branch_name, "branch_id": branch_id, "history": history}) diff --git a/src/poly/tests/cli_test.py b/src/poly/tests/cli_test.py index ee6d8d38..540ad1ad 100644 --- a/src/poly/tests/cli_test.py +++ b/src/poly/tests/cli_test.py @@ -2834,6 +2834,48 @@ def test_nonexistent_branch_shows_warning(self, mock_warning): mock_warning.assert_called_once() self.assertIn("does not exist", mock_warning.call_args[0][0]) + @patch("poly.output.console.print_branch_history") + @patch("poly.output.console.plain") + def test_limit_truncates_history(self, mock_plain, mock_print_history): + """--limit truncates history to the given number of entries.""" + self.proj.get_branch_history.return_value = [ + {"mergedAt": f"2026-07-{i:02d}"} for i in range(1, 21) + ] + + BranchCommand.branch_history(TEST_DIR, branch_name="feature-a", limit=5) + + mock_print_history.assert_called_once() + printed = mock_print_history.call_args[0][0] + self.assertEqual(len(printed), 5) + + @patch("poly.output.console.print_branch_history") + @patch("poly.output.console.plain") + def test_default_limit_is_10(self, mock_plain, mock_print_history): + """Without --limit, history is capped at 10 entries.""" + self.proj.get_branch_history.return_value = [ + {"mergedAt": f"2026-07-{i:02d}"} for i in range(1, 21) + ] + + BranchCommand.branch_history(TEST_DIR, branch_name="feature-a") + + mock_print_history.assert_called_once() + printed = mock_print_history.call_args[0][0] + self.assertEqual(len(printed), 10) + + @patch("poly.cli_commands.branch.json_print") + def test_limit_applies_to_json_output(self, mock_json): + """--limit also truncates history in JSON mode.""" + self.proj.get_branch_history.return_value = [ + {"mergedAt": f"2026-07-{i:02d}"} for i in range(1, 21) + ] + + BranchCommand.branch_history( + TEST_DIR, branch_name="feature-a", output_json=True, limit=3 + ) + + payload = mock_json.call_args[0][0] + self.assertEqual(len(payload["history"]), 3) + class BranchRenameTest(unittest.TestCase): """Tests for BranchCommand.branch_rename CLI handler.""" From ec0dbf8f39f318143caa7a0881702c55958e20e5 Mon Sep 17 00:00:00 2001 From: Ruari Phipps Date: Thu, 23 Jul 2026 11:49:49 +0100 Subject: [PATCH 6/9] Add page to list commands --- src/poly/cli_commands/branch.py | 17 +++++++++-------- src/poly/cli_commands/conversations.py | 5 +++-- src/poly/cli_commands/deployments.py | 23 +++++++++++++---------- src/poly/cli_commands/testing.py | 5 +++-- src/poly/output/console.py | 11 +++++++++++ src/poly/tests/cli_test.py | 25 +++++++++++++++++-------- 6 files changed, 56 insertions(+), 30 deletions(-) diff --git a/src/poly/cli_commands/branch.py b/src/poly/cli_commands/branch.py index acafa27a..89858e94 100644 --- a/src/poly/cli_commands/branch.py +++ b/src/poly/cli_commands/branch.py @@ -251,10 +251,9 @@ def add_arguments(cls, subparsers: _SubParsersAction[ArgumentParser], parents: P branch_history_parser.add_argument( "--limit", type=int, - default=10, - help="Number of history entries to show. Defaults to 10.", + default=None, + help="Maximum number of history entries to show. Shows all by default.", ) - branch_history_parser.set_defaults(branch_subcommand="history") branch_rename_parser = branch_subparsers.add_parser( @@ -988,10 +987,10 @@ def branch_history( base_path: str, branch_name: Optional[str] = None, output_json: bool = False, - limit: int = 10, + limit: Optional[int] = None, ) -> None: """Show the history of a branch in the Agent Studio project.""" - from poly.output.console import plain, print_branch_history, warning + from poly.output.console import paged_output, plain, print_branch_history, warning project = load_project(base_path, output_json=output_json) @@ -1016,7 +1015,8 @@ def branch_history( return history = project.get_branch_history(branch_id) - history = history[:limit] + if limit is not None: + history = history[:limit] if output_json: json_print({"branch_name": branch_name, "branch_id": branch_id, "history": history}) @@ -1026,8 +1026,9 @@ def branch_history( plain(f"[muted]No history found for branch '{branch_name}'.[/muted]") return - plain(f"History for branch '{branch_name}':") - print_branch_history(history) + with paged_output(): + plain(f"History for branch '{branch_name}':") + print_branch_history(history) @classmethod def branch_rename( diff --git a/src/poly/cli_commands/conversations.py b/src/poly/cli_commands/conversations.py index 52e8de4d..27d22e44 100644 --- a/src/poly/cli_commands/conversations.py +++ b/src/poly/cli_commands/conversations.py @@ -159,7 +159,7 @@ def conversations_list( offset: Number of conversations to skip. output_json: If True, emit machine-readable JSON. """ - from poly.output.console import info, print_conversations + from poly.output.console import info, paged_output, print_conversations project = load_project(base_path, output_json=output_json) result = AgentStudioInterface.list_conversations( @@ -176,7 +176,8 @@ def conversations_list( if not conversations: info("No conversations found.") return - print_conversations(conversations, url_builder=project.get_conversation_url) + with paged_output(): + print_conversations(conversations, url_builder=project.get_conversation_url) @classmethod def conversations_get( diff --git a/src/poly/cli_commands/deployments.py b/src/poly/cli_commands/deployments.py index 389d55df..d3f6aac2 100644 --- a/src/poly/cli_commands/deployments.py +++ b/src/poly/cli_commands/deployments.py @@ -62,8 +62,8 @@ def add_arguments(cls, subparsers: _SubParsersAction[ArgumentParser], parents: P deployment_list_parser.add_argument( "--limit", type=int, - default=10, - help="Number of versions to show. Defaults to 10.", + default=None, + help="Maximum number of versions to show. Shows all by default.", ) deployment_list_parser.add_argument( "--offset", @@ -479,7 +479,7 @@ def deployments_list( cls, base_path: str, environment: str = "sandbox", - limit: int = 10, + limit: Optional[int] = None, offset: int = 0, version_hash: str = None, output_json: bool = False, @@ -487,20 +487,20 @@ def deployments_list( ) -> None: """List deployment history for the project. - By default shows the 10 most recent deployments for the sandbox environment. + By default shows all deployments for the sandbox environment. Pass version_hash to start the listing from a specific version. Use details for full per-deployment metadata. Args: base_path: Base path for the project. environment: Environment to query — sandbox, pre-release, or live. - limit: Maximum number of versions to show. + limit: Maximum number of versions to show. Shows all by default. offset: Number of versions to skip before showing results. version_hash: Start listing from this version hash (overrides offset). output_json: If True, print result as JSON instead of rich text. details: If True, print full metadata for each deployment. """ - from poly.output.console import error, print_deployments + from poly.output.console import error, paged_output, print_deployments project = load_project(base_path) versions, active_deployment_hashes = project.get_deployments(client_env=environment) @@ -524,7 +524,8 @@ def deployments_list( return offset = version_idx - versions = versions[offset : offset + limit] + end = offset + limit if limit is not None else None + versions = versions[offset:end] if output_json: json_output = { "versions": versions, @@ -532,7 +533,8 @@ def deployments_list( } json_print(json_output) else: - print_deployments(versions, active_deployment_hashes, details=details) + with paged_output(): + print_deployments(versions, active_deployment_hashes, details=details) @classmethod def deployments_show( @@ -972,7 +974,7 @@ def ab_test_list( output_json: bool = False, ) -> None: """List A/B tests for the project.""" - from poly.output.console import print_ab_tests + from poly.output.console import paged_output, print_ab_tests project = load_project(base_path, output_json=output_json) ab_tests = project.list_ab_tests(limit=limit) @@ -980,7 +982,8 @@ def ab_test_list( json_print({"success": True, "ab_tests": ab_tests}) else: dep_map = cls._fetch_deployment_map(project) if ab_tests else {} - print_ab_tests(ab_tests, deployments=dep_map) + with paged_output(): + print_ab_tests(ab_tests, deployments=dep_map) @classmethod def ab_test_active( diff --git a/src/poly/cli_commands/testing.py b/src/poly/cli_commands/testing.py index 960f180c..7d254ea1 100644 --- a/src/poly/cli_commands/testing.py +++ b/src/poly/cli_commands/testing.py @@ -252,7 +252,7 @@ def testing_list( output_json: bool = False, ) -> None: """List test runs.""" - from poly.output.console import print_test_run_list + from poly.output.console import paged_output, print_test_run_list project = load_project(base_path) result = project.list_test_runs(limit=limit, offset=offset) @@ -261,7 +261,8 @@ def testing_list( json_print({"success": True, "test_runs": result}) return - print_test_run_list(result) + with paged_output(): + print_test_run_list(result) @classmethod def testing_show( diff --git a/src/poly/output/console.py b/src/poly/output/console.py index 841c4a7c..17684593 100644 --- a/src/poly/output/console.py +++ b/src/poly/output/console.py @@ -9,6 +9,7 @@ import os import sys from collections.abc import Callable +from contextlib import contextmanager from datetime import datetime from typing import Any from zoneinfo import ZoneInfo, ZoneInfoNotFoundError @@ -51,6 +52,16 @@ def set_verbose(verbose: bool) -> None: _verbose = verbose +@contextmanager +def paged_output(enabled: bool = True): + """Pipe output through the system pager when enabled and stdout is a TTY.""" + if enabled and console.is_terminal: + with console.pager(styles=True): + yield + else: + yield + + # ── Helpers ────────────────────────────────────────────────────────── diff --git a/src/poly/tests/cli_test.py b/src/poly/tests/cli_test.py index 540ad1ad..56380248 100644 --- a/src/poly/tests/cli_test.py +++ b/src/poly/tests/cli_test.py @@ -1248,17 +1248,27 @@ def test_no_versions_calls_error(self, mock_error): self.assertIn("No versions found", mock_error.call_args[0][0]) @patch("poly.output.console.print_deployments") - def test_default_call_shows_first_ten(self, mock_print_dep): - """Default call (no hash, no json) displays the first 10 versions.""" + def test_default_call_shows_all(self, mock_print_dep): + """Default call (no hash, no json, no limit) displays all versions.""" self.proj.get_deployments.return_value = (self.versions, self.active_hashes) DeploymentsCommand.deployments_list(TEST_DIR) mock_print_dep.assert_called_once() displayed_versions = mock_print_dep.call_args[0][0] - self.assertEqual(len(displayed_versions), 10) + self.assertEqual(len(displayed_versions), 15) self.assertEqual(displayed_versions[0]["name"], "v0") + @patch("poly.output.console.print_deployments") + def test_explicit_limit(self, mock_print_dep): + """--limit caps the number of displayed versions.""" + self.proj.get_deployments.return_value = (self.versions, self.active_hashes) + + DeploymentsCommand.deployments_list(TEST_DIR, limit=5) + + displayed_versions = mock_print_dep.call_args[0][0] + self.assertEqual(len(displayed_versions), 5) + @patch("poly.cli_commands.deployments.json_print") def test_output_json_calls_json_print(self, mock_json_print): """print_deployments with output_json=True calls json_print.""" @@ -1270,7 +1280,7 @@ def test_output_json_calls_json_print(self, mock_json_print): output = mock_json_print.call_args[0][0] self.assertIn("versions", output) self.assertIn("active_deployment_hashes", output) - self.assertEqual(len(output["versions"]), 10) + self.assertEqual(len(output["versions"]), 15) @patch("poly.output.console.print_deployments") def test_hash_sets_offset(self, mock_print_dep): @@ -2850,8 +2860,8 @@ def test_limit_truncates_history(self, mock_plain, mock_print_history): @patch("poly.output.console.print_branch_history") @patch("poly.output.console.plain") - def test_default_limit_is_10(self, mock_plain, mock_print_history): - """Without --limit, history is capped at 10 entries.""" + def test_no_limit_shows_all(self, mock_plain, mock_print_history): + """Without --limit, all history entries are shown.""" self.proj.get_branch_history.return_value = [ {"mergedAt": f"2026-07-{i:02d}"} for i in range(1, 21) ] @@ -2860,7 +2870,7 @@ def test_default_limit_is_10(self, mock_plain, mock_print_history): mock_print_history.assert_called_once() printed = mock_print_history.call_args[0][0] - self.assertEqual(len(printed), 10) + self.assertEqual(len(printed), 20) @patch("poly.cli_commands.branch.json_print") def test_limit_applies_to_json_output(self, mock_json): @@ -2876,7 +2886,6 @@ def test_limit_applies_to_json_output(self, mock_json): payload = mock_json.call_args[0][0] self.assertEqual(len(payload["history"]), 3) - class BranchRenameTest(unittest.TestCase): """Tests for BranchCommand.branch_rename CLI handler.""" From 91f9067df8e2f79e28c43170399e6fa25f04c17a Mon Sep 17 00:00:00 2001 From: Ruari Phipps Date: Wed, 26 Aug 2026 12:13:32 +0100 Subject: [PATCH 7/9] refactor: add return type hint to paged_output Co-Authored-By: Claude Opus 5 (1M context) --- src/poly/output/console.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/poly/output/console.py b/src/poly/output/console.py index e2536a7d..e1aaced6 100644 --- a/src/poly/output/console.py +++ b/src/poly/output/console.py @@ -9,7 +9,7 @@ import logging import os import sys -from collections.abc import Callable +from collections.abc import Callable, Iterator from contextlib import contextmanager from datetime import datetime from typing import Any @@ -57,7 +57,7 @@ def set_verbose(verbose: bool) -> None: @contextmanager -def paged_output(enabled: bool = True): +def paged_output(enabled: bool = True) -> Iterator[None]: """Pipe output through the system pager when enabled and stdout is a TTY.""" if enabled and console.is_terminal: with console.pager(styles=True): From 6f33a84ab3fba90c1068e80dc64efd557ce1023d Mon Sep 17 00:00:00 2001 From: Ruari Phipps Date: Wed, 26 Aug 2026 13:57:23 +0100 Subject: [PATCH 8/9] fix: only page output that overflows the screen Rich's default pager routes through pydoc, which hardcodes LESS without F (quit-if-one-screen) or X (stay out of the alternate screen), and overwrites any LESS the user has set. Short listings opened less, demanded q, and then vanished on quit. Decide in Python instead of delegating to less flags, so the behaviour holds for whichever pager is configured, and default LESS to git's FRX rather than clobbering it. Co-Authored-By: Claude Opus 5 (1M context) --- src/poly/output/console.py | 49 ++++++++++- src/poly/tests/console_test.py | 156 +++++++++++++++++++++++++++++++++ 2 files changed, 204 insertions(+), 1 deletion(-) diff --git a/src/poly/output/console.py b/src/poly/output/console.py index e1aaced6..2ed8f04f 100644 --- a/src/poly/output/console.py +++ b/src/poly/output/console.py @@ -8,6 +8,7 @@ import json import logging import os +import subprocess import sys from collections.abc import Callable, Iterator from contextlib import contextmanager @@ -18,6 +19,7 @@ from rich import box from rich.console import Console, Group from rich.live import Live +from rich.pager import Pager from rich.panel import Panel from rich.spinner import Spinner from rich.syntax import Syntax @@ -56,11 +58,56 @@ def set_verbose(verbose: bool) -> None: _verbose = verbose +class _TerminalPager(Pager): + """Pager that only engages when the content overflows the screen. + + Rich's default pager pipes everything to ``less`` via ``pydoc``, which + hardcodes ``LESS`` without ``F`` (quit-if-one-screen) or ``X`` (stay out of + the alternate screen). A three-row table would open ``less``, demand ``q``, + and then vanish on quit. git avoids that by setting ``LESS=FRX`` and letting + ``less -F`` decide; we make the same decision here instead, so it holds for + whichever pager the user has configured. + """ + + def show(self, content: str) -> None: + """Write content straight out if it fits, otherwise page it.""" + # Rich has already wrapped to the console width, so newlines are rows. + if content.count("\n") < console.size.height: + console.file.write(content) + return + + env = os.environ.copy() + # git's defaults: F quits if it fits, R keeps colour, X leaves output in + # the scrollback. setdefault, not assignment — a user-set LESS wins. + env.setdefault("LESS", "FRX") + try: + proc = subprocess.Popen( + os.environ.get("PAGER") or "less", + shell=True, + stdin=subprocess.PIPE, + env=env, + text=True, + errors="backslashreplace", + ) + except OSError: + # No usable pager — better to dump the output than to lose it. + console.file.write(content) + return + + try: + with proc.stdin as pipe: + pipe.write(content) + except OSError: + # The user quit before we finished writing; the pager has the rest. + pass + proc.wait() + + @contextmanager def paged_output(enabled: bool = True) -> Iterator[None]: """Pipe output through the system pager when enabled and stdout is a TTY.""" if enabled and console.is_terminal: - with console.pager(styles=True): + with console.pager(pager=_TerminalPager(), styles=True): yield else: yield diff --git a/src/poly/tests/console_test.py b/src/poly/tests/console_test.py index e56c7940..9b34b7ed 100644 --- a/src/poly/tests/console_test.py +++ b/src/poly/tests/console_test.py @@ -4,10 +4,15 @@ """ import unittest +from unittest.mock import MagicMock, PropertyMock, patch + +from rich.console import Console, ConsoleDimensions from poly.output.console import ( + _TerminalPager, console, flatten_branch_tree, + paged_output, print_archived_branches, print_branch_history, print_releases_branches, @@ -380,3 +385,154 @@ def test_table_does_not_mark_a_live_parent(self): self.assertIn("Active Release", output) self.assertNotIn("(archived)", output) + + +class TerminalPagerTest(unittest.TestCase): + """Tests for _TerminalPager, which pages only when output overflows the screen.""" + + SCREEN_HEIGHT = 25 + + def setUp(self): + size_patcher = patch.object( + Console, + "size", + new_callable=PropertyMock, + return_value=ConsoleDimensions(80, self.SCREEN_HEIGHT), + ) + size_patcher.start() + self.addCleanup(size_patcher.stop) + + def _show(self, content: str, env: dict | None = None): + """Run the pager over content, returning (written_directly, popen_mock).""" + written = [] + with ( + patch.object( + Console, + "file", + new_callable=PropertyMock, + return_value=MagicMock(write=written.append), + ), + patch("poly.output.console.subprocess.Popen") as mock_popen, + patch.dict("os.environ", env or {}, clear=True), + ): + _TerminalPager().show(content) + return "".join(written), mock_popen + + def test_content_shorter_than_screen_is_written_directly(self): + """Output that fits on one screen bypasses the pager entirely.""" + content = "row\n" * (self.SCREEN_HEIGHT - 1) + + written, mock_popen = self._show(content) + + self.assertEqual(written, content) + mock_popen.assert_not_called() + + def test_content_longer_than_screen_is_piped_to_the_pager(self): + """Output that overflows the screen is handed to the pager, not stdout.""" + content = "row\n" * (self.SCREEN_HEIGHT + 10) + + written, mock_popen = self._show(content) + + self.assertEqual(written, "") + mock_popen.assert_called_once() + # The pipe is entered as a context manager, so the write lands on __enter__. + pipe = mock_popen.return_value.stdin.__enter__.return_value + pipe.write.assert_called_once_with(content) + + def test_less_defaults_to_git_flags_when_unset(self): + """With LESS unset the pager inherits git's FRX, so it behaves like git.""" + content = "row\n" * (self.SCREEN_HEIGHT + 10) + + _, mock_popen = self._show(content) + + self.assertEqual(mock_popen.call_args.kwargs["env"]["LESS"], "FRX") + + def test_user_set_less_is_preserved(self): + """A user's own LESS wins over our default, matching git's behaviour.""" + content = "row\n" * (self.SCREEN_HEIGHT + 10) + + _, mock_popen = self._show(content, env={"LESS": "S"}) + + self.assertEqual(mock_popen.call_args.kwargs["env"]["LESS"], "S") + + def test_pager_command_honours_the_pager_env_var(self): + """PAGER selects the pager binary, falling back to less when unset.""" + content = "row\n" * (self.SCREEN_HEIGHT + 10) + + _, with_pager = self._show(content, env={"PAGER": "bat"}) + _, without_pager = self._show(content) + + self.assertEqual(with_pager.call_args[0][0], "bat") + self.assertEqual(without_pager.call_args[0][0], "less") + + def test_unusable_pager_falls_back_to_writing_output(self): + """If the pager cannot be spawned the output is dumped rather than lost.""" + content = "row\n" * (self.SCREEN_HEIGHT + 10) + written = [] + with ( + patch.object( + Console, + "file", + new_callable=PropertyMock, + return_value=MagicMock(write=written.append), + ), + patch("poly.output.console.subprocess.Popen", side_effect=OSError("no less")), + patch.dict("os.environ", {}, clear=True), + ): + _TerminalPager().show(content) + + self.assertEqual("".join(written), content) + + def test_quitting_the_pager_early_is_not_an_error(self): + """A broken pipe from quitting mid-write is swallowed, not raised.""" + content = "row\n" * (self.SCREEN_HEIGHT + 10) + with ( + patch.object(Console, "file", new_callable=PropertyMock, return_value=MagicMock()), + patch("poly.output.console.subprocess.Popen") as mock_popen, + patch.dict("os.environ", {}, clear=True), + ): + pipe = mock_popen.return_value.stdin.__enter__.return_value + pipe.write.side_effect = BrokenPipeError() + + _TerminalPager().show(content) + + mock_popen.return_value.wait.assert_called_once() + + +class PagedOutputTest(unittest.TestCase): + """Tests for the paged_output context manager's TTY and enabled guards.""" + + def test_no_paging_when_stdout_is_not_a_terminal(self): + """Piped and redirected output is never paged, matching git.""" + with ( + patch.object(Console, "is_terminal", new_callable=PropertyMock, return_value=False), + patch.object(console, "pager") as mock_pager, + ): + with paged_output(): + pass + + mock_pager.assert_not_called() + + def test_no_paging_when_explicitly_disabled(self): + """Passing enabled=False opts a caller out even on a terminal.""" + with ( + patch.object(Console, "is_terminal", new_callable=PropertyMock, return_value=True), + patch.object(console, "pager") as mock_pager, + ): + with paged_output(enabled=False): + pass + + mock_pager.assert_not_called() + + def test_paging_uses_the_terminal_pager_with_styles(self): + """On a terminal the custom pager is used, with styles kept for colour.""" + with ( + patch.object(Console, "is_terminal", new_callable=PropertyMock, return_value=True), + patch.object(console, "pager") as mock_pager, + ): + with paged_output(): + pass + + mock_pager.assert_called_once() + self.assertIsInstance(mock_pager.call_args.kwargs["pager"], _TerminalPager) + self.assertTrue(mock_pager.call_args.kwargs["styles"]) From e13f0f2273db47ee52e0fbfa3ab924e86981be50 Mon Sep 17 00:00:00 2001 From: Ruari Phipps Date: Wed, 26 Aug 2026 16:28:58 +0100 Subject: [PATCH 9/9] Redo pager --- src/poly/output/console.py | 54 +++++-------------- src/poly/tests/console_test.py | 97 ++++++++-------------------------- 2 files changed, 35 insertions(+), 116 deletions(-) diff --git a/src/poly/output/console.py b/src/poly/output/console.py index 2ed8f04f..9a350fa5 100644 --- a/src/poly/output/console.py +++ b/src/poly/output/console.py @@ -8,7 +8,6 @@ import json import logging import os -import subprocess import sys from collections.abc import Callable, Iterator from contextlib import contextmanager @@ -19,7 +18,7 @@ from rich import box from rich.console import Console, Group from rich.live import Live -from rich.pager import Pager +from rich.pager import Pager, SystemPager from rich.panel import Panel from rich.spinner import Spinner from rich.syntax import Syntax @@ -58,56 +57,31 @@ def set_verbose(verbose: bool) -> None: _verbose = verbose -class _TerminalPager(Pager): +class _OverflowPager(Pager): """Pager that only engages when the content overflows the screen. - Rich's default pager pipes everything to ``less`` via ``pydoc``, which - hardcodes ``LESS`` without ``F`` (quit-if-one-screen) or ``X`` (stay out of - the alternate screen). A three-row table would open ``less``, demand ``q``, - and then vanish on quit. git avoids that by setting ``LESS=FRX`` and letting - ``less -F`` decide; we make the same decision here instead, so it holds for - whichever pager the user has configured. + Rich's default pager sends everything to ``less``, so even a three-row table + would open the pager and demand ``q``. git avoids that with ``less -F`` + (quit-if-one-screen); deciding it here instead means it also holds for a + reader whose ``PAGER`` is ``bat`` or ``more``, or whose ``LESS`` has no + ``F``. Spawning the pager, colour passthrough, ctrl-c and quitting early are + all left to Rich and pydoc, which already handle them. """ def show(self, content: str) -> None: - """Write content straight out if it fits, otherwise page it.""" + """Write content straight out if it fits, otherwise hand it to the pager.""" # Rich has already wrapped to the console width, so newlines are rows. if content.count("\n") < console.size.height: console.file.write(content) - return - - env = os.environ.copy() - # git's defaults: F quits if it fits, R keeps colour, X leaves output in - # the scrollback. setdefault, not assignment — a user-set LESS wins. - env.setdefault("LESS", "FRX") - try: - proc = subprocess.Popen( - os.environ.get("PAGER") or "less", - shell=True, - stdin=subprocess.PIPE, - env=env, - text=True, - errors="backslashreplace", - ) - except OSError: - # No usable pager — better to dump the output than to lose it. - console.file.write(content) - return - - try: - with proc.stdin as pipe: - pipe.write(content) - except OSError: - # The user quit before we finished writing; the pager has the rest. - pass - proc.wait() + else: + SystemPager().show(content) @contextmanager def paged_output(enabled: bool = True) -> Iterator[None]: """Pipe output through the system pager when enabled and stdout is a TTY.""" if enabled and console.is_terminal: - with console.pager(pager=_TerminalPager(), styles=True): + with console.pager(pager=_OverflowPager(), styles=True): yield else: yield @@ -365,7 +339,7 @@ def print_branch_history(commits: list[dict[str, Any]]) -> None: console.print("[muted]No commits found for this branch.[/muted]") return - table = Table(box=None, show_header=False, header_style="bold", padding=(0, 1)) + table = Table(box=None, show_header=True, header_style="bold", padding=(0, 1)) table.add_column("Merged At", no_wrap=True) table.add_column("Branch", no_wrap=True) table.add_column("Merged By", no_wrap=True) @@ -757,7 +731,7 @@ def print_deployments( if not details: table = Table( box=None, - show_header=False, + show_header=True, header_style="bold", padding=(0, 1), ) diff --git a/src/poly/tests/console_test.py b/src/poly/tests/console_test.py index 9b34b7ed..a57e7d16 100644 --- a/src/poly/tests/console_test.py +++ b/src/poly/tests/console_test.py @@ -9,7 +9,7 @@ from rich.console import Console, ConsoleDimensions from poly.output.console import ( - _TerminalPager, + _OverflowPager, console, flatten_branch_tree, paged_output, @@ -387,8 +387,8 @@ def test_table_does_not_mark_a_live_parent(self): self.assertNotIn("(archived)", output) -class TerminalPagerTest(unittest.TestCase): - """Tests for _TerminalPager, which pages only when output overflows the screen.""" +class OverflowPagerTest(unittest.TestCase): + """Tests for _OverflowPager, which pages only when output overflows.""" SCREEN_HEIGHT = 25 @@ -402,8 +402,8 @@ def setUp(self): size_patcher.start() self.addCleanup(size_patcher.stop) - def _show(self, content: str, env: dict | None = None): - """Run the pager over content, returning (written_directly, popen_mock).""" + def _show(self, content: str): + """Run the pager over content, returning (written_directly, system_pager).""" written = [] with ( patch.object( @@ -412,91 +412,36 @@ def _show(self, content: str, env: dict | None = None): new_callable=PropertyMock, return_value=MagicMock(write=written.append), ), - patch("poly.output.console.subprocess.Popen") as mock_popen, - patch.dict("os.environ", env or {}, clear=True), + patch("poly.output.console.SystemPager") as mock_system_pager, ): - _TerminalPager().show(content) - return "".join(written), mock_popen + _OverflowPager().show(content) + return "".join(written), mock_system_pager def test_content_shorter_than_screen_is_written_directly(self): """Output that fits on one screen bypasses the pager entirely.""" content = "row\n" * (self.SCREEN_HEIGHT - 1) - written, mock_popen = self._show(content) + written, mock_system_pager = self._show(content) self.assertEqual(written, content) - mock_popen.assert_not_called() + mock_system_pager.assert_not_called() - def test_content_longer_than_screen_is_piped_to_the_pager(self): - """Output that overflows the screen is handed to the pager, not stdout.""" + def test_content_longer_than_screen_is_handed_to_the_pager(self): + """Output that overflows the screen goes to the pager, not stdout.""" content = "row\n" * (self.SCREEN_HEIGHT + 10) - written, mock_popen = self._show(content) + written, mock_system_pager = self._show(content) self.assertEqual(written, "") - mock_popen.assert_called_once() - # The pipe is entered as a context manager, so the write lands on __enter__. - pipe = mock_popen.return_value.stdin.__enter__.return_value - pipe.write.assert_called_once_with(content) + mock_system_pager.return_value.show.assert_called_once_with(content) - def test_less_defaults_to_git_flags_when_unset(self): - """With LESS unset the pager inherits git's FRX, so it behaves like git.""" - content = "row\n" * (self.SCREEN_HEIGHT + 10) - - _, mock_popen = self._show(content) - - self.assertEqual(mock_popen.call_args.kwargs["env"]["LESS"], "FRX") - - def test_user_set_less_is_preserved(self): - """A user's own LESS wins over our default, matching git's behaviour.""" - content = "row\n" * (self.SCREEN_HEIGHT + 10) - - _, mock_popen = self._show(content, env={"LESS": "S"}) - - self.assertEqual(mock_popen.call_args.kwargs["env"]["LESS"], "S") - - def test_pager_command_honours_the_pager_env_var(self): - """PAGER selects the pager binary, falling back to less when unset.""" - content = "row\n" * (self.SCREEN_HEIGHT + 10) - - _, with_pager = self._show(content, env={"PAGER": "bat"}) - _, without_pager = self._show(content) - - self.assertEqual(with_pager.call_args[0][0], "bat") - self.assertEqual(without_pager.call_args[0][0], "less") - - def test_unusable_pager_falls_back_to_writing_output(self): - """If the pager cannot be spawned the output is dumped rather than lost.""" - content = "row\n" * (self.SCREEN_HEIGHT + 10) - written = [] - with ( - patch.object( - Console, - "file", - new_callable=PropertyMock, - return_value=MagicMock(write=written.append), - ), - patch("poly.output.console.subprocess.Popen", side_effect=OSError("no less")), - patch.dict("os.environ", {}, clear=True), - ): - _TerminalPager().show(content) - - self.assertEqual("".join(written), content) - - def test_quitting_the_pager_early_is_not_an_error(self): - """A broken pipe from quitting mid-write is swallowed, not raised.""" - content = "row\n" * (self.SCREEN_HEIGHT + 10) - with ( - patch.object(Console, "file", new_callable=PropertyMock, return_value=MagicMock()), - patch("poly.output.console.subprocess.Popen") as mock_popen, - patch.dict("os.environ", {}, clear=True), - ): - pipe = mock_popen.return_value.stdin.__enter__.return_value - pipe.write.side_effect = BrokenPipeError() + def test_content_exactly_filling_the_screen_is_not_paged(self): + """The boundary case stays inline rather than opening the pager.""" + content = "row\n" * (self.SCREEN_HEIGHT - 1) - _TerminalPager().show(content) + _, mock_system_pager = self._show(content) - mock_popen.return_value.wait.assert_called_once() + mock_system_pager.assert_not_called() class PagedOutputTest(unittest.TestCase): @@ -524,7 +469,7 @@ def test_no_paging_when_explicitly_disabled(self): mock_pager.assert_not_called() - def test_paging_uses_the_terminal_pager_with_styles(self): + def test_paging_uses_the_overflow_pager_with_styles(self): """On a terminal the custom pager is used, with styles kept for colour.""" with ( patch.object(Console, "is_terminal", new_callable=PropertyMock, return_value=True), @@ -534,5 +479,5 @@ def test_paging_uses_the_terminal_pager_with_styles(self): pass mock_pager.assert_called_once() - self.assertIsInstance(mock_pager.call_args.kwargs["pager"], _TerminalPager) + self.assertIsInstance(mock_pager.call_args.kwargs["pager"], _OverflowPager) self.assertTrue(mock_pager.call_args.kwargs["styles"])