Add prompts for plan and run

- Move text styling code from steps/execute.py to a newly created
utils/text_styler.py to be shared by multiple modules
- Add prompt to warn user about the potential discrepency between
plan and actual upgrade steps
- Add prompt to let the user confirm about proceeding with plan

cou/cli.py

@@ -22,6 +22,7 @@
 from signal import SIGINT, SIGTERM
 from typing import Optional
 
+from aioconsole import ainput
 from juju.errors import JujuError
 
 from cou.commands import parse_args
@@ -34,6 +35,7 @@
 from cou.utils import progress_indicator
 from cou.utils.cli import interrupt_handler
 from cou.utils.juju_utils import COUModel
+from cou.utils.text_styler import bold, normal
 
 AVAILABLE_OPTIONS = "cas"
 
@@ -72,6 +74,17 @@ def _missing_(cls, value: object) -> Enum:
         raise ValueError(f"{value} is not a valid member of VerbosityLevel.")
 
 
+def prompt_message(parameter: str) -> str:
+    """Generate eye-catching prompt.
+
+    :param parameter: String to show at the prompt with the user options.
+    :type parameter: str
+    :return: Prompt string with the user options.
+    :rtype: str
+    """
+    return normal("\n" + parameter + " (") + bold("y") + normal("/") + bold("N") + normal("): ")
+
+
 def get_log_level(quiet: bool = False, verbosity: int = 0) -> str:
     """Get a log level based on input options.
 
@@ -126,9 +139,13 @@ async def get_upgrade_plan(model_name: Optional[str], backup_database: bool) ->
     :type backup_database: bool
     """
     analysis_result, upgrade_plan = await analyze_and_plan(model_name, backup_database)
-    logger.info(upgrade_plan)
+    logger.debug(upgrade_plan)
     print(upgrade_plan)  # print plan to console even in quiet mode
     manually_upgrade_data_plane(analysis_result)
