See also: deferring events: details and dilemmas
Scenario allows you to accurately simulate the Operator Framework’s event queue. The event queue is responsible for keeping track of the deferred events.
Using scenario.state.DeferredEvent, you can verify that if the charm triggers with this and that event in its queue (they would be there because they had been deferred in the previous run), then the output state is valid. Also, you can verify that a certain event has been deferred by a certain charm execution.
Typical use cases will not require you to deal with the ‘raw’ DeferredEvent object: the data it holds is rather low-level and deals with specific ops implementation details. Instead, for the most part, you should be able to use the scenario.deferred helper, which helps you construct a DeferredEvent, and the scenario.state.Event.deferred method, which transforms a regular Event into a DeferredEvent data structure.
Example usage: testing charm code if deferred events are present
from scenario import State, deferred, Context
class MyCharm(...):
...
def _on_update_status(self, e):
e.defer()
def _on_start(self, e):
e.defer()
def test_start_on_deferred_update_status(MyCharm):
"""Test charm execution if a 'start' is dispatched when in the previous run an update-status had been deferred."""
state_in = State(
deferred=[
deferred('update_status',
handler=MyCharm._on_update_status)
]
)
state_out = Context(MyCharm).run('start', state_in)
assert len(state_out.deferred) == 1
assert state_out.deferred[0].name == 'start'
[technical implementation note]: the way deferred events are implemented in ops, they hold a reference to the charm method (event handler) that the event was emitted on when it was deferred. So, in order to simulate a deferred event, we need to tell to the runtime what event handler to reemit it on as soon as the Framework wakes up and processes its deferral queue. For this reason, we need to pass a mandatory handler argument to deferred.
Example usage: asserting that an event has been deferred
On the output side, you can verify that an event that you expect to have been deferred during this trigger, has indeed been deferred.
from scenario import State, Context
class MyCharm(...):
...
def _on_start(self, e):
e.defer()
def test_defer(MyCharm):
out = Context(MyCharm).run('start', State())
assert len(out.deferred) == 1
assert out.deferred[0].name == 'start'
scenario.state.Event.deferred
You can generate the DeferredEvent data structure from the corresponding Event:
from scenario import Event, Relation
class MyCharm(...):
...
deferred_start = Event('start').deferred(MyCharm._on_start)
deferred_install = Event('install').deferred(MyCharm._on_start)
This is especially handy when deferring events that have some mandatory metadata, such as relation and workload events (which need respectively a Relation and a Container to work):
foo_relation = Relation('foo')
deferred_relation_changed_evt = foo_relation.changed_event.deferred(handler=MyCharm._on_foo_relation_changed)
Deferring events owned by other Objects
The deferred helper Scenario provides will not support out of the box all custom event subclasses, or events emitted by charm libraries or objects other than the main charm class.
For general-purpose usage, you will need to instantiate scenario.state.DeferredEvent directly.
from scenario import DeferredEvent
my_deferred_event = DeferredEvent(
handle_path='MyCharm/MyCharmLib/on/database_ready[1]', # `ops.Handle` path of the event.
owner='MyCharmLib', # the unique identifier of the `ops.Object` observing the event.
observer='_on_database_ready' # (observer) method name of the owner that the event was emitted on when it was deferred
)