[feature] Add interactive recovery for docstrfmt errors

Author: Dhanus3133

openwisp_utils/releaser/changelog.py

@@ -5,6 +5,8 @@
 import sys
 import tempfile
 
+import questionary
+
 
 def find_cliff_config():
     # Locates the cliff.toml file packaged within 'openwisp_utils'.
@@ -175,12 +177,29 @@ def format_rst_block(content):
         ).strip()
         return formatted_block
 
-    except (subprocess.CalledProcessError, FileNotFoundError) as e:
-        print("\nWarning: Could not format content with `docstrfmt`.", file=sys.stderr)
-        if isinstance(e, subprocess.CalledProcessError):
-            print(f"{e.stderr}", file=sys.stderr)
-        # Return original content if formatting fails
-        return content.strip()
+    except subprocess.CalledProcessError as e:
+        print(
+            "\n❌ ERROR: Could not format changelog content with `docstrfmt`.",
+            file=sys.stderr,
+        )
+        print("Linter output:", file=sys.stderr)
+        indented_stderr = "\n".join(
+            [f"    {line}" for line in e.stderr.strip().split("\n")]
+        )
+        print(indented_stderr, file=sys.stderr)
+        print(
+            "\nThis is likely due to a syntax error in the generated changelog from git-cliff."
+        )
+        decision = questionary.confirm(
+            "Continue with the unformatted changelog? You can fix the syntax manually later.",
+            default=True,
+        ).ask()
+
+        if decision:
+            return content.strip()
+        else:
+            print("\n❌ Release process aborted by user.")
+            sys.exit(1)
     finally:
         if temp_file_path and os.path.exists(temp_file_path):
             os.remove(temp_file_path)

openwisp_utils/releaser/tests/test_changelog.py

@@ -148,14 +148,42 @@ def test_run_git_cliff_called_process_error(mock_run, mock_find_config):
         run_git_cliff()
 
 
