feat(status): allow to nest statuses and prompt inside

hexagon/support/printer/logger.py

@@ -1,4 +1,5 @@
-from typing import Optional, Union
+from types import TracebackType
+from typing import Optional, Union, Type
 
 from rich.console import Console
 from rich.syntax import Syntax
@@ -12,6 +13,7 @@ def __init__(
  ) -> None:
  self.__console = None
  self.__decorations = None
+ self.__live_status = None
  if isinstance(theme, str):
  self.load_theme(theme)
  else:
@@ -73,7 +75,11 @@ def finish(self, message: str = None):
  self.__console.print(f"{self.__decorations.finish}{message or ''}")
 
  def status(self, message: str = None):
- return self.__console.status(message)
+ if self.__live_status:
+ self.__live_status.update(message)
+ else:
+ self.__live_status = NestableStatus(self.__console.status(message))
+ return self.__live_status
 
  def load_theme(self, theme: Union[str, LoggingTheme], console: Console = None):
  self.__decorations = (
@@ -85,3 +91,75 @@ def load_theme(self, theme: Union[str, LoggingTheme], console: Console = None):
 
  def use_borders(self):
  return self.__decorations.prompt_border
+
+ def status_aware(self, decorated):
+ """
+ Decorator that stops the status bar while the decorated function is running.
+
+ :param decorated: the function to be decorated
+ :return: the decorated function
+ """
+
+ def decorator(cls, *args, **kwargs):
+ if self.__live_status:
+ self.__live_status.stop()
+ res = decorated(*args, **kwargs)
+ if self.__live_status:
+ self.__live_status.start()
+ return res
+
+ return decorator
+
+
+class NestableStatus:
+ """
+ A wrapper for rich's Status class that allows nesting status calls.
+
+ ie.:
+ with log.status("foo"):
+ # status should show "foo"
+ with log.status("bar"):
+ # status should show "bar"
+ # status should show "foo"
+ # status should be hidden
+
+ """
+
+ def __init__(self, status):
+ self.__status = status
+ self.__levels = 1
+
+ @property
+ def renderable(self):
+ return self.__status.renderable
+
+ @property
+ def console(self):
+ return self.__status.console
+
+ def update(self, *args, **kwargs):
+ self.__levels += 1
+ self.__status.update(*args, **kwargs)
+
+ def start(self):
+ self.__status.start()
+
+ def stop(self) -> None:
+ self.__status.stop()
+
+ def __rich__(self):
+ return self.__status.renderable
+
+ def __enter__(self):
+ self.__status.start()
+ return self
+
+ def __exit__(
+ self,
+ exc_type: Optional[Type[BaseException]],
+ exc_val: Optional[BaseException],
+ exc_tb: Optional[TracebackType],
+ ):
+ self.__levels -= 1
+ if self.__levels == 0:
+ self.stop()

hexagon/support/prompt.py

@@ -10,6 +10,7 @@
 
 from hexagon.domain.args import HexagonArg
 from hexagon.support.printer import log
+from hexagon.utils.decorators import for_all_methods
 from hexagon.utils.typing import field_info
 
 
@@ -60,6 +61,7 @@ def set_default(options, model_field: ModelField):
  return {}
 
 
+@for_all_methods(log.status_aware)
 class Prompt:
  def query_field(self, model_field: ModelField, model_class, **kwargs):
  inq = self.text

hexagon/support/update/cli.py

@@ -11,7 +11,7 @@
 from hexagon.support.printer import log
 from hexagon.support.prompt import prompt
 from hexagon.support.update.shared import already_checked_for_updates
-from hexagon.utils.silent_fail import silent_fail
+from hexagon.utils.decorators import silent_fail
 
 
 @silent_fail()

hexagon/support/update/hexagon.py

@@ -11,7 +11,7 @@
 from hexagon.support.update.changelog.format import format_entries
 from hexagon.support.update.changelog.parse import parse_changelog
 from hexagon.support.update.shared import already_checked_for_updates
-from hexagon.utils.silent_fail import silent_fail
+from hexagon.utils.decorators import silent_fail
 
 CHANGELOG_MAX_PRINT_ENTRIES = 10
 

hexagon/utils/silent_fail.py → hexagon/utils/decorators.py

@@ -16,3 +16,13 @@ def wrapper():
  return wrapper
 
  return decorator
+
+
+def for_all_methods(decorator):
+ def decorate(cls):
+ for attr in cls.__dict__:
+ if callable(getattr(cls, attr)):
+ setattr(cls, attr, decorator(getattr(cls, attr)))
+ return cls
+
+ return decorate

tests_e2e/__specs/status.py

@@ -0,0 +1,37 @@
+from tests_e2e.__specs.utils.hexagon_spec import as_a_user
+
+
+def test_status_is_shown_correctly():
+ """
+ We are not actually testing the display of the spinner,
+ rich hides it when not running in a TTY.
+ But we are testing that status nesting and prompting works correctly.
+ """
+ (
+ as_a_user(__file__)
+ .run_hexagon(["status"], os_env_vars={"HEXAGON_THEME": "no_border"})
+ .then_output_should_be(
+ ["Test", "", "some info 1", "some info 2", "update?"],
+ ignore_blank_lines=False,
+ )
+ .input("y")
+ .then_output_should_be(
+ [
+ "update? Yes",
+ "about to show nested status",
+ "inside nested status",
+ "prompting...",
+ "enter something",
+ ]
+ )
+ .input("asdf")
+ .then_output_should_be(
+ [
+ "entered",
+ "back to main status",
+ "showing a new status...",
+ "inside new status",
+ ],
+ )
+ .exit(status=0)
+ )

tests_e2e/__specs/utils/hexagon_spec.py

@@ -54,7 +54,6 @@ class HexagonSpec:
  HEXAGON_THEME = "HEXAGON_THEME"
  HEXAGON_UPDATE_DISABLED = "HEXAGON_UPDATE_DISABLED"
  HEXAGON_CLI_UPDATE_DISABLED = "HEXAGON_CLI_UPDATE_DISABLED"
- HEXAGON_DISABLE_SPINNER = "HEXAGON_DISABLE_SPINNER"
  HEXAGON_SEND_TELEMETRY = "HEXAGON_SEND_TELEMETRY"
  HEXAGON_STORAGE_PATH = "HEXAGON_STORAGE_PATH"
  HEXAGON_CONFIG_FILE = "HEXAGON_CONFIG_FILE"

tests_e2e/__specs/utils/run.py

@@ -45,9 +45,6 @@ def run_hexagon_e2e_test(
  if "HEXAGON_CLI_UPDATE_DISABLED" not in os_env_vars:
  os_env_vars["HEXAGON_CLI_UPDATE_DISABLED"] = "1"
 
- if "HEXAGON_DISABLE_SPINNER" not in os_env_vars:
- os_env_vars["HEXAGON_DISABLE_SPINNER"] = "1"
-
  if "HEXAGON_SEND_TELEMETRY" not in os_env_vars:
  os_env_vars["HEXAGON_SEND_TELEMETRY"] = "0"
 

tests_e2e/status/app.yml

@@ -0,0 +1,13 @@
+cli:
+ name: Test
+ command: hexagon-test
+ custom_tools_dir: .
+
+tools:
+ - name: status
+ action: status
+ type: shell
+ alias: s
+ long_name: Show different statuses
+
+envs: []

tests_e2e/status/status.py

@@ -0,0 +1,24 @@
+import time
+
+from hexagon.support.printer import log
+from hexagon.support.prompt import prompt
+
+
+def main(*_):
+ with log.status("first status shown..."):
+ log.info("some info 1")
+ log.info("some info 2")
+ time.sleep(1)
+ if not prompt.confirm("update?", default=True):
+ return
+ log.info("about to show nested status")
+ with log.status("nested status..."):
+ log.info("inside nested status")
+ log.info("prompting...")
+ text = prompt.text(message="enter something")
+ log.info(f"entered {text}")
+ log.info("back to main status")
+
+ log.info("showing a new status...")
+ with log.status("new status..."):
+ log.info("inside new status")