+    print(
+        "Please note that the actually upgrade steps could be different "
+        "because the plan will be re-calculated at upgrade time."
+    )
 
 
 async def run_upgrade(
@@ -149,7 +166,24 @@ async def run_upgrade(
     :type quiet: bool
     """
     analysis_result, upgrade_plan = await analyze_and_plan(model_name, backup_database)
-    logger.info(upgrade_plan)
+    logger.debug(upgrade_plan)
+    print(upgrade_plan)
+
+    if interactive:
+        prompt_input = (
+            await ainput(prompt_message("Would you like to start the upgrade?"))
+        ).casefold()
+
+        match prompt_input:
+            case "y":
+                logger.info("Start the upgrade.")
+            case "n":
+                logger.info("Exiting COU without running upgrades.")
+                return
+            case _:
+                print("No valid input provided! Exiting COU without upgrades.")
+                logger.debug("No valid input provided! Exiting COU without upgrades.")
+                return
 
     # NOTE(rgildein): add handling upgrade plan canceling for SIGINT (ctrl+c) and SIGTERM
     loop = asyncio.get_event_loop()
@@ -158,7 +192,6 @@ async def run_upgrade(
 
     # don't print plan if in quiet mode
     if not quiet:
-        print(upgrade_plan)
         print("Running cloud upgrade...")
 
     await apply_step(upgrade_plan, interactive)

cou/steps/execute.py

@@ -18,45 +18,24 @@
 import sys
 
 from aioconsole import ainput
-from colorama import Style
 
 from cou.steps import ApplicationUpgradePlan, BaseStep, UpgradeStep
 from cou.utils import progress_indicator
+from cou.utils.text_styler import bold, normal
 
 AVAILABLE_OPTIONS = ["y", "n"]
 
 logger = logging.getLogger(__name__)
 
 
-def prompt(parameter: str) -> str:
+def prompt_message(parameter: str) -> str:
     """Generate eye-catching prompt.
 
     :param parameter: String to show at the prompt with the user options.
     :type parameter: str
     :return: Prompt string with the user options.
     :rtype: str
     """
-
-    def bold(text: str) -> str:
-        """Transform the text in bold format.
-
-        :param text: text to format.
-        :type text: str
-        :return: text formatted.
-        :rtype: str
-        """
-        return Style.RESET_ALL + Style.BRIGHT + text + Style.RESET_ALL
-
-    def normal(text: str) -> str:
-        """Transform the text in normal format.
-
-        :param text: text to format.
-        :type text: str
-        :return: text formatted.
-        :rtype: str
-        """
-        return Style.RESET_ALL + text + Style.RESET_ALL
-
     return (
         normal("\n" + parameter + "\nContinue (")
         + bold("y")
@@ -135,7 +114,7 @@ async def apply_step(step: BaseStep, interactive: bool, overwrite_progress: bool
         if not interactive or not step.prompt:
             result = "y"
         else:
-            result = (await ainput(prompt(description_to_prompt))).casefold()
+            result = (await ainput(prompt_message(description_to_prompt))).casefold()
 
         match result:
             case "y":

cou/utils/text_styler.py

@@ -0,0 +1,39 @@
+# Copyright 2023 Canonical Limited
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Command line text styling utilities."""
+
+from colorama import Style
+
+
+def bold(text: str) -> str:
+    """Transform the text in bold format.
+
+    :param text: text to format.
+    :type text: str
+    :return: text formatted.
+    :rtype: str
+    """
+    return Style.RESET_ALL + Style.BRIGHT + text + Style.RESET_ALL
+
+
+def normal(text: str) -> str:
+    """Transform the text in normal format.
+
+    :param text: text to format.
+    :type text: str
+    :return: text formatted.
+    :rtype: str
+    """
+    return Style.RESET_ALL + text + Style.RESET_ALL

tests/unit/steps/test_steps_execute.py

@@ -27,7 +27,7 @@
     UpgradePlan,
     UpgradeStep,
 )
-from cou.steps.execute import _run_step, apply_step, prompt
+from cou.steps.execute import _run_step, apply_step, prompt_message
 
 
 @pytest.mark.asyncio
@@ -109,7 +109,7 @@ async def test_apply_step_abort(mock_run_step, mock_input, input_value):
     with pytest.raises(SystemExit):
         await apply_step(upgrade_step, True)
 
-    mock_input.assert_awaited_once_with(prompt("Test Step"))
+    mock_input.assert_awaited_once_with(prompt_message("Test Step"))
     mock_run_step.assert_not_awaited()
 
 
@@ -137,7 +137,7 @@ async def test_apply_step_continue(mock_run_step, mock_input, input_value):
 
     await apply_step(upgrade_step, True)
 
-    mock_input.assert_awaited_once_with(prompt("Test Step"))
+    mock_input.assert_awaited_once_with(prompt_message("Test Step"))
     mock_run_step.assert_awaited_once_with(upgrade_step, True, False)
 
 
@@ -152,7 +152,9 @@ async def test_apply_step_nonsense(mock_run_step, mock_input):
     with pytest.raises(SystemExit, match="1"):
         await apply_step(upgrade_step, True)
 
-    mock_input.assert_has_awaits([call(prompt("Test Step")), call(prompt("Test Step"))])
+    mock_input.assert_has_awaits(
+        [call(prompt_message("Test Step")), call(prompt_message("Test Step"))]
+    )
     mock_run_step.assert_not_awaited()
 
 
@@ -173,7 +175,7 @@ async def test_apply_application_upgrade_plan(mock_run_step, mock_input):
     mock_input.side_effect = ["y"]
     await apply_step(upgrade_plan, True)
 
-    mock_input.assert_has_awaits([call(prompt(expected_prompt))])
+    mock_input.assert_has_awaits([call(prompt_message(expected_prompt))])
 
 
 @pytest.mark.asyncio

tests/unit/test_cli.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, call, patch
 
 import pytest
 from juju.errors import JujuError
@@ -90,7 +90,7 @@ async def test_get_upgrade_plan(mock_logger, mock_analyze_and_plan, mock_manuall
     await cli.get_upgrade_plan(None, True)
 
     mock_analyze_and_plan.assert_awaited_once_with(None, True)
-    mock_logger.info.assert_called_once_with(plan)
+    mock_logger.debug.assert_called_once_with(plan)
     mock_manually_upgrade.assert_called_once()
 
 
@@ -107,7 +107,9 @@ async def test_get_upgrade_plan(mock_logger, mock_analyze_and_plan, mock_manuall
 @patch("cou.cli.apply_step")
 @patch("builtins.print")
 @patch("cou.cli.logger")
+@patch("cou.cli.ainput")
 async def test_run_upgrade_quiet(
+    mock_input,
     mock_logger,
     mock_print,
     mock_apply_step,
@@ -122,15 +124,72 @@ async def test_run_upgrade_quiet(
     mock_analysis_result = MagicMock()
     mock_analyze_and_plan.return_value = (mock_analysis_result, plan)
 
-    await cli.run_upgrade(model_name=None, backup_database=True, interactive=True, quiet=quiet)
+    await cli.run_upgrade(model_name=None, backup_database=True, interactive=False, quiet=quiet)
 
+    mock_input.assert_not_awaited()
     mock_analyze_and_plan.assert_awaited_once_with(None, True)
-    mock_logger.info.assert_called_once_with(plan)
-    mock_apply_step.assert_called_once_with(plan, True)
+    mock_logger.debug.assert_called_once_with(plan)
+    mock_apply_step.assert_called_once_with(plan, False)
     mock_print.call_count == expected_print_count
     mock_manually_upgrade.assert_called_once()
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("input_value", ["n", "x"])
+@patch("cou.cli.manually_upgrade_data_plane")
+@patch("cou.cli.analyze_and_plan", new_callable=AsyncMock)
+@patch("cou.cli.apply_step")
+@patch("cou.cli.ainput")
+async def test_run_upgrade_with_prompt_abort(
+    mock_input,
+    mock_apply_step,
+    mock_analyze_and_plan,
+    mock_manually_upgrade,
+    input_value,
+):
+    plan = UpgradePlan(description="Upgrade cloud from 'ussuri' to 'victoria'")
+    plan.add_step(PreUpgradeStep(description="backup mysql databases", parallel=False))
+    mock_analysis_result = MagicMock()
+    mock_analyze_and_plan.return_value = (mock_analysis_result, plan)
+    mock_input.side_effect = input_value
+
+    await cli.run_upgrade(model_name=None, backup_database=True, interactive=True, quiet=False)
+
+    mock_analyze_and_plan.assert_awaited_once_with(None, True)
+    mock_input.assert_has_awaits(
+        [call(cli.prompt_message("Would you like to start the upgrade?"))]
+    )
+    mock_apply_step.assert_not_awaited()
+    mock_manually_upgrade.assert_not_called()
+
+
+@pytest.mark.asyncio
+@patch("cou.cli.manually_upgrade_data_plane")
+@patch("cou.cli.analyze_and_plan", new_callable=AsyncMock)
+@patch("cou.cli.apply_step")
+@patch("cou.cli.ainput")
+async def test_run_upgrade_with_prompt_continue(
+    mock_input,
+    mock_apply_step,
+    mock_analyze_and_plan,
+    mock_manually_upgrade,
+):
+    plan = UpgradePlan(description="Upgrade cloud from 'ussuri' to 'victoria'")
+    plan.add_step(PreUpgradeStep(description="backup mysql databases", parallel=False))
+    mock_analysis_result = MagicMock()
+    mock_analyze_and_plan.return_value = (mock_analysis_result, plan)
+    mock_input.side_effect = "y"
+
+    await cli.run_upgrade(model_name=None, backup_database=True, interactive=True, quiet=False)
+
+    mock_analyze_and_plan.assert_awaited_once_with(None, True)
+    mock_input.assert_has_awaits(
+        [call(cli.prompt_message("Would you like to start the upgrade?"))]
+    )
+    mock_apply_step.assert_called_once_with(plan, True)
+    mock_manually_upgrade.assert_called_once()
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("command", ["plan", "run", "other1", "other2"])
 @patch("cou.cli.get_upgrade_plan")