From 6557b103eba401748aff1d1ef9611f790fa4db30 Mon Sep 17 00:00:00 2001 From: Tiexin Guo Date: Thu, 16 May 2024 11:34:35 +0200 Subject: [PATCH] test: refactor test_main.py and test_storage.py (#1199) Refactor test_storage.py and test_main.py to pytest style. --- .github/workflows/framework-tests.yaml | 1 + test/test_helpers.py | 35 -- test/test_main.py | 478 +++++++++++++++---------- test/test_storage.py | 167 +++++---- 4 files changed, 389 insertions(+), 292 deletions(-) diff --git a/.github/workflows/framework-tests.yaml b/.github/workflows/framework-tests.yaml index 4832ba8c6..2a89c85e6 100644 --- a/.github/workflows/framework-tests.yaml +++ b/.github/workflows/framework-tests.yaml @@ -103,6 +103,7 @@ jobs: python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"] exclude: - {python-version: "3.8", os: "macos-latest"} # macos-14 is arm64, and there's no Python 3.8 build for arm64 + - {python-version: "3.9", os: "macos-latest"} # macos-14 is arm64, and there's no Python 3.9 build for arm64 steps: - uses: actions/checkout@v3 diff --git a/test/test_helpers.py b/test/test_helpers.py index 38e52e92a..ee75268f9 100644 --- a/test/test_helpers.py +++ b/test/test_helpers.py @@ -201,38 +201,3 @@ def test_fake_script_clear(self): assert fake_script_calls(self, clear=True) == [['bar', 'd e', 'f']] assert fake_script_calls(self, clear=True) == [] - - -class BaseTestCase(unittest.TestCase): - - def create_framework(self, - *, - model: typing.Optional[ops.Model] = None, - tmpdir: typing.Optional[pathlib.Path] = None): - """Create a Framework object. - - By default operate in-memory; pass a temporary directory via the 'tmpdir' - parameter if you wish to instantiate several frameworks sharing the - same dir (e.g. for storing state). - """ - if tmpdir is None: - data_fpath = ":memory:" - charm_dir = 'non-existant' - else: - data_fpath = tmpdir / "framework.data" - charm_dir = tmpdir - - framework = ops.Framework( - SQLiteStorage(data_fpath), - charm_dir, - meta=model._cache._meta if model else ops.CharmMeta(), - model=model) # type: ignore - self.addCleanup(framework.close) - return framework - - def create_model(self): - """Create a Model object.""" - backend = _ModelBackend(unit_name='myapp/0') - meta = ops.CharmMeta() - model = ops.Model(meta, backend) - return model diff --git a/test/test_main.py b/test/test_main.py index e81f343ee..a13dd0bf1 100644 --- a/test/test_main.py +++ b/test/test_main.py @@ -23,7 +23,6 @@ import sys import tempfile import typing -import unittest import warnings from pathlib import Path from unittest.mock import patch @@ -35,7 +34,7 @@ from ops.storage import SQLiteStorage from .charms.test_main.src.charm import MyCharmEvents -from .test_helpers import fake_script, fake_script_calls +from .test_helpers import FakeScript # This relies on the expected repository structure to find a path to # source of the charm under test. @@ -54,24 +53,25 @@ class SymlinkTargetError(Exception): class EventSpec: - def __init__(self, - event_type: typing.Type[ops.EventBase], - event_name: str, - env_var: typing.Optional[str] = None, - relation_id: typing.Optional[int] = None, - remote_app: typing.Optional[str] = None, - remote_unit: typing.Optional[str] = None, - model_name: typing.Optional[str] = None, - set_in_env: typing.Optional[typing.Dict[str, - str]] = None, - workload_name: typing.Optional[str] = None, - notice_id: typing.Optional[str] = None, - notice_type: typing.Optional[str] = None, - notice_key: typing.Optional[str] = None, - departing_unit_name: typing.Optional[str] = None, - secret_id: typing.Optional[str] = None, - secret_label: typing.Optional[str] = None, - secret_revision: typing.Optional[str] = None): + def __init__( + self, + event_type: typing.Type[ops.EventBase], + event_name: str, + env_var: typing.Optional[str] = None, + relation_id: typing.Optional[int] = None, + remote_app: typing.Optional[str] = None, + remote_unit: typing.Optional[str] = None, + model_name: typing.Optional[str] = None, + set_in_env: typing.Optional[typing.Dict[str, str]] = None, + workload_name: typing.Optional[str] = None, + notice_id: typing.Optional[str] = None, + notice_type: typing.Optional[str] = None, + notice_key: typing.Optional[str] = None, + departing_unit_name: typing.Optional[str] = None, + secret_id: typing.Optional[str] = None, + secret_label: typing.Optional[str] = None, + secret_revision: typing.Optional[str] = None, + ): self.event_type = event_type self.event_name = event_name self.env_var = env_var @@ -93,7 +93,7 @@ def __init__(self, @patch('ops.main.setup_root_logging', new=lambda *a, **kw: None) # type: ignore @patch('ops.main._emit_charm_event', new=lambda *a, **kw: None) # type: ignore @patch('ops.charm._evaluate_status', new=lambda *a, **kw: None) # type: ignore -class CharmInitTestCase(unittest.TestCase): +class TestCharmInit: @patch('sys.stderr', new_callable=io.StringIO) def test_breakpoint(self, fake_stderr: io.StringIO): @@ -118,11 +118,11 @@ class MyCharm(ops.CharmBase): assert mock.call_count == 0 def _check( - self, - charm_class: typing.Type[ops.CharmBase], - *, - extra_environ: typing.Optional[typing.Dict[str, str]] = None, - **kwargs: typing.Any + self, + charm_class: typing.Type[ops.CharmBase], + *, + extra_environ: typing.Optional[typing.Dict[str, str]] = None, + **kwargs: typing.Any, ): """Helper for below tests.""" fake_environ = { @@ -146,8 +146,8 @@ def _check( def test_init_signature_passthrough(self): class MyCharm(ops.CharmBase): - def __init__(self, *args): # type: ignore - super().__init__(*args) # type: ignore + def __init__(self, framework: ops.Framework): + super().__init__(framework) with warnings.catch_warnings(record=True) as warn_cm: self._check(MyCharm) @@ -192,7 +192,7 @@ def test_storage_with_storage(self): def test_controller_storage_deprecated(self): with patch('ops.storage.juju_backend_available') as juju_backend_available: juju_backend_available.return_value = True - with self.assertWarnsRegex(DeprecationWarning, 'Controller storage'): + with pytest.warns(DeprecationWarning, match='Controller storage'): with pytest.raises(FileNotFoundError, match='state-get'): self._check(ops.CharmBase, use_juju_for_storage=True) @@ -200,7 +200,7 @@ def test_controller_storage_deprecated(self): @patch('sys.argv', new=("hooks/config-changed",)) @patch('ops.main._Manager._setup_root_logging', new=lambda *a, **kw: None) # type: ignore @patch('ops.charm._evaluate_status', new=lambda *a, **kw: None) # type: ignore -class TestDispatch(unittest.TestCase): +class TestDispatch: def _check(self, *, with_dispatch: bool = False, dispatch_path: str = ''): """Helper for below tests.""" class MyCharm(ops.CharmBase): @@ -257,6 +257,11 @@ def test_with_dispatch_path_but_no_dispatch(self): typing.Dict[str, typing.Union[str, int, None]]]] +@pytest.fixture +def fake_script(request: pytest.FixtureRequest): + return FakeScript(request) + + class _TestMain(abc.ABC): @abc.abstractmethod @@ -270,11 +275,17 @@ def _setup_entry_point(self, directory: Path, entry_point: str): return NotImplemented @abc.abstractmethod - def _call_event(self, rel_path: Path, env: typing.Dict[str, str]): + def _call_event( + self, + fake_script: FakeScript, + rel_path: Path, + env: typing.Dict[str, str], + ): """Set up the environment and call (i.e. run) the given event.""" return NotImplemented @abc.abstractmethod + @pytest.mark.usefixtures("setup_charm") def test_setup_event_links(self): """Test auto-creation of symlinks caused by initial events. @@ -283,19 +294,9 @@ def test_setup_event_links(self): """ return NotImplemented - addCleanup = unittest.TestCase.addCleanup # noqa - assertEqual = unittest.TestCase.assertEqual # noqa - assertFalse = unittest.TestCase.assertFalse # noqa - assertIn = unittest.TestCase.assertIn # noqa - assertIsNotNone = unittest.TestCase.assertIsNotNone # noqa - assertRaises = unittest.TestCase.assertRaises # noqa - assertRegex = unittest.TestCase.assertRegex # noqa - assertNotIn = unittest.TestCase.assertNotIn # noqa - fail = unittest.TestCase.fail - subTest = unittest.TestCase.subTest # noqa - - def setUp(self): - self._setup_charm_dir() + @pytest.fixture + def setup_charm(self, request: pytest.FixtureRequest, fake_script: FakeScript): + self._setup_charm_dir(request) # Relations events are defined dynamically and modify the class attributes. # We use a subclass temporarily to prevent these side effects from leaking. @@ -305,18 +306,23 @@ class TestCharmEvents(ops.CharmEvents): def cleanup(): ops.CharmBase.on = ops.CharmEvents() # type: ignore - self.addCleanup(cleanup) + request.addfinalizer(cleanup) - fake_script(typing.cast(unittest.TestCase, self), 'is-leader', 'echo true') - fake_script(typing.cast(unittest.TestCase, self), 'juju-log', 'exit 0') + fake_script.write('is-leader', 'echo true') + fake_script.write('juju-log', 'exit 0') # set to something other than None for tests that care self.stdout = None self.stderr = None - def _setup_charm_dir(self): + def _setup_charm_dir(self, request: pytest.FixtureRequest): self._tmpdir = Path(tempfile.mkdtemp(prefix='tmp-ops-test-')).resolve() - self.addCleanup(shutil.rmtree, str(self._tmpdir)) + + def cleanup(): + shutil.rmtree(self._tmpdir, ignore_errors=True) + + request.addfinalizer(cleanup) + self.JUJU_CHARM_DIR = self._tmpdir / 'test_main' self._charm_state_file = self.JUJU_CHARM_DIR / '.unit-state.db' self.hooks_dir = self.JUJU_CHARM_DIR / 'hooks' @@ -347,9 +353,10 @@ def _prepare_actions(self): for action_name in ('start', 'foo-bar', 'get-model-name', 'get-status', 'keyerror'): self._setup_entry_point(actions_dir, action_name) - def _read_and_clear_state(self, - event_name: str) -> typing.Union[ops.BoundStoredState, - ops.StoredStateData]: + def _read_and_clear_state( + self, + event_name: str, + ) -> typing.Union[ops.BoundStoredState, ops.StoredStateData]: if self._charm_state_file.stat().st_size: storage = SQLiteStorage(self._charm_state_file) with (self.JUJU_CHARM_DIR / 'metadata.yaml').open() as m: @@ -381,7 +388,7 @@ class Charm(self.charm_module.Charm, _HasStored): # type: ignore stored = ops.StoredStateData(None, None) # type: ignore return stored - def _simulate_event(self, event_spec: EventSpec): + def _simulate_event(self, fake_script: FakeScript, event_spec: EventSpec): ppath = Path(__file__).parent pypath = str(ppath.parent) if 'PYTHONPATH' in os.environ: @@ -460,47 +467,62 @@ def _simulate_event(self, event_spec: EventSpec): if event_spec.model_name is not None: env['JUJU_MODEL_NAME'] = event_spec.model_name - self._call_event(Path(event_dir, event_filename), env) + self._call_event(fake_script, Path(event_dir, event_filename), env) return self._read_and_clear_state(event_spec.event_name) - def test_event_reemitted(self): + @pytest.mark.usefixtures("setup_charm") + def test_event_reemitted(self, fake_script: FakeScript): # First run "install" to make sure all hooks are set up. - state = self._simulate_event(EventSpec(ops.InstallEvent, 'install')) + state = self._simulate_event(fake_script, EventSpec(ops.InstallEvent, 'install')) assert isinstance(state, ops.BoundStoredState) assert list(state.observed_event_types) == ['InstallEvent'] - state = self._simulate_event(EventSpec(ops.ConfigChangedEvent, 'config-changed')) + state = self._simulate_event( + fake_script, + EventSpec(ops.ConfigChangedEvent, 'config-changed') + ) assert isinstance(state, ops.BoundStoredState) assert list(state.observed_event_types) == ['ConfigChangedEvent'] # Re-emit should pick the deferred config-changed. - state = self._simulate_event(EventSpec(ops.UpdateStatusEvent, 'update-status')) + state = self._simulate_event( + fake_script, + EventSpec(ops.UpdateStatusEvent, 'update-status') + ) assert isinstance(state, ops.BoundStoredState) assert list(state.observed_event_types) == \ ['ConfigChangedEvent', 'UpdateStatusEvent'] - def test_no_reemission_on_collect_metrics(self): - fake_script(typing.cast(unittest.TestCase, self), 'add-metric', 'exit 0') + @pytest.mark.usefixtures("setup_charm") + def test_no_reemission_on_collect_metrics(self, fake_script: FakeScript): + fake_script.write('add-metric', 'exit 0') # First run "install" to make sure all hooks are set up. - state = self._simulate_event(EventSpec(ops.InstallEvent, 'install')) + state = self._simulate_event(fake_script, EventSpec(ops.InstallEvent, 'install')) assert isinstance(state, ops.BoundStoredState) assert list(state.observed_event_types) == ['InstallEvent'] - state = self._simulate_event(EventSpec(ops.ConfigChangedEvent, 'config-changed')) + state = self._simulate_event( + fake_script, + EventSpec(ops.ConfigChangedEvent, 'config-changed') + ) assert isinstance(state, ops.BoundStoredState) assert list(state.observed_event_types) == ['ConfigChangedEvent'] # Re-emit should not pick the deferred config-changed because # collect-metrics runs in a restricted context. - state = self._simulate_event(EventSpec(ops.CollectMetricsEvent, 'collect-metrics')) + state = self._simulate_event( + fake_script, + EventSpec(ops.CollectMetricsEvent, 'collect-metrics') + ) assert isinstance(state, ops.BoundStoredState) assert list(state.observed_event_types) == ['CollectMetricsEvent'] - def test_multiple_events_handled(self): + @pytest.mark.usefixtures("setup_charm") + def test_multiple_events_handled(self, fake_script: FakeScript): self._prepare_actions() - fake_script(typing.cast(unittest.TestCase, self), 'action-get', "echo '{}'") + fake_script.write('action-get', "echo '{}'") # Sample events with a different amount of dashes used # and with endpoints from different sections of metadata.yaml @@ -643,11 +665,11 @@ def test_multiple_events_handled(self): logger.debug('Expected events %s', events_under_test) # First run "install" to make sure all hooks are set up. - self._simulate_event(EventSpec(ops.InstallEvent, 'install')) + self._simulate_event(fake_script, EventSpec(ops.InstallEvent, 'install')) # Simulate hook executions for every event. for event_spec, expected_event_data in events_under_test: - state = self._simulate_event(event_spec) + state = self._simulate_event(fake_script, event_spec) assert isinstance(state, ops.BoundStoredState) state_key = f"on_{event_spec.event_name}" @@ -665,7 +687,8 @@ def test_multiple_events_handled(self): assert getattr(state, f"{event_spec.event_name}_data") == \ expected_event_data - def test_event_not_implemented(self): + @pytest.mark.usefixtures("setup_charm") + def test_event_not_implemented(self, fake_script: FakeScript): """Make sure events without implementation do not cause non-zero exit.""" # Simulate a scenario where there is a symlink for an event that # a charm does not know how to handle. @@ -674,25 +697,34 @@ def test_event_not_implemented(self): hook_path.symlink_to('install') try: - self._simulate_event(EventSpec(ops.HookEvent, 'not-implemented-event')) + self._simulate_event( + fake_script, + EventSpec(ops.HookEvent, 'not-implemented-event') + ) except subprocess.CalledProcessError: - self.fail('Event simulation for an unsupported event' - ' results in a non-zero exit code returned') + pytest.fail('Event simulation for an unsupported event' + ' results in a non-zero exit code returned') - def test_no_actions(self): + @pytest.mark.usefixtures("setup_charm") + def test_no_actions(self, fake_script: FakeScript): (self.JUJU_CHARM_DIR / 'actions.yaml').unlink() - self._simulate_event(EventSpec(ops.InstallEvent, 'install')) + self._simulate_event(fake_script, EventSpec(ops.InstallEvent, 'install')) - def test_empty_actions(self): + @pytest.mark.usefixtures("setup_charm") + def test_empty_actions(self, fake_script: FakeScript): (self.JUJU_CHARM_DIR / 'actions.yaml').write_text('') - self._simulate_event(EventSpec(ops.InstallEvent, 'install')) + self._simulate_event(fake_script, EventSpec(ops.InstallEvent, 'install')) - def test_collect_metrics(self): - fake_script(typing.cast(unittest.TestCase, self), 'add-metric', 'exit 0') - self._simulate_event(EventSpec(ops.InstallEvent, 'install')) + @pytest.mark.usefixtures("setup_charm") + def test_collect_metrics(self, fake_script: FakeScript): + fake_script.write('add-metric', 'exit 0') + self._simulate_event(fake_script, EventSpec(ops.InstallEvent, 'install')) # Clear the calls during 'install' - fake_script_calls(typing.cast(unittest.TestCase, self), clear=True) - self._simulate_event(EventSpec(ops.CollectMetricsEvent, 'collect_metrics')) + fake_script.calls(clear=True) + self._simulate_event( + fake_script, + EventSpec(ops.CollectMetricsEvent, 'collect_metrics') + ) expected = [ VERSION_LOGLINE, @@ -700,18 +732,25 @@ def test_collect_metrics(self): ['add-metric', '--labels', 'bar=4.2', 'foo=42'], ['is-leader', '--format=json'], ] - calls = fake_script_calls(typing.cast(unittest.TestCase, self)) + calls = fake_script.calls() assert calls == expected - def test_custom_event(self): - self._simulate_event(EventSpec(ops.InstallEvent, 'install')) + @pytest.mark.usefixtures("setup_charm") + def test_custom_event(self, fake_script: FakeScript): + self._simulate_event(fake_script, EventSpec(ops.InstallEvent, 'install')) # Clear the calls during 'install' - fake_script_calls(typing.cast(unittest.TestCase, self), clear=True) - self._simulate_event(EventSpec(ops.UpdateStatusEvent, 'update-status', - set_in_env={'EMIT_CUSTOM_EVENT': "1"})) + fake_script.calls(clear=True) + self._simulate_event( + fake_script, + EventSpec( + ops.UpdateStatusEvent, + 'update-status', + set_in_env={'EMIT_CUSTOM_EVENT': "1"} + ) + ) - calls = fake_script_calls(typing.cast(unittest.TestCase, self)) + calls = fake_script.calls() custom_event_prefix = 'Emitting custom event ops.Framework: + def create_framework( + self, + request: pytest.FixtureRequest, + fake_script: FakeScript, + ) -> ops.Framework: """Create a Framework that we can use to test the backend storage.""" - return ops.Framework(self.create_storage(), None, None, None) # type: ignore + storage = self.create_storage(request, fake_script) + return ops.Framework(storage, None, None, None) # type: ignore @abc.abstractmethod - def create_storage(self) -> ops.storage.SQLiteStorage: + def create_storage( + self, + request: pytest.FixtureRequest, + fake_script: FakeScript, + ) -> ops.storage.SQLiteStorage: """Create a Storage backend that we can interact with.""" return NotImplemented - def test_save_and_load_snapshot(self): - f = self.create_framework() + def test_save_and_load_snapshot( + self, + request: pytest.FixtureRequest, + fake_script: FakeScript, + ): + f = self.create_framework(request, fake_script) class Sample(ops.StoredStateData): - def __init__(self, parent: ops.Object, key: str, - content: typing.Dict[str, typing.Any]): + def __init__( + self, + parent: ops.Object, + key: str, + content: typing.Dict[str, typing.Any], + ): super().__init__(parent, key) self.content = content @@ -81,8 +103,8 @@ def restore(self, snapshot: typing.Dict[str, typing.Any]): res = f.load_snapshot(handle) assert data == res.content # type: ignore - def test_emit_event(self): - f = self.create_framework() + def test_emit_event(self, request: pytest.FixtureRequest, fake_script: FakeScript): + f = self.create_framework(request, fake_script) class Evt(ops.EventBase): def __init__(self, handle: ops.Handle, content: typing.Any): @@ -125,23 +147,31 @@ def restore(self, snapshot: typing.Dict[str, typing.Any]) -> None: s.on.event.emit(None) assert s.observed_content is None - def test_save_and_overwrite_snapshot(self): - store = self.create_storage() + def test_save_and_overwrite_snapshot( + self, + request: pytest.FixtureRequest, + fake_script: FakeScript, + ): + store = self.create_storage(request, fake_script) store.save_snapshot('foo', {1: 2}) assert store.load_snapshot('foo') == {1: 2} store.save_snapshot('foo', {'three': 4}) assert store.load_snapshot('foo') == {'three': 4} - def test_drop_snapshot(self): - store = self.create_storage() + def test_drop_snapshot(self, request: pytest.FixtureRequest, fake_script: FakeScript): + store = self.create_storage(request, fake_script) store.save_snapshot('foo', {1: 2}) assert store.load_snapshot('foo') == {1: 2} store.drop_snapshot('foo') with pytest.raises(ops.storage.NoSnapshotError): store.load_snapshot('foo') - def test_save_snapshot_empty_string(self): - store = self.create_storage() + def test_save_snapshot_empty_string( + self, + request: pytest.FixtureRequest, + fake_script: FakeScript, + ): + store = self.create_storage(request, fake_script) with pytest.raises(ops.storage.NoSnapshotError): store.load_snapshot('foo') store.save_snapshot('foo', '') @@ -150,8 +180,12 @@ def test_save_snapshot_empty_string(self): with pytest.raises(ops.storage.NoSnapshotError): store.load_snapshot('foo') - def test_save_snapshot_none(self): - store = self.create_storage() + def test_save_snapshot_none( + self, + request: pytest.FixtureRequest, + fake_script: FakeScript, + ): + store = self.create_storage(request, fake_script) with pytest.raises(ops.storage.NoSnapshotError): store.load_snapshot('bar') store.save_snapshot('bar', None) @@ -160,8 +194,12 @@ def test_save_snapshot_none(self): with pytest.raises(ops.storage.NoSnapshotError): store.load_snapshot('bar') - def test_save_snapshot_zero(self): - store = self.create_storage() + def test_save_snapshot_zero( + self, + request: pytest.FixtureRequest, + fake_script: FakeScript, + ): + store = self.create_storage(request, fake_script) with pytest.raises(ops.storage.NoSnapshotError): store.load_snapshot('zero') store.save_snapshot('zero', 0) @@ -170,15 +208,14 @@ def test_save_snapshot_zero(self): with pytest.raises(ops.storage.NoSnapshotError): store.load_snapshot('zero') - def test_save_notice(self): - store = self.create_storage() + def test_save_notice(self, request: pytest.FixtureRequest, fake_script: FakeScript): + store = self.create_storage(request, fake_script) store.save_notice('event', 'observer', 'method') - assert list(store.notices('event')) == \ - [('event', 'observer', 'method')] + assert list(store.notices('event')) == [('event', 'observer', 'method')] - def test_all_notices(self): + def test_all_notices(self, request: pytest.FixtureRequest, fake_script: FakeScript): notices = [('e1', 'o1', 'm1'), ('e1', 'o2', 'm2'), ('e2', 'o3', 'm3')] - store = self.create_storage() + store = self.create_storage(request, fake_script) for notice in notices: store.save_notice(*notice) @@ -195,28 +232,36 @@ def test_all_notices(self): assert list(store.notices(None)) == notices assert list(store.notices('')) == notices - def test_load_notices(self): - store = self.create_storage() + def test_load_notices(self, request: pytest.FixtureRequest, fake_script: FakeScript): + store = self.create_storage(request, fake_script) assert list(store.notices('path')) == [] - def test_save_one_load_another_notice(self): - store = self.create_storage() + def test_save_one_load_another_notice( + self, + request: pytest.FixtureRequest, + fake_script: FakeScript, + ): + store = self.create_storage(request, fake_script) store.save_notice('event', 'observer', 'method') assert list(store.notices('other')) == [] - def test_save_load_drop_load_notices(self): - store = self.create_storage() + def test_save_load_drop_load_notices( + self, + request: pytest.FixtureRequest, + fake_script: FakeScript, + ): + store = self.create_storage(request, fake_script) store.save_notice('event', 'observer', 'method') store.save_notice('event', 'observer', 'method2') - assert list(store.notices('event')) == \ - [('event', 'observer', 'method'), - ('event', 'observer', 'method2'), - ] + assert list(store.notices('event')) == [ + ('event', 'observer', 'method'), + ('event', 'observer', 'method2') + ] -class TestSQLiteStorage(StoragePermutations, BaseTestCase): +class TestSQLiteStorage(StoragePermutations): - def create_storage(self): + def create_storage(self, request: pytest.FixtureRequest, fake_script: FakeScript): return ops.storage.SQLiteStorage(':memory:') def test_permissions_new(self): @@ -255,7 +300,7 @@ def test_permissions_failure(self, chmod: unittest.mock.MagicMock): pytest.raises(RuntimeError, ops.storage.SQLiteStorage, filename) -def setup_juju_backend(test_case: unittest.TestCase, state_file: pathlib.Path): +def setup_juju_backend(fake_script: FakeScript, state_file: pathlib.Path): """Create fake scripts for pretending to be state-set and state-get.""" template_args = { 'executable': str(pathlib.Path(sys.executable).as_posix()), @@ -263,7 +308,7 @@ def setup_juju_backend(test_case: unittest.TestCase, state_file: pathlib.Path): 'state_file': str(state_file.as_posix()), } - fake_script(test_case, 'state-set', dedent('''\ + fake_script.write('state-set', dedent('''\ {executable} -c ' import sys if "{pthpth}" not in sys.path: @@ -284,7 +329,7 @@ def setup_juju_backend(test_case: unittest.TestCase, state_file: pathlib.Path): ' "$@" ''').format(**template_args)) - fake_script(test_case, 'state-get', dedent('''\ + fake_script.write('state-get', dedent('''\ {executable} -Sc ' import sys if "{pthpth}" not in sys.path: @@ -302,7 +347,7 @@ def setup_juju_backend(test_case: unittest.TestCase, state_file: pathlib.Path): ' "$@" ''').format(**template_args)) - fake_script(test_case, 'state-delete', dedent('''\ + fake_script.write('state-delete', dedent('''\ {executable} -Sc ' import sys if "{pthpth}" not in sys.path: @@ -322,18 +367,18 @@ def setup_juju_backend(test_case: unittest.TestCase, state_file: pathlib.Path): ''').format(**template_args)) -class TestJujuStorage(StoragePermutations, BaseTestCase): +class TestJujuStorage(StoragePermutations): - def create_storage(self): + def create_storage(self, request: pytest.FixtureRequest, fake_script: FakeScript): fd, fn = tempfile.mkstemp(prefix='tmp-ops-test-state-') os.close(fd) state_file = pathlib.Path(fn) - self.addCleanup(state_file.unlink) - setup_juju_backend(self, state_file) + request.addfinalizer(state_file.unlink) + setup_juju_backend(fake_script, state_file) return ops.storage.JujuStorage() -class TestSimpleLoader(BaseTestCase): +class TestSimpleLoader: def test_is_c_loader(self): loader = ops.storage._SimpleLoader(io.StringIO('')) @@ -376,25 +421,25 @@ class Foo: self.assertRefused(f) -class TestJujuStateBackend(BaseTestCase): +class TestJujuStateBackend: def test_is_not_available(self): assert not ops.storage.juju_backend_available() - def test_is_available(self): - fake_script(self, 'state-get', 'echo ""') + def test_is_available(self, fake_script: FakeScript): + fake_script.write('state-get', 'echo ""') assert ops.storage.juju_backend_available() - assert fake_script_calls(self, clear=True) == [] + assert fake_script.calls(clear=True) == [] - def test_set_encodes_args(self): + def test_set_encodes_args(self, fake_script: FakeScript): t = tempfile.NamedTemporaryFile() try: - fake_script(self, 'state-set', dedent(""" + fake_script.write('state-set', dedent(""" cat >> {} """).format(pathlib.Path(t.name).as_posix())) backend = ops.storage._JujuStorageBackend() backend.set('key', {'foo': 2}) - assert fake_script_calls(self, clear=True) == [ + assert fake_script.calls(clear=True) == [ ['state-set', '--file', '-'], ] t.seek(0) @@ -406,21 +451,21 @@ def test_set_encodes_args(self): {foo: 2} """) - def test_get(self): - fake_script(self, 'state-get', dedent(""" + def test_get(self, fake_script: FakeScript): + fake_script.write('state-get', dedent(""" echo 'foo: "bar"' """)) backend = ops.storage._JujuStorageBackend() value = backend.get('key') assert value == {'foo': 'bar'} - assert fake_script_calls(self, clear=True) == [ + assert fake_script.calls(clear=True) == [ ['state-get', 'key'], ] - def test_set_and_get_complex_value(self): + def test_set_and_get_complex_value(self, fake_script: FakeScript): t = tempfile.NamedTemporaryFile() try: - fake_script(self, 'state-set', dedent(""" + fake_script.write('state-set', dedent(""" cat >> {} """).format(pathlib.Path(t.name).as_posix())) backend = ops.storage._JujuStorageBackend() @@ -433,7 +478,7 @@ def test_set_and_get_complex_value(self): 'seven': b'1234', } backend.set('Class[foo]/_stored', complex_val) - assert fake_script_calls(self, clear=True) == [ + assert fake_script.calls(clear=True) == [ ['state-set', '--file', '-'], ] t.seek(0) @@ -457,7 +502,7 @@ def test_set_and_get_complex_value(self): """) # Note that the content is yaml in a string, embedded inside YAML to declare the Key: # Value of where to store the entry. - fake_script(self, 'state-get', dedent(""" + fake_script.write('state-get', dedent(""" echo "foo: 2 3: [1, 2, '3'] four: !!set {2: null, 3: null}