|
class TestSecretCharm(ops.CharmBase): |
|
_stored = ops.StoredState() |
|
|
|
def __init__(self, framework: ops.Framework): |
|
super().__init__(framework) |
|
self._stored.setdefault(secret_id=None) |
|
framework.observe(self.on['eval'].action, self._on_eval) |
|
|
|
def _on_eval(self, event: ops.ActionEvent): |
|
"""Action to evaluate arbitrary Python code in specific context.""" |
|
assert event.params['code'] |
|
rv = {} |
|
|
|
rv['_before'] = self.snapshot() |
|
|
|
eval(event.patams['code'], globals(), locals()) # noqa |
|
|
|
rv['_after'] = self._snapshot() |
|
|
|
event.set_results({'rv': json.dumps(rv)}) |
|
|
|
def _snapshot(self): |
|
secret_id = self._stored_state.secret_id |
|
if not secret_id: |
|
return None |
|
secret = self.model.get_secret(id=secret_id) |
|
return {"meta": secret.get_info(), |
|
"current": secret.get_content(), |
|
"newest": secret.peek_content()} |
|
|
|
|
|
def create_secret(self: ops.CharmBase): |
|
self._stored.secret_id = self.app.add_secret({'foo': 'bar'}).id |
|
|
|
def set_label1(self: ops.CharmBase): |
|
assert self._stored.secret_id |
|
self.model.get_secret(id=self._storted.secret_id, label="lalala") |
|
|
|
def set_label1_assertions(results: dict): |
|
assert results['meta'] == {'revision': 0, labe='lalala', ...} |
|
|
|
def set_label2(self: ops.CharmBase): |
|
assert self._stored.secret_id |
|
self.model.get_secret(id=self._storted.secret_id).set_info(label="lalala") |
|
|
|
def set_label2_assertions(results: dict): |
|
assert results['meta'] == {'revision': 0, labe='lalala', ...} |
|
|
|
def set_content_and_description(...): |
|
secret = ... |
|
secret.set_content({"c42": "42"}) |
|
secret.set_info(description="big secret") |
|
|
|
def set_content_and_description_assertions(results: dict): |
|
... |
|
|
|
def cleanup(...): |
|
assert self._stored.secret_id |
|
self.model.get_secret(id=self._stored.secret_id).remove_all_revisions() |
|
self._stored.secret_id = None |
|
|
|
|
|
@pytest.mark.parametrise("test_case, assertions", |
|
[ |
|
(set_label1, set_label1_assertions), |
|
(set_label2, set_label2_assertions), |
|
(set_content_and_description, set_content_and_description_assertions), |
|
]) |
|
def test_ops_testing_secret(secret_charm: type[ops.CharmBase], secret_charm_meta: dict[str, Any], test_case=..., assertions=...) |
|
ctx = ops.testing.Context(secret_charm, meta=secret_charm_meta) |
|
setup = {'code': pyfunc_to_python_source(create_secret)} |
|
params = {'code': pyfunc_to_python_source(test_case)} |
|
cleanup = {'code': pyfunc_to_python_source(cleanup)} |
|
|
|
state = ops.testing.State(leader=True) |
|
state = ctx.run(ctx.on.action('eval', params=setup), state) |
|
state = ctx.run(ctx.on.action('eval', params=params), state) |
|
|
|
assertions(ctx.action_results) |
|
|
|
# optional |
|
ctx.run(ctx.on.action('eval', params=cleanup), state) |
|
|
|
@pytest.mark.parametrise("test_case, assertions", |
|
[ |
|
(set_label1, set_label1_assertions), |
|
(set_label2, set_label2_assertions), |
|
(set_content_and_description, set_content_and_description_assertions), |
|
]) |
|
def test_juju_secret(charm_path: Path, juju: jubilant.Juju, test_case=..., assertions=...) |
|
juju.deploy(charm_path) |
|
status = juju.wait(jubulant.all_active) |
|
unit, unit_obj = next(iter(status.apps['test-secret'].units.items())) |
|
assert unit_obj.leader |
|
|
|
setup = {'code': pyfunc_to_python_source(create_secret)} |
|
params = {'code': pyfunc_to_python_source(test_case)} |
|
cleanup = {'code': pyfunc_to_python_source(cleanup)} |
|
|
|
juju.run(unit, 'eval', {'code': create_secret}) |
|
rv = juju.run(unit, 'eval', {'code': params}) |
|
assertions(rv.results) |
|
|
|
# Kinda necessary if the deployed app is reused between test cases |
|
juju.run(unit, 'eval', {'code': cleanup}) |