KeyError:尝试模拟DynamoDB时出现“操作”

待测试代码:

class ElasticSearchReporter:
    def __init__(...):
        ...
        self._dynamodb_client = session.client('dynamodb')

    def upload_record_to_dynamodb(self, es_index, dataclass_instance):
        record_id = str(uuid.uuid4())
        ...
        dataclass_attributes[current_field.name] = {'Value': attribute_value}
        record_key = {'record_id': {'S': record_id}}
        self._dynamodb_client.update_item(TableName=es_index, Key=record_key, AttributeUpdates=dataclass_attributes)

为了测试它,我这样做:

import unittest
import boto3

from moto import mock_dynamodb2
from unittest.mock import MagicMock, patch

class TestElasticSearch(unittest.TestCase):
    def setUp(self):
        self.es_record = ...

    @mock_dynamodb2
    @patch('boto3.client')
    def test_upload_record_to_dynamodb(self, mock_boto3_sts_client):
        table_name = 'my_table'
        dynamodb = boto3.resource('dynamodb', region_name='eu-west-1')

        table = dynamodb.create_table(
            TableName=table_name,
            KeySchema=[
                {
                    'AttributeName': 'record_id',
                    'KeyType': 'HASH'
                },
            ],
            AttributeDefinitions=[
                {
                    'AttributeName': 'record_id',
                    'AttributeType': 'S'
                },

            ]
        )
        # Wait until the table exists.
        table.meta.client.get_waiter('table_exists').wait(TableName=table_name)

        rtn_val = {
            'Credentials': {
                'AccessKeyId': 'test_access_key',
                'SecretAccessKey': 'test',
                'SessionToken': 'test_session',
            }
        }
        mock_sts_client = MagicMock()
        mock_sts_client.assume_role.return_value = rtn_val
        mock_boto3_sts_client.return_value = mock_sts_client

        es_reporter = ElasticSearchReporter(...)
        es_reporter.upload_record_to_dynamodb(table_name, self.es_record)

当我运行测试时,它失败了:

         es_reporter = ElasticSearchReporter(self.role_arn, None)
         es_reporter.upload_record_to_dynamodb(
             ElasticsearchIndices.ANALYSIS_BACKEND_SESSIONS_ALPHA,
 >           self.es_record
         )
 
 test/test_common/test_helpers/test_elasticsearch.py:167: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
 ../../build/DEV.STD.PTHREAD/build/lib/python3.7/site-packages/atvvqadac_common/helpers/elasticsearch.py:143: in upload_record_to_dynamodb
    self._dynamodb_client.update_item(TableName=es_index, Key=record_key, AttributeUpdates=dataclass_attributes)
<several lines here that I skip>
    item.update_with_attribute_updates(attribute_updates)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _  
 self = Item: {'Attributes': {'record_id': {'S': '6266046e-95c5-4448-aa2d-8d4959b6e306'}}}
 attribute_updates = {'aggregator_results_s3_path': {'Value': {'S': 's3:/my_bucket/2020-11-06T110210251207_81_my_session/inference-results/...:02:10.390290'}}, 'audio_s3_path': {'Value': {'S': 's3:/my_bucket/2020-11-06T110210251207_81_my_session/audio/'}}, ...}
 
     def update_with_attribute_updates(self, attribute_updates):
         for attribute_name, update_action in attribute_updates.items():
 >           action = update_action['Action']
 E           KeyError: 'Action'
 
 /home/gsamaras/pkg-cache/packages/Python-moto/Python-moto-1.x.67811.0/AL2012/DEV.STD.PTHREAD/build/lib/python3.7/site-packages/moto/dynamodb2/models.py:293: KeyError
----------------------------- Captured stderr call -----------------------------
020-11-06 15:55:29,175 INFO [botocore.credentials] Found credentials in environment variables.
2020-11-06 15:55:29 INFO [elastic_search] Record about to be sent to DynamoDB (record ID: 6266046e-95c5-4448-aa2d-8d4959b6e306): {'stage': {<and so on..>

你能帮我修一下吗?

参考:

1. how-to-mock-aws-dynamodb-service