feat(parameters): add clear_cache method for providers (#1194)

aws_lambda_powertools/utilities/parameters/__init__.py

@@ -5,7 +5,7 @@
 """
 
 from .appconfig import AppConfigProvider, get_app_config
-from .base import BaseProvider
+from .base import BaseProvider, clear_caches
 from .dynamodb import DynamoDBProvider
 from .exceptions import GetParameterError, TransformParameterError
 from .secrets import SecretsProvider, get_secret
@@ -23,4 +23,5 @@
     "get_parameter",
     "get_parameters",
     "get_secret",
+    "clear_caches",
 ]

aws_lambda_powertools/utilities/parameters/base.py

@@ -177,6 +177,9 @@ def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]:
         """
         raise NotImplementedError()
 
+    def clear_cache(self):
+        self.store.clear()
+
 
 def get_transform_method(key: str, transform: Optional[str] = None) -> Optional[str]:
     """
@@ -251,3 +254,8 @@ def transform_value(
         if raise_on_transform_error:
             raise TransformParameterError(str(exc))
         return None
+
+
+def clear_caches():
+    """Clear cached parameter values from all providers"""
+    DEFAULT_PROVIDERS.clear()

docs/utilities/parameters.md

@@ -518,9 +518,9 @@ The **`config`** and **`boto3_session`** parameters enable you to pass in a cust
 
 ## Testing your code
 
-For unit testing your applications, you can mock the calls to the parameters utility to avoid calling AWS APIs. This
-can be achieved in a number of ways - in this example, we use the [pytest monkeypatch fixture](https://docs.pytest.org/en/latest/how-to/monkeypatch.html)
-to patch the `parameters.get_parameter` method:
+### Mocking parameter values
+
+For unit testing your applications, you can mock the calls to the parameters utility to avoid calling AWS APIs. This can be achieved in a number of ways - in this example, we use the [pytest monkeypatch fixture](https://docs.pytest.org/en/latest/how-to/monkeypatch.html) to patch the `parameters.get_parameter` method:
 
 === "tests.py"
 	```python
@@ -588,3 +588,75 @@ object named `get_parameter_mock`.
 		assert return_val.get('message') == 'mock_value'
 
 	```
+
+
+### Clearing cache
+
+Parameters utility caches all parameter values for performance and cost reasons. However, this can have unintended interference in tests using the same parameter name.
+
+Within your tests, you can use `clear_cache` method available in [every provider](#built-in-provider-class). When using multiple providers or higher level functions like `get_parameter`, use `clear_caches` standalone function to clear cache globally.
+
+
+=== "clear_cache method"
+	```python hl_lines="9"
+    import pytest
+
+    from src import app
+
+
+    @pytest.fixture(scope="function", autouse=True)
+	def clear_parameters_cache():
+		yield
+		app.ssm_provider.clear_cache() # This will clear SSMProvider cache
+
+	@pytest.fixture
+    def mock_parameter_response(monkeypatch):
+        def mockreturn(name):
+            return "mock_value"
+
+        monkeypatch.setattr(app.ssm_provider, "get", mockreturn)
+
+    # Pass our fixture as an argument to all tests where we want to mock the get_parameter response
+    def test_handler(mock_parameter_response):
+        return_val = app.handler({}, {})
+        assert return_val.get('message') == 'mock_value'
+	```
+
+=== "global clear_caches"
+	```python hl_lines="10"
+    import pytest
+
+	from aws_lambda_powertools.utilities import parameters
+    from src import app
+
+
+    @pytest.fixture(scope="function", autouse=True)
+	def clear_parameters_cache():
+		yield
+		parameters.clear_caches() # This will clear all providers cache
+
+	@pytest.fixture
+    def mock_parameter_response(monkeypatch):
+        def mockreturn(name):
+            return "mock_value"
+
+        monkeypatch.setattr(app.ssm_provider, "get", mockreturn)
+
+    # Pass our fixture as an argument to all tests where we want to mock the get_parameter response
+    def test_handler(mock_parameter_response):
+        return_val = app.handler({}, {})
+        assert return_val.get('message') == 'mock_value'
+	```
+
+=== "app.py"
+	```python
+    from aws_lambda_powertools.utilities import parameters
+    from botocore.config import Config
+
+    ssm_provider = parameters.SSMProvider(config=Config(region_name="us-west-1"))
+
+
+    def handler(event, context):
+        value = ssm_provider.get("/my/parameter")
+		return {"message": value}
+	```

tests/functional/test_utilities_parameters.py

@@ -505,6 +505,79 @@ def test_ssm_provider_get_cached(mock_name, mock_value, config):
         stubber.deactivate()
 
 
+def test_providers_global_clear_cache(mock_name, mock_value, monkeypatch):
+    # GIVEN all providers are previously initialized
+    # and parameters, secrets, and app config are fetched
+    class TestProvider(BaseProvider):
+        def _get(self, name: str, **kwargs) -> str:
+            return mock_value
+
+        def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]:
+            ...
+
+    monkeypatch.setitem(parameters.base.DEFAULT_PROVIDERS, "ssm", TestProvider())
+    monkeypatch.setitem(parameters.base.DEFAULT_PROVIDERS, "secrets", TestProvider())
+    monkeypatch.setitem(parameters.base.DEFAULT_PROVIDERS, "appconfig", TestProvider())
+
+    parameters.get_parameter(mock_name)
+    parameters.get_secret(mock_name)
+    parameters.get_app_config(name=mock_name, environment="test", application="test")
+
+    # WHEN clear_caches is called
+    parameters.clear_caches()
+
+    # THEN all providers cache should be reset
+    assert parameters.base.DEFAULT_PROVIDERS == {}
+
+
+def test_ssm_provider_clear_cache(mock_name, mock_value, config):
+    # GIVEN a provider is initialized with a cached value
+    provider = parameters.SSMProvider(config=config)
+    provider.store[(mock_name, None)] = ExpirableValue(mock_value, datetime.now() + timedelta(seconds=60))
+
+    # WHEN clear_cache is called from within the provider instance
+    provider.clear_cache()
+
+    # THEN store should be empty
+    assert provider.store == {}
+
+
+def test_dynamodb_provider_clear_cache(mock_name, mock_value, config):
+    # GIVEN a provider is initialized with a cached value
+    provider = parameters.DynamoDBProvider(table_name="test", config=config)
+    provider.store[(mock_name, None)] = ExpirableValue(mock_value, datetime.now() + timedelta(seconds=60))
+
+    # WHEN clear_cache is called from within the provider instance
+    provider.clear_cache()
+
+    # THEN store should be empty
+    assert provider.store == {}
+
+
+def test_secrets_provider_clear_cache(mock_name, mock_value, config):
+    # GIVEN a provider is initialized with a cached value
+    provider = parameters.SecretsProvider(config=config)
+    provider.store[(mock_name, None)] = ExpirableValue(mock_value, datetime.now() + timedelta(seconds=60))
+
+    # WHEN clear_cache is called from within the provider instance
+    provider.clear_cache()
+
+    # THEN store should be empty
+    assert provider.store == {}
+
+
+def test_appconf_provider_clear_cache(mock_name, config):
+    # GIVEN a provider is initialized with a cached value
+    provider = parameters.AppConfigProvider(environment="test", application="test", config=config)
+    provider.store[(mock_name, None)] = ExpirableValue(mock_value, datetime.now() + timedelta(seconds=60))
+
+    # WHEN clear_cache is called from within the provider instance
+    provider.clear_cache()
+
+    # THEN store should be empty
+    assert provider.store == {}
+
+
 def test_ssm_provider_get_expired(mock_name, mock_value, mock_version, config):
     """
     Test SSMProvider.get() with a cached but expired value