diff --git a/lib/charms/data_platform_libs/v0/data_secrets.py b/lib/charms/data_platform_libs/v0/data_secrets.py new file mode 100644 index 0000000000..254b9af3df --- /dev/null +++ b/lib/charms/data_platform_libs/v0/data_secrets.py @@ -0,0 +1,143 @@ +"""Secrets related helper classes/functions.""" +# Copyright 2023 Canonical Ltd. +# See LICENSE file for licensing details. + +from typing import Dict, Literal, Optional + +from ops import Secret, SecretInfo +from ops.charm import CharmBase +from ops.model import SecretNotFoundError + +# The unique Charmhub library identifier, never change it +LIBID = "d77fb3d01aba41ed88e837d0beab6be5" + +# Increment this major API version when introducing breaking changes +LIBAPI = 0 + +# Increment this PATCH version before using `charmcraft publish-lib` or reset +# to 0 if you are raising the major API version +LIBPATCH = 1 + + +APP_SCOPE = "app" +UNIT_SCOPE = "unit" +Scopes = Literal["app", "unit"] + + +class DataSecretsError(Exception): + """A secret that we want to create already exists.""" + + +class SecretAlreadyExistsError(DataSecretsError): + """A secret that we want to create already exists.""" + + +def generate_secret_label(charm: CharmBase, scope: Scopes) -> str: + """Generate unique group_mappings for secrets within a relation context. + + Defined as a standalone function, as the choice on secret labels definition belongs to the + Application Logic. To be kept separate from classes below, which are simply to provide a + (smart) abstraction layer above Juju Secrets. + """ + members = [charm.app.name, scope] + return f"{'.'.join(members)}" + + +# Secret cache + + +class CachedSecret: + """Abstraction layer above direct Juju access with caching. + + The data structure is precisely re-using/simulating Juju Secrets behavior, while + also making sure not to fetch a secret multiple times within the same event scope. + """ + + def __init__(self, charm: CharmBase, label: str, secret_uri: Optional[str] = None): + self._secret_meta = None + self._secret_content = {} + self._secret_uri = secret_uri + self.label = label + self.charm = charm + + def add_secret(self, content: Dict[str, str], scope: Scopes) -> Secret: + """Create a new secret.""" + if self._secret_uri: + raise SecretAlreadyExistsError( + "Secret is already defined with uri %s", self._secret_uri + ) + + if scope == APP_SCOPE: + secret = self.charm.app.add_secret(content, label=self.label) + else: + secret = self.charm.unit.add_secret(content, label=self.label) + self._secret_uri = secret.id + self._secret_meta = secret + return self._secret_meta + + @property + def meta(self) -> Optional[Secret]: + """Getting cached secret meta-information.""" + if self._secret_meta: + return self._secret_meta + + if not (self._secret_uri or self.label): + return + + try: + self._secret_meta = self.charm.model.get_secret(label=self.label) + except SecretNotFoundError: + if self._secret_uri: + self._secret_meta = self.charm.model.get_secret( + id=self._secret_uri, label=self.label + ) + return self._secret_meta + + def get_content(self) -> Dict[str, str]: + """Getting cached secret content.""" + if not self._secret_content: + if self.meta: + self._secret_content = self.meta.get_content() + return self._secret_content + + def set_content(self, content: Dict[str, str]) -> None: + """Setting cached secret content.""" + if self.meta: + self.meta.set_content(content) + self._secret_content = content + + def get_info(self) -> Optional[SecretInfo]: + """Wrapper function for get the corresponding call on the Secret object if any.""" + if self.meta: + return self.meta.get_info() + + +class SecretCache: + """A data structure storing CachedSecret objects.""" + + def __init__(self, charm): + self.charm = charm + self._secrets: Dict[str, CachedSecret] = {} + + def get(self, label: str, uri: Optional[str] = None) -> Optional[CachedSecret]: + """Getting a secret from Juju Secret store or cache.""" + if not self._secrets.get(label): + secret = CachedSecret(self.charm, label, uri) + + # Checking if the secret exists, otherwise we don't register it in the cache + if secret.meta: + self._secrets[label] = secret + return self._secrets.get(label) + + def add(self, label: str, content: Dict[str, str], scope: Scopes) -> CachedSecret: + """Adding a secret to Juju Secret.""" + if self._secrets.get(label): + raise SecretAlreadyExistsError(f"Secret {label} already exists") + + secret = CachedSecret(self.charm, label) + secret.add_secret(content, scope) + self._secrets[label] = secret + return self._secrets[label] + + +# END: Secret cache diff --git a/src/charm.py b/src/charm.py index 999b3bb722..39bdb50bc2 100755 --- a/src/charm.py +++ b/src/charm.py @@ -8,9 +8,10 @@ import os import subprocess import time -from typing import Dict, List, Optional, Set +from typing import Dict, List, Literal, Optional, Set, get_args from charms.data_platform_libs.v0.data_models import TypedCharmBase +from charms.data_platform_libs.v0.data_secrets import SecretCache, generate_secret_label from charms.grafana_agent.v0.cos_agent import COSAgentProvider from charms.operator_libs_linux.v2 import snap from charms.postgresql_k8s.v0.postgresql import ( @@ -38,7 +39,6 @@ MaintenanceStatus, ModelError, Relation, - SecretNotFoundError, Unit, WaitingStatus, ) @@ -68,11 +68,9 @@ POSTGRESQL_SNAP_NAME, REPLICATION_PASSWORD_KEY, REWIND_PASSWORD_KEY, - SECRET_CACHE_LABEL, SECRET_DELETED_LABEL, SECRET_INTERNAL_LABEL, SECRET_KEY_OVERRIDES, - SECRET_LABEL, SNAP_PACKAGES, SYSTEM_USERS, TLS_CA_FILE, @@ -91,6 +89,8 @@ NO_PRIMARY_MESSAGE = "no primary in the cluster" +Scopes = Literal[APP_SCOPE, UNIT_SCOPE] + class PostgresqlOperatorCharm(TypedCharmBase[CharmConfig]): """Charmed Operator for the PostgreSQL database.""" @@ -101,7 +101,7 @@ class PostgresqlOperatorCharm(TypedCharmBase[CharmConfig]): def __init__(self, *args): super().__init__(*args) - self.secrets = {APP_SCOPE: {}, UNIT_SCOPE: {}} + self.secrets = SecretCache(self) juju_version = JujuVersion.from_environ() if juju_version.major > 2: @@ -191,180 +191,114 @@ def unit_peer_data(self) -> Dict: return relation.data[self.unit] - def _scope_obj(self, scope: str): - if scope == APP_SCOPE: - return self.framework.model.app - if scope == UNIT_SCOPE: - return self.framework.model.unit - - def _juju_secrets_get(self, scope: str) -> Optional[bool]: - """Helper function to get Juju secret.""" - if scope == UNIT_SCOPE: - peer_data = self.unit_peer_data - else: - peer_data = self.app_peer_data - - if not peer_data.get(SECRET_INTERNAL_LABEL): - return - - if SECRET_CACHE_LABEL not in self.secrets[scope]: - try: - # NOTE: Secret contents are not yet available! - secret = self.model.get_secret(id=peer_data[SECRET_INTERNAL_LABEL]) - except SecretNotFoundError as e: - logging.debug(f"No secret found for ID {peer_data[SECRET_INTERNAL_LABEL]}, {e}") - return - - logging.debug(f"Secret {peer_data[SECRET_INTERNAL_LABEL]} downloaded") + def _peer_data(self, scope: Scopes) -> Dict: + """Return corresponding databag for app/unit.""" + relation = self.model.get_relation(PEER) + if relation is None: + return {} - # We keep the secret object around -- needed when applying modifications - self.secrets[scope][SECRET_LABEL] = secret + return relation.data[self._scope_obj(scope)] - # We retrieve and cache actual secret data for the lifetime of the event scope - self.secrets[scope][SECRET_CACHE_LABEL] = secret.get_content() + def _scope_obj(self, scope: Scopes): + if scope == APP_SCOPE: + return self.app + if scope == UNIT_SCOPE: + return self.unit - return bool(self.secrets[scope].get(SECRET_CACHE_LABEL)) + def _translate_field_to_secret_key(self, key: str) -> str: + """Change 'key' to secrets-compatible key field.""" + key = SECRET_KEY_OVERRIDES.get(key, key) + new_key = key.replace("_", "-") + return new_key.strip("-") - def _juju_secret_get_key(self, scope: str, key: str) -> Optional[str]: - if not key: - return + def _safe_get_secret(self, scope: Scopes, label: str) -> SecretCache: + """Safety measure, for upgrades between versions based on secret URI usage to others with labels usage. - key = SECRET_KEY_OVERRIDES.get(key, key) + If the secret can't be retrieved by label, we search for the uri -- and if found, we "stick" the + label on the secret for further usage. + """ + secret_uri = self._peer_data(scope).get(SECRET_INTERNAL_LABEL, None) + secret = self.secrets.get(label, secret_uri) - if self._juju_secrets_get(scope): - secret_cache = self.secrets[scope].get(SECRET_CACHE_LABEL) - if secret_cache: - secret_data = secret_cache.get(key) - if secret_data and secret_data != SECRET_DELETED_LABEL: - logging.debug(f"Getting secret {scope}:{key}") - return secret_data - logging.debug(f"No value found for secret {scope}:{key}") + # Since now we switched to labels, the databag reference can be removed + if secret_uri and secret and scope == APP_SCOPE and self.unit.is_leader(): + self._peer_data(scope).pop(SECRET_INTERNAL_LABEL, None) + return secret - def get_secret(self, scope: str, key: str) -> Optional[str]: + def get_secret(self, scope: Scopes, key: str) -> Optional[str]: """Get secret from the secret storage.""" - if scope not in [APP_SCOPE, UNIT_SCOPE]: + if scope not in get_args(Scopes): raise RuntimeError("Unknown secret scope.") - if scope == UNIT_SCOPE: - result = self.unit_peer_data.get(key, None) - else: - result = self.app_peer_data.get(key, None) - - # TODO change upgrade to switch to secrets once minor version upgrades is done - if result: - return result - - juju_version = JujuVersion.from_environ() - if juju_version.has_secrets: - return self._juju_secret_get_key(scope, key) - - def _juju_secret_set(self, scope: str, key: str, value: str) -> str: - """Helper function setting Juju secret.""" - if scope == UNIT_SCOPE: - peer_data = self.unit_peer_data - else: - peer_data = self.app_peer_data - self._juju_secrets_get(scope) - - key = SECRET_KEY_OVERRIDES.get(key, key) - - secret = self.secrets[scope].get(SECRET_LABEL) + if value := self._peer_data(scope).get(key, None): + return value - # It's not the first secret for the scope, we can reuse the existing one - # that was fetched in the previous call - if secret: - secret_cache = self.secrets[scope][SECRET_CACHE_LABEL] - - if secret_cache.get(key) == value: - logging.debug(f"Key {scope}:{key} has this value defined already") - else: - secret_cache[key] = value - try: - secret.set_content(secret_cache) - except OSError as error: - logging.error( - f"Error in attempt to set {scope}:{key}. " - f"Existing keys were: {list(secret_cache.keys())}. {error}" - ) - logging.debug(f"Secret {scope}:{key} was {key} set") - - # We need to create a brand-new secret for this scope - else: - scope_obj = self._scope_obj(scope) + if JujuVersion.from_environ().has_secrets: + label = generate_secret_label(self, scope) + secret = self._safe_get_secret(scope, label) - secret = scope_obj.add_secret({key: value}) if not secret: - raise RuntimeError(f"Couldn't set secret {scope}:{key}") - - self.secrets[scope][SECRET_LABEL] = secret - self.secrets[scope][SECRET_CACHE_LABEL] = {key: value} - logging.debug(f"Secret {scope}:{key} published (as first). ID: {secret.id}") - peer_data.update({SECRET_INTERNAL_LABEL: secret.id}) - - # TODO change upgrade to switch to secrets once minor version upgrades is done - if key in peer_data: - del peer_data[key] + return - return self.secrets[scope][SECRET_LABEL].id + secret_key = self._translate_field_to_secret_key(key) + value = secret.get_content().get(secret_key) + if value != SECRET_DELETED_LABEL: + return value - def set_secret(self, scope: str, key: str, value: Optional[str]) -> Optional[str]: + def set_secret(self, scope: Scopes, key: str, value: Optional[str]) -> Optional[str]: """Set secret from the secret storage.""" - if scope not in [APP_SCOPE, UNIT_SCOPE]: + if scope not in get_args(Scopes): raise RuntimeError("Unknown secret scope.") if not value: return self.remove_secret(scope, key) - juju_version = JujuVersion.from_environ() + if JujuVersion.from_environ().has_secrets: + # Charm must have been upgraded since last run + # We move from databag to secrets + self._peer_data(scope).pop(key, None) - if juju_version.has_secrets: - self._juju_secret_set(scope, key, value) - return - if scope == UNIT_SCOPE: - self.unit_peer_data.update({key: value}) + secret_key = self._translate_field_to_secret_key(key) + label = generate_secret_label(self, scope) + secret = self._safe_get_secret(scope, label) + if not secret: + self.secrets.add(label, {secret_key: value}, scope) + else: + content = secret.get_content() + content.update({secret_key: value}) + secret.set_content(content) + return label else: - self.app_peer_data.update({key: value}) - - def _juju_secret_remove(self, scope: str, key: str) -> None: - """Remove a Juju 3.x secret.""" - self._juju_secrets_get(scope) - - key = SECRET_KEY_OVERRIDES.get(key, key) + self._peer_data(scope).update({key: value}) - secret = self.secrets[scope].get(SECRET_LABEL) - if not secret: - logging.error(f"Secret {scope}:{key} wasn't deleted: no secrets are available") - return + def remove_secret(self, scope: Scopes, key: str) -> None: + """Removing a secret.""" + if scope not in get_args(Scopes): + raise RuntimeError("Unknown secret scope.") - secret_cache = self.secrets[scope].get(SECRET_CACHE_LABEL) - if not secret_cache or key not in secret_cache: - logging.error(f"No secret {scope}:{key}") - return + if JujuVersion.from_environ().has_secrets: + secret_key = self._translate_field_to_secret_key(key) + label = generate_secret_label(self, scope) + secret = self.secrets.get(label) - secret_cache[key] = SECRET_DELETED_LABEL - secret.set_content(secret_cache) - logging.debug(f"Secret {scope}:{key}") + if not secret: + return - # TODO change upgrade to switch to secrets once minor version upgrades is done - if scope == UNIT_SCOPE: - peer_data = self.unit_peer_data - else: - peer_data = self.app_peer_data - if key in peer_data: - del peer_data[key] + content = secret.get_content() - def remove_secret(self, scope: str, key: str) -> None: - """Removing a secret.""" - if scope not in [APP_SCOPE, UNIT_SCOPE]: - raise RuntimeError("Unknown secret scope.") + if not content.get(secret_key) or content[secret_key] == SECRET_DELETED_LABEL: + logger.error(f"Non-existing secret {scope}:{key} was attempted to be removed.") + return - juju_version = JujuVersion.from_environ() - if juju_version.has_secrets: - return self._juju_secret_remove(scope, key) - if scope == UNIT_SCOPE: - del self.unit_peer_data[key] + content[secret_key] = SECRET_DELETED_LABEL + secret.set_content(content) + # Just in case we started on databag + self._peer_data(scope).pop(key, None) else: - del self.app_peer_data[key] + try: + self._peer_data(scope).pop(key) + except KeyError: + logger.error(f"Non-existing secret {scope}:{key} was attempted to be removed.") @property def is_cluster_initialised(self) -> bool: diff --git a/src/constants.py b/src/constants.py index 02a0b40a9c..76090da662 100644 --- a/src/constants.py +++ b/src/constants.py @@ -55,8 +55,6 @@ METRICS_PORT = "9187" -SECRET_LABEL = "secret" -SECRET_CACHE_LABEL = "cache" SECRET_INTERNAL_LABEL = "internal-secret" SECRET_DELETED_LABEL = "None" diff --git a/tests/integration/test_password_rotation.py b/tests/integration/test_password_rotation.py index cada825418..d8de9080f3 100644 --- a/tests/integration/test_password_rotation.py +++ b/tests/integration/test_password_rotation.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 # Copyright 2022 Canonical Ltd. # See LICENSE file for licensing details. +import json import time import pytest @@ -10,6 +11,7 @@ from tests.integration.helpers import ( CHARM_SERIES, check_patroni, + get_leader_unit, get_password, restart_patroni, set_password, @@ -74,3 +76,53 @@ async def test_password_rotation(ops_test: OpsTest): if not await unit.is_leader_from_status(): restart_patroni(ops_test, unit.name) assert check_patroni(ops_test, unit.name, restart_time) + + +@pytest.mark.juju3 +async def test_password_from_secret_same_as_cli(ops_test: OpsTest): + """Checking if password is same as returned by CLI. + + I.e. we're manipulating the secret we think we're manipulating. + """ + # + # No way to retrieve a secet by label for now (https://bugs.launchpad.net/juju/+bug/2037104) + # Therefore we take advantage of the fact, that we only have ONE single secret a this point + # So we take the single member of the list + # NOTE: This would BREAK if for instance units had secrets at the start... + # + leader_unit = await get_leader_unit(ops_test, APP_NAME) + leader = leader_unit.name + password = await get_password(ops_test, unit_name=leader, username="replication") + complete_command = "list-secrets" + _, stdout, _ = await ops_test.juju(*complete_command.split()) + secret_id = stdout.split("\n")[1].split(" ")[0] + + # Getting back the pw from juju CLI + complete_command = f"show-secret {secret_id} --reveal --format=json" + _, stdout, _ = await ops_test.juju(*complete_command.split()) + data = json.loads(stdout) + assert data[secret_id]["content"]["Data"]["replication-password"] == password + + +async def test_empty_password(ops_test: OpsTest) -> None: + """Test that the password can't be set to an empty string.""" + leader_unit = await get_leader_unit(ops_test, APP_NAME) + leader = leader_unit.name + await set_password(ops_test, unit_name=leader, username="replication", password="") + password = await get_password(ops_test, unit_name=leader, username="replication") + # The password is 'None', BUT NOT because of SECRET_DELETED_LABEL + # `get_secret()` returns a None value (as the field in the secret is set to string value "None") + # And this true None value is turned to a string when the event is setting results. + assert password == "None" + + +async def test_no_password_change_on_invalid_password(ops_test: OpsTest) -> None: + """Test that in general, there is no change when password validation fails.""" + leader_unit = await get_leader_unit(ops_test, APP_NAME) + leader = leader_unit.name + password1 = await get_password(ops_test, unit_name=leader, username="replication") + # The password has to be minimum 3 characters + await set_password(ops_test, unit_name=leader, username="replication", password="ca" * 1000000) + password2 = await get_password(ops_test, unit_name=leader, username="replication") + # The password didn't change + assert password1 == password2 diff --git a/tests/unit/test_charm.py b/tests/unit/test_charm.py index 35b7ba5943..c768262a23 100644 --- a/tests/unit/test_charm.py +++ b/tests/unit/test_charm.py @@ -2,7 +2,7 @@ # See LICENSE file for licensing details. import subprocess import unittest -from unittest.mock import MagicMock, Mock, PropertyMock, mock_open, patch +from unittest.mock import MagicMock, Mock, PropertyMock, mock_open, patch, sentinel from charms.operator_libs_linux.v2 import snap from charms.postgresql_k8s.v0.postgresql import ( @@ -11,7 +11,7 @@ PostgreSQLUpdateUserPasswordError, ) from ops.framework import EventBase -from ops.model import ActiveStatus, BlockedStatus, SecretNotFoundError, WaitingStatus +from ops.model import ActiveStatus, BlockedStatus, WaitingStatus from ops.testing import Harness from tenacity import RetryError @@ -20,10 +20,7 @@ from constants import ( PEER, POSTGRESQL_SNAP_NAME, - SECRET_CACHE_LABEL, SECRET_DELETED_LABEL, - SECRET_INTERNAL_LABEL, - SECRET_LABEL, SNAP_PACKAGES, ) from tests.helpers import patch_network_get @@ -914,62 +911,47 @@ def test_get_secret(self, _): assert self.charm.get_secret("unit", "password") == "test-password" @patch_network_get(private_address="1.1.1.1") - @patch("ops.charm.model.Model.get_secret") @patch("charm.JujuVersion.has_secrets", new_callable=PropertyMock, return_value=True) @patch("charm.PostgresqlOperatorCharm._on_leader_elected") - def test_get_secret_juju_error(self, _, __, _get_secret): + def test_get_secret_juju(self, _, __): self.harness.set_leader() + with patch.object(self.charm, "secrets") as _secret_cache: + # Test application scope. + _secret_cache.get.return_value = None + assert self.charm.get_secret("app", "password") is None + _secret_cache.get.assert_called_once_with("postgresql.app", None) + _secret_cache.reset_mock() + + _secret_cache.get.return_value = Mock() + _secret_cache.get.return_value.get_content.return_value.get.return_value = ( + sentinel.test_password + ) + assert self.charm.get_secret("app", "password") == sentinel.test_password + _secret_cache.get.assert_called_once_with("postgresql.app", None) + _secret_cache.get.return_value.get_content.return_value.get.assert_called_once_with( + "password" + ) + _secret_cache.reset_mock() - # clean the caches - if SECRET_INTERNAL_LABEL in self.charm.app_peer_data: - del self.charm.app_peer_data[SECRET_INTERNAL_LABEL] - self.charm.secrets["app"] = {} - - # general tests - self.harness.update_relation_data( - self.rel_id, self.charm.app.name, {SECRET_INTERNAL_LABEL: "secret_key"} - ) - _get_secret.side_effect = SecretNotFoundError - assert self.charm.get_secret("app", "password") is None - self.harness.update_relation_data(self.rel_id, self.charm.app.name, {}) - - @patch_network_get(private_address="1.1.1.1") - @patch("ops.charm.model.Model.get_secret") - @patch("charm.JujuVersion.has_secrets", new_callable=PropertyMock, return_value=True) - @patch("charm.PostgresqlOperatorCharm._on_leader_elected") - def test_get_secret_juju(self, _, __, _get_secret): - self.harness.set_leader() - _get_secret.return_value.get_content.return_value = {"password": "test-password"} - - # clean the caches - if SECRET_INTERNAL_LABEL in self.charm.app_peer_data: - del self.charm.app_peer_data[SECRET_INTERNAL_LABEL] - self.charm.secrets["app"] = {} - - assert self.charm.get_secret("app", None) is None - assert not _get_secret.called - _get_secret.reset_mock() - - # Test application scope. - assert self.charm.get_secret("app", "password") is None - self.harness.update_relation_data( - self.rel_id, self.charm.app.name, {SECRET_INTERNAL_LABEL: "secret_key"} - ) - assert self.charm.get_secret("app", "password") == "test-password" - _get_secret.assert_called_once_with(id="secret_key") - - _get_secret.reset_mock() + # Test unit scope. + _secret_cache.get.return_value = None + assert self.charm.get_secret("unit", "password") is None + _secret_cache.get.assert_called_once_with("postgresql.unit", None) + _secret_cache.reset_mock() - # Test unit scope. - assert self.charm.get_secret("unit", "password") is None - self.harness.update_relation_data( - self.rel_id, self.charm.unit.name, {SECRET_INTERNAL_LABEL: "secret_key"} - ) - assert self.charm.get_secret("unit", "password") == "test-password" - _get_secret.assert_called_once_with(id="secret_key") + _secret_cache.get.return_value = Mock() + _secret_cache.get.return_value.get_content.return_value.get.return_value = ( + sentinel.test_password + ) + assert self.charm.get_secret("unit", "password") == sentinel.test_password + _secret_cache.get.assert_called_once_with("postgresql.unit", None) + _secret_cache.get.return_value.get_content.return_value.get.assert_called_once_with( + "password" + ) + _secret_cache.reset_mock() - with self.assertRaises(RuntimeError): - self.charm.get_secret("test", "password") + with self.assertRaises(RuntimeError): + self.charm.get_secret("test", "password") @patch_network_get(private_address="1.1.1.1") @patch("charm.PostgresqlOperatorCharm._on_leader_elected") @@ -1004,43 +986,36 @@ def test_set_secret(self, _): @patch("charm.PostgresqlOperatorCharm._on_leader_elected") def test_set_secret_juju(self, _, __): self.harness.set_leader() - secret_mock = Mock() - self.charm.secrets["app"][SECRET_LABEL] = secret_mock - self.charm.secrets["unit"][SECRET_LABEL] = secret_mock - self.charm.secrets["app"][SECRET_CACHE_LABEL] = {} - self.charm.secrets["unit"][SECRET_CACHE_LABEL] = {} - - # Test application scope. - assert "password" not in self.charm.secrets["app"].get(SECRET_CACHE_LABEL, {}) - self.charm.set_secret("app", "password", "test-password") - assert self.charm.secrets["app"][SECRET_CACHE_LABEL]["password"] == "test-password" - secret_mock.set_content.assert_called_once_with( - self.charm.secrets["app"][SECRET_CACHE_LABEL] - ) - secret_mock.reset_mock() - - self.charm.set_secret("app", "password", None) - assert self.charm.secrets["app"][SECRET_CACHE_LABEL]["password"] == SECRET_DELETED_LABEL - secret_mock.set_content.assert_called_once_with( - self.charm.secrets["app"][SECRET_CACHE_LABEL] - ) - secret_mock.reset_mock() - - # Test unit scope. - assert "password" not in self.charm.secrets["unit"].get(SECRET_CACHE_LABEL, {}) - self.charm.set_secret("unit", "password", "test-password") - assert self.charm.secrets["unit"][SECRET_CACHE_LABEL]["password"] == "test-password" - secret_mock.set_content.assert_called_once_with( - self.charm.secrets["unit"][SECRET_CACHE_LABEL] - ) - secret_mock.reset_mock() - - self.charm.set_secret("unit", "password", None) - assert self.charm.secrets["unit"][SECRET_CACHE_LABEL]["password"] == SECRET_DELETED_LABEL - secret_mock.set_content.assert_called_once_with( - self.charm.secrets["unit"][SECRET_CACHE_LABEL] - ) - secret_mock.reset_mock() + with patch.object(self.charm, "secrets") as _secret_cache: + # Test application scope. + self.charm.set_secret("app", "password", "test-password") + _secret_cache.get.assert_called_once_with("postgresql.app", None) + _secret_cache.get().get_content().update.assert_called_once_with( + {"password": "test-password"} + ) + _secret_cache.reset_mock() + + self.charm.set_secret("app", "password", None) + _secret_cache.get.assert_called_once_with("postgresql.app") + content = _secret_cache.get().get_content() + content.__setitem__.assert_called_once_with("password", SECRET_DELETED_LABEL) + _secret_cache.get().set_content.assert_called_once_with(content) + _secret_cache.reset_mock() + + # Test unit scope. + self.charm.set_secret("unit", "password", "test-password") + _secret_cache.get.assert_called_once_with("postgresql.unit", None) + _secret_cache.get().get_content().update.assert_called_once_with( + {"password": "test-password"} + ) + _secret_cache.reset_mock() + + self.charm.set_secret("unit", "password", None) + _secret_cache.get.assert_called_once_with("postgresql.unit") + content = _secret_cache.get().get_content() + content.__setitem__.assert_called_once_with("password", SECRET_DELETED_LABEL) + _secret_cache.get().set_content.assert_called_once_with(content) + _secret_cache.reset_mock() @patch( "subprocess.check_call",