jubilant¶

Jubilant is a Pythonic wrapper around the Juju CLI for writing charm integration tests.

exception jubilant.CLIError(returncode, cmd, output=None, stderr=None)¶

Bases: CalledProcessError

Subclass of CalledProcessError that includes stdout and stderr in the __str__.

class jubilant.Juju( *, model: str | None = None, wait_timeout: float = 180.0, cli_binary: str | PathLike[str] | None = None, )¶

Bases: object

Instantiate this class to run Juju commands.

Most methods directly call a single Juju CLI command. If a CLI command doesn’t yet exist as a method, use cli().

Example:

juju = jubilant.Juju()
juju.deploy('snappass-test')

Parameters:

model – If specified, operate on this Juju model, otherwise use the current Juju model.
wait_timeout – The default timeout for wait() (in seconds) if that method’s timeout parameter is not specified.
cli_binary – Path to the Juju CLI binary. If not specified, uses juju and assumes it is in the PATH.

Add a named model and set this instance’s model to it.

To avoid interfering with CLI users, this won’t switch the Juju CLI to the newly-created model. However, because model is set to the name of the new model, all subsequent operations on this instance will use the new model.

Parameters:

model – Name of model to add.
controller – Name of controller to operate in. If not specified, use the current controller.
config – Model configuration as key-value pairs, for example, {'image-stream': 'daily'}.

add_unit( app: str, *, attach_storage: str | Iterable[str] | None = None, num_units: int = 1, to: str | Iterable[str] | None = None, )¶

Add one or more units to a deployed application.

Parameters:

app – Name of application to add units to.
attach_storage – Existing storage(s) to attach to the deployed unit, for example, foo/0 or mydisk/1. Not available for Kubernetes models.
num_units – Number of units to add.
to – Machine or container to deploy the unit in (bypasses constraints). For example, to deploy to a new LXD container on machine 25, use lxd:25.

cli( *args: str, include_model: bool = True, stdin: str | None = None, ) → str¶

Run a Juju CLI command and return its standard output.

Parameters:

args – Command-line arguments (excluding juju).
include_model – If true and model is set, insert the --model argument after the first argument in args.
stdin – Standard input to send to the process, if any.

cli_binary: str¶: Path to the Juju CLI binary. If None, uses juju and assumes it is in the PATH.

config( app: str, *, app_config: bool = False, ) → Mapping[str, bool | int | float | str | SecretURI]¶

Get or set the configuration of a deployed application.

If called with only the app argument, get the config and return it. If called with the values argument, set the config values and return None.

Parameters:

app – Application name to get or set config for.
values – Mapping of config names to values. Reset values that are None.
app_config – When getting config, set this to True to get the (poorly-named) “application-config” values instead of charm config.

debug_log(*, limit: int = 0) → str¶

Return debug log messages from a model.

Parameters:: limit – Limit the result to the most recent limit lines. Defaults to 0, meaning return all lines in the log.

Deploy an application or bundle.

Parameters:

charm – Name of charm or bundle to deploy, or path to a local file (must start with / or .).
app – Optional application name within the model; defaults to the charm name.
attach_storage – Existing storage(s) to attach to the deployed unit, for example, foo/0 or mydisk/1. Not available for Kubernetes models.
base – The base on which to deploy, for example, ubuntu@22.04.
channel – Channel to use when deploying from Charmhub, for example, latest/edge.
config – Application configuration as key-value pairs, for example, {'name': 'My Wiki'}.
constraints – Hardware constraints for new machines, for example, {'mem': '8G'}.
force – If true, bypass checks such as supported bases.
num_units – Number of units to deploy for principal charms.
resources – Specify named resources to use for deployment, for example: {'bin': '/path/to/some/binary'}.
revision – Charmhub revision number to deploy.
storage – Constraints for named storage(s), for example, {'data': 'tmpfs,1G'}.
to – Machine or container to deploy the unit in (bypasses constraints). For example, to deploy to a new LXD container on machine 25, use lxd:25.
trust – If true, allows charm to run hooks that require access to cloud credentials.

