When testing a Kubernetes charm, you can mock container interactions. When using the null state (State()
), there will be no containers. So if the charm were to self.unit.containers
, it would get back an empty dict.
To give the charm access to some containers, you need to pass them to the input state, like so:
State(containers=[...])
Example usage: container connectivity
from scenario.state import Container, State
state = State(containers=[
Container(name="foo", can_connect=True),
Container(name="bar", can_connect=False)
])
In this case, self.unit.get_container('foo').can_connect()
would return True
, while for ‘bar’ it would give False
.
Filesystem mocks
Typically, a container will have some files that the charm will want to read or write. This is how you configure a container to give it access to filesystem mounts:
from pathlib import Path
from scenario.state import Container, State, Mount
local_file = Path('/path/to/local/real/file.txt')
state = State(containers=[
Container(name="foo",
can_connect=True,
mounts={'local': Mount('/local/share/config.yaml', local_file)})
])
Typically you’d use a temporary directory for that.
In this case, if the charm were to:
def _on_start(self, _):
foo = self.unit.get_container('foo')
content = foo.pull('/local/share/config.yaml').read()
then content
would be the contents of our locally-supplied file.txt
. You can use tempdir
for nicely wrapping
strings and passing them to the charm via the container.
container.push
works similarly, so you can write a test like:
import tempfile
from ops.charm import CharmBase
from scenario import State, Container, Mount, Context
class MyCharm(CharmBase):
def __init__(self, *args):
super().__init__(*args)
self.framework.observe(self.on.foo_pebble_ready, self._on_pebble_ready)
def _on_pebble_ready(self, _):
foo = self.unit.get_container('foo')
foo.push('/local/share/config.yaml', "TEST", make_dirs=True)
def test_pebble_push():
with tempfile.NamedTemporaryFile() as local_file:
container = Container(name='foo',
can_connect=True,
mounts={'local': Mount('/local/share/config.yaml', local_file.name)})
state_in = State(
containers=[container]
)
Context(
MyCharm,
meta={"name": "foo", "containers": {"foo": {}}}).run(
"start",
state_in,
)
assert local_file.read().decode() == "TEST"
pebble exec
mocks
Mocking pebble exec
means that you need to specify, for each possible command the charm might run on the container, what the result of that would be: its return code and what will be written to stdout/stderr.
from ops.charm import CharmBase
from scenario import State, Container, ExecOutput, Context
LS_LL = """
.rw-rw-r-- 228 ubuntu ubuntu 18 jan 12:05 -- foo.yaml
.rw-rw-r-- 497 ubuntu ubuntu 18 jan 12:05 -- bar.yaml
drwxrwxr-x - ubuntu ubuntu 18 jan 12:06 -- baz
"""
class MyCharm(CharmBase):
def _on_start(self, _):
foo = self.unit.get_container('foo')
proc = foo.exec(['ls', '-ll'])
stdout, _ = proc.wait_output()
assert stdout == LS_LL
def test_pebble_exec():
container = Container(
name='foo',
exec_mock={
('ls', '-ll'): # this is the command we're mocking
ExecOutput(return_code=0, # this data structure contains all we need to mock the call.
stdout=LS_LL)
}
)
state_in = State(
containers=[container]
)
state_out = Context(
MyCharm,
meta={"name": "foo", "containers": {"foo": {}}},
).run(
container.pebble_ready_event,
state_in,
)