chore: add tests for static_pk_value feature

Author: heitorlessa

tests/functional/idempotency/conftest.py

@@ -215,6 +215,13 @@ def persistence_store_compound(config):
     return DynamoDBPersistenceLayer(table_name=TABLE_NAME, boto_config=config, key_attr="id", sort_key_attr="sk")
 
 
+@pytest.fixture
+def persistence_store_compound_static_pk_value(config, static_pk_value):
+    return DynamoDBPersistenceLayer(
+        table_name=TABLE_NAME, boto_config=config, key_attr="id", sort_key_attr="sk", static_pk_value=static_pk_value
+    )
+
+
 @pytest.fixture
 def idempotency_config(config, request, default_jmespath):
     return IdempotencyConfig(
@@ -246,3 +253,58 @@ def _func_echo_decoder(self, value):
 @pytest.fixture
 def mock_function():
     return mock.MagicMock()
+
+
+@pytest.fixture
+def static_pk_value():
+    return "static-value"
+
+
+@pytest.fixture
+def expected_params_update_item_compound_key_static_pk_value(
+    serialized_lambda_response, hashed_idempotency_key, static_pk_value
+):
+    return {
+        "ExpressionAttributeNames": {
+            "#expiry": "expiration",
+            "#response_data": "data",
+            "#status": "status",
+        },
+        "ExpressionAttributeValues": {
+            ":expiry": {"N": stub.ANY},
+            ":response_data": {"S": serialized_lambda_response},
+            ":status": {"S": "COMPLETED"},
+        },
+        "Key": {"id": {"S": static_pk_value}, "sk": {"S": hashed_idempotency_key}},
+        "TableName": "TEST_TABLE",
+        "UpdateExpression": "SET #response_data = :response_data, " "#expiry = :expiry, #status = :status",
+    }
+
+
+@pytest.fixture
+def expected_params_put_item_compound_key_static_pk_value(hashed_idempotency_key, static_pk_value):
+    return {
+        "ConditionExpression": (
+            "attribute_not_exists(#id) OR #expiry < :now OR "
+            "(#status = :inprogress AND attribute_exists(#in_progress_expiry) AND #in_progress_expiry < :now_in_millis)"
+        ),
+        "ExpressionAttributeNames": {
+            "#id": "id",
+            "#expiry": "expiration",
+            "#status": "status",
+            "#in_progress_expiry": "in_progress_expiration",
+        },
+        "ExpressionAttributeValues": {
+            ":now": {"N": stub.ANY},
+            ":now_in_millis": {"N": stub.ANY},
+            ":inprogress": {"S": "INPROGRESS"},
+        },
+        "Item": {
+            "expiration": {"N": stub.ANY},
+            "in_progress_expiration": {"N": stub.ANY},
+            "id": {"S": static_pk_value},
+            "sk": {"S": hashed_idempotency_key},
+            "status": {"S": "INPROGRESS"},
+        },
+        "TableName": "TEST_TABLE",
+    }

tests/functional/idempotency/test_idempotency.py

@@ -1504,3 +1504,34 @@ def lambda_handler(event, context):
 
     stubber.assert_no_pending_responses()
     stubber.deactivate()
+
+
+@pytest.mark.parametrize("idempotency_config", [{"use_local_cache": False}], indirect=True)
+def test_idempotent_lambda_compound_static_pk_value_has_correct_pk(
+    idempotency_config: IdempotencyConfig,
+    persistence_store_compound_static_pk_value: DynamoDBPersistenceLayer,
+    lambda_apigw_event,
+    expected_params_put_item_compound_key_static_pk_value,
+    expected_params_update_item_compound_key_static_pk_value,
+    lambda_response,
+    lambda_context,
+):
+    """
+    Test idempotent decorator having a DynamoDBPersistenceLayer with a compound key and a static PK value
+    """
+
+    stubber = stub.Stubber(persistence_store_compound_static_pk_value._client)
+    ddb_response = {}
+
+    stubber.add_response("put_item", ddb_response, expected_params_put_item_compound_key_static_pk_value)
+    stubber.add_response("update_item", ddb_response, expected_params_update_item_compound_key_static_pk_value)
+    stubber.activate()
+
+    @idempotent(config=idempotency_config, persistence_store=persistence_store_compound_static_pk_value)
+    def lambda_handler(event, context):
+        return lambda_response
+
+    lambda_handler(lambda_apigw_event, lambda_context)
+
+    stubber.assert_no_pending_responses()
+    stubber.deactivate()