destroy_model( model: str, *, destroy_storage: bool = False, force: bool = False, ) → None¶

Terminate all machines (or containers) and resources for a model.

If the given model is this instance’s model, also sets this instance’s model to None.

Parameters:

model – Name of model to destroy.
destroy_storage – If true, destroy all storage instances in the model.
force – If true, force model destruction and ignore any errors.

exec( *command: str, machine: int, wait: float | None = None, ) → Task¶

exec( *command: str, unit: str, wait: float | None = None, ) → Task

Run the command on the remote target specified.

You must specify either machine or unit, but not both.

Note: this method does not support running a command on multiple units at once. If you need that, let us know, and we’ll consider adding it with a new exec_multiple method or similar.

Parameters:

command – Command to run, along with its arguments.
machine – ID of machine to run the command on.
unit – Name of unit to run the command on, for example mysql/0 or mysql/leader.
wait – Maximum time to wait for command to finish; TimeoutError is raised if this is reached. Default is to wait indefinitely.

Returns:

The task created to run the command, including logs, failure message, and so on.

Raises:

ValueError – if the machine or unit doesn’t exist.
TaskError – if the command failed.
TimeoutError – if wait was specified and the wait time was reached.

integrate( app1: str, app2: str, *, via: str | Iterable[str] | None = None, ) → None¶

Integrate two applications, creating a relation between them.

The order of app1 and app2 is not significant. Each of them should be in the format <application>[:<endpoint>]. The endpoint is only required if there’s more than one possible integration between the two applications.

To integrate an application in the current model with an application in another model (cross-model), prefix app1 or app2 with <model>.. To integrate with an application on another controller, app1 or app2 must be an offer endpoint. See juju integrate --help for details.

Parameters:

app1 – One of the applications (and endpoints) to integrate.
app2 – The other of the applications (and endpoints) to integrate.
via – Inform the offering side (the remote application) of the source of traffic, to enable network ports to be opened. This is in CIDR notation, for example 192.0.2.0/24.

model: str | None¶: If not None, operate on this Juju model, otherwise use the current Juju model.

remove_application( *app: str, destroy_storage: bool = False, force: bool = False, ) → None¶

Remove applications from the model.

Parameters:

app – Name of the application or applications to remove.
destroy_storage – If True, also destroy storage attached to application units.
force – Force removal even if an application is in an error state.

remove_relation(app1: str, app2: str, *, force: bool = False) → None¶

Remove an existing relation between two applications (opposite of integrate()).

Parameters:

app1 – One of the applications (and endpoints) to integrate.
app2 – The other of the applications (and endpoints) to integrate.
force – Force removal, ignoring operational errors.

remove_unit( app_or_unit: str | Iterable[str], *, destroy_storage: bool = False, force: bool = False, num_units: int = 0, ) → None¶

Remove application units from the model.

Examples:

# Kubernetes model:
juju.remove_unit('wordpress', num_units=2)

# Machine model:
juju.remove_unit('wordpress/1')
juju.remove_unit(['wordpress/2', 'wordpress/3'])

Parameters:

app_or_unit – On machine models, this is the name of the unit or units to remove. On Kubernetes models, this is actually the application name (a single string), as individual units are not named; you must use num_units to remove more than one unit on a Kubernetes model.
destroy_storage – If True, also destroy storage attached to units.
force – Force removal even if a unit is in an error state.
num_units – Number of units to remove (Kubernetes models only).

run( unit: str, action: str, params: Mapping[str, Any] | None = None, *, wait: float | None = None, ) → Task¶

Run an action on the given unit and wait for the result.

Note: this method does not support running an action on multiple units at once. If you need that, let us know, and we’ll consider adding it with a new run_multiple method or similar.

Example:

