diff --git a/pyproject.toml b/pyproject.toml index 131ad6492..99a02e2a4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -254,8 +254,31 @@ convention = "google" builtins-ignorelist = ["id", "min", "map", "range", "type", "TimeoutError", "ConnectionError", "Warning", "input", "format"] [tool.pyright] -include = ["ops/*.py", "ops/_private/*.py", "test/*.py", "test/charms/*/src/*.py", "testing/src/*.py"] -exclude = ["tracing/*"] +include = ["ops/*.py", "ops/_private/*.py", "test/*.py", "test/charms/*/src/*.py", "testing/src/*.py", "testing/tests/*.py", "testing/tests/test_e2e/*.py"] +exclude = [ + "tracing/*", + "testing/tests/helpers.py", + "testing/tests/test_charm_spec_autoload.py", + "testing/tests/test_consistency_checker.py", + "testing/tests/test_context_on.py", + "testing/tests/test_context.py", + "testing/tests/test_emitted_events_util.py", + "testing/tests/test_plugin.py", + "testing/tests/test_runtime.py", + "testing/tests/test_e2e/test_network.py", + "testing/tests/test_e2e/test_cloud_spec.py", + "testing/tests/test_e2e/test_play_assertions.py", + "testing/tests/test_e2e/test_vroot.py", + "testing/tests/test_e2e/__init__.py", + "testing/tests/test_e2e/test_resource.py", + "testing/tests/test_e2e/test_state.py", + "testing/tests/test_e2e/test_actions.py", + "testing/tests/test_e2e/test_config.py", + "testing/tests/test_e2e/test_event.py", + "testing/tests/test_e2e/test_deferred.py", + "testing/tests/test_e2e/test_stored_state.py", + "testing/tests/test_e2e/conftest.py", +] extraPaths = ["testing", "tracing"] pythonVersion = "3.10" # check no python > 3.10 features are used pythonPlatform = "All" diff --git a/testing/src/scenario/state.py b/testing/src/scenario/state.py index 06b2b57c6..014335c22 100644 --- a/testing/src/scenario/state.py +++ b/testing/src/scenario/state.py @@ -1139,7 +1139,7 @@ def services(self) -> dict[str, pebble.ServiceInfo]: infos[name] = info return infos - def get_filesystem(self, ctx: Context) -> pathlib.Path: + def get_filesystem(self, ctx: Context[Any]) -> pathlib.Path: """Simulated Pebble filesystem in this context. Returns: @@ -1490,7 +1490,7 @@ def __eq__(self, other: object) -> bool: return (self.name, self.index) == (other.name, other.index) return False - def get_filesystem(self, ctx: Context) -> pathlib.Path: + def get_filesystem(self, ctx: Context[Any]) -> pathlib.Path: """Simulated filesystem root in this context.""" return ctx._get_storage_root(self.name, self.index) diff --git a/testing/tests/test_e2e/test_juju_log.py b/testing/tests/test_e2e/test_juju_log.py index 2e18446f0..cdc90222b 100644 --- a/testing/tests/test_e2e/test_juju_log.py +++ b/testing/tests/test_e2e/test_juju_log.py @@ -4,39 +4,29 @@ from __future__ import annotations import logging -from collections.abc import Mapping -from typing import Any -import pytest -from scenario import Context -from scenario.state import JujuLogLine, State +from scenario import Context, JujuLogLine, State -from ops.charm import CharmBase, CollectStatusEvent +import ops logger = logging.getLogger('testing logger') -@pytest.fixture(scope='function') -def mycharm(): - class MyCharm(CharmBase): - META: Mapping[str, Any] = {'name': 'mycharm'} +class Charm(ops.CharmBase): + def __init__(self, framework: ops.Framework): + super().__init__(framework) + for evt in self.on.events().values(): + framework.observe(evt, self._on_event) - def __init__(self, framework): - super().__init__(framework) - for evt in self.on.events().values(): - self.framework.observe(evt, self._on_event) + def _on_event(self, event: ops.EventBase): + if isinstance(event, ops.CollectStatusEvent): + return + print('foo!') + logger.warning('bar!') - def _on_event(self, event): - if isinstance(event, CollectStatusEvent): - return - print('foo!') - logger.warning('bar!') - return MyCharm - - -def test_juju_log(mycharm): - ctx = Context(mycharm, meta=mycharm.META) +def test_juju_log(): + ctx = Context(Charm, meta={'name': 'foo'}) ctx.run(ctx.on.start(), State()) assert JujuLogLine(level='DEBUG', message='Emitting Juju event start.') in ctx.juju_log assert JujuLogLine(level='WARNING', message='bar!') in ctx.juju_log diff --git a/testing/tests/test_e2e/test_manager.py b/testing/tests/test_e2e/test_manager.py index 816d28acb..ce62a27df 100644 --- a/testing/tests/test_e2e/test_manager.py +++ b/testing/tests/test_e2e/test_manager.py @@ -3,74 +3,62 @@ from __future__ import annotations -from collections.abc import Mapping -from typing import Any - import pytest -from scenario import Context, State -from scenario.context import AlreadyEmittedError, Manager - -from ops import ActiveStatus -from ops.charm import CharmBase, CollectStatusEvent +from scenario import Context, Manager, State +from scenario.errors import AlreadyEmittedError +import ops -@pytest.fixture(scope='function') -def mycharm(): - class MyCharm(CharmBase): - META: Mapping[str, Any] = {'name': 'mycharm'} - ACTIONS: Mapping[str, Any] = {'do-x': {}} - def __init__(self, framework): - super().__init__(framework) - for evt in self.on.events().values(): - self.framework.observe(evt, self._on_event) +class Charm(ops.CharmBase): + def __init__(self, framework: ops.Framework): + super().__init__(framework) + for evt in self.on.events().values(): + framework.observe(evt, self._on_event) - def _on_event(self, e): - if isinstance(e, CollectStatusEvent): - return + def _on_event(self, e: ops.EventBase): + if isinstance(e, ops.CollectStatusEvent): + return + self.unit.status = ops.ActiveStatus(e.handle.kind) - self.unit.status = ActiveStatus(e.handle.kind) - return MyCharm +@pytest.fixture +def ctx() -> Context[Charm]: + return Context(Charm, meta={'name': 'foo'}, actions={'do-x': {}}) -def test_manager(mycharm): - ctx = Context(mycharm, meta=mycharm.META) +def test_manager(ctx: Context[Charm]): with Manager(ctx, ctx.on.start(), State()) as manager: - assert isinstance(manager.charm, mycharm) + assert isinstance(manager.charm, Charm) state_out = manager.run() assert isinstance(state_out, State) -def test_manager_implicit(mycharm): - ctx = Context(mycharm, meta=mycharm.META) +def test_manager_implicit(ctx: Context[Charm]): with Manager(ctx, ctx.on.start(), State()) as manager: - assert isinstance(manager.charm, mycharm) + assert isinstance(manager.charm, Charm) # do not call .run() # run is called automatically assert manager._emitted -def test_manager_reemit_fails(mycharm): - ctx = Context(mycharm, meta=mycharm.META) +def test_manager_reemit_fails(ctx: Context[Charm]): with Manager(ctx, ctx.on.start(), State()) as manager: manager.run() with pytest.raises(AlreadyEmittedError): manager.run() -def test_context_manager(mycharm): - ctx = Context(mycharm, meta=mycharm.META) +def test_context_manager(ctx: Context[Charm]): with ctx(ctx.on.start(), State()) as manager: state_out = manager.run() assert isinstance(state_out, State) assert ctx.emitted_events[0].handle.kind == 'start' -def test_context_action_manager(mycharm): - ctx = Context(mycharm, meta=mycharm.META, actions=mycharm.ACTIONS) +def test_context_action_manager(ctx: Context[Charm]): with ctx(ctx.on.action('do-x'), State()) as manager: state_out = manager.run() assert isinstance(state_out, State) diff --git a/testing/tests/test_e2e/test_pebble.py b/testing/tests/test_e2e/test_pebble.py index b3bb52243..e20f5108f 100644 --- a/testing/tests/test_e2e/test_pebble.py +++ b/testing/tests/test_e2e/test_pebble.py @@ -6,56 +6,53 @@ import dataclasses import datetime import io -from pathlib import Path +import pathlib +from typing import TYPE_CHECKING import pytest from scenario import Context from scenario.state import CheckInfo, Container, Exec, Mount, Notice, State -from ops import PebbleCustomNoticeEvent, PebbleReadyEvent, pebble -from ops.charm import CharmBase -from ops.framework import Framework +import ops from ops.log import _get_juju_log_and_app_id -from ops.pebble import ExecError, Layer, ServiceStartup, ServiceStatus -from ..helpers import jsonpatch_delta, trigger +from ..helpers import jsonpatch_delta, trigger # type: ignore +if TYPE_CHECKING: + from ops.pebble import LayerDict, ServiceDict -@pytest.fixture(scope='function') -def charm_cls(): - class MyCharm(CharmBase): - def __init__(self, framework: Framework): - super().__init__(framework) - for evt in self.on.events().values(): - self.framework.observe(evt, self._on_event) - def _on_event(self, event): - pass +class Charm(ops.CharmBase): + def __init__(self, framework: ops.Framework): + super().__init__(framework) + for evt in self.on.events().values(): + framework.observe(evt, self._on_event) - return MyCharm + def _on_event(self, event: ops.EventBase): + pass -def test_no_containers(charm_cls): - def callback(self: CharmBase): +def test_no_containers(): + def callback(self: ops.CharmBase): assert not self.unit.containers trigger( State(), - charm_type=charm_cls, + charm_type=Charm, meta={'name': 'foo'}, event='start', post_event=callback, ) -def test_containers_from_meta(charm_cls): - def callback(self: CharmBase): +def test_containers_from_meta(): + def callback(self: ops.CharmBase): assert self.unit.containers assert self.unit.get_container('foo') trigger( State(), - charm_type=charm_cls, + charm_type=Charm, meta={'name': 'foo', 'containers': {'foo': {}}}, event='start', post_event=callback, @@ -63,26 +60,26 @@ def callback(self: CharmBase): @pytest.mark.parametrize('can_connect', (True, False)) -def test_connectivity(charm_cls, can_connect): - def callback(self: CharmBase): +def test_connectivity(can_connect: bool): + def callback(self: ops.CharmBase): assert can_connect == self.unit.get_container('foo').can_connect() trigger( State(containers={Container(name='foo', can_connect=can_connect)}), - charm_type=charm_cls, + charm_type=Charm, meta={'name': 'foo', 'containers': {'foo': {}}}, event='start', post_event=callback, ) -def test_fs_push(tmp_path, charm_cls): +def test_fs_push(tmp_path: pathlib.Path): text = 'lorem ipsum/n alles amat gloriae foo' pth = tmp_path / 'textfile' pth.write_text(text) - def callback(self: CharmBase): + def callback(self: ops.CharmBase): container = self.unit.get_container('foo') baz = container.pull('/bar/baz.txt') assert baz.read() == text @@ -97,7 +94,7 @@ def callback(self: CharmBase): ) } ), - charm_type=charm_cls, + charm_type=Charm, meta={'name': 'foo', 'containers': {'foo': {}}}, event='start', post_event=callback, @@ -105,10 +102,10 @@ def callback(self: CharmBase): @pytest.mark.parametrize('make_dirs', (True, False)) -def test_fs_pull(tmp_path, charm_cls, make_dirs): +def test_fs_pull(tmp_path: pathlib.Path, make_dirs: bool): text = 'lorem ipsum/n alles amat gloriae foo' - def callback(self: CharmBase): + def callback(self: ops.CharmBase): container = self.unit.get_container('foo') if make_dirs: container.push('/foo/bar/baz.txt', text, make_dirs=make_dirs) @@ -116,11 +113,11 @@ def callback(self: CharmBase): baz = container.pull('/foo/bar/baz.txt') assert baz.read() == text else: - with pytest.raises(pebble.PathError): + with pytest.raises(ops.pebble.PathError): container.push('/foo/bar/baz.txt', text, make_dirs=make_dirs) # check that nothing was changed - with pytest.raises((FileNotFoundError, pebble.PathError)): + with pytest.raises((FileNotFoundError, ops.pebble.PathError)): container.pull('/foo/bar/baz.txt') container = Container( @@ -131,7 +128,7 @@ def callback(self: CharmBase): state = State(containers={container}) ctx = Context( - charm_type=charm_cls, + charm_type=Charm, meta={'name': 'foo', 'containers': {'foo': {}}}, ) with ctx(ctx.on.start(), state=state) as mgr: @@ -143,12 +140,12 @@ def callback(self: CharmBase): file = tmp_path / 'bar' / 'baz.txt' # another is: - assert file == Path(out.get_container('foo').mounts['foo'].source) / 'bar' / 'baz.txt' + base = pathlib.Path(out.get_container('foo').mounts['foo'].source) + assert file == base / 'bar' / 'baz.txt' # but that is actually a symlink to the context's root tmp folder: - assert ( - Path(ctx._tmp.name) / 'containers' / 'foo' / 'foo' / 'bar' / 'baz.txt' - ).read_text() == text + base = pathlib.Path(ctx._tmp.name) + assert (base / 'containers' / 'foo' / 'foo' / 'bar' / 'baz.txt').read_text() == text assert file.read_text() == text # shortcut for API niceness purposes: @@ -189,11 +186,12 @@ def callback(self: CharmBase): ('ps', PS), ), ) -def test_exec(charm_cls, cmd, out): - def callback(self: CharmBase): +def test_exec(cmd: str, out: str): + def callback(self: ops.CharmBase): container = self.unit.get_container('foo') proc = container.exec([cmd]) proc.wait() + assert proc.stdout is not None assert proc.stdout.read() == out trigger( @@ -206,13 +204,29 @@ def callback(self: CharmBase): ) } ), - charm_type=charm_cls, + charm_type=Charm, meta={'name': 'foo', 'containers': {'foo': {}}}, event='start', post_event=callback, ) +class ExecCharm(ops.CharmBase): + stdin: str | io.StringIO | None + write: str | None + + def __init__(self, framework: ops.Framework): + super().__init__(framework) + self.framework.observe(self.on.foo_pebble_ready, self._on_ready) + + def _on_ready(self, _: ops.EventBase): + proc = self.unit.get_container('foo').exec(['ls'], stdin=self.stdin) + if self.write: + assert proc.stdin is not None + proc.stdin.write(self.write) + proc.wait() + + @pytest.mark.parametrize( 'stdin,write', ( @@ -221,26 +235,19 @@ def callback(self: CharmBase): [io.StringIO('hello world!'), None], ), ) -def test_exec_history_stdin(stdin, write): - class MyCharm(CharmBase): - def __init__(self, framework: Framework): - super().__init__(framework) - self.framework.observe(self.on.foo_pebble_ready, self._on_ready) - - def _on_ready(self, _): - proc = self.unit.get_container('foo').exec(['ls'], stdin=stdin) - if write: - proc.stdin.write(write) - proc.wait() - - ctx = Context(MyCharm, meta={'name': 'foo', 'containers': {'foo': {}}}) +def test_exec_history_stdin( + monkeypatch: pytest.MonkeyPatch, stdin: str | io.StringIO | None, write: str | None +): + monkeypatch.setattr(ExecCharm, 'stdin', stdin, raising=False) + monkeypatch.setattr(ExecCharm, 'write', write, raising=False) + ctx = Context(ExecCharm, meta={'name': 'foo', 'containers': {'foo': {}}}) container = Container(name='foo', can_connect=True, execs={Exec([])}) ctx.run(ctx.on.pebble_ready(container=container), State(containers={container})) assert ctx.exec_history[container.name][0].stdin == 'hello world!' -def test_pebble_ready(charm_cls): - def callback(self: CharmBase): +def test_pebble_ready(): + def callback(self: ops.CharmBase): foo = self.unit.get_container('foo') assert foo.can_connect() @@ -248,62 +255,71 @@ def callback(self: CharmBase): trigger( State(containers={container}), - charm_type=charm_cls, + charm_type=Charm, meta={'name': 'foo', 'containers': {'foo': {}}}, event='pebble_ready', post_event=callback, ) -@pytest.mark.parametrize('starting_service_status', pebble.ServiceStatus) -def test_pebble_plan(charm_cls, starting_service_status): - class PlanCharm(charm_cls): - def __init__(self, framework): - super().__init__(framework) - framework.observe(self.on.foo_pebble_ready, self._on_ready) - - def _on_ready(self, event): - foo = event.workload - - assert foo.get_plan().to_dict() == {'services': {'fooserv': {'startup': 'enabled'}}} - fooserv = foo.get_services('fooserv')['fooserv'] - assert fooserv.startup == ServiceStartup.ENABLED - assert fooserv.current == ServiceStatus.ACTIVE - - foo.add_layer( - 'bar', - { - 'summary': 'bla', - 'description': 'deadbeef', - 'services': {'barserv': {'startup': 'disabled'}}, - }, - ) +class PlanCharm(ops.CharmBase): + starting_service_status: ops.pebble.ServiceStatus - foo.replan() - assert foo.get_plan().to_dict() == { - 'services': { - 'barserv': {'startup': 'disabled'}, - 'fooserv': {'startup': 'enabled'}, - } + def __init__(self, framework: ops.Framework): + super().__init__(framework) + framework.observe(self.on.foo_pebble_ready, self._on_ready) + + def _on_ready(self, event: ops.PebbleReadyEvent): + foo = event.workload + + assert foo.get_plan().to_dict() == {'services': {'fooserv': {'startup': 'enabled'}}} + fooserv = foo.get_services('fooserv')['fooserv'] + assert fooserv.startup == ops.pebble.ServiceStartup.ENABLED + assert fooserv.current == ops.pebble.ServiceStatus.ACTIVE + + foo.add_layer( + 'bar', + { + 'summary': 'bla', + 'description': 'deadbeef', + 'services': {'barserv': {'startup': 'disabled'}}, + }, + ) + + foo.replan() + assert foo.get_plan().to_dict() == { + 'services': { + 'barserv': {'startup': 'disabled'}, + 'fooserv': {'startup': 'enabled'}, } + } + + assert foo.get_service('barserv').current == self.starting_service_status + foo.start('barserv') + # whatever the original state, starting a service sets it to active + assert foo.get_service('barserv').current == ops.pebble.ServiceStatus.ACTIVE - assert foo.get_service('barserv').current == starting_service_status - foo.start('barserv') - # whatever the original state, starting a service sets it to active - assert foo.get_service('barserv').current == ServiceStatus.ACTIVE + +@pytest.mark.parametrize('starting_service_status', ops.pebble.ServiceStatus) +def test_pebble_plan( + monkeypatch: pytest.MonkeyPatch, starting_service_status: ops.pebble.ServiceStatus +): + monkeypatch.setattr( + PlanCharm, 'starting_service_status', starting_service_status, raising=False + ) container = Container( name='foo', can_connect=True, layers={ - 'foo': pebble.Layer({ + 'foo': ops.pebble.Layer({ 'summary': 'bla', 'description': 'deadbeef', 'services': {'fooserv': {'startup': 'enabled'}}, }) }, service_statuses={ - 'fooserv': pebble.ServiceStatus.ACTIVE, + 'fooserv': ops.pebble.ServiceStatus.ACTIVE, # todo: should we disallow setting status for services that aren't known YET? 'barserv': starting_service_status, }, @@ -316,22 +332,22 @@ def _on_ready(self, event): event='pebble_ready', ) - def serv(name, obj): - return pebble.Service(name, raw=obj) + def serv(name: str, obj: ServiceDict) -> ops.pebble.Service: + return ops.pebble.Service(name, raw=obj) container = out.get_container(container.name) assert container.plan.services == { 'barserv': serv('barserv', {'startup': 'disabled'}), 'fooserv': serv('fooserv', {'startup': 'enabled'}), } - assert container.services['fooserv'].current == pebble.ServiceStatus.ACTIVE - assert container.services['fooserv'].startup == pebble.ServiceStartup.ENABLED + assert container.services['fooserv'].current == ops.pebble.ServiceStatus.ACTIVE + assert container.services['fooserv'].startup == ops.pebble.ServiceStartup.ENABLED - assert container.services['barserv'].current == pebble.ServiceStatus.ACTIVE - assert container.services['barserv'].startup == pebble.ServiceStartup.DISABLED + assert container.services['barserv'].current == ops.pebble.ServiceStatus.ACTIVE + assert container.services['barserv'].startup == ops.pebble.ServiceStartup.DISABLED -def test_exec_wait_error(charm_cls): +def test_exec_wait_error(): state = State( containers={ Container( @@ -342,17 +358,17 @@ def test_exec_wait_error(charm_cls): } ) - ctx = Context(charm_cls, meta={'name': 'foo', 'containers': {'foo': {}}}) + ctx = Context(Charm, meta={'name': 'foo', 'containers': {'foo': {}}}) with ctx(ctx.on.start(), state) as mgr: container = mgr.charm.unit.get_container('foo') proc = container.exec(['foo']) - with pytest.raises(ExecError) as exc_info: + with pytest.raises(ops.pebble.ExecError) as exc_info: # type: ignore proc.wait_output() - assert exc_info.value.stdout == 'hello pebble' + assert exc_info.value.stdout == 'hello pebble' # type: ignore @pytest.mark.parametrize('command', (['foo'], ['foo', 'bar'], ['foo', 'bar', 'baz'])) -def test_exec_wait_output(charm_cls, command): +def test_exec_wait_output(command: list[str]): state = State( containers={ Container( @@ -363,7 +379,7 @@ def test_exec_wait_output(charm_cls, command): } ) - ctx = Context(charm_cls, meta={'name': 'foo', 'containers': {'foo': {}}}) + ctx = Context(Charm, meta={'name': 'foo', 'containers': {'foo': {}}}) with ctx(ctx.on.start(), state) as mgr: container = mgr.charm.unit.get_container('foo') proc = container.exec(command) @@ -373,7 +389,7 @@ def test_exec_wait_output(charm_cls, command): assert ctx.exec_history[container.name][0].command == command -def test_exec_wait_output_error(charm_cls): +def test_exec_wait_output_error(): state = State( containers={ Container( @@ -384,15 +400,15 @@ def test_exec_wait_output_error(charm_cls): } ) - ctx = Context(charm_cls, meta={'name': 'foo', 'containers': {'foo': {}}}) + ctx = Context(Charm, meta={'name': 'foo', 'containers': {'foo': {}}}) with ctx(ctx.on.start(), state) as mgr: container = mgr.charm.unit.get_container('foo') proc = container.exec(['foo']) - with pytest.raises(ExecError): + with pytest.raises(ops.pebble.ExecError): proc.wait_output() -def test_pebble_custom_notice(charm_cls): +def test_pebble_custom_notice(): notices = [ Notice(key='example.com/foo'), Notice(key='example.com/bar', last_data={'a': 'b'}), @@ -405,40 +421,61 @@ def test_pebble_custom_notice(charm_cls): ) state = State(containers=[container]) - ctx = Context(charm_cls, meta={'name': 'foo', 'containers': {'foo': {}}}) + ctx = Context(Charm, meta={'name': 'foo', 'containers': {'foo': {}}}) with ctx(ctx.on.pebble_custom_notice(container=container, notice=notices[-1]), state) as mgr: container = mgr.charm.unit.get_container('foo') assert container.get_notices() == [n._to_ops() for n in notices] -def test_pebble_custom_notice_in_charm(): +class CustomNoticeCharm(ops.CharmBase): + key: str + data: dict[str, str] + user_id: int + first_occurred: datetime.datetime + last_occurred: datetime.datetime + last_repeated: datetime.datetime + occurrences: int + repeat_after: datetime.timedelta + expire_after: datetime.timedelta + + def __init__(self, framework: ops.Framework): + super().__init__(framework) + framework.observe(self.on.foo_pebble_custom_notice, self._on_custom_notice) + + def _on_custom_notice(self, event: ops.PebbleCustomNoticeEvent): + notice = event.notice + assert notice.type == ops.pebble.NoticeType.CUSTOM + assert notice.key == self.key + assert notice.last_data == self.data + assert notice.user_id == self.user_id + assert notice.first_occurred == self.first_occurred + assert notice.last_occurred == self.last_occurred + assert notice.last_repeated == self.last_repeated + assert notice.occurrences == self.occurrences + assert notice.repeat_after == self.repeat_after + assert notice.expire_after == self.expire_after + + +def test_pebble_custom_notice_in_charm(monkeypatch: pytest.MonkeyPatch): key = 'example.com/test/charm' data = {'foo': 'bar'} user_id = 100 first_occurred = datetime.datetime(1979, 1, 25, 11, 0, 0) - last_occured = datetime.datetime(2006, 8, 28, 13, 28, 0) + last_occurred = datetime.datetime(2006, 8, 28, 13, 28, 0) last_repeated = datetime.datetime(2023, 9, 4, 9, 0, 0) occurrences = 42 repeat_after = datetime.timedelta(days=7) expire_after = datetime.timedelta(days=365) - class MyCharm(CharmBase): - def __init__(self, framework): - super().__init__(framework) - framework.observe(self.on.foo_pebble_custom_notice, self._on_custom_notice) - - def _on_custom_notice(self, event: PebbleCustomNoticeEvent): - notice = event.notice - assert notice.type == pebble.NoticeType.CUSTOM - assert notice.key == key - assert notice.last_data == data - assert notice.user_id == user_id - assert notice.first_occurred == first_occurred - assert notice.last_occurred == last_occured - assert notice.last_repeated == last_repeated - assert notice.occurrences == occurrences - assert notice.repeat_after == repeat_after - assert notice.expire_after == expire_after + monkeypatch.setattr(CustomNoticeCharm, 'key', key, raising=False) + monkeypatch.setattr(CustomNoticeCharm, 'data', data, raising=False) + monkeypatch.setattr(CustomNoticeCharm, 'user_id', user_id, raising=False) + monkeypatch.setattr(CustomNoticeCharm, 'first_occurred', first_occurred, raising=False) + monkeypatch.setattr(CustomNoticeCharm, 'last_occurred', last_occurred, raising=False) + monkeypatch.setattr(CustomNoticeCharm, 'last_repeated', last_repeated, raising=False) + monkeypatch.setattr(CustomNoticeCharm, 'occurrences', occurrences, raising=False) + monkeypatch.setattr(CustomNoticeCharm, 'repeat_after', repeat_after, raising=False) + monkeypatch.setattr(CustomNoticeCharm, 'expire_after', expire_after, raising=False) notices = [ Notice('example.com/test/other'), @@ -448,7 +485,7 @@ def _on_custom_notice(self, event: PebbleCustomNoticeEvent): last_data=data, user_id=user_id, first_occurred=first_occurred, - last_occurred=last_occured, + last_occurred=last_occurred, last_repeated=last_repeated, occurrences=occurrences, repeat_after=repeat_after, @@ -461,23 +498,32 @@ def _on_custom_notice(self, event: PebbleCustomNoticeEvent): notices=notices, ) state = State(containers=[container]) - ctx = Context(MyCharm, meta={'name': 'foo', 'containers': {'foo': {}}}) + ctx = Context(CustomNoticeCharm, meta={'name': 'foo', 'containers': {'foo': {}}}) ctx.run(ctx.on.pebble_custom_notice(container=container, notice=notices[-1]), state) -def test_pebble_check_failed(): - infos = [] +class CheckFailedCharm(ops.CharmBase): + infos: list[ops.LazyCheckInfo] + + def __init__(self, framework: ops.Framework): + super().__init__(framework) + framework.observe(self.on.foo_pebble_check_failed, self._on_check_failed) - class MyCharm(CharmBase): - def __init__(self, framework): - super().__init__(framework) - framework.observe(self.on.foo_pebble_check_failed, self._on_check_failed) + def _on_check_failed(self, event: ops.PebbleCheckFailedEvent): + self.infos.append(event.info) - def _on_check_failed(self, event): - infos.append(event.info) - ctx = Context(MyCharm, meta={'name': 'foo', 'containers': {'foo': {}}}) - layer = pebble.Layer({ +@pytest.fixture +def capture_info_failure_charm(monkeypatch: pytest.MonkeyPatch) -> type[CheckFailedCharm]: + monkeypatch.setattr(CheckFailedCharm, 'infos', [], raising=False) + return CheckFailedCharm + + +def test_pebble_check_failed(capture_info_failure_charm: CheckFailedCharm): + ctx: Context[CheckFailedCharm] = Context( + capture_info_failure_charm, meta={'name': 'foo', 'containers': {'foo': {}}} + ) + layer = ops.pebble.Layer({ 'checks': {'http-check': {'override': 'replace', 'startup': 'enabled', 'threshold': 3}} }) assert layer.checks['http-check'].threshold is not None @@ -485,154 +531,177 @@ def _on_check_failed(self, event): 'http-check', successes=3, failures=7, - status=pebble.CheckStatus.DOWN, - level=layer.checks['http-check'].level, - startup=layer.checks['http-check'].startup, + status=ops.pebble.CheckStatus.DOWN, + level=ops.pebble.CheckLevel(layer.checks['http-check'].level), + startup=ops.pebble.CheckStartup(layer.checks['http-check'].startup), threshold=layer.checks['http-check'].threshold, ) container = Container('foo', check_infos={check}, layers={'layer1': layer}) state = State(containers={container}) ctx.run(ctx.on.pebble_check_failed(container, check), state=state) + infos = capture_info_failure_charm.infos assert len(infos) == 1 assert infos[0].name == 'http-check' - assert infos[0].status == pebble.CheckStatus.DOWN + assert infos[0].status == ops.pebble.CheckStatus.DOWN assert infos[0].successes == 3 assert infos[0].failures == 7 -def test_pebble_check_recovered(): - infos = [] +class CheckRecoveredCharm(ops.CharmBase): + infos: list[ops.LazyCheckInfo] - class MyCharm(CharmBase): - def __init__(self, framework): - super().__init__(framework) - framework.observe(self.on.foo_pebble_check_recovered, self._on_check_recovered) + def __init__(self, framework: ops.Framework): + super().__init__(framework) + framework.observe(self.on.foo_pebble_check_recovered, self._on_check_recovered) - def _on_check_recovered(self, event): - infos.append(event.info) + def _on_check_recovered(self, event: ops.PebbleCheckRecoveredEvent): + self.infos.append(event.info) - ctx = Context(MyCharm, meta={'name': 'foo', 'containers': {'foo': {}}}) - layer = pebble.Layer({ + +@pytest.fixture +def capture_info_recovered_charm(monkeypatch: pytest.MonkeyPatch) -> type[CheckRecoveredCharm]: + monkeypatch.setattr(CheckRecoveredCharm, 'infos', [], raising=False) + return CheckRecoveredCharm + + +def test_pebble_check_recovered(capture_info_recovered_charm: CheckRecoveredCharm): + ctx = Context(capture_info_recovered_charm, meta={'name': 'foo', 'containers': {'foo': {}}}) + layer = ops.pebble.Layer({ 'checks': {'http-check': {'override': 'replace', 'startup': 'enabled', 'threshold': 3}} }) assert layer.checks['http-check'].threshold is not None check = CheckInfo( 'http-check', successes=None, - status=pebble.CheckStatus.UP, - level=layer.checks['http-check'].level, - startup=layer.checks['http-check'].startup, + status=ops.pebble.CheckStatus.UP, + level=ops.pebble.CheckLevel(layer.checks['http-check'].level), + startup=ops.pebble.CheckStartup(layer.checks['http-check'].startup), threshold=layer.checks['http-check'].threshold, ) container = Container('foo', check_infos={check}, layers={'layer1': layer}) state = State(containers={container}) ctx.run(ctx.on.pebble_check_recovered(container, check), state=state) + infos = capture_info_recovered_charm.infos assert len(infos) == 1 assert infos[0].name == 'http-check' - assert infos[0].status == pebble.CheckStatus.UP + assert infos[0].status == ops.pebble.CheckStatus.UP assert infos[0].successes is None assert infos[0].failures == 0 -def test_pebble_check_failed_two_containers(): - foo_infos = [] - bar_infos = [] +class DoubleCharm(ops.CharmBase): + foo_infos: list[ops.LazyCheckInfo] + bar_infos: list[ops.LazyCheckInfo] - class MyCharm(CharmBase): - def __init__(self, framework: Framework): - super().__init__(framework) - framework.observe(self.on.foo_pebble_check_failed, self._on_foo_check_failed) - framework.observe(self.on.bar_pebble_check_failed, self._on_bar_check_failed) + def __init__(self, framework: ops.Framework): + super().__init__(framework) + framework.observe(self.on.foo_pebble_check_failed, self._on_foo_check_failed) + framework.observe(self.on.bar_pebble_check_failed, self._on_bar_check_failed) - def _on_foo_check_failed(self, event): - foo_infos.append(event.info) + def _on_foo_check_failed(self, event: ops.PebbleCheckFailedEvent): + self.foo_infos.append(event.info) + + def _on_bar_check_failed(self, event: ops.PebbleCheckFailedEvent): + self.bar_infos.append(event.info) + + +@pytest.fixture +def capture_info_double_charm(monkeypatch: pytest.MonkeyPatch) -> type[DoubleCharm]: + monkeypatch.setattr(DoubleCharm, 'foo_infos', [], raising=False) + monkeypatch.setattr(DoubleCharm, 'bar_infos', [], raising=False) + return DoubleCharm - def _on_bar_check_failed(self, event): - bar_infos.append(event.info) - ctx = Context(MyCharm, meta={'name': 'foo', 'containers': {'foo': {}, 'bar': {}}}) +def test_pebble_check_failed_two_containers(capture_info_recovered_charm: DoubleCharm): + ctx = Context( + capture_info_recovered_charm, meta={'name': 'foo', 'containers': {'foo': {}, 'bar': {}}} + ) - layer = pebble.Layer({ + layer = ops.pebble.Layer({ 'checks': {'http-check': {'override': 'replace', 'startup': 'enabled', 'threshold': 3}} }) assert layer.checks['http-check'].threshold is not None check = CheckInfo( 'http-check', failures=7, - status=pebble.CheckStatus.DOWN, - level=layer.checks['http-check'].level, - startup=layer.checks['http-check'].startup, + status=ops.pebble.CheckStatus.DOWN, + level=ops.pebble.CheckLevel(layer.checks['http-check'].level), + startup=ops.pebble.CheckStartup(layer.checks['http-check'].startup), threshold=layer.checks['http-check'].threshold, ) foo_container = Container('foo', check_infos={check}, layers={'layer1': layer}) bar_container = Container('bar', check_infos={check}, layers={'layer1': layer}) state = State(containers={foo_container, bar_container}) ctx.run(ctx.on.pebble_check_failed(foo_container, check), state=state) + foo_infos = DoubleCharm.foo_infos + bar_infos = DoubleCharm.bar_infos assert len(foo_infos) == 1 assert foo_infos[0].name == 'http-check' - assert foo_infos[0].status == pebble.CheckStatus.DOWN + assert foo_infos[0].status == ops.pebble.CheckStatus.DOWN assert foo_infos[0].successes == 0 assert foo_infos[0].failures == 7 assert len(bar_infos) == 0 -def test_pebble_add_layer(): - class MyCharm(CharmBase): - def __init__(self, framework: Framework): - super().__init__(framework) - framework.observe(self.on.foo_pebble_ready, self._on_foo_ready) - - def _on_foo_ready(self, _): - self.unit.get_container('foo').add_layer( - 'foo', - {'checks': {'chk1': {'override': 'replace'}}}, - ) +class LayerCharm(ops.CharmBase): + def __init__(self, framework: ops.Framework): + super().__init__(framework) + framework.observe(self.on.foo_pebble_ready, self._on_foo_ready) + + def _on_foo_ready(self, _: ops.EventBase): + self.unit.get_container('foo').add_layer( + 'foo', + {'checks': {'chk1': {'override': 'replace'}}}, + ) - ctx = Context(MyCharm, meta={'name': 'foo', 'containers': {'foo': {}}}) + +def test_pebble_add_layer(): + ctx = Context(LayerCharm, meta={'name': 'foo', 'containers': {'foo': {}}}) container = Container('foo', can_connect=True) state_out = ctx.run(ctx.on.pebble_ready(container), state=State(containers={container})) chk1_info = state_out.get_container('foo').get_check_info('chk1') - assert chk1_info.status == pebble.CheckStatus.UP + assert chk1_info.status == ops.pebble.CheckStatus.UP -def test_pebble_start_check(): - class MyCharm(CharmBase): - def __init__(self, framework: Framework): - super().__init__(framework) - framework.observe(self.on.foo_pebble_ready, self._on_foo_ready) - framework.observe(self.on.config_changed, self._on_config_changed) - - def _on_foo_ready(self, _): - container = self.unit.get_container('foo') - container.add_layer( - 'foo', - { - 'checks': { - 'chk1': { - 'override': 'replace', - 'startup': 'disabled', - 'threshold': 3, - } +class StartCharm(ops.CharmBase): + def __init__(self, framework: ops.Framework): + super().__init__(framework) + framework.observe(self.on.foo_pebble_ready, self._on_foo_ready) + framework.observe(self.on.config_changed, self._on_config_changed) + + def _on_foo_ready(self, _: ops.EventBase): + container = self.unit.get_container('foo') + container.add_layer( + 'foo', + { + 'checks': { + 'chk1': { + 'override': 'replace', + 'startup': 'disabled', + 'threshold': 3, } - }, - ) + } + }, + ) + + def _on_config_changed(self, _: ops.EventBase): + container = self.unit.get_container('foo') + container.start_checks('chk1') - def _on_config_changed(self, _): - container = self.unit.get_container('foo') - container.start_checks('chk1') - ctx = Context(MyCharm, meta={'name': 'foo', 'containers': {'foo': {}}}) +def test_pebble_start_check(): + ctx = Context(StartCharm, meta={'name': 'foo', 'containers': {'foo': {}}}) container = Container('foo', can_connect=True) # Ensure that it starts as inactive. state_out = ctx.run(ctx.on.pebble_ready(container), state=State(containers={container})) chk1_info = state_out.get_container('foo').get_check_info('chk1') - assert chk1_info.status == pebble.CheckStatus.INACTIVE + assert chk1_info.status == ops.pebble.CheckStatus.INACTIVE # Verify that start_checks works. state_out = ctx.run(ctx.on.config_changed(), state=state_out) chk1_info = state_out.get_container('foo').get_check_info('chk1') - assert chk1_info.status == pebble.CheckStatus.UP + assert chk1_info.status == ops.pebble.CheckStatus.UP @pytest.fixture @@ -643,27 +712,28 @@ def reset_security_logging(): _get_juju_log_and_app_id.cache_clear() -def test_pebble_stop_check(reset_security_logging: None): - class MyCharm(CharmBase): - def __init__(self, framework: Framework): - super().__init__(framework) - framework.observe(self.on.config_changed, self._on_config_changed) +class StopCharm(ops.CharmBase): + def __init__(self, framework: ops.Framework): + super().__init__(framework) + framework.observe(self.on.config_changed, self._on_config_changed) + + def _on_config_changed(self, _: ops.EventBase): + container = self.unit.get_container('foo') + container.stop_checks('chk1') - def _on_config_changed(self, _): - container = self.unit.get_container('foo') - container.stop_checks('chk1') - ctx = Context(MyCharm, meta={'name': 'foo', 'containers': {'foo': {}}}) +def test_pebble_stop_check(reset_security_logging: None): + ctx = Context(StopCharm, meta={'name': 'foo', 'containers': {'foo': {}}}) - layer = pebble.Layer({ + layer = ops.pebble.Layer({ 'checks': {'chk1': {'override': 'replace', 'startup': 'enabled', 'threshold': 3}} }) assert layer.checks['chk1'].threshold is not None info_in = CheckInfo( 'chk1', - status=pebble.CheckStatus.UP, - level=layer.checks['chk1'].level, - startup=layer.checks['chk1'].startup, + status=ops.pebble.CheckStatus.UP, + level=ops.pebble.CheckLevel(layer.checks['chk1'].level), + startup=ops.pebble.CheckStartup(layer.checks['chk1'].startup), threshold=layer.checks['chk1'].threshold, ) container = Container( @@ -674,29 +744,30 @@ def _on_config_changed(self, _): ) state_out = ctx.run(ctx.on.config_changed(), state=State(containers={container})) info_out = state_out.get_container('foo').get_check_info('chk1') - assert info_out.status == pebble.CheckStatus.INACTIVE + assert info_out.status == ops.pebble.CheckStatus.INACTIVE -def test_pebble_replan_checks(): - class MyCharm(CharmBase): - def __init__(self, framework: Framework): - super().__init__(framework) - framework.observe(self.on.config_changed, self._on_config_changed) +class ReplanCharm(ops.CharmBase): + def __init__(self, framework: ops.Framework): + super().__init__(framework) + framework.observe(self.on.config_changed, self._on_config_changed) + + def _on_config_changed(self, _: ops.EventBase): + container = self.unit.get_container('foo') + container.replan() - def _on_config_changed(self, _): - container = self.unit.get_container('foo') - container.replan() - ctx = Context(MyCharm, meta={'name': 'foo', 'containers': {'foo': {}}}) - layer = pebble.Layer({ +def test_pebble_replan_checks(): + ctx = Context(ReplanCharm, meta={'name': 'foo', 'containers': {'foo': {}}}) + layer = ops.pebble.Layer({ 'checks': {'chk1': {'override': 'replace', 'startup': 'enabled', 'threshold': 3}} }) assert layer.checks['chk1'].threshold is not None info_in = CheckInfo( 'chk1', - status=pebble.CheckStatus.INACTIVE, - level=layer.checks['chk1'].level, - startup=layer.checks['chk1'].startup, + status=ops.pebble.CheckStatus.INACTIVE, + level=ops.pebble.CheckLevel(layer.checks['chk1'].level), + startup=ops.pebble.CheckStartup(layer.checks['chk1'].startup), threshold=layer.checks['chk1'].threshold, ) container = Container( @@ -707,7 +778,23 @@ def _on_config_changed(self, _): ) state_out = ctx.run(ctx.on.config_changed(), state=State(containers={container})) info_out = state_out.get_container('foo').get_check_info('chk1') - assert info_out.status == pebble.CheckStatus.UP + assert info_out.status == ops.pebble.CheckStatus.UP + + +class CombineLayerCharm(ops.CharmBase): + layer_name: str + layer_dict: LayerDict + combine: bool + + def __init__(self, framework: ops.Framework): + super().__init__(framework) + framework.observe(self.on['my-container'].pebble_ready, self._on_pebble_ready) + + def _on_pebble_ready(self, _: ops.PebbleReadyEvent): + container = self.unit.get_container('my-container') + container.add_layer( + self.layer_name, ops.pebble.Layer(self.layer_dict), combine=self.combine + ) @pytest.mark.parametrize( @@ -743,19 +830,14 @@ def _on_config_changed(self, _): ], ) def test_add_layer_merge_check( - new_layer_name: str, combine: bool, new_layer_dict: pebble.LayerDict + monkeypatch: pytest.MonkeyPatch, new_layer_name: str, combine: bool, new_layer_dict: LayerDict ): - class MyCharm(CharmBase): - def __init__(self, framework: Framework): - super().__init__(framework) - framework.observe(self.on['my-container'].pebble_ready, self._on_pebble_ready) - - def _on_pebble_ready(self, _: PebbleReadyEvent): - container = self.unit.get_container('my-container') - container.add_layer(new_layer_name, Layer(new_layer_dict), combine=combine) + monkeypatch.setattr(CombineLayerCharm, 'layer_name', new_layer_name, raising=False) + monkeypatch.setattr(CombineLayerCharm, 'layer_dict', new_layer_dict, raising=False) + monkeypatch.setattr(CombineLayerCharm, 'combine', combine, raising=False) - ctx = Context(MyCharm, meta={'name': 'foo', 'containers': {'my-container': {}}}) - layer_in = pebble.Layer({ + ctx = Context(CombineLayerCharm, meta={'name': 'foo', 'containers': {'my-container': {}}}) + layer_in = ops.pebble.Layer({ 'checks': { 'server-ready': { 'override': 'replace', @@ -769,9 +851,9 @@ def _on_pebble_ready(self, _: PebbleReadyEvent): assert layer_in.checks['server-ready'].threshold is not None check_in = CheckInfo( 'server-ready', - level=layer_in.checks['server-ready'].level, + level=ops.pebble.CheckLevel(layer_in.checks['server-ready'].level), threshold=layer_in.checks['server-ready'].threshold, - startup=layer_in.checks['server-ready'].startup, + startup=ops.pebble.CheckStartup(layer_in.checks['server-ready'].startup), ) container_in = Container( 'my-container', @@ -779,21 +861,21 @@ def _on_pebble_ready(self, _: PebbleReadyEvent): layers={'base': layer_in}, check_infos={check_in}, ) - assert container_in.get_check_info('server-ready').level == pebble.CheckLevel.READY + assert container_in.get_check_info('server-ready').level == ops.pebble.CheckLevel.READY state_in = State(containers={container_in}) state_out = ctx.run(ctx.on.pebble_ready(container_in), state_in) check_out = state_out.get_container(container_in.name).get_check_info('server-ready') - new_layer_check = new_layer_dict['checks']['server-ready'] - assert check_out.level == pebble.CheckLevel(new_layer_check.get('level', 'ready')) - assert check_out.startup == pebble.CheckStartup(new_layer_check.get('startup', 'enabled')) + new_layer_check = new_layer_dict.get('checks', {}).get('server-ready', {}) + assert check_out.level == ops.pebble.CheckLevel(new_layer_check.get('level', 'ready')) + assert check_out.startup == ops.pebble.CheckStartup(new_layer_check.get('startup', 'enabled')) assert check_out.threshold == new_layer_check.get('threshold', 10) @pytest.mark.parametrize('layer1_name,layer2_name', [('a-base', 'b-base'), ('b-base', 'a-base')]) -def test_layers_merge_in_plan(layer1_name, layer2_name): - layer1 = pebble.Layer({ +def test_layers_merge_in_plan(layer1_name: str, layer2_name: str): + layer1_dict: LayerDict = { 'services': { 'server': { 'override': 'replace', @@ -809,8 +891,8 @@ def test_layers_merge_in_plan(layer1_name, layer2_name): 'level': 'ready', 'startup': 'enabled', 'threshold': 10, - 'period': 1, - 'timeout': 28, + 'period': '1s', + 'timeout': '28s', 'http': {'url': 'http://localhost:5000/version'}, } }, @@ -823,8 +905,8 @@ def test_layers_merge_in_plan(layer1_name, layer2_name): 'labels': {'foo': 'bar'}, } }, - }) - layer2 = pebble.Layer({ + } + layer2_dict: LayerDict = { 'services': { 'server': { 'override': 'merge', @@ -844,9 +926,11 @@ def test_layers_merge_in_plan(layer1_name, layer2_name): 'location': 'https://loki2.example.com', }, }, - }) + } + layer1 = ops.pebble.Layer(layer1_dict) + layer2 = ops.pebble.Layer(layer2_dict) - ctx = Context(CharmBase, meta={'name': 'foo', 'containers': {'my-container': {}}}) + ctx = Context(ops.CharmBase, meta={'name': 'foo', 'containers': {'my-container': {}}}) # TODO also a starting layer. container = Container('my-container', can_connect=True) @@ -861,18 +945,20 @@ def test_layers_merge_in_plan(layer1_name, layer2_name): assert service.summary == 'sum' assert service.description == 'desc' # Service.startup is always a string, even though we have the enum. - assert service.startup == pebble.ServiceStartup.ENABLED.value + assert service.startup == ops.pebble.ServiceStartup.ENABLED.value assert service.override == 'merge' assert service.command == '/bin/sleep 20' check = plan.checks['server-ready'] - assert check.startup == pebble.CheckStartup.ENABLED + assert check.startup == ops.pebble.CheckStartup.ENABLED assert check.threshold == 10 - assert check.period == 1 - assert check.timeout == 28 + assert check.period == '1s' + assert check.timeout == '28s' assert check.override == 'merge' - assert check.level == pebble.CheckLevel.ALIVE - assert check.http['url'] == 'http://localhost:5050/version' + assert check.level == ops.pebble.CheckLevel.ALIVE + http = check.http + assert http is not None + assert http.get('url') == 'http://localhost:5050/version' log_target = plan.log_targets['loki'] assert log_target.type == 'loki' diff --git a/testing/tests/test_e2e/test_ports.py b/testing/tests/test_e2e/test_ports.py index 9b37f5fb0..c1c6ac60f 100644 --- a/testing/tests/test_e2e/test_ports.py +++ b/testing/tests/test_e2e/test_ports.py @@ -3,19 +3,14 @@ from __future__ import annotations -from collections.abc import Mapping -from typing import Any - import pytest -from scenario import Context, State -from scenario.state import Port, StateValidationError, TCPPort, UDPPort +from scenario import Context, Port, State, TCPPort, UDPPort +from scenario.errors import StateValidationError from ops import CharmBase, Framework, StartEvent, StopEvent -class MyCharm(CharmBase): - META: Mapping[str, Any] = {'name': 'edgar'} - +class Charm(CharmBase): def __init__(self, framework: Framework): super().__init__(framework) framework.observe(self.on.start, self._open_port) @@ -30,20 +25,21 @@ def _close_port(self, _: StopEvent): @pytest.fixture -def ctx(): - return Context(MyCharm, meta=MyCharm.META) +def ctx() -> Context[Charm]: + return Context(Charm, meta={'name': 'edgar'}) -def test_open_port(ctx): +def test_open_port(ctx: Context[Charm]): out = ctx.run(ctx.on.start(), State()) - assert len(out.opened_ports) == 1 - port = next(iter(out.opened_ports)) + ports = tuple(out.opened_ports) + assert len(ports) == 1 + port = ports[0] assert port.protocol == 'tcp' assert port.port == 12 -def test_close_port(ctx): +def test_close_port(ctx: Context[Charm]): out = ctx.run(ctx.on.stop(), State(opened_ports={TCPPort(42)})) assert not out.opened_ports @@ -54,7 +50,7 @@ def test_port_no_arguments(): @pytest.mark.parametrize('klass', (TCPPort, UDPPort)) -def test_port_port(klass): +def test_port_port(klass: type[Port]): with pytest.raises(StateValidationError): klass(port=0) with pytest.raises(StateValidationError): diff --git a/testing/tests/test_e2e/test_relations.py b/testing/tests/test_e2e/test_relations.py index bbf7db34f..2b821a7be 100644 --- a/testing/tests/test_e2e/test_relations.py +++ b/testing/tests/test_e2e/test_relations.py @@ -4,66 +4,55 @@ from __future__ import annotations from collections.abc import Callable +from typing import Any, cast import pytest -import scenario -from scenario import Context -from scenario.errors import UncaughtCharmError +from scenario.context import Context +from scenario.errors import StateValidationError, UncaughtCharmError from scenario.state import ( _DEFAULT_JUJU_DATABAG, PeerRelation, Relation, RelationBase, State, - StateValidationError, SubordinateRelation, _Event, _next_relation_id, ) import ops -from ops.charm import ( - CharmBase, - CharmEvents, - CollectStatusEvent, - RelationBrokenEvent, - RelationCreatedEvent, - RelationDepartedEvent, - RelationEvent, -) -from ops.framework import EventBase, Framework from tests.helpers import trigger @pytest.fixture(scope='function') def mycharm(): - class MyCharmEvents(CharmEvents): + class MyCharmEvents(ops.CharmEvents): @classmethod - def define_event(cls, event_kind: str, event_type: type[EventBase]): + def define_event(cls, event_kind: str, event_type: type[ops.EventBase]): if getattr(cls, event_kind, None): delattr(cls, event_kind) return super().define_event(event_kind, event_type) - class MyCharm(CharmBase): - _call: Callable[[MyCharm, _Event], None] | None = None + class MyCharm(ops.CharmBase): + _call: Callable[[_Event], None] | None = None called = False - on = MyCharmEvents() + on: ops.CharmEvents = MyCharmEvents() - def __init__(self, framework: Framework): + def __init__(self, framework: ops.Framework): super().__init__(framework) for evt in self.on.events().values(): self.framework.observe(evt, self._on_event) - def _on_event(self, event): + def _on_event(self, event: ops.EventBase): if self._call: MyCharm.called = True - self._call(event) + self._call(cast('Any', event)) return MyCharm -def test_get_relation(mycharm): - def pre_event(charm: CharmBase): +def test_get_relation(mycharm: Any): + def pre_event(charm: ops.CharmBase): assert charm.model.get_relation('foo') assert charm.model.get_relation('bar') is None assert charm.model.get_relation('qux') @@ -151,7 +140,7 @@ def test_validation(self, context: str): }, actions={'my-act': {}}, ) - rel_in = scenario.Relation( + rel_in = Relation( endpoint='my-rel', local_app_data={'k': 'local val'}, remote_app_data={'k': 'remote val'}, @@ -164,12 +153,12 @@ def test_validation(self, context: str): def test_relation_set_single_add_del_change(): relation_name = 'relation-name' - class Charm(CharmBase): - def __init__(self, framework: Framework): + class Charm(ops.CharmBase): + def __init__(self, framework: ops.Framework): super().__init__(framework) framework.observe(self.on.update_status, self._update_status) - def _update_status(self, event: EventBase): + def _update_status(self, event: ops.EventBase): rel = self.model.get_relation(relation_name) assert rel is not None data = rel.data[self.unit] @@ -307,12 +296,12 @@ def test_relation_set_bulk_update( ): relation_name = 'relation-name' - class Charm(CharmBase): - def __init__(self, framework: Framework): + class Charm(ops.CharmBase): + def __init__(self, framework: ops.Framework): super().__init__(framework) framework.observe(self.on.update_status, self._update_status) - def _update_status(self, event: EventBase): + def _update_status(self, event: ops.EventBase): rel = self.model.get_relation(relation_name) assert rel is not None data = rel.data[self.unit] @@ -339,19 +328,21 @@ def _update_status(self, event: EventBase): 'remote_app_name', ('remote', 'prometheus', 'aodeok123'), ) -def test_relation_events(mycharm, evt_name, remote_app_name): +def test_relation_events(mycharm: Any, evt_name: str, remote_app_name: str): relation = Relation(endpoint='foo', interface='foo', remote_app_name=remote_app_name) - def callback(charm: CharmBase, e): - if not isinstance(e, RelationEvent): + def callback(charm: ops.CharmBase, e: ops.EventBase): + if not isinstance(e, ops.RelationEvent): return # filter out collect status events if evt_name == 'broken': assert charm.model.get_relation('foo') is None assert e.relation.app.name == remote_app_name else: - assert charm.model.get_relation('foo').app is not None - assert charm.model.get_relation('foo').app.name == remote_app_name + rel = charm.model.get_relation('foo') + assert rel is not None + assert rel.app is not None + assert rel.app.name == remote_app_name mycharm._call = callback @@ -390,18 +381,25 @@ def callback(charm: CharmBase, e): 'remote_unit_id', (0, 1), ) -def test_relation_events_attrs(mycharm, evt_name, has_unit, remote_app_name, remote_unit_id): +def test_relation_events_attrs( + mycharm: Any, + evt_name: str, + has_unit: bool, + remote_app_name: str, + remote_unit_id: int, +): relation = Relation(endpoint='foo', interface='foo', remote_app_name=remote_app_name) - def callback(charm: CharmBase, event): - if isinstance(event, CollectStatusEvent): + def callback(charm: ops.CharmBase, event: ops.EventBase): + if isinstance(event, ops.CollectStatusEvent): return - assert event.app - if not isinstance(event, (RelationCreatedEvent, RelationBrokenEvent)): - assert event.unit - if isinstance(event, RelationDepartedEvent): - assert event.departing_unit + if isinstance(event, ops.RelationEvent): + assert event.app + if not isinstance(event, ops.RelationCreatedEvent | ops.RelationBrokenEvent): + assert event.unit + if isinstance(event, ops.RelationDepartedEvent): + assert event.departing_unit mycharm._call = callback @@ -430,7 +428,9 @@ def callback(charm: CharmBase, event): 'remote_app_name', ('remote', 'prometheus', 'aodeok123'), ) -def test_relation_events_no_attrs(mycharm, evt_name, remote_app_name, caplog): +def test_relation_events_no_attrs( + mycharm: Any, evt_name: str, remote_app_name: str, caplog: pytest.LogCaptureFixture +): relation = Relation( endpoint='foo', interface='foo', @@ -438,17 +438,18 @@ def test_relation_events_no_attrs(mycharm, evt_name, remote_app_name, caplog): remote_units_data={0: {}, 1: {}}, # 2 units ) - def callback(charm: CharmBase, event): - if isinstance(event, CollectStatusEvent): + def callback(charm: ops.CharmBase, event: ops.EventBase): + if isinstance(event, ops.CollectStatusEvent): return - assert event.app # that's always present - # .unit is always None for created and broken. - if isinstance(event, (RelationCreatedEvent, RelationBrokenEvent)): - assert event.unit is None - else: - assert event.unit - assert (evt_name == 'departed') is bool(getattr(event, 'departing_unit', False)) + if isinstance(event, ops.RelationEvent): + assert event.app # that's always present + # .unit is always None for created and broken. + if isinstance(event, ops.RelationCreatedEvent | ops.RelationBrokenEvent): + assert event.unit is None + else: + assert event.unit + assert (evt_name == 'departed') is bool(getattr(event, 'departing_unit', False)) mycharm._call = callback @@ -490,19 +491,22 @@ def test_relation_default_unit_data_peer(): @pytest.mark.parametrize('evt_name', ('broken', 'created')) -def test_relation_events_no_remote_units(mycharm, evt_name, caplog): +def test_relation_events_no_remote_units( + mycharm: Any, evt_name: str, caplog: pytest.LogCaptureFixture +): relation = Relation( endpoint='foo', interface='foo', remote_units_data={}, # no units ) - def callback(charm: CharmBase, event): - if isinstance(event, CollectStatusEvent): + def callback(charm: ops.CharmBase, event: ops.EventBase): + if isinstance(event, ops.CollectStatusEvent): return - assert event.app # that's always present - assert not event.unit + if isinstance(event, ops.RelationEvent): + assert event.app # that's always present + assert not event.unit mycharm._call = callback @@ -526,16 +530,16 @@ def callback(charm: CharmBase, event): assert 'remote unit ID unset; no remote unit data present' in caplog.text -@pytest.mark.parametrize('data', (set(), {}, [], (), 1, 1.0, None, b'')) -def test_relation_unit_data_bad_types(mycharm, data): +@pytest.mark.parametrize('data', (set(), {}, [], (), 1, 1.0, None, b'')) # type: ignore +def test_relation_unit_data_bad_types(mycharm: Any, data: object): with pytest.raises(StateValidationError): - Relation(endpoint='foo', interface='foo', remote_units_data={0: {'a': data}}) + Relation(endpoint='foo', interface='foo', remote_units_data={0: {'a': cast('Any', data)}}) -@pytest.mark.parametrize('data', (set(), {}, [], (), 1, 1.0, None, b'')) -def test_relation_app_data_bad_types(mycharm, data): +@pytest.mark.parametrize('data', (set(), {}, [], (), 1, 1.0, None, b'')) # type: ignore +def test_relation_app_data_bad_types(mycharm: Any, data: object): with pytest.raises(StateValidationError): - Relation(endpoint='foo', interface='foo', local_app_data={'a': data}) + Relation(endpoint='foo', interface='foo', local_app_data={'a': cast('Any', data)}) @pytest.mark.parametrize( @@ -550,7 +554,7 @@ def test_relation_app_data_bad_types(mycharm, data): SubordinateRelation('c'), ), ) -def test_relation_event_trigger(relation, evt_name, mycharm): +def test_relation_event_trigger(relation: RelationBase, evt_name: str, mycharm: Any): meta = { 'name': 'mycharm', 'requires': {'a': {'interface': 'i1'}}, @@ -571,7 +575,7 @@ def test_relation_event_trigger(relation, evt_name, mycharm): ) -def test_trigger_sub_relation(mycharm): +def test_trigger_sub_relation(mycharm: Any): meta = { 'name': 'mycharm', 'provides': { @@ -586,7 +590,7 @@ def test_trigger_sub_relation(mycharm): sub1 = SubordinateRelation('foo', remote_unit_data={'1': '2'}, remote_app_name='primary1') sub2 = SubordinateRelation('foo', remote_unit_data={'3': '4'}, remote_app_name='primary2') - def post_event(charm: CharmBase): + def post_event(charm: ops.CharmBase): b_relations = charm.model.relations['foo'] assert len(b_relations) == 2 for relation in b_relations: @@ -615,7 +619,7 @@ def test_relation_ids(): assert rel.id == initial_id + i -def test_broken_relation_not_in_model_relations(mycharm): +def test_broken_relation_not_in_model_relations(mycharm: Any): rel = Relation('foo') ctx = Context(mycharm, meta={'name': 'local', 'requires': {'foo': {'interface': 'foo'}}}) @@ -627,18 +631,20 @@ def test_broken_relation_not_in_model_relations(mycharm): def test_get_relation_when_missing(): - class MyCharm(CharmBase): - def __init__(self, framework): + class MyCharm(ops.CharmBase): + def __init__(self, framework: ops.Framework): super().__init__(framework) self.framework.observe(self.on.update_status, self._on_update_status) self.framework.observe(self.on.config_changed, self._on_config_changed) self.relation = None - def _on_update_status(self, _): + def _on_update_status(self, _: ops.EventBase): self.relation = self.model.get_relation('foo') - def _on_config_changed(self, _): - self.relation = self.model.get_relation('foo', self.config['relation-id']) + def _on_config_changed(self, _: ops.EventBase): + relation_id = self.config['relation-id'] + assert isinstance(relation_id, int) + self.relation = self.model.get_relation('foo', relation_id) ctx = Context( MyCharm, @@ -655,6 +661,7 @@ def _on_config_changed(self, _): rel = Relation('foo') with ctx(ctx.on.update_status(), State(relations={rel})) as mgr: mgr.run() + assert mgr.charm.relation is not None assert mgr.charm.relation.id == rel.id # If a relation that doesn't exist is requested, that should also not raise @@ -662,6 +669,7 @@ def _on_config_changed(self, _): with ctx(ctx.on.config_changed(), State(config={'relation-id': 42})) as mgr: mgr.run() rel = mgr.charm.relation + assert rel is not None assert rel.id == 42 assert not rel.active @@ -673,9 +681,9 @@ def _on_config_changed(self, _): @pytest.mark.parametrize('klass', (Relation, PeerRelation, SubordinateRelation)) -def test_relation_positional_arguments(klass): +def test_relation_positional_arguments(klass: type[RelationBase]): with pytest.raises(TypeError): - klass('foo', 'bar', None) + cast('Any', klass)('foo', 'bar', None) def test_relation_default_values(): @@ -724,12 +732,12 @@ def test_peer_relation_default_values(): def test_relation_remote_model(): - class MyCharm(CharmBase): - def __init__(self, framework): + class MyCharm(ops.CharmBase): + def __init__(self, framework: ops.Framework): super().__init__(framework) self.framework.observe(self.on.start, self._on_start) - def _on_start(self, event): + def _on_start(self, event: ops.EventBase): relation = self.model.get_relation('foo') assert relation is not None self.remote_model_uuid = relation.remote_model.uuid @@ -751,12 +759,12 @@ def _on_start(self, event): def test_peer_relation_units_does_not_contain_this_unit(): relation_name = 'relation-name' - class Charm(CharmBase): - def __init__(self, framework: Framework): + class Charm(ops.CharmBase): + def __init__(self, framework: ops.Framework): super().__init__(framework) framework.observe(self.on.update_status, self._update_status) - def _update_status(self, _: EventBase): + def _update_status(self, _: ops.EventBase): rel = self.model.get_relation(relation_name) assert rel is not None assert self.unit not in rel.units diff --git a/testing/tests/test_e2e/test_rubbish_events.py b/testing/tests/test_e2e/test_rubbish_events.py index b94403e03..c3e08c96e 100644 --- a/testing/tests/test_e2e/test_rubbish_events.py +++ b/testing/tests/test_e2e/test_rubbish_events.py @@ -31,10 +31,10 @@ class MySubEvents(CharmEvents): sub = EventSource(SubEvent) class Sub(Object): - on = MySubEvents() + on: ClassVar[MySubEvents] = MySubEvents() class MyCharm(CharmBase): - on = MyCharmEvents() + on: ClassVar[MyCharmEvents] = MyCharmEvents() evts: ClassVar[list[EventBase]] = [] def __init__(self, framework: Framework): @@ -43,19 +43,21 @@ def __init__(self, framework: Framework): self.framework.observe(self.sub.on.sub, self._on_event) self.framework.observe(self.on.qux, self._on_event) - def _on_event(self, e): + def _on_event(self, e: EventBase): MyCharm.evts.append(e) return MyCharm @pytest.mark.parametrize('evt_name', ('rubbish', 'foo', 'bar')) -def test_rubbish_event_raises(mycharm: CharmBase, evt_name: str): +def test_rubbish_event_raises(mycharm: type[CharmBase], evt_name: str): with pytest.raises(AttributeError): trigger(State(), evt_name, mycharm, meta={'name': 'foo'}) -def test_rubbish_pebble_ready_event_raises(mycharm: CharmBase, monkeypatch: pytest.MonkeyPatch): +def test_rubbish_pebble_ready_event_raises( + mycharm: type[CharmBase], monkeypatch: pytest.MonkeyPatch +): monkeypatch.setenv('SCENARIO_SKIP_CONSISTENCY_CHECKS', '1') # else it will whine about the container not being in state and meta; # but if we put the container in meta, it will actually register an event! @@ -64,14 +66,14 @@ def test_rubbish_pebble_ready_event_raises(mycharm: CharmBase, monkeypatch: pyte @pytest.mark.parametrize('evt_name', ('qux',)) -def test_custom_events_fail(mycharm, evt_name): +def test_custom_events_fail(mycharm: type[CharmBase], evt_name: str): with pytest.raises(AttributeError): trigger(State(), evt_name, mycharm, meta={'name': 'foo'}) # cfr: https://github.com/PietroPasotti/ops-scenario/pull/11#discussion_r1101694961 @pytest.mark.parametrize('evt_name', ('sub',)) -def test_custom_events_sub_raise(mycharm, evt_name): +def test_custom_events_sub_raise(mycharm: type[CharmBase], evt_name: str): with pytest.raises(AttributeError): trigger(State(), evt_name, mycharm, meta={'name': 'foo'}) @@ -88,6 +90,8 @@ def test_custom_events_sub_raise(mycharm, evt_name): ('bar-relation-changed', True), ), ) -def test_is_custom_event(mycharm, evt_name, expected): - spec = _CharmSpec(charm_type=mycharm, meta={'name': 'mycharm', 'requires': {'foo': {}}}) +def test_is_custom_event(mycharm: type[CharmBase], evt_name: str, expected: bool): + spec: _CharmSpec[CharmBase] = _CharmSpec( + charm_type=mycharm, meta={'name': 'mycharm', 'requires': {'foo': {}}} + ) assert _Event(evt_name)._is_builtin_event(spec) is expected diff --git a/testing/tests/test_e2e/test_secrets.py b/testing/tests/test_e2e/test_secrets.py index c0170d2b2..99513defc 100644 --- a/testing/tests/test_e2e/test_secrets.py +++ b/testing/tests/test_e2e/test_secrets.py @@ -5,46 +5,43 @@ import collections import datetime -from typing import Literal, cast +from typing import Any, Literal, cast from unittest.mock import ANY import pytest from scenario import Context from scenario.state import Relation, Secret, State -from ops.charm import CharmBase -from ops.framework import Framework -from ops.model import ModelError, SecretNotFoundError, SecretRotate -from ops.model import Secret as ops_Secret +import ops from test.charms.test_secrets.src.charm import Result, SecretsCharm from tests.helpers import trigger @pytest.fixture(scope='function') -def mycharm(): - class MyCharm(CharmBase): - def __init__(self, framework: Framework): +def mycharm() -> type[ops.CharmBase]: + class MyCharm(ops.CharmBase): + def __init__(self, framework: ops.Framework): super().__init__(framework) for evt in self.on.events().values(): self.framework.observe(evt, self._on_event) - def _on_event(self, event): + def _on_event(self, event: ops.EventBase): pass return MyCharm -def test_get_secret_no_secret(mycharm): +def test_get_secret_no_secret(mycharm: type[ops.CharmBase]): ctx = Context(mycharm, meta={'name': 'local'}) with ctx(ctx.on.update_status(), State()) as mgr: - with pytest.raises(SecretNotFoundError): + with pytest.raises(ops.SecretNotFoundError): assert mgr.charm.model.get_secret(id='foo') - with pytest.raises(SecretNotFoundError): + with pytest.raises(ops.SecretNotFoundError): assert mgr.charm.model.get_secret(label='foo') @pytest.mark.parametrize('owner', ('app', 'unit')) -def test_get_secret(mycharm, owner): +def test_get_secret(mycharm: type[ops.CharmBase], owner: Literal['app', 'unit']): ctx = Context(mycharm, meta={'name': 'local'}) secret = Secret({'a': 'b'}, owner=owner) with ctx( @@ -55,7 +52,7 @@ def test_get_secret(mycharm, owner): @pytest.mark.parametrize('owner', ('app', 'unit')) -def test_get_secret_get_refresh(mycharm, owner): +def test_get_secret_get_refresh(mycharm: type[ops.CharmBase], owner: Literal['app', 'unit']): ctx = Context(mycharm, meta={'name': 'local'}) secret = Secret( tracked_content={'a': 'b'}, @@ -71,7 +68,7 @@ def test_get_secret_get_refresh(mycharm, owner): @pytest.mark.parametrize('app', (True, False)) -def test_get_secret_nonowner_peek_update(mycharm, app): +def test_get_secret_nonowner_peek_update(mycharm: type[ops.CharmBase], app: bool): ctx = Context(mycharm, meta={'name': 'local'}) secret = Secret( tracked_content={'a': 'b'}, @@ -95,7 +92,7 @@ def test_get_secret_nonowner_peek_update(mycharm, app): @pytest.mark.parametrize('owner', ('app', 'unit')) -def test_get_secret_owner_peek_update(mycharm, owner): +def test_get_secret_owner_peek_update(mycharm: type[ops.CharmBase], owner: Literal['app', 'unit']): ctx = Context(mycharm, meta={'name': 'local'}) secret = Secret( tracked_content={'a': 'b'}, @@ -117,7 +114,9 @@ def test_get_secret_owner_peek_update(mycharm, owner): @pytest.mark.parametrize('owner', ('app', 'unit')) -def test_secret_changed_owner_evt_fails(mycharm, owner): +def test_secret_changed_owner_evt_fails( + mycharm: type[ops.CharmBase], owner: Literal['app', 'unit'] +): ctx = Context(mycharm, meta={'name': 'local'}) secret = Secret( tracked_content={'a': 'b'}, @@ -136,13 +135,15 @@ def test_secret_changed_owner_evt_fails(mycharm, owner): ('remove', 1), ], ) -def test_consumer_events_failures(mycharm, evt_suffix, revision): +def test_consumer_events_failures( + mycharm: type[ops.CharmBase], evt_suffix: str, revision: int | None +): ctx = Context(mycharm, meta={'name': 'local'}) secret = Secret( tracked_content={'a': 'b'}, latest_content={'a': 'c'}, ) - kwargs = {'secret': secret} + kwargs: dict[str, Any] = {'secret': secret} if revision is not None: kwargs['revision'] = revision with pytest.raises(ValueError): @@ -150,7 +151,7 @@ def test_consumer_events_failures(mycharm, evt_suffix, revision): @pytest.mark.parametrize('app', (True, False)) -def test_add(mycharm, app): +def test_add(mycharm: type[ops.CharmBase], app: bool): ctx = Context(mycharm, meta={'name': 'local'}) with ctx( ctx.on.update_status(), @@ -169,7 +170,7 @@ def test_add(mycharm, app): assert secret.label == 'mylabel' -def test_set_legacy_behaviour(mycharm): +def test_set_legacy_behaviour(mycharm: type[ops.CharmBase]): # in juju < 3.1.7, secret owners always used to track the latest revision. # ref: https://bugs.launchpad.net/juju/+bug/2037120 ctx = Context(mycharm, meta={'name': 'local'}, juju_version='3.1.6') @@ -179,7 +180,7 @@ def test_set_legacy_behaviour(mycharm): State(), ) as mgr: charm = mgr.charm - secret: ops_Secret = charm.unit.add_secret(rev1, label='mylabel') + secret: ops.Secret = charm.unit.add_secret(rev1, label='mylabel') assert ( secret.get_content() == secret.peek_content() @@ -190,7 +191,7 @@ def test_set_legacy_behaviour(mycharm): secret.set_content(rev2) # We need to get the secret again, because ops caches the content in # the object. - secret: ops_Secret = charm.model.get_secret(label='mylabel') + secret: ops.Secret = charm.model.get_secret(label='mylabel') assert ( secret.get_content() == secret.peek_content() @@ -207,7 +208,7 @@ def test_set_legacy_behaviour(mycharm): ) -def test_set(mycharm): +def test_set(mycharm: type[ops.CharmBase]): ctx = Context(mycharm, meta={'name': 'local'}) rev1, rev2 = {'foo': 'bar'}, {'foo': 'baz', 'qux': 'roz'} with ctx( @@ -215,7 +216,7 @@ def test_set(mycharm): State(), ) as mgr: charm = mgr.charm - secret: ops_Secret = charm.unit.add_secret(rev1, label='mylabel') + secret: ops.Secret = charm.unit.add_secret(rev1, label='mylabel') assert ( secret.get_content() == secret.peek_content() @@ -239,7 +240,7 @@ def test_set(mycharm): ) -def test_set_juju33(mycharm): +def test_set_juju33(mycharm: type[ops.CharmBase]): ctx = Context(mycharm, meta={'name': 'local'}, juju_version='3.3.1') rev1, rev2 = {'foo': 'bar'}, {'foo': 'baz', 'qux': 'roz'} with ctx( @@ -247,7 +248,7 @@ def test_set_juju33(mycharm): State(), ) as mgr: charm = mgr.charm - secret: ops_Secret = charm.unit.add_secret(rev1, label='mylabel') + secret: ops.Secret = charm.unit.add_secret(rev1, label='mylabel') assert secret.get_content() == rev1 secret.set_content(rev2) @@ -265,14 +266,14 @@ def test_set_juju33(mycharm): @pytest.mark.parametrize('app', (True, False)) -def test_meta(mycharm, app): +def test_meta(mycharm: type[ops.CharmBase], app: bool): ctx = Context(mycharm, meta={'name': 'local'}) secret = Secret( {'a': 'b'}, owner='app' if app else 'unit', label='mylabel', description='foobarbaz', - rotate=SecretRotate.HOURLY, + rotate=ops.SecretRotate.HOURLY, ) with ctx( ctx.on.update_status(), @@ -290,12 +291,14 @@ def test_meta(mycharm, app): assert secret.label is None assert info.description == 'foobarbaz' assert info.label == 'mylabel' - assert info.rotation == SecretRotate.HOURLY + assert info.rotation == ops.SecretRotate.HOURLY @pytest.mark.parametrize('leader', (True, False)) @pytest.mark.parametrize('owner', ('app', 'unit', None)) -def test_secret_permission_model(mycharm, leader, owner): +def test_secret_permission_model( + mycharm: type[ops.CharmBase], leader: bool, owner: Literal['app', 'unit'] | None +): expect_manage = bool( # if you're the leader and own this app secret (owner == 'app' and leader) @@ -304,23 +307,23 @@ def test_secret_permission_model(mycharm, leader, owner): ) ctx = Context(mycharm, meta={'name': 'local'}) - secret = Secret( + scenario_secret = Secret( {'a': 'b'}, label='mylabel', owner=owner, description='foobarbaz', - rotate=SecretRotate.HOURLY, + rotate=ops.SecretRotate.HOURLY, ) - secret_id = secret.id + secret_id = scenario_secret.id with ctx( ctx.on.update_status(), State( leader=leader, - secrets={secret}, + secrets={scenario_secret}, ), ) as mgr: # can always view - secret: ops_Secret = mgr.charm.model.get_secret(id=secret_id) + secret: ops.Secret = mgr.charm.model.get_secret(id=secret_id) assert secret.get_content()['a'] == 'b' assert secret.peek_content() assert secret.get_content(refresh=True) @@ -340,22 +343,22 @@ def test_secret_permission_model(mycharm, leader, owner): else: # cannot manage # nothing else to do directly if you can't get a hold of the Secret instance # but we can try some raw backend calls - with pytest.raises(ModelError): + with pytest.raises(ops.ModelError): secret.get_info() - with pytest.raises(ModelError): + with pytest.raises(ops.ModelError): secret.set_content(content={'boo': 'foo'}) @pytest.mark.parametrize('app', (True, False)) -def test_grant(mycharm, app): +def test_grant(mycharm: type[ops.CharmBase], app: bool): ctx = Context(mycharm, meta={'name': 'local', 'requires': {'foo': {'interface': 'bar'}}}) secret = Secret( {'a': 'b'}, owner='unit', label='mylabel', description='foobarbaz', - rotate=SecretRotate.HOURLY, + rotate=ops.SecretRotate.HOURLY, ) with ctx( ctx.on.update_status(), @@ -367,6 +370,7 @@ def test_grant(mycharm, app): charm = mgr.charm secret = charm.model.get_secret(label='mylabel') foo = charm.model.get_relation('foo') + assert foo is not None if app: secret.grant(relation=foo) else: @@ -376,7 +380,7 @@ def test_grant(mycharm, app): assert vals == [{'remote'}] if app else [{'remote/0'}] -def test_update_metadata(mycharm): +def test_update_metadata(mycharm: type[ops.CharmBase]): exp = datetime.datetime(2050, 12, 12) ctx = Context(mycharm, meta={'name': 'local'}) @@ -396,25 +400,25 @@ def test_update_metadata(mycharm): label='babbuccia', description='blu', expire=exp, - rotate=SecretRotate.DAILY, + rotate=ops.SecretRotate.DAILY, ) output = mgr.run() secret_out = output.get_secret(label='babbuccia') assert secret_out.label == 'babbuccia' - assert secret_out.rotate == SecretRotate.DAILY + assert secret_out.rotate == ops.SecretRotate.DAILY assert secret_out.description == 'blu' assert secret_out.expire == exp @pytest.mark.parametrize('leader', (True, False)) -def test_grant_after_add(leader): - class GrantingCharm(CharmBase): - def __init__(self, *args): +def test_grant_after_add(leader: bool): + class GrantingCharm(ops.CharmBase): + def __init__(self, *args: Any): super().__init__(*args) self.framework.observe(self.on.start, self._on_start) - def _on_start(self, _): + def _on_start(self, _: ops.EventBase): if leader: secret = self.app.add_secret({'foo': 'bar'}) else: @@ -426,22 +430,22 @@ def _on_start(self, _): ctx.run(ctx.on.start(), state) -def test_grant_nonowner(mycharm): +def test_grant_nonowner(mycharm: type[ops.CharmBase]): secret = Secret( {'a': 'b'}, label='mylabel', description='foobarbaz', - rotate=SecretRotate.HOURLY, + rotate=ops.SecretRotate.HOURLY, ) secret_id = secret.id - def post_event(charm: CharmBase): + def post_event(charm: ops.CharmBase): secret = charm.model.get_secret(id=secret_id) secret = charm.model.get_secret(label='mylabel') foo = charm.model.get_relation('foo') assert foo is not None - with pytest.raises(ModelError): + with pytest.raises(ops.ModelError): secret.grant(relation=foo) trigger( @@ -457,7 +461,7 @@ def post_event(charm: CharmBase): def test_add_grant_revoke_remove(): - class GrantingCharm(CharmBase): + class GrantingCharm(ops.CharmBase): pass ctx = Context(GrantingCharm, meta={'name': 'foo', 'provides': {'bar': {'interface': 'bar'}}}) @@ -501,12 +505,12 @@ class GrantingCharm(CharmBase): def test_secret_removed_event(): - class SecretCharm(CharmBase): - def __init__(self, framework): + class SecretCharm(ops.CharmBase): + def __init__(self, framework: ops.Framework): super().__init__(framework) self.framework.observe(self.on.secret_remove, self._on_secret_remove) - def _on_secret_remove(self, event): + def _on_secret_remove(self, event: Any): event.secret.remove_revision(event.revision) ctx = Context(SecretCharm, meta={'name': 'foo'}) @@ -521,12 +525,12 @@ def _on_secret_remove(self, event): def test_secret_expired_event(): - class SecretCharm(CharmBase): - def __init__(self, framework): + class SecretCharm(ops.CharmBase): + def __init__(self, framework: ops.Framework): super().__init__(framework) self.framework.observe(self.on.secret_expired, self._on_secret_expired) - def _on_secret_expired(self, event): + def _on_secret_expired(self, event: Any): event.secret.set_content({'password': 'newpass'}) event.secret.remove_revision(event.revision) @@ -542,12 +546,12 @@ def _on_secret_expired(self, event): def test_remove_bad_revision(): - class SecretCharm(CharmBase): - def __init__(self, framework): + class SecretCharm(ops.CharmBase): + def __init__(self, framework: ops.Framework): super().__init__(framework) self.framework.observe(self.on.secret_remove, self._on_secret_remove) - def _on_secret_remove(self, event): + def _on_secret_remove(self, event: Any): with pytest.raises(ValueError): event.secret.remove_revision(event.revision) @@ -564,12 +568,12 @@ def _on_secret_remove(self, event): def test_set_label_on_get(): - class SecretCharm(CharmBase): - def __init__(self, framework): + class SecretCharm(ops.CharmBase): + def __init__(self, framework: ops.Framework): super().__init__(framework) self.framework.observe(self.on.start, self._on_start) - def _on_start(self, _): + def _on_start(self, _: ops.EventBase): id = self.unit.add_secret({'foo': 'bar'}).id secret = self.model.get_secret(id=id, label='label1') assert secret.label == 'label1' @@ -583,7 +587,7 @@ def _on_start(self, _): def test_no_additional_positional_arguments(): with pytest.raises(TypeError): - Secret({}, {}) + Secret({}, {}) # type: ignore def test_default_values(): @@ -605,7 +609,7 @@ def test_add_secret(secrets_context: Context[SecretsCharm]): result = cast('Result', secrets_context.action_results) assert result is not None - assert result['secretid'] + assert result.get('secretid') scenario_secret = next(iter(state.secrets)) @@ -660,8 +664,8 @@ def test_add_secret_with_metadata(secrets_context: Context[SecretsCharm], fields assert scenario_secret.expire == datetime.datetime(2020, 1, 1, 0, 0, 0) assert info['expires'] == datetime.datetime(2020, 1, 1, 0, 0, 0) if 'rotate' in fields: - assert scenario_secret.rotate == SecretRotate.DAILY - assert info['rotation'] == SecretRotate.DAILY + assert scenario_secret.rotate == ops.SecretRotate.DAILY + assert info['rotation'] == ops.SecretRotate.DAILY # https://github.com/canonical/operator/issues/2104 assert info['rotates'] is None @@ -704,7 +708,7 @@ def test_set_secret( if counts['expire']: assert info['expires'] == datetime.datetime(2010 + counts['expire'], 1, 1, 0, 0) if counts['rotate']: - rotation_values = ['sentinel', *SecretRotate.__members__.values()] + rotation_values = ['sentinel', *ops.SecretRotate.__members__.values()] assert info['rotation'] == rotation_values[counts['rotate']] @@ -713,8 +717,10 @@ def common_assertions(scenario_secret: Secret | None, result: Result): assert scenario_secret.owner == 'app' assert not scenario_secret.remote_grants - assert result.get('after') - info = result['after']['info'] + after = result.get('after') + assert after is not None + info = after['info'] + assert info is not None # Verify that the unit and the scaffolding see the same data # # Scenario presents a secret with a full secret URI to the charm @@ -728,5 +734,5 @@ def common_assertions(scenario_secret: Secret | None, result: Result): # https://github.com/canonical/operator/issues/2104 assert info['rotates'] is None - assert scenario_secret.tracked_content == result['after']['tracked'] - assert scenario_secret.latest_content == result['after']['latest'] + assert scenario_secret.tracked_content == after['tracked'] + assert scenario_secret.latest_content == after['latest'] diff --git a/testing/tests/test_e2e/test_status.py b/testing/tests/test_e2e/test_status.py index 4c45241b1..274aff6f0 100644 --- a/testing/tests/test_e2e/test_status.py +++ b/testing/tests/test_e2e/test_status.py @@ -17,28 +17,26 @@ ) import ops -from ops.charm import CharmBase -from ops.framework import Framework from ..helpers import trigger @pytest.fixture(scope='function') -def mycharm(): - class MyCharm(CharmBase): - def __init__(self, framework: Framework): +def mycharm() -> type[ops.CharmBase]: + class MyCharm(ops.CharmBase): + def __init__(self, framework: ops.Framework): super().__init__(framework) for evt in self.on.events().values(): - self.framework.observe(evt, self._on_event) + framework.observe(evt, self._on_event) - def _on_event(self, event): + def _on_event(self, event: ops.EventBase): pass return MyCharm -def test_initial_status(mycharm): - def post_event(charm: CharmBase): +def test_initial_status(mycharm: type[ops.CharmBase]): + def post_event(charm: ops.CharmBase): assert charm.unit.status == UnknownStatus() out = trigger( @@ -52,13 +50,13 @@ def post_event(charm: CharmBase): assert out.unit_status == UnknownStatus() -def test_status_history(mycharm): +def test_status_history(mycharm: type[ops.CharmBase]): class StatusCharm(mycharm): - def __init__(self, framework): + def __init__(self, framework: ops.Framework): super().__init__(framework) framework.observe(self.on.update_status, self._on_update_status) - def _on_update_status(self, _): + def _on_update_status(self, _: ops.EventBase): for obj in (self.unit, self.app): obj.status = ops.ActiveStatus('1') obj.status = ops.BlockedStatus('2') @@ -86,13 +84,13 @@ def _on_update_status(self, _): ] -def test_status_history_preservation(mycharm): +def test_status_history_preservation(mycharm: type[ops.CharmBase]): class StatusCharm(mycharm): - def __init__(self, framework): + def __init__(self, framework: ops.Framework): super().__init__(framework) framework.observe(self.on.update_status, self._on_update_status) - def _on_update_status(self, _): + def _on_update_status(self, _: ops.EventBase): for obj in (self.unit, self.app): obj.status = WaitingStatus('3') @@ -117,21 +115,21 @@ def _on_update_status(self, _): assert ctx.app_status_history == [ActiveStatus('bar')] -def test_workload_history(mycharm): +def test_workload_history(mycharm: type[ops.CharmBase]): class WorkloadCharm(mycharm): - def __init__(self, framework): + def __init__(self, framework: ops.Framework): super().__init__(framework) framework.observe(self.on.install, self._on_install) framework.observe(self.on.start, self._on_start) framework.observe(self.on.update_status, self._on_update_status) - def _on_install(self, _): + def _on_install(self, _: ops.EventBase): self.unit.set_workload_version('1') - def _on_start(self, _): + def _on_start(self, _: ops.EventBase): self.unit.set_workload_version('1.1') - def _on_update_status(self, _): + def _on_update_status(self, _: ops.EventBase): self.unit.set_workload_version('1.2') ctx = Context( @@ -158,7 +156,7 @@ def _on_update_status(self, _): UnknownStatus(), ), ) -def test_status_comparison(status): +def test_status_comparison(status: ops.StatusBase): if isinstance(status, UnknownStatus): ops_status = ops.UnknownStatus() else: @@ -188,12 +186,12 @@ def test_status_comparison(status): ), ) def test_status_success(status: ops.StatusBase): - class MyCharm(CharmBase): - def __init__(self, framework: Framework): + class MyCharm(ops.CharmBase): + def __init__(self, framework: ops.Framework): super().__init__(framework) framework.observe(self.on.update_status, self._on_update_status) - def _on_update_status(self, _): + def _on_update_status(self, _: ops.EventBase): self.unit.status = status ctx = Context(MyCharm, meta={'name': 'foo'}) @@ -208,12 +206,12 @@ def _on_update_status(self, _): ), ) def test_status_error(status: ops.StatusBase, monkeypatch: pytest.MonkeyPatch): - class MyCharm(CharmBase): - def __init__(self, framework: Framework): + class MyCharm(ops.CharmBase): + def __init__(self, framework: ops.Framework): super().__init__(framework) framework.observe(self.on.update_status, self._on_update_status) - def _on_update_status(self, _): + def _on_update_status(self, _: ops.EventBase): self.unit.status = status monkeypatch.setenv('SCENARIO_BARE_CHARM_ERRORS', 'false') diff --git a/testing/tests/test_e2e/test_storage.py b/testing/tests/test_e2e/test_storage.py index 604be52f3..820248d58 100644 --- a/testing/tests/test_e2e/test_storage.py +++ b/testing/tests/test_e2e/test_storage.py @@ -24,22 +24,22 @@ class MyCharmWithoutStorage(CharmBase): @pytest.fixture -def storage_ctx(): - return Context(MyCharmWithStorage, meta=MyCharmWithStorage.META) +def storage_ctx() -> Context[MyCharmWithStorage]: + return Context(MyCharmWithStorage, meta=dict(MyCharmWithStorage.META)) @pytest.fixture -def no_storage_ctx(): - return Context(MyCharmWithoutStorage, meta=MyCharmWithoutStorage.META) +def no_storage_ctx() -> Context[MyCharmWithoutStorage]: + return Context(MyCharmWithoutStorage, meta=dict(MyCharmWithoutStorage.META)) -def test_storage_get_null(no_storage_ctx): +def test_storage_get_null(no_storage_ctx: Context[MyCharmWithoutStorage]): with no_storage_ctx(no_storage_ctx.on.update_status(), State()) as mgr: storages = mgr.charm.model.storages assert not len(storages) -def test_storage_get_unknown_name(storage_ctx): +def test_storage_get_unknown_name(storage_ctx: Context[MyCharmWithStorage]): with storage_ctx(storage_ctx.on.update_status(), State()) as mgr: storages = mgr.charm.model.storages # not in metadata @@ -47,7 +47,7 @@ def test_storage_get_unknown_name(storage_ctx): storages['bar'] -def test_storage_request_unknown_name(storage_ctx): +def test_storage_request_unknown_name(storage_ctx: Context[MyCharmWithStorage]): with storage_ctx(storage_ctx.on.update_status(), State()) as mgr: storages = mgr.charm.model.storages # not in metadata @@ -55,7 +55,7 @@ def test_storage_request_unknown_name(storage_ctx): storages.request('bar') -def test_storage_get_some(storage_ctx): +def test_storage_get_some(storage_ctx: Context[MyCharmWithStorage]): with storage_ctx(storage_ctx.on.update_status(), State()) as mgr: storages = mgr.charm.model.storages # known but none attached @@ -63,7 +63,7 @@ def test_storage_get_some(storage_ctx): @pytest.mark.parametrize('n', (1, 3, 5)) -def test_storage_add(storage_ctx, n): +def test_storage_add(storage_ctx: Context[MyCharmWithStorage], n: int): with storage_ctx(storage_ctx.on.update_status(), State()) as mgr: storages = mgr.charm.model.storages storages.request('foo', n) @@ -71,10 +71,10 @@ def test_storage_add(storage_ctx, n): assert storage_ctx.requested_storages['foo'] == n -def test_storage_usage(storage_ctx): +def test_storage_usage(storage_ctx: Context[MyCharmWithStorage]): storage = Storage('foo') # setup storage with some content - (storage.get_filesystem(storage_ctx) / 'myfile.txt').write_text('helloworld') + (storage.get_filesystem(storage_ctx) / 'myfile.txt').write_text('helloworld') # type: ignore with storage_ctx(storage_ctx.on.update_status(), State(storages={storage})) as mgr: foo = mgr.charm.model.storages['foo'][0] @@ -87,14 +87,16 @@ def test_storage_usage(storage_ctx): myfile.write_text('helloworlds') # post-mortem: inspect fs contents. - assert (storage.get_filesystem(storage_ctx) / 'path.py').read_text() == 'helloworlds' + assert ( + storage.get_filesystem(storage_ctx) / 'path.py' # type: ignore + ).read_text() == 'helloworlds' -def test_storage_attached_event(storage_ctx): +def test_storage_attached_event(storage_ctx: Context[MyCharmWithStorage]): storage = Storage('foo') storage_ctx.run(storage_ctx.on.storage_attached(storage), State(storages={storage})) -def test_storage_detaching_event(storage_ctx): +def test_storage_detaching_event(storage_ctx: Context[MyCharmWithStorage]): storage = Storage('foo') storage_ctx.run(storage_ctx.on.storage_detaching(storage), State(storages={storage}))