Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/howto/legacy/migrate-unit-tests-from-harness.md
Original file line number Diff line number Diff line change
Expand Up @@ -541,5 +541,5 @@ For more information about state-transition testing, see:

For more examples of collect-status event handlers, see:

- The [httpbin-demo charm](https://github.com/canonical/operator/blob/main/examples/httpbin-demo/src/charm.py)
- The [httpbin-demo charm](https://github.com/canonical/operator/blob/main/examples/httpbin-demo/src/charm.py) and [its unit tests](https://github.com/canonical/operator/blob/main/examples/httpbin-demo/tests/unit/test_charm.py)
- [Update the unit status to reflect the relation state](#integrate-your-charm-with-postgresql-update-unit-status) in the Kubernetes charm tutorial
8 changes: 5 additions & 3 deletions examples/httpbin-demo/src/charm.py
Original file line number Diff line number Diff line change
Expand Up @@ -67,9 +67,7 @@ def _on_collect_status(self, event: ops.CollectStatusEvent):
except ValueError as e:
event.add_status(ops.BlockedStatus(str(e)))
try:
if not self.container.get_service(SERVICE_NAME).is_running():
# We can connect to Pebble in the container, but the service hasn't started yet.
event.add_status(ops.MaintenanceStatus("waiting for workload"))
service = self.container.get_service(SERVICE_NAME)
except ops.ModelError:
# We can connect to Pebble in the container, but the service doesn't exist. This is
# most likely because we haven't added a layer yet.
Expand All @@ -82,6 +80,10 @@ def _on_collect_status(self, event: ops.CollectStatusEvent):
# It's technically possible (but unlikely) for Pebble to have an internal error.
logger.error("Unable to fetch service info from Pebble")
raise
else:
if not service.is_running():
# We can connect to Pebble in the container, but the service hasn't started yet.
event.add_status(ops.MaintenanceStatus("waiting for workload"))
event.add_status(ops.ActiveStatus())

def _on_httpbin_pebble_ready(self, event: ops.PebbleReadyEvent):
Expand Down
133 changes: 81 additions & 52 deletions examples/httpbin-demo/tests/unit/test_charm.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,20 +13,34 @@
# limitations under the License.

import ops
from ops import testing
import pytest
from ops import pebble, testing

from charm import CONTAINER_NAME, SERVICE_NAME, HttpbinDemoCharm

# A mock Pebble layer - useful for testing the charm's status reporting code. The status reporting
# code shouldn't care how the service is started, so the layer doesn't need the real command.
MOCK_LAYER = pebble.Layer(
{
"services": {
SERVICE_NAME: {
"override": "replace",
"command": "mock-command",
"startup": "enabled",
},
},
}
)

def test_httpbin_pebble_ready():

def test_pebble_ready():
"""Check that the charm correctly starts the service in the container."""
# Arrange:
ctx = testing.Context(HttpbinDemoCharm)
container = testing.Container(CONTAINER_NAME, can_connect=True)
state_in = testing.State(containers={container})

# Act:
state_out = ctx.run(ctx.on.pebble_ready(container), state_in)

# Assert:
updated_plan = state_out.get_container(container.name).plan
expected_plan = {
Expand All @@ -48,72 +62,87 @@ def test_httpbin_pebble_ready():
assert state_out.unit_status == testing.ActiveStatus()


def test_config_changed_valid_can_connect():
"""Test a config-changed event when the config is valid and the container can be reached."""
# Arrange:
ctx = testing.Context(HttpbinDemoCharm) # The default config will be read from charmcraft.yaml
container = testing.Container(CONTAINER_NAME, can_connect=True)
state_in = testing.State(
containers={container},
config={"log-level": "debug"}, # This is the config the charmer passed with `juju config`
)

# Act:
state_out = ctx.run(ctx.on.config_changed(), state_in)
def test_status_service_active():
"""Check that the charm goes into active status if the service is active.

# Assert:
updated_plan = state_out.get_container(container.name).plan
gunicorn_args = updated_plan.services[SERVICE_NAME].environment["GUNICORN_CMD_ARGS"]
assert gunicorn_args == "--log-level debug"
This test is useful in addition to ``test_pebble_ready()`` because it checks that the charm
consistently sets active status, regardless of which event was handled.
"""
ctx = testing.Context(HttpbinDemoCharm)
container = testing.Container(
CONTAINER_NAME,
layers={"base": MOCK_LAYER},
service_statuses={SERVICE_NAME: pebble.ServiceStatus.ACTIVE},
can_connect=True,
)
state_in = testing.State(containers={container})
state_out = ctx.run(ctx.on.update_status(), state_in)
assert state_out.unit_status == testing.ActiveStatus()


def test_config_changed_valid_cannot_connect():
"""Test a config-changed event when the config is valid but the container cannot be reached.

We expect to end up in MaintenanceStatus waiting for the deferred event to
be retried.
"""
# Arrange:
def test_status_service_inactive():
"""Check that the charm goes into maintenance status if the service isn't active."""
ctx = testing.Context(HttpbinDemoCharm)
container = testing.Container(CONTAINER_NAME, can_connect=False)
state_in = testing.State(containers={container}, config={"log-level": "debug"})
container = testing.Container(
CONTAINER_NAME,
layers={"base": MOCK_LAYER},
service_statuses={SERVICE_NAME: pebble.ServiceStatus.INACTIVE},
can_connect=True,
)
state_in = testing.State(containers={container})
state_out = ctx.run(ctx.on.update_status(), state_in)
assert state_out.unit_status == testing.MaintenanceStatus("waiting for workload")

# Act:
state_out = ctx.run(ctx.on.config_changed(), state_in)

# Assert:
assert isinstance(state_out.unit_status, testing.MaintenanceStatus)
def test_status_no_service():
"""Check that the charm goes into maintenance status if the service hasn't been defined."""
ctx = testing.Context(HttpbinDemoCharm)
container = testing.Container(CONTAINER_NAME, can_connect=True)
state_in = testing.State(containers={container})
state_out = ctx.run(ctx.on.update_status(), state_in)
assert state_out.unit_status == testing.MaintenanceStatus("waiting for workload container")


def test_config_changed_valid_uppercase():
"""Test a config-changed event when the config is valid and uppercase."""
# Arrange:
def test_status_no_pebble():
"""Check that the charm goes into maintenance status if the container is down."""
ctx = testing.Context(HttpbinDemoCharm)
container = testing.Container(CONTAINER_NAME, can_connect=False)
state_in = testing.State(containers={container})
state_out = ctx.run(ctx.on.update_status(), state_in)
assert state_out.unit_status == testing.MaintenanceStatus("waiting for workload container")


@pytest.mark.parametrize(
"user_log_level, gunicorn_log_level",
[
("debug", "debug"),
("DEBUG", "debug"),
],
)
def test_config_changed(user_log_level: str, gunicorn_log_level: str):
"""Test a config-changed event when the config is valid."""
ctx = testing.Context(HttpbinDemoCharm)
container = testing.Container(CONTAINER_NAME, can_connect=True)
state_in = testing.State(containers={container}, config={"log-level": "DEBUG"})

# Act:
state_in = testing.State(containers={container}, config={"log-level": user_log_level})
state_out = ctx.run(ctx.on.config_changed(), state_in)

# Assert:
updated_plan = state_out.get_container(container.name).plan
gunicorn_args = updated_plan.services[SERVICE_NAME].environment["GUNICORN_CMD_ARGS"]
assert gunicorn_args == "--log-level debug"
assert isinstance(state_out.unit_status, testing.ActiveStatus)
assert gunicorn_args == f"--log-level {gunicorn_log_level}"
assert state_out.unit_status == testing.ActiveStatus()


def test_config_changed_invalid():
@pytest.mark.parametrize(
"user_log_level",
[
"",
"foobar",
],
)
def test_config_changed_invalid(user_log_level: str):
"""Test a config-changed event when the config is invalid."""
# Arrange:
ctx = testing.Context(HttpbinDemoCharm)
container = testing.Container(CONTAINER_NAME, can_connect=True)
invalid_level = "foobar"
state_in = testing.State(containers={container}, config={"log-level": invalid_level})

# Act:
state_in = testing.State(containers={container}, config={"log-level": user_log_level})
state_out = ctx.run(ctx.on.config_changed(), state_in)

# Assert:
assert isinstance(state_out.unit_status, testing.BlockedStatus)
assert invalid_level in state_out.unit_status.message
assert f"'{user_log_level}'" in state_out.unit_status.message