juju = jubilant.Juju()
result = juju.run('mysql/0', 'get-password')
assert result.results['username'] == 'USER0'

Parameters:

unit – Name of unit to run the action on, for example mysql/0 or mysql/leader.
action – Name of action to run.
params – Optional named parameters to pass to the action.
wait – Maximum time to wait for action to finish; TimeoutError is raised if this is reached. Default is to wait indefinitely.

Returns:

The task created to run the action, including logs, failure message, and so on.

Raises:

ValueError – if the action or the unit doesn’t exist.
TaskError – if the action failed.
TimeoutError – if wait was specified and the wait time was reached.

status() → Status¶: Fetch the status of the current model, including its applications and units.

trust( app: str, *, remove: bool = False, scope: Literal['cluster'] | None = None, ) → None¶

Set the trust status of a deployed application.

Parameters:

app – Application name to set trust status for.
remove – Set to True to remove trust status.
scope – On Kubernetes models, this must be set to “cluster”, as the trust operation grants the charm full access to the cluster.

wait( ready: Callable[[Status], bool], *, error: Callable[[Status], bool] | None = None, delay: float = 1.0, timeout: float | None = None, successes: int = 3, ) → Status¶

Wait until ready(status) returns true.

This fetches the Juju status repeatedly (waiting delay seconds between each call), and returns the last status after the ready callable returns true for successes times in a row.

This function logs the status object after the first status call, and after subsequent calls if the status object has changed.

Example:

juju = jubilant.Juju()
juju.deploy('snappass-test')
juju.wait(
    lambda status: status.apps['snappass-test'].is_active,
    error=jubilant.any_error,
)

Parameters:

ready – Callable that takes a Status object and returns true when the wait should be considered ready. It needs to return true successes times in a row before wait returns.
error – Callable that takes a Status object and returns true when wait should raise an error (WaitError).
delay – Delay in seconds between status calls.
timeout – Overall timeout; TimeoutError is raised if this is reached. If not specified, uses the wait_timeout specified when the instance was created.
successes – Number of times ready must return true for the wait to succeed.

Raises:

TimeoutError – If the timeout is reached. A string representation of the last status, if any, is added as an exception note.
WaitError – If the error callable returns True. A string representation of the last status is added as an exception note.

wait_timeout: float¶: The default timeout for wait() (in seconds) if that method’s timeout parameter is not specified.

class jubilant.SecretURI¶

Bases: str

A string subclass that represents a secret URI (“secret:…”).

class jubilant.Status(model: ~jubilant.statustypes.ModelStatus, machines: dict[str, ~jubilant.statustypes.MachineStatus], apps: dict[str, ~jubilant.statustypes.AppStatus], app_endpoints: dict[str, ~jubilant.statustypes.RemoteAppStatus] = <factory>, offers: dict[str, ~jubilant.statustypes.OfferStatus] = <factory>, storage: ~jubilant.statustypes.CombinedStorage = <factory>, controller: ~jubilant.statustypes.ControllerStatus = <factory>)¶

Bases: object

Parsed version of the status object returned by juju status --format=json.

app_endpoints: dict[str, RemoteAppStatus]¶

apps: dict[str, AppStatus]¶

controller: ControllerStatus¶

machines: dict[str, MachineStatus]¶

model: ModelStatus¶

offers: dict[str, OfferStatus]¶

storage: CombinedStorage¶

class jubilant.Task(id: str, status: ~typing.Literal['aborted', 'cancelled', 'completed', 'error', 'failed'], results: dict[str, ~typing.Any] = <factory>, return_code: int = 0, stdout: str = '', stderr: str = '', message: str = '', log: list[str] = <factory>)¶

Bases: object

A task holds the results of Juju running an action or exec command on a single unit.

id: str¶: Task ID of the action, for use with juju show-task.

log: list[str]¶: List of messages logged by the action hook.

