CODEWITHSUNDEEP
pythonamazon-web-servicespytestboto3moto
Alyx
Alyx

Reputation: 9

How to mock correctly a class' aws client with moto?

So I'm doing tests for my lambdas using pytest and moto, the issue is that every lambda has parameters stored in Parameter Store (some also in Secrets Manager) but for the sake of caching I use a class that compiles all params:

    # Initializing the class for a global variable
    _cache_params = {}

    def __init__(self, region_name="us-east-1") -> None:
        self.region_name = region_name
        self.ssm = boto3.client("ssm", region_name=self.region_name)
        self.secrets = boto3.client("secretsmanager", region_name=self.region_name)

    def set_config(self, config_name: str):
        # After pre-setupping, and called from within a lambda/function params_getter.set_config("/environment-project/configuration")
        self._config_name = config_name
        params = self._load_params()
        secrets = self._load_secrets()
        self._cache_params = {
            **params,
            **secrets,
        }  # If keys between both dictionaries match secrets will overwrite params

    def _load_params(self):
        # this will fetch Parameter Store json values
        params = self.ssm.get_parameter(
            Name=f"{self._config_name}"  # For future use, the param must be named according to terraform such as "dev-project""project_config_params"
        )
        # make a log when no entity, return empty
        if "Parameter" not in params:
            logger.warning(f"Empty params for {self._config_name}")

        return json.loads(params.get("Parameter", {})["Value"])

    def _load_secrets(self):
        # this will fetch Secrets Manager values
        secrets = self.secrets.get_secret_value(SecretId=self._config_name)
        if "SecretString" not in secrets:
            logger.warning(f"Empty secrets for {self.config_name}")
        return json.loads(secrets.get("SecretString", {}))

    def get_param(self, paramName: str) -> str:
        try:
            return self._cache_params[paramName]  # Exception when None
        except Exception as e:
            raise e


config = ParamsGetter()

And it's just called at the beginning of every lambda, where CONFIG_NAME is the Name to a json containing everything:

import boto3
import email
from email import policy
from config.params_getter import ParamsGetter, config
import os
import logging

CONFIG_NAME = os.environ["CONFIG_NAME"]
config.set_config(CONFIG_NAME)
bucket = config.get_param("nameOfValue")

logging.basicConfig()
logger = logging.getLogger("lambda_function.py")
logger.setLevel("DEBUG")

def lambda_handler(event,_):

However I find it very difficult to mock ssm values in the conftest.py:

# ENV setup for lambdas
os.environ["CONFIG_NAME"] = "/test-myProject/configuration"


# Changed ENV setup to adapt to config get_params() ssm and secrets configuration
@pytest.fixture(autouse=True, scope="session")
def ssm_mock():
    with mock_ssm():
        ssm = boto3.client("ssm")
        # Mock all conf params
        ssm.put_parameter(
            Name="/test-myProject/configuration",
            Value=json.encoder(
                emailDataTable="VALUE",
                emailMetadataTable="VALUE",
                attachmentMetadataTable="VALUE",
                snsTopicArn="VALUE",
                snsRoleArn="VALUE",
                intakeBucket="VALUE",
                mulesoftAddress="VALUE",
            ),
            Type="String",
        )


@pytest.fixture(autouse=True, scope="session")
def secrets_mock():
    with mock_secretsmanager():
        secrets = boto3.client("secretsmanager")
        # Mock secrets
        secrets.create_secret(
            Name="/test-myProject/configuration",
            SecretString=json.encoder(
                clientId="VALUE",
                clientSecret="VALUE",
            ),
        )

In the end I keep getting this error: botocore.errorfactory.ParameterNotFound: An error occurred (ParameterNotFound) when calling the GetParameter operation Which obviously means that there are no Parameters with that name. I don't know much about moto or how should I configure my other test besides the conftest to handle these variables.

Upvotes: 0

Views: 1312

Answers (1)

Bert Blommers
Bert Blommers

Reputation: 2123

The SSM mock is started inside ssm_mock(), but the mock ends the moment the method ends. That's why the parameters cannot be reached, because Moto is no longer active.

If you yield the SSM-client, that will ensure that the test-method is executed while the mock_ssm is still active.

def ssm_client():
    with mock_ssm():
        ssm = boto3.client("ssm")
        # Mock all conf params
        ...
        yield ssm


def test_my_things(ssm_client):
    ssm_client.get_parameter(...)

See the documentation here: http://docs.getmoto.org/en/latest/docs/getting_started.html#example-on-usage

Upvotes: 0

Related Questions