Imported function is not using mocked services

[MauriceHeinze]

Summary: When using Moto 5.0.9 with Python 3 on macOS, the put_item function is trying to connect to the real AWS DynamoDB service instead of using the Moto mock despite using the @mock_aws decorator.

The repository that int_repo is using uses boto3 to talk to DynamoDB.

What happens? The test is failing. botocore.errorfactory.ResourceNotFoundException: An error occurred (ResourceNotFoundException) when calling the PutItem operation: Requested resource not found

What should happen? The test should pass, because dynamodb connects to the mocked up one.

Lambda to test:

# pylint: disable=C0114
import json
import os
import time
import uuid

from repositories.kpi_repository import increment_kpi_counter
from common_dynamo_repository import (  # pylint: disable=E0401
    InternalDynamoDBRepository,
)

def lambda_handler(event, _context):
    int_repo = InternalDynamoDBRepository()
    user_id = event["requestContext"]["authorizer"]["claims"]["sub"]

    assessment_details = event["body"]
    assessment_details = json.loads(assessment_details)

    assessment_id = str(uuid.uuid4())

    new_assessment = {
        "PK": "USER#" + user_id,
        "SK": "ASSESSMENT#HEADER#" + assessment_id,
        "ASSESSMENT_name": assessment_details["name"],
        "GSI_ASSESSMENT": 1,
    }
    response = int_repo.put_item(os.environ["DATABASE_TABLE"], new_assessment)

    if response != 200:
        return {
            "statusCode": response,
            "body": "error",
            "headers": {"Access-Control-Allow-Origin": "*"},
        }

    return {
        "statusCode": 200,
        "body": json.dumps({"assessmentId": assessment_id}),
        "headers": {"Access-Control-Allow-Origin": "*"},
    }

Test:

import importlib.util
import json
import os
import sys
import time

path = os.path.abspath(__file__)
root_path = path.split("Functions")[0]
root_path += "DatabaseLayers//python//lib//python3.10//site-packages"
sys.path.append(root_path)

from decimal import Decimal

import boto3
from moto import mock_aws

os.environ["ENV_NAME"] = "dev"
os.environ["DATABASE_TABLE"] = os.environ["ENV_NAME"] + "PROJECTDatabaseTable"
os.environ["FILTER_DATABASE_TABLE"] = os.environ["ENV_NAME"] + "PROJECTFilterDatabaseTable"

@mock_aws
class TestResource:

    def test_create_assessment(self):
        # test set up
        script_path = self.get_script_path()
        spec = importlib.util.spec_from_file_location("lambda_function", script_path)
        lambda_function = spec.loader.load_module()
        self.table = self.create_table(os.environ["DATABASE_TABLE"])
        from common_dynamo_repository import InternalDynamoDBRepository

        user_list = [
            {
                "user_id": "1",
                "email": "test@email.com",
            }
        ]
        int_repo = InternalDynamoDBRepository()

        for user in user_list:
            self.add_user_application(int_repo, user["email"], user)

        context = None
        event = {
            "requestContext": {
                "authorizer": {
                    "claims": {
                        "sub": user_list[0]["user_id"],
                        "email": user_list[0]["email"],
                    }
                }
            }
        }

        assessment_details = {
            "name": "Assessment 012024",
            "building_name": "Building 01",
        }
        event["body"] = json.dumps(assessment_details)

        # test
        test = lambda_function.lambda_handler(event, context)
        response = json.loads(test["body"])
        assert "assessmentId" in response
        assert test["statusCode"] == 200
        dynamo_db = int_repo.scan_table(os.environ["DATABASE_TABLE"])

        print("DYNAMO DB CREATE ASSESSMENT")
        print(dynamo_db)

        assert len(dynamo_db) == 3

        assert dynamo_db[2]["PK"] == "USER#" + user_list[0]["user_id"]
        assert dynamo_db[2]["SK"].startswith("ASSESSMENT#HEADER#")
        assert dynamo_db[2]["GSI_ASSESSMENT"] == Decimal(1)
        assert dynamo_db[2]["ASSESSMENT_name"] == assessment_details["name"]

    def get_script_path(self):
        script_path = ""
        script_path_parts = path.rsplit("\\", 1)
        if len(script_path_parts) == 2:
            script_path = script_path_parts[0]
            script_path += "\\src\\lambda_function.py"
        else:
            script_path_parts = path.rsplit("/", 1)
            script_path = script_path_parts[0]
            script_path += "/src/lambda_function.py"
        return script_path

    def create_table(self, table_name):
        import boto3

        region = "us-east-1"
        dynamodb_client = boto3.client("dynamodb", region_name=region)
        input = self.create_table_input(table_name)
        try:
            response = dynamodb_client.create_table(**input)
            print("Successfully created table.")
        except BaseException as error:
            print(
                "Unknown error while creating table: "
                + error.response["Error"]["Message"]
            )

    def create_table_input(self, table_name):
        return {
            "TableName": table_name,
            "KeySchema": [
                {"AttributeName": "PK", "KeyType": "HASH"},
                {"AttributeName": "SK", "KeyType": "RANGE"},
            ],
            "BillingMode": "PROVISIONED",
            "AttributeDefinitions": [
                {"AttributeName": "PK", "AttributeType": "S"},
                {"AttributeName": "SK", "AttributeType": "S"},
            ],
            "ProvisionedThroughput": {"ReadCapacityUnits": 1, "WriteCapacityUnits": 1},
        }

    def add_user_application(self, int_repo, email, user_details):
        new_user = {
            "PK": "USER",
            "SK": user_details["user_id"],
            "User_Email": email,
        }
        dynamo_response = int_repo.put_item_if_not_exists(
            os.environ["DATABASE_TABLE"], "PK", new_user
        )
        return

[GustonMari]

I have the same error, but sometimes it works and the mocking operation is working

[bblommers]

@MauriceHeinze Based on the provided code I can't see anything wrong - but I also can't reproduce it like this.

Can you provide an MRE (Minimal Reproducible Example)? That would make debugging a little easier.