-@patch(
-    "subprocess.run",
-    side_effect=subprocess.CalledProcessError(1, "cmd", stderr="error"),
-)
-def test_format_rst_block_failure(mock_run):
-    original_content = "Some content"
-    formatted_content = format_rst_block(original_content)
-    assert formatted_content == original_content.strip()
+@patch("openwisp_utils.releaser.changelog.questionary.confirm")
+@patch("openwisp_utils.releaser.changelog.subprocess.run")
+def test_format_rst_block_failure_user_continues(mock_subprocess, mock_questionary):
+    """Tests that if formatting fails, the user can choose to continue with the unformatted content."""
+    # Simulate user choosing to continue
+    mock_questionary.return_value.ask.return_value = True
+    # Simulate a docstrfmt failure
+    mock_subprocess.side_effect = subprocess.CalledProcessError(
+        1, "docstrfmt", stderr="Syntax error on line 5."
+    )
+    original_content = "* a list"
+
+    result = format_rst_block(original_content)
+
+    # The function should return the original, unformatted content
+    assert result == original_content
+    # The user should have been prompted
+    mock_questionary.assert_called_once()
+
+
+@patch("openwisp_utils.releaser.changelog.questionary.confirm")
+@patch("openwisp_utils.releaser.changelog.subprocess.run")
+def test_format_rst_block_failure_user_aborts(mock_subprocess, mock_questionary):
+    """Tests that if formatting fails, the user can choose to abort the release process."""
+    # Simulate user choosing to abort
+    mock_questionary.return_value.ask.return_value = False
+    # Simulate a docstrfmt failure
+    mock_subprocess.side_effect = subprocess.CalledProcessError(
+        1, "docstrfmt", stderr="Syntax error on line 5."
+    )
+
+    with pytest.raises(SystemExit):
+        format_rst_block("* a list")
+
+    # The user should have been prompted
+    mock_questionary.assert_called_once()
 
 
 SAMPLE_CHANGELOG = """Changelog

openwisp_utils/releaser/tests/test_utils.py

@@ -1,3 +1,4 @@
+import subprocess
 from unittest.mock import MagicMock, mock_open, patch
 
 import pytest
@@ -155,8 +156,8 @@ def test_get_release_block_from_md_file(mock_file):
 
 @patch("openwisp_utils.releaser.utils.subprocess.run")
 @patch("builtins.print")
-def test_format_file_with_docstrfmt(mock_print, mock_subprocess):
-    """Tests the `format_file_with_docstrfmt` function."""
+def test_format_file_with_docstrfmt_success(mock_print, mock_subprocess):
+    """Tests the successful `format_file_with_docstrfmt` function."""
     file_path = "CHANGES.rst"
     format_file_with_docstrfmt(file_path)
 
@@ -178,6 +179,66 @@ def test_format_file_with_docstrfmt(mock_print, mock_subprocess):
     mock_print.assert_called_once_with(f"✅ Formatted {file_path} successfully.")
 
 
+@patch("openwisp_utils.releaser.utils.questionary.select")
+@patch("openwisp_utils.releaser.utils.subprocess.run")
+def test_format_file_interactive_retry_success(mock_subprocess, mock_questionary):
+    # Tests the interactive retry loop in `format_file_with_docstrfmt`.
+    # The first format attempt fails, the user "fixes" it, and the second succeeds.
+
+    # First call fails, second call succeeds
+    mock_subprocess.side_effect = [
+        subprocess.CalledProcessError(1, "docstrfmt", stderr="Syntax error"),
+        MagicMock(),  # Represents a successful run
+    ]
+    # User chooses to retry
+    mock_questionary.return_value.ask.return_value = "I have fixed the file, try again."
+
+    format_file_with_docstrfmt("CHANGES.rst")
+
+    # subprocess.run should have been called twice
+    assert mock_subprocess.call_count == 2
+    mock_questionary.assert_called_once()
+
+
+@patch("openwisp_utils.releaser.utils.questionary.select")
+@patch("openwisp_utils.releaser.utils.subprocess.run")
+def test_format_file_interactive_skip(mock_subprocess, mock_questionary):
+    """Tests that the user can skip formatting in `format_file_with_docstrfmt`."""
+    # The call to docstrfmt fails
+    mock_subprocess.side_effect = subprocess.CalledProcessError(
+        1, "docstrfmt", stderr="Syntax error"
+    )
+    # User chooses to skip
+    mock_questionary.return_value.ask.return_value = (
+        "Skip formatting and continue with the unformatted file."
+    )
+
+    format_file_with_docstrfmt("CHANGES.rst")
+
+    # subprocess.run should only have been called once
+    assert mock_subprocess.call_count == 1
+    mock_questionary.assert_called_once()
+
+
+@patch("openwisp_utils.releaser.utils.questionary.select")
+@patch("openwisp_utils.releaser.utils.subprocess.run")
+def test_format_file_interactive_abort(mock_subprocess, mock_questionary):
+    """Tests that the user can abort the release in `format_file_with_docstrfmt`."""
+    # The call to docstrfmt fails
+    mock_subprocess.side_effect = subprocess.CalledProcessError(
+        1, "docstrfmt", stderr="Syntax error"
+    )
+    # User chooses to abort
+    mock_questionary.return_value.ask.return_value = "Abort the release process."
+
+    with pytest.raises(SystemExit):
+        format_file_with_docstrfmt("CHANGES.rst")
+
+    # subprocess.run should have been called once before aborting
+    assert mock_subprocess.call_count == 1
+    mock_questionary.assert_called_once()
+
+
 @patch("requests.request")
 def test_retryable_request_success(mock_request):
     """Tests that the function returns a response on the first successful call."""

openwisp_utils/releaser/utils.py

@@ -95,11 +95,47 @@ def rst_to_markdown(text):
 
 def format_file_with_docstrfmt(file_path):
     """Format a file using `docstrfmt`."""
-    subprocess.run(
-        ["docstrfmt", "--ignore-cache", "--line-length", "74", file_path],
-        check=True,
-        capture_output=True,
-        text=True,
-        encoding="utf-8",
-    )
-    print(f"✅ Formatted {file_path} successfully.")
+    while True:
+        try:
+            subprocess.run(
+                ["docstrfmt", "--ignore-cache", "--line-length", "74", file_path],
+                check=True,
+                capture_output=True,
+                text=True,
+                encoding="utf-8",
+            )
+            print(f"✅ Formatted {file_path} successfully.")
+            break
+        except subprocess.CalledProcessError as e:
+            print(
+                f"\n❌ ERROR: `docstrfmt` failed to format '{file_path}'.",
+                file=sys.stderr,
+            )
+            print("Linter output:", file=sys.stderr)
+            # Indent the error for readability
+            indented_stderr = "\n".join(
+                [f"    {line}" for line in e.stderr.strip().split("\n")]
+            )
+            print(indented_stderr, file=sys.stderr)
+
+            print(
+                f"\nThis is likely caused by a syntax error in '{file_path}'.\n"
+                "Please open the file in another terminal, fix the issue, and save it."
+            )
+            decision = questionary.select(
+                "How would you like to proceed?",
+                choices=[
+                    "I have fixed the file, try again.",
+                    "Skip formatting and continue with the unformatted file.",
+                    "Abort the release process.",
+                ],
+            ).ask()
+
+            if decision == "I have fixed the file, try again.":
+                continue
+            elif decision == "Skip formatting and continue with the unformatted file.":
+                print(f"⚠️  Skipped formatting for '{file_path}'.")
+                break
+            else:  # Abort or None
+                print("\n❌ Release process aborted by user.")
+                sys.exit(1)