使用moto对Boto3 DynamoDB代码单元测试时出现Scan操作ResourceNotFoundException错误

问题根因 #

• fixture执行顺序错误：你设置了dynamodb_instance为autouse=True，它会在dynamodb_client、aws_credentials以及mock_dynamodb2上下文生效之前就完成UsagePlan的初始化，此时UsagePlan父类中创建的boto3连接未被moto拦截，走的是真实AWS访问链路。
• 表名不匹配：业务代码中UsagePlan固定使用Usage作为表名，但测试代码里创建的测试表名为SwuTemp，即使mock生效也找不到对应表。
• 初始化上下文错误：UsagePlan的初始化过程没有被包裹在mock_dynamodb2的上下文范围内，导致内部的boto3调用未被mock。

修复步骤 #

1. 调整fixture依赖关系，取消dynamodb_instance的autouse配置，让它依赖dynamodb_client保证mock先生效，初始化过程放在mock上下文内。
2. 统一表名：将测试代码中创建的表名改为和业务代码一致的Usage，或者给UsagePlan类增加自定义表名的传参能力，测试时传入测试表名。
3. 补充moto创建表的必填参数ProvisionedThroughput，并增加表状态等待逻辑，避免表未创建完成就发起请求。

修改后的测试代码示例 #

from dynamodb_layer.dynamodb_layer import UsagePlan
import os
import boto3
import pytest
from moto import mock_dynamodb2

# 统一表名和业务代码一致
def table_name():
    return 'Usage'

def s1_details():
    return {'api_keys': [{'api_key': 'abcdefghijk987654321', 'api_key_id': '1newosh86', 'api_key_name': 'Quota Test'}, {'api_key': '53HGyaDTej57777nxnyE9KoGMnfl8tW21dPmRuPh6', 'api_key_id': 'wyengereu0j', 'api_key_name': 'mytestapi'}], 'usage_plan_id': 'jvpoe7', 'last_updated_date': '2021-09-07T20:07:35.695015', 'quota': {'limit': 150, 'offset': 0, 'period': 'MONTH'}, 'recently_updated': True, 'used_quota': {'wyengereu0j': [[0, 150], [0, 150], [0, 150], [0, 150], [5, 0], [0, 0], [0, 0]], '1newosh86': [[31, 9], [0, 9], [0, 9], [0, 9], [0, 9], [0, 9], [0, 9]]}, 'throttle': {'burstLimit': 3, 'rateLimit': 2}, 'stages': [{'name': 'myapitest', 'throttle': {'/camps/{id}/POST': {'burstLimit': 2, 'rateLimit': '1.0'}}, 'stage': 'Prod', 'apiId': 'qjia099d0'}], 'description': 'This is my description', 'name': 'Quota Test Usage plan'}

# 补充原代码缺失的s2_details实现
def s2_details():
    test_data = s1_details()
    test_data['usage_plan_id'] = 'test_id_2'
    return test_data

def create_table(init, table_name):
    table = init.create_table(
                TableName=table_name,
                KeySchema=[
                    {
                        'AttributeName': 'usage_plan_id',
                        'KeyType': 'HASH'
                    }
                ],
                AttributeDefinitions=[
                    {
                        'AttributeName': 'usage_plan_id',
                        'AttributeType': 'S'
                    }
                ],
                # moto要求必填吞吐配置参数
                ProvisionedThroughput={
                    'ReadCapacityUnits': 1,
                    'WriteCapacityUnits': 1
                }
            )
    # 等待表创建完成
    table.wait_until_exists()
    return table

def put_item(init, usage_plan_dict):
    return init.put_item(Item={'usage_plan_id': usage_plan_dict['usage_plan_id'],
                        'api_keys': usage_plan_dict['api_keys'],
                        'description': usage_plan_dict['description'],
                        'last_updated_date': usage_plan_dict['last_updated_date'],
                        'name': usage_plan_dict['name'],
                        'quota': usage_plan_dict['quota'],
                        'recently_updated': usage_plan_dict['recently_updated'],
                        'stages': usage_plan_dict['stages'],
                        'throttle': usage_plan_dict['throttle'],
                        'used_quota': usage_plan_dict['used_quota']
                        })

@pytest.fixture
def aws_credentials():
    """Mocked AWS Credentials for moto."""
    os.environ["AWS_ACCESS_KEY_ID"] = "testing"
    os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
    os.environ["AWS_SECURITY_TOKEN"] = "testing"
    os.environ["AWS_SESSION_TOKEN"] = "testing"

@pytest.fixture
def dynamodb_client(aws_credentials):
    """Mock and initialize connection"""
    with mock_dynamodb2():
        boto3.setup_default_session()
        client = boto3.resource('dynamodb', region_name='us-east-1')
        table = create_table(client, table_name())
        for plan in [s1_details(), s2_details()]:
            put_item(table, plan)
        yield table

# 调整fixture依赖，保证mock生效后再初始化UsagePlan
@pytest.fixture
def dynamodb_instance(dynamodb_client):
    yield UsagePlan('usage_plan_id')

def test_get_plans_from_dynamodb(dynamodb_client, dynamodb_instance):
    response = dynamodb_instance.get_plans_from_dynamodb()
    print(response)
    assert len(response['Items']) == 2

内容的提问来源于stack exchange，提问作者user2993021