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..db185c68f 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[CharmBase]) -> pathlib.Path: """Simulated Pebble filesystem in this context. Returns: diff --git a/testing/tests/test_e2e/test_juju_log.py b/testing/tests/test_e2e/test_juju_log.py index 2e18446f0..061fd1dd7 100644 --- a/testing/tests/test_e2e/test_juju_log.py +++ b/testing/tests/test_e2e/test_juju_log.py @@ -11,23 +11,23 @@ from scenario import Context from scenario.state import JujuLogLine, State -from ops.charm import CharmBase, CollectStatusEvent +import ops logger = logging.getLogger('testing logger') @pytest.fixture(scope='function') def mycharm(): - class MyCharm(CharmBase): + class MyCharm(ops.CharmBase): META: Mapping[str, Any] = {'name': 'mycharm'} - def __init__(self, framework): + 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): - if isinstance(event, CollectStatusEvent): + def _on_event(self, event: ops.EventBase): + if isinstance(event, ops.CollectStatusEvent): return print('foo!') logger.warning('bar!') @@ -35,7 +35,7 @@ def _on_event(self, event): return MyCharm -def test_juju_log(mycharm): +def test_juju_log(mycharm: Any): ctx = Context(mycharm, meta=mycharm.META) ctx.run(ctx.on.start(), State()) assert JujuLogLine(level='DEBUG', message='Emitting Juju event start.') in ctx.juju_log diff --git a/testing/tests/test_e2e/test_manager.py b/testing/tests/test_e2e/test_manager.py index 816d28acb..e68ad06cd 100644 --- a/testing/tests/test_e2e/test_manager.py +++ b/testing/tests/test_e2e/test_manager.py @@ -8,33 +8,33 @@ import pytest from scenario import Context, State -from scenario.context import AlreadyEmittedError, Manager +from scenario.context import Manager +from scenario.errors import AlreadyEmittedError -from ops import ActiveStatus -from ops.charm import CharmBase, CollectStatusEvent +import ops @pytest.fixture(scope='function') def mycharm(): - class MyCharm(CharmBase): + class MyCharm(ops.CharmBase): META: Mapping[str, Any] = {'name': 'mycharm'} ACTIONS: Mapping[str, Any] = {'do-x': {}} - def __init__(self, framework): + 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, e): - if isinstance(e, CollectStatusEvent): + def _on_event(self, e: ops.EventBase): + if isinstance(e, ops.CollectStatusEvent): return - self.unit.status = ActiveStatus(e.handle.kind) + self.unit.status = ops.ActiveStatus(e.handle.kind) return MyCharm -def test_manager(mycharm): +def test_manager(mycharm: Any): ctx = Context(mycharm, meta=mycharm.META) with Manager(ctx, ctx.on.start(), State()) as manager: assert isinstance(manager.charm, mycharm) @@ -43,7 +43,7 @@ def test_manager(mycharm): assert isinstance(state_out, State) -def test_manager_implicit(mycharm): +def test_manager_implicit(mycharm: Any): ctx = Context(mycharm, meta=mycharm.META) with Manager(ctx, ctx.on.start(), State()) as manager: assert isinstance(manager.charm, mycharm) @@ -53,7 +53,7 @@ def test_manager_implicit(mycharm): assert manager._emitted -def test_manager_reemit_fails(mycharm): +def test_manager_reemit_fails(mycharm: Any): ctx = Context(mycharm, meta=mycharm.META) with Manager(ctx, ctx.on.start(), State()) as manager: manager.run() @@ -61,7 +61,7 @@ def test_manager_reemit_fails(mycharm): manager.run() -def test_context_manager(mycharm): +def test_context_manager(mycharm: Any): ctx = Context(mycharm, meta=mycharm.META) with ctx(ctx.on.start(), State()) as manager: state_out = manager.run() @@ -69,7 +69,7 @@ def test_context_manager(mycharm): assert ctx.emitted_events[0].handle.kind == 'start' -def test_context_action_manager(mycharm): +def test_context_action_manager(mycharm: Any): ctx = Context(mycharm, meta=mycharm.META, actions=mycharm.ACTIONS) with ctx(ctx.on.action('do-x'), State()) as manager: state_out = manager.run() diff --git a/testing/tests/test_e2e/test_pebble.py b/testing/tests/test_e2e/test_pebble.py index b3bb52243..3411c7b98 100644 --- a/testing/tests/test_e2e/test_pebble.py +++ b/testing/tests/test_e2e/test_pebble.py @@ -6,37 +6,38 @@ 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): +def charm_cls() -> 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_no_containers(charm_cls): - def callback(self: CharmBase): +def test_no_containers(charm_cls: type[ops.CharmBase]): + def callback(self: ops.CharmBase): assert not self.unit.containers trigger( @@ -48,8 +49,8 @@ def callback(self: CharmBase): ) -def test_containers_from_meta(charm_cls): - def callback(self: CharmBase): +def test_containers_from_meta(charm_cls: type[ops.CharmBase]): + def callback(self: ops.CharmBase): assert self.unit.containers assert self.unit.get_container('foo') @@ -63,8 +64,8 @@ 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(charm_cls: type[ops.CharmBase], can_connect: bool): + def callback(self: ops.CharmBase): assert can_connect == self.unit.get_container('foo').can_connect() trigger( @@ -76,13 +77,13 @@ def callback(self: CharmBase): ) -def test_fs_push(tmp_path, charm_cls): +def test_fs_push(tmp_path: pathlib.Path, charm_cls: type[ops.CharmBase]): 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 @@ -105,10 +106,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, charm_cls: type[ops.CharmBase], 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 +117,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( @@ -143,12 +144,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 +190,12 @@ def callback(self: CharmBase): ('ps', PS), ), ) -def test_exec(charm_cls, cmd, out): - def callback(self: CharmBase): +def test_exec(charm_cls: type[ops.CharmBase], 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( @@ -221,15 +223,16 @@ def callback(self: CharmBase): [io.StringIO('hello world!'), None], ), ) -def test_exec_history_stdin(stdin, write): - class MyCharm(CharmBase): - def __init__(self, framework: Framework): +def test_exec_history_stdin(stdin: str | io.StringIO | None, write: str | None): + class MyCharm(ops.CharmBase): + def __init__(self, framework: ops.Framework): super().__init__(framework) self.framework.observe(self.on.foo_pebble_ready, self._on_ready) - def _on_ready(self, _): + def _on_ready(self, _: ops.EventBase): proc = self.unit.get_container('foo').exec(['ls'], stdin=stdin) if write: + assert proc.stdin is not None proc.stdin.write(write) proc.wait() @@ -239,8 +242,8 @@ def _on_ready(self, _): assert ctx.exec_history[container.name][0].stdin == 'hello world!' -def test_pebble_ready(charm_cls): - def callback(self: CharmBase): +def test_pebble_ready(charm_cls: type[ops.CharmBase]): + def callback(self: ops.CharmBase): foo = self.unit.get_container('foo') assert foo.can_connect() @@ -255,20 +258,22 @@ def callback(self: CharmBase): ) -@pytest.mark.parametrize('starting_service_status', pebble.ServiceStatus) -def test_pebble_plan(charm_cls, starting_service_status): +@pytest.mark.parametrize('starting_service_status', ops.pebble.ServiceStatus) +def test_pebble_plan( + charm_cls: type[ops.CharmBase], starting_service_status: ops.pebble.ServiceStatus +): class PlanCharm(charm_cls): - def __init__(self, framework): + def __init__(self, framework: ops.Framework): super().__init__(framework) framework.observe(self.on.foo_pebble_ready, self._on_ready) - def _on_ready(self, event): + 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 == ServiceStartup.ENABLED - assert fooserv.current == ServiceStatus.ACTIVE + assert fooserv.startup == ops.pebble.ServiceStartup.ENABLED + assert fooserv.current == ops.pebble.ServiceStatus.ACTIVE foo.add_layer( 'bar', @@ -290,20 +295,20 @@ def _on_ready(self, event): 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 + assert foo.get_service('barserv').current == ops.pebble.ServiceStatus.ACTIVE 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 +321,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(charm_cls: type[ops.CharmBase]): state = State( containers={ Container( @@ -346,13 +351,13 @@ def test_exec_wait_error(charm_cls): 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(charm_cls: type[ops.CharmBase], command: list[str]): state = State( containers={ Container( @@ -373,7 +378,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(charm_cls: type[ops.CharmBase]): state = State( containers={ Container( @@ -388,11 +393,11 @@ def test_exec_wait_output_error(charm_cls): 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(charm_cls: type[ops.CharmBase]): notices = [ Notice(key='example.com/foo'), Notice(key='example.com/bar', last_data={'a': 'b'}), @@ -422,14 +427,14 @@ def test_pebble_custom_notice_in_charm(): repeat_after = datetime.timedelta(days=7) expire_after = datetime.timedelta(days=365) - class MyCharm(CharmBase): - def __init__(self, framework): + class MyCharm(ops.CharmBase): + 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: PebbleCustomNoticeEvent): + def _on_custom_notice(self, event: ops.PebbleCustomNoticeEvent): notice = event.notice - assert notice.type == pebble.NoticeType.CUSTOM + assert notice.type == ops.pebble.NoticeType.CUSTOM assert notice.key == key assert notice.last_data == data assert notice.user_id == user_id @@ -466,18 +471,18 @@ def _on_custom_notice(self, event: PebbleCustomNoticeEvent): def test_pebble_check_failed(): - infos = [] + infos: list[ops.LazyCheckInfo] = [] - class MyCharm(CharmBase): - def __init__(self, framework): + class MyCharm(ops.CharmBase): + def __init__(self, framework: ops.Framework): super().__init__(framework) framework.observe(self.on.foo_pebble_check_failed, self._on_check_failed) - def _on_check_failed(self, event): + def _on_check_failed(self, event: ops.PebbleCheckFailedEvent): infos.append(event.info) ctx = Context(MyCharm, meta={'name': 'foo', 'containers': {'foo': {}}}) - 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 @@ -485,9 +490,9 @@ 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}) @@ -495,33 +500,33 @@ def _on_check_failed(self, event): ctx.run(ctx.on.pebble_check_failed(container, check), state=state) 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 = [] + infos: list[ops.LazyCheckInfo] = [] - class MyCharm(CharmBase): - def __init__(self, framework): + class MyCharm(ops.CharmBase): + 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): + def _on_check_recovered(self, event: ops.PebbleCheckRecoveredEvent): infos.append(event.info) ctx = Context(MyCharm, meta={'name': 'foo', 'containers': {'foo': {}}}) - 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', 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}) @@ -529,39 +534,39 @@ def _on_check_recovered(self, event): ctx.run(ctx.on.pebble_check_recovered(container, check), state=state) 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 = [] + foo_infos: list[ops.LazyCheckInfo] = [] + bar_infos: list[ops.LazyCheckInfo] = [] - 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.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): + def _on_foo_check_failed(self, event: ops.PebbleCheckFailedEvent): foo_infos.append(event.info) - def _on_bar_check_failed(self, event): + def _on_bar_check_failed(self, event: ops.PebbleCheckFailedEvent): bar_infos.append(event.info) ctx = Context(MyCharm, 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}) @@ -570,19 +575,19 @@ def _on_bar_check_failed(self, event): ctx.run(ctx.on.pebble_check_failed(foo_container, check), state=state) 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): + class MyCharm(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, _): + def _on_foo_ready(self, _: ops.EventBase): self.unit.get_container('foo').add_layer( 'foo', {'checks': {'chk1': {'override': 'replace'}}}, @@ -592,17 +597,17 @@ def _on_foo_ready(self, _): 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): + class MyCharm(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, _): + def _on_foo_ready(self, _: ops.EventBase): container = self.unit.get_container('foo') container.add_layer( 'foo', @@ -617,7 +622,7 @@ def _on_foo_ready(self, _): }, ) - def _on_config_changed(self, _): + def _on_config_changed(self, _: ops.EventBase): container = self.unit.get_container('foo') container.start_checks('chk1') @@ -627,12 +632,12 @@ def _on_config_changed(self, _): # 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 @@ -644,26 +649,26 @@ def reset_security_logging(): def test_pebble_stop_check(reset_security_logging: None): - 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.config_changed, self._on_config_changed) - def _on_config_changed(self, _): + def _on_config_changed(self, _: ops.EventBase): container = self.unit.get_container('foo') container.stop_checks('chk1') ctx = Context(MyCharm, 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 +679,29 @@ 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): + class MyCharm(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, _): + def _on_config_changed(self, _: ops.EventBase): container = self.unit.get_container('foo') container.replan() ctx = Context(MyCharm, 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.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 +712,7 @@ 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 @pytest.mark.parametrize( @@ -742,20 +747,18 @@ def _on_config_changed(self, _): }, ], ) -def test_add_layer_merge_check( - new_layer_name: str, combine: bool, new_layer_dict: pebble.LayerDict -): - class MyCharm(CharmBase): - def __init__(self, framework: Framework): +def test_add_layer_merge_check(new_layer_name: str, combine: bool, new_layer_dict: LayerDict): + class MyCharm(ops.CharmBase): + 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, _: PebbleReadyEvent): + def _on_pebble_ready(self, _: ops.PebbleReadyEvent): container = self.unit.get_container('my-container') - container.add_layer(new_layer_name, Layer(new_layer_dict), combine=combine) + container.add_layer(new_layer_name, ops.pebble.Layer(new_layer_dict), combine=combine) ctx = Context(MyCharm, meta={'name': 'foo', 'containers': {'my-container': {}}}) - layer_in = pebble.Layer({ + layer_in = ops.pebble.Layer({ 'checks': { 'server-ready': { 'override': 'replace', @@ -769,9 +772,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 +782,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 +812,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 +826,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 +847,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 +866,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..f2c69c01f 100644 --- a/testing/tests/test_e2e/test_ports.py +++ b/testing/tests/test_e2e/test_ports.py @@ -8,7 +8,8 @@ import pytest from scenario import Context, State -from scenario.state import Port, StateValidationError, TCPPort, UDPPort +from scenario.errors import StateValidationError +from scenario.state import Port, TCPPort, UDPPort from ops import CharmBase, Framework, StartEvent, StopEvent @@ -30,20 +31,21 @@ def _close_port(self, _: StopEvent): @pytest.fixture -def ctx(): - return Context(MyCharm, meta=MyCharm.META) +def ctx() -> Context[MyCharm]: + return Context(MyCharm, meta=dict(MyCharm.META)) -def test_open_port(ctx): +def test_open_port(ctx: Context[MyCharm]): 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[MyCharm]): out = ctx.run(ctx.on.stop(), State(opened_ports={TCPPort(42)})) assert not out.opened_ports @@ -54,7 +56,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}))