feat: rename wait_status to wait and adjust the API (#2)

Author: benhoyt

.gitignore

@@ -1,4 +1,5 @@
 __pycache__
 jubilant.egg-info
 /dist
+/SCRATCH.md
 /docs/_build

README.md

@@ -3,3 +3,11 @@
 Jubilant is a Python library that wraps the [Juju](https://juju.is/) CLI for use in charm integration tests. It provides methods that map 1:1 to Juju CLI commands, but with a type-annotated, Pythonic interface.
 
 **NOTE:** Jubilant is in very early stages of development. This is pre-alpha code. Our intention is to release a 1.0.0 version early to mid 2025.
+
+
+## Design goals
+
+- Familiar to users of the Juju CLI. We try to ensure methods, argument names, and response field names match the Juju CLI and its responses, with minor exceptions (such as "application" being shortened to "app" in `Status` fields).
+- Simple API. Any cleverness in helpers and fixtures.
+- TODO: versioning (Juju 3 and 4, maybe jubilant.compat for Juju 5 changes)
+- TODO: no async/await

docs/conf.py

@@ -6,6 +6,7 @@
 project = 'jubilant'
 extensions = [
     'sphinx.ext.autodoc',
+    'sphinx.ext.napoleon',
 ]
 templates_path = ['_templates']
 html_theme = 'sphinx_rtd_theme'

jubilant/__init__.py

@@ -1,13 +1,36 @@
 """Jubilant is a Pythonic wrapper around the Juju CLI for integration testing."""
 
-from ._errors import RunError
+from ._errors import CLIError, WaitError
+from ._helpers import (
+    all_active,
+    all_blocked,
+    all_error,
+    all_maintenance,
+    all_waiting,
+    any_active,
+    any_blocked,
+    any_error,
+    any_maintenance,
+    any_waiting,
+)
 from ._juju import Juju
 from ._types import Status
 
 __all__ = [
+    'CLIError',
     'Juju',
-    'RunError',
     'Status',
+    'WaitError',
+    'all_active',
+    'all_blocked',
+    'all_error',
+    'all_maintenance',
+    'all_waiting',
+    'any_active',
+    'any_blocked',
+    'any_error',
+    'any_maintenance',
+    'any_waiting',
 ]
 
 __version__ = '0.0.0a1'

jubilant/_errors.py

@@ -1,8 +1,8 @@
 import subprocess
 
 
-class RunError(subprocess.CalledProcessError):
-    """Subclass of CalledProcessError that includes stdout/stderr in the string."""
+class CLIError(subprocess.CalledProcessError):
+    """Subclass of CalledProcessError that includes stdout and stderr in the __str__."""
 
     def __str__(self):
         s = super().__str__()
@@ -11,3 +11,9 @@ def __str__(self):
         if self.stderr:
             s += '\nStderr:\n' + self.stderr
         return s
+
+
+class WaitError(Exception):
+    """Raised when :meth:`Juju.wait` returns false."""
+
+    # TODO: want a "status" or other attributes?

jubilant/_helpers.py

@@ -0,0 +1,133 @@
+from collections.abc import Iterable
+
+from ._types import Status
+
+# TODO: think further about how these should handle subordinate applications;
+#       they should probably automatically include subordinate status, but per
+#       Dima that might come from a different place in the status object, so not
+#       be handled automatically with the code below. See also:
+#       https://github.com/juju/python-libjuju/blob/138d8f9058e1023e01a01d587b81949433da61ce/tests/unit/data/subordinate-fullstatus.json
+
+
+def all_active(status: Status, apps: Iterable[str] | None = None) -> bool:
+    """Report whether all applications or units in *status* are in "active" status.
+
+    Args:
+        status: The :class:`Status` object being tested.
+        apps: An optional list of application names. If provided, only these applications
+            (and their units) are tested.
+    """
+    return _all_statuses_are('active', status, apps)
+
+
+def all_error(status: Status, apps: Iterable[str] | None = None) -> bool:
+    """Report whether all applications or units in *status* are in "error" status.
+
+    TODO: Args
+    """
+    return _all_statuses_are('error', status, apps)
+
+
+def all_blocked(status: Status, apps: Iterable[str] | None = None) -> bool:
+    """Report whether all applications or units in *status* are in "blocked" status.
+
+    TODO: Args
+    """
+    return _all_statuses_are('blocked', status, apps)
+
+
+def all_maintenance(status: Status, apps: Iterable[str] | None = None) -> bool:
+    """Report whether all applications or units in *status* are in "maintenance" status.
+
+    TODO: Args
+    """
+    return _all_statuses_are('maintenance', status, apps)
+
+
+def all_waiting(status: Status, apps: Iterable[str] | None = None) -> bool:
+    """Report whether all applications or units in *status* are in "waiting" status.
+
+    TODO: Args
+    """
+    return _all_statuses_are('waiting', status, apps)
+
+
+def any_active(status: Status, apps: Iterable[str] | None = None) -> bool:
+    """Report whether any application or unit in *status* is in "active" status.
+
+    Args:
+        status: The :class:`Status` object being tested.
+        apps: An optional list of application names. If provided, only these applications
+            (and their units) are tested.
+    """
+    return _any_status_is('active', status, apps)
+
+
+def any_error(status: Status, apps: Iterable[str] | None = None) -> bool:
+    """Report whether any application or unit in *status* is in "error" status.
+
+    TODO: Args
+    """
+    return _any_status_is('error', status, apps)
+
+
+def any_blocked(status: Status, apps: Iterable[str] | None = None) -> bool:
+    """Report whether any application or unit in *status* is in "blocked" status.
+
+    TODO: Args
+    """
+    return _any_status_is('blocked', status, apps)
+
+
+def any_maintenance(status: Status, apps: Iterable[str] | None = None) -> bool:
+    """Report whether any application or unit in *status* is in "maintenance" status.
+
+    TODO: Args
+    """
+    return _any_status_is('maintenance', status, apps)
+
+
+def any_waiting(status: Status, apps: Iterable[str] | None = None) -> bool:
+    """Report whether any application or unit in *status* is in "waiting" status.
+
+    TODO: Args
+    """
+    return _any_status_is('waiting', status, apps)
+
+
+def _all_statuses_are(expected: str, status: Status, apps: Iterable[str] | None) -> bool:
+    if isinstance(apps, str | bytes):
+        raise TypeError('"apps" must be an iterable of names (like a list), not a string')
+
+    if apps is None:
+        apps = status.apps.keys()
+
+    for app in apps:
+        app_info = status.apps.get(app)
+        if app_info is None:
+            return False
+        if app_info.app_status.current != expected:
+            return False
+        for unit_info in app_info.units.values():
+            if unit_info.workload_status.current != expected:
+                return False
+    return True
+
+
+def _any_status_is(expected: str, status: Status, apps: Iterable[str] | None) -> bool:
+    if isinstance(apps, str | bytes):
+        raise TypeError('"apps" must be an iterable of names (like a list), not a string')
+
+    if apps is None:
+        apps = status.apps.keys()
+
+    for app in apps:
+        app_info = status.apps.get(app)
+        if app_info is None:
+            continue
+        if app_info.app_status.current == expected:
+            return True
+        for unit_info in app_info.units.values():
+            if unit_info.workload_status.current == expected:
+                return True
+    return False