message: str = ''¶: Failure message, if the charm provided a message when it failed the action.

raise_on_failure()¶: If task was not successful, raise a TaskError.

results: dict[str, Any]¶

Results of the action provided by the charm.

This excludes the special “return-code”, “stdout”, and “stderr” keys inserted by Juju; those values are provided by separate attributes.

return_code: int = 0¶: Return code from executing the charm action hook.

status: Literal['aborted', 'cancelled', 'completed', 'error', 'failed']¶: Status of the action (Juju operation). Typically “completed” or “failed”.

stderr: str = ''¶: Stderr printed by the action hook.

stdout: str = ''¶: Stdout printed by the action hook.

property success: bool¶: Whether the action was successful.

exception jubilant.TaskError(task: Task)¶

Bases: Exception

Exception raised when an action or exec command fails.

task: Task¶: Associated task.

exception jubilant.WaitError¶

Bases: Exception

Raised when Juju.wait()’s error callable returns False.

jubilant.all_active( status: Status, apps: Iterable[str] | None = None, ) → bool¶

Report whether all applications or units in status are in “active” status.

Parameters:

status – The status object being tested.
apps – An optional list of application names. If provided, only these applications (and their units) are tested.

jubilant.all_blocked( status: Status, apps: Iterable[str] | None = None, ) → bool¶

Report whether all applications or units in status are in “blocked” status.

Parameters:

status – The status object being tested.
apps – An optional list of application names. If provided, only these applications (and their units) are tested.

jubilant.all_error( status: Status, apps: Iterable[str] | None = None, ) → bool¶

Report whether all applications or units in status are in “error” status.

Parameters:

status – The status object being tested.
apps – An optional list of application names. If provided, only these applications (and their units) are tested.

jubilant.all_maintenance( status: Status, apps: Iterable[str] | None = None, ) → bool¶

Report whether all applications or units in status are in “maintenance” status.

Parameters:

status – The status object being tested.
apps – An optional list of application names. If provided, only these applications (and their units) are tested.

jubilant.all_waiting( status: Status, apps: Iterable[str] | None = None, ) → bool¶

Report whether all applications or units in status are in “waiting” status.

Parameters:

status – The status object being tested.
apps – An optional list of application names. If provided, only these applications (and their units) are tested.

jubilant.any_active( status: Status, apps: Iterable[str] | None = None, ) → bool¶

Report whether any application or unit in status is in “active” status.

Parameters:

status – The status object being tested.
apps – An optional list of application names. If provided, only these applications (and their units) are tested.

jubilant.any_blocked( status: Status, apps: Iterable[str] | None = None, ) → bool¶

Report whether any application or unit in status is in “blocked” status.

Parameters:

status – The status object being tested.
apps – An optional list of application names. If provided, only these applications (and their units) are tested.

jubilant.any_error( status: Status, apps: Iterable[str] | None = None, ) → bool¶

Report whether any application or unit in status is in “error” status.

Parameters:

status – The status object being tested.
apps – An optional list of application names. If provided, only these applications (and their units) are tested.

jubilant.any_maintenance( status: Status, apps: Iterable[str] | None = None, ) → bool¶

Report whether any application or unit in status is in “maintenance” status.

Parameters:

status – The status object being tested.
apps – An optional list of application names. If provided, only these applications (and their units) are tested.

jubilant.any_waiting( status: Status, apps: Iterable[str] | None = None, ) → bool¶

Report whether any application or unit in status is in “waiting” status.

Parameters:

status – The status object being tested.
apps – An optional list of application names. If provided, only these applications (and their units) are tested.

jubilant.temp_model( keep: bool = False, ) → Generator[Juju, None, None]¶

Context manager to create a temporary model for running tests in.

This creates a new model with a random name in the format jubilant-abcd1234, and destroys it and its storage when the context manager exits.

Provides a Juju instance to operate on.

Parameters:: keep – If true, keep the created model around when the context manager exits.