test_dynamodb_condition_expressions.py

import re
from decimal import Decimal
from uuid import uuid4

import boto3
import pytest
from botocore.exceptions import ClientError

from moto import mock_aws


@mock_aws
def test_condition_expression_with_dot_in_attr_name():
    dynamodb = boto3.resource("dynamodb", region_name="us-east-2")
    table_name = f"T{uuid4()}"
    dynamodb.create_table(
        TableName=table_name,
        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
        AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
        BillingMode="PAY_PER_REQUEST",
    )
    table = dynamodb.Table(table_name)

    email_like_str = "test@foo.com"
    record = {"id": "key-0", "first": {email_like_str: {"third": {"VALUE"}}}}
    table.put_item(Item=record)

    table.update_item(
        Key={"id": "key-0"},
        UpdateExpression="REMOVE #first.#second, #other",
        ExpressionAttributeNames={
            "#first": "first",
            "#second": email_like_str,
            "#third": "third",
            "#other": "other",
        },
        ExpressionAttributeValues={":value": "VALUE", ":one": 1},
        ConditionExpression="size(#first.#second.#third) = :one AND contains(#first.#second.#third, :value)",
        ReturnValues="ALL_NEW",
    )

    item = table.get_item(Key={"id": "key-0"})["Item"]
    assert item == {"id": "key-0", "first": {}}


@mock_aws
def test_condition_expressions():
    client = boto3.client("dynamodb", region_name="us-east-1")
    table_name = f"T{uuid4()}"

    # Create the DynamoDB table.
    client.create_table(
        TableName=table_name,
        AttributeDefinitions=[
            {"AttributeName": "client", "AttributeType": "S"},
            {"AttributeName": "app", "AttributeType": "S"},
        ],
        KeySchema=[
            {"AttributeName": "client", "KeyType": "HASH"},
            {"AttributeName": "app", "KeyType": "RANGE"},
        ],
        ProvisionedThroughput={"ReadCapacityUnits": 123, "WriteCapacityUnits": 123},
    )
    client.put_item(
        TableName=table_name,
        Item={
            "client": {"S": "client1"},
            "app": {"S": "app1"},
            "match": {"S": "match"},
            "existing": {"S": "existing"},
        },
    )

    client.put_item(
        TableName=table_name,
        Item={
            "client": {"S": "client1"},
            "app": {"S": "app1"},
            "match": {"S": "match"},
            "existing": {"S": "existing"},
        },
        ConditionExpression="attribute_exists(#existing) AND attribute_not_exists(#nonexistent) AND #match = :match",
        ExpressionAttributeNames={
            "#existing": "existing",
            "#nonexistent": "nope",
            "#match": "match",
        },
        ExpressionAttributeValues={":match": {"S": "match"}},
    )

    client.put_item(
        TableName=table_name,
        Item={
            "client": {"S": "client1"},
            "app": {"S": "app1"},
            "match": {"S": "match"},
            "existing": {"S": "existing"},
        },
        ConditionExpression="NOT(attribute_exists(#nonexistent1) AND attribute_exists(#nonexistent2))",
        ExpressionAttributeNames={"#nonexistent1": "nope", "#nonexistent2": "nope2"},
    )

    client.put_item(
        TableName=table_name,
        Item={
            "client": {"S": "client1"},
            "app": {"S": "app1"},
            "match": {"S": "match"},
            "existing": {"S": "existing"},
        },
        ConditionExpression="attribute_exists(#nonexistent) OR attribute_exists(#existing)",
        ExpressionAttributeNames={"#nonexistent": "nope", "#existing": "existing"},
    )

    client.put_item(
        TableName=table_name,
        Item={
            "client": {"S": "client1"},
            "app": {"S": "app1"},
            "match": {"S": "match"},
            "existing": {"S": "existing"},
        },
        ConditionExpression="#client BETWEEN :a AND :z",
        ExpressionAttributeNames={"#client": "client"},
        ExpressionAttributeValues={":a": {"S": "a"}, ":z": {"S": "z"}},
    )

    client.put_item(
        TableName=table_name,
        Item={
            "client": {"S": "client1"},
            "app": {"S": "app1"},
            "match": {"S": "match"},
            "existing": {"S": "existing"},
        },
        ConditionExpression="#client IN (:client1, :client2)",
        ExpressionAttributeNames={"#client": "client"},
        ExpressionAttributeValues={
            ":client1": {"S": "client1"},
            ":client2": {"S": "client2"},
        },
    )

    with pytest.raises(client.exceptions.ConditionalCheckFailedException):
        client.put_item(
            TableName=table_name,
            Item={
                "client": {"S": "client1"},
                "app": {"S": "app1"},
                "match": {"S": "match"},
                "existing": {"S": "existing"},
            },
            ConditionExpression="attribute_exists(#nonexistent1) AND attribute_exists(#nonexistent2)",
            ExpressionAttributeNames={
                "#nonexistent1": "nope",
                "#nonexistent2": "nope2",
            },
        )

    with pytest.raises(client.exceptions.ConditionalCheckFailedException):
        client.put_item(
            TableName=table_name,
            Item={
                "client": {"S": "client1"},
                "app": {"S": "app1"},
                "match": {"S": "match"},
                "existing": {"S": "existing"},
            },
            ConditionExpression="NOT(attribute_not_exists(#nonexistent1) AND attribute_not_exists(#nonexistent2))",
            ExpressionAttributeNames={
                "#nonexistent1": "nope",
                "#nonexistent2": "nope2",
            },
        )

    with pytest.raises(client.exceptions.ConditionalCheckFailedException):
        client.put_item(
            TableName=table_name,
            Item={
                "client": {"S": "client1"},
                "app": {"S": "app1"},
                "match": {"S": "match"},
                "existing": {"S": "existing"},
            },
            ConditionExpression="attribute_exists(#existing) AND attribute_not_exists(#nonexistent) AND #match = :match",
            ExpressionAttributeNames={
                "#existing": "existing",
                "#nonexistent": "nope",
                "#match": "match",
            },
            ExpressionAttributeValues={":match": {"S": "match2"}},
        )

    # Make sure update_item honors ConditionExpression as well
    client.update_item(
        TableName=table_name,
        Key={"client": {"S": "client1"}, "app": {"S": "app1"}},
        UpdateExpression="set #match=:match",
        ConditionExpression="attribute_exists(#existing)",
        ExpressionAttributeNames={"#existing": "existing", "#match": "match"},
        ExpressionAttributeValues={":match": {"S": "match"}},
    )

    with pytest.raises(client.exceptions.ConditionalCheckFailedException) as exc:
        client.update_item(
            TableName=table_name,
            Key={"client": {"S": "client1"}, "app": {"S": "app1"}},
            UpdateExpression="set #match=:match",
            ConditionExpression="attribute_not_exists(#existing)",
            ExpressionAttributeValues={":match": {"S": "match"}},
            ExpressionAttributeNames={"#existing": "existing", "#match": "match"},
        )
    _assert_conditional_check_failed_exception(exc)

    with pytest.raises(client.exceptions.ConditionalCheckFailedException) as exc:
        client.update_item(
            TableName=table_name,
            Key={"client": {"S": "client2"}, "app": {"S": "app1"}},
            UpdateExpression="set #match=:match",
            ConditionExpression="attribute_exists(#existing)",
            ExpressionAttributeValues={":match": {"S": "match"}},
            ExpressionAttributeNames={"#existing": "existing", "#match": "match"},
        )
    _assert_conditional_check_failed_exception(exc)

    with pytest.raises(client.exceptions.ConditionalCheckFailedException):
        client.delete_item(
            TableName=table_name,
            Key={"client": {"S": "client1"}, "app": {"S": "app1"}},
            ConditionExpression="attribute_not_exists(#existing)",
            ExpressionAttributeValues={":match": {"S": "match"}},
            ExpressionAttributeNames={"#existing": "existing"},
        )


def _assert_conditional_check_failed_exception(exc):
    err = exc.value.response["Error"]
    assert err["Code"] == "ConditionalCheckFailedException"
    assert err["Message"] == "The conditional request failed"


@mock_aws
def test_condition_expression_numerical_attribute():
    dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
    table = dynamodb.create_table(
        TableName=f"T{uuid4()}",
        KeySchema=[{"AttributeName": "partitionKey", "KeyType": "HASH"}],
        AttributeDefinitions=[{"AttributeName": "partitionKey", "AttributeType": "S"}],
        BillingMode="PAY_PER_REQUEST",
    )
    table.put_item(Item={"partitionKey": "pk-pos", "myAttr": 5})
    table.put_item(Item={"partitionKey": "pk-neg", "myAttr": -5})

    # try to update the item we put in the table using numerical condition expression
    # Specifically, verify that we can compare with a zero-value
    # First verify that > and >= work on positive numbers
    update_numerical_con_expr(
        key="pk-pos", con_expr="myAttr > :zero", res="6", table=table
    )
    update_numerical_con_expr(
        key="pk-pos", con_expr="myAttr >= :zero", res="7", table=table
    )
    # Second verify that < and <= work on negative numbers
    update_numerical_con_expr(
        key="pk-neg", con_expr="myAttr < :zero", res="-4", table=table
    )
    update_numerical_con_expr(
        key="pk-neg", con_expr="myAttr <= :zero", res="-3", table=table
    )


def update_numerical_con_expr(key, con_expr, res, table):
    table.update_item(
        Key={"partitionKey": key},
        UpdateExpression="ADD myAttr :one",
        ExpressionAttributeValues={":zero": 0, ":one": 1},
        ConditionExpression=con_expr,
    )
    assert table.get_item(Key={"partitionKey": key})["Item"]["myAttr"] == Decimal(res)


@mock_aws
def test_condition_expression__attr_doesnt_exist():
    client = boto3.client("dynamodb", region_name="us-east-1")
    table_name = f"T{uuid4()}"

    client.create_table(
        TableName=table_name,
        KeySchema=[{"AttributeName": "forum_name", "KeyType": "HASH"}],
        AttributeDefinitions=[{"AttributeName": "forum_name", "AttributeType": "S"}],
        ProvisionedThroughput={"ReadCapacityUnits": 1, "WriteCapacityUnits": 1},
    )

    client.put_item(
        TableName=table_name, Item={"forum_name": {"S": "foo"}, "ttl": {"N": "4567.89"}}
    )

    def update_if_attr_doesnt_exist():
        # Test nonexistent top-level attribute.
        client.update_item(
            TableName=table_name,
            Key={"forum_name": {"S": "the-key"}},
            UpdateExpression="set #new_state=:new_state, #ttl=:ttl",
            ConditionExpression="attribute_not_exists(#new_state)",
            ExpressionAttributeNames={"#new_state": "foobar", "#ttl": "ttl"},
            ExpressionAttributeValues={
                ":new_state": {"S": "some-value"},
                ":ttl": {"N": "12345.67"},
            },
            ReturnValues="ALL_NEW",
        )

    update_if_attr_doesnt_exist()

    # Second time should fail
    with pytest.raises(client.exceptions.ConditionalCheckFailedException) as exc:
        update_if_attr_doesnt_exist()
    _assert_conditional_check_failed_exception(exc)


@mock_aws
def test_condition_expression__or_order():
    client = boto3.client("dynamodb", region_name="us-east-1")
    table_name = f"T{uuid4()}"

    client.create_table(
        TableName=table_name,
        KeySchema=[{"AttributeName": "forum_name", "KeyType": "HASH"}],
        AttributeDefinitions=[{"AttributeName": "forum_name", "AttributeType": "S"}],
        ProvisionedThroughput={"ReadCapacityUnits": 1, "WriteCapacityUnits": 1},
    )

    # ensure that the RHS of the OR expression is not evaluated if the LHS
    # returns true (as it would result an error)
    client.update_item(
        TableName=table_name,
        Key={"forum_name": {"S": "the-key"}},
        UpdateExpression="set #ttl=:ttl",
        ConditionExpression="attribute_not_exists(#ttl) OR #ttl <= :old_ttl",
        ExpressionAttributeNames={"#ttl": "ttl"},
        ExpressionAttributeValues={":ttl": {"N": "6"}, ":old_ttl": {"N": "5"}},
    )


@mock_aws
def test_condition_expression__and_order():
    client = boto3.client("dynamodb", region_name="us-east-1")
    table_name = f"T{uuid4()}"

    client.create_table(
        TableName=table_name,
        KeySchema=[{"AttributeName": "forum_name", "KeyType": "HASH"}],
        AttributeDefinitions=[{"AttributeName": "forum_name", "AttributeType": "S"}],
        ProvisionedThroughput={"ReadCapacityUnits": 1, "WriteCapacityUnits": 1},
    )

    # ensure that the RHS of the AND expression is not evaluated if the LHS
    # returns true (as it would result an error)
    with pytest.raises(client.exceptions.ConditionalCheckFailedException) as exc:
        client.update_item(
            TableName=table_name,
            Key={"forum_name": {"S": "the-key"}},
            UpdateExpression="set #ttl=:ttl",
            ConditionExpression="attribute_exists(#ttl) AND #ttl <= :old_ttl",
            ExpressionAttributeNames={"#ttl": "ttl"},
            ExpressionAttributeValues={":ttl": {"N": "6"}, ":old_ttl": {"N": "5"}},
        )
    _assert_conditional_check_failed_exception(exc)


@mock_aws
def test_condition_expression_with_reserved_keyword_as_attr_name():
    dynamodb = boto3.resource("dynamodb", region_name="us-east-2")
    table_name = f"T{uuid4()}"
    dynamodb.create_table(
        TableName=table_name,
        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
        AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
        BillingMode="PAY_PER_REQUEST",
    )
    table = dynamodb.Table(table_name)

    email_like_str = "test@foo.com"
    record = {"id": "key-0", "first": {email_like_str: {"end": {"VALUE"}}}}
    table.put_item(Item=record)

    expected_error_message = re.escape(
        "An error occurred (ValidationException) when "
        "calling the UpdateItem operation: Invalid ConditionExpression: Attribute name "
        "is a reserved keyword; reserved keyword: end"
    )
    with pytest.raises(
        dynamodb.meta.client.exceptions.ClientError, match=expected_error_message
    ):
        table.update_item(
            Key={"id": "key-0"},
            UpdateExpression="REMOVE #first.#second, #other",
            ExpressionAttributeNames={
                "#first": "first",
                "#second": email_like_str,
                "#other": "other",
            },
            ExpressionAttributeValues={":value": "VALUE", ":one": 1},
            ConditionExpression="size(#first.#second.end) = :one AND contains(#first.#second.end, :value)",
            ReturnValues="ALL_NEW",
        )

    # table is unchanged
    item = table.get_item(Key={"id": "key-0"})["Item"]
    assert item == record

    # using attribute names solves the issue
    table.update_item(
        Key={"id": "key-0"},
        UpdateExpression="REMOVE #first.#second, #other",
        ExpressionAttributeNames={
            "#first": "first",
            "#second": email_like_str,
            "#other": "other",
            "#end": "end",
        },
        ExpressionAttributeValues={":value": "VALUE", ":one": 1},
        ConditionExpression="size(#first.#second.#end) = :one AND contains(#first.#second.#end, :value)",
        ReturnValues="ALL_NEW",
    )

    item = table.get_item(Key={"id": "key-0"})["Item"]
    assert item == {"id": "key-0", "first": {}}


@mock_aws
def test_condition_expression_parentheses_behavior():
    client = boto3.client("dynamodb", region_name="us-east-1")
    table_name = f"T{uuid4()}"

    client.create_table(
        TableName=table_name,
        KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}],
        AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}],
        BillingMode="PAY_PER_REQUEST",
    )
    client.put_item(
        TableName=table_name,
        Item={
            "pk": {"S": "pk"},
            "a": {"N": "1"},
            "b": {"N": "2"},
            "c": {"N": "3"},
            "e": {"N": "4"},
        },
    )

    # Test Case 1: #a = :b OR (#c = :d AND #e = :f)
    # AWS DDB allows this. Moto should too.
    client.update_item(
        TableName=table_name,
        Key={"pk": {"S": "pk"}},
        UpdateExpression="SET z = :z",
        ConditionExpression="#a = :b OR (#c = :d AND #e = :f)",
        ExpressionAttributeNames={"#a": "pk", "#c": "pk", "#e": "pk"},
        ExpressionAttributeValues={
            ":b": {"S": "pk"},
            ":d": {"S": "pk"},
            ":f": {"S": "pk"},
            ":z": {"S": "updated1"},
        },
    )

    # Test Case 2: (attribute_exists (#0)) AND (((#1 <> :0) AND (#1 <> :1)) AND (#2 = :3))
    # AWS DDB allows this. Moto should too.
    client.update_item(
        TableName=table_name,
        Key={"pk": {"S": "pk"}},
        UpdateExpression="SET z = :z",
        ConditionExpression="(attribute_exists (#0)) AND (((#1 <> :0) AND (#1 <> :1)) AND (#2 = :3))",
        ExpressionAttributeNames={"#0": "pk", "#1": "pk", "#2": "pk"},
        ExpressionAttributeValues={
            ":0": {"S": "nope"},
            ":1": {"S": "nope"},
            ":3": {"S": "pk"},
            ":z": {"S": "updated2"},
        },
    )

    # Test Case 3: ((((a < b))))
    # AWS DDB fails this. Moto should too.
    with pytest.raises(ClientError) as exc:
        client.update_item(
            TableName=table_name,
            Key={"pk": {"S": "pk"}},
            UpdateExpression="SET z = :z",
            ConditionExpression="((((a < b))))",
            ExpressionAttributeValues={":z": {"S": "updated3"}},
        )

    err = exc.value.response["Error"]
    assert err["Code"] == "ValidationException"
    assert (
        err["Message"]
        == "Invalid ConditionExpression: The expression has redundant parentheses;"
    )

    # Test Case 4: ((#a = :b) OR (#c = :d) AND (#e = :f))
    # AWS DDB allows this. Moto should too.
    client.update_item(
        TableName=table_name,
        Key={"pk": {"S": "pk"}},
        UpdateExpression="SET z = :z",
        ConditionExpression="((#a = :b) OR (#c = :d) AND (#e = :f))",
        ExpressionAttributeNames={"#a": "pk", "#c": "pk", "#e": "pk"},
        ExpressionAttributeValues={
            ":b": {"S": "pk"},
            ":d": {"S": "pk"},
            ":f": {"S": "pk"},
            ":z": {"S": "updated4"},
        },
    )


@mock_aws
def test_condition_expression_allows_required_parentheses():
    client = boto3.client("dynamodb", region_name="us-east-1")
    table_name = f"T{uuid4()}"

    client.create_table(
        TableName=table_name,
        KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}],
        AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}],
        BillingMode="PAY_PER_REQUEST",
    )
    client.put_item(
        TableName=table_name,
        Item={
            "pk": {"S": "pk"},
            "a": {"S": "match"},
            "c": {"S": "match"},
            "e": {"S": "match"},
        },
    )

    client.update_item(
        TableName=table_name,
        Key={"pk": {"S": "pk"}},
        UpdateExpression="SET #z = :z",
        ConditionExpression="#a = :b AND (#c = :d OR #e = :f)",
        ExpressionAttributeNames={"#a": "a", "#c": "c", "#e": "e", "#z": "z"},
        ExpressionAttributeValues={
            ":b": {"S": "match"},
            ":d": {"S": "match"},
            ":f": {"S": "nope"},
            ":z": {"S": "updated"},
        },
    )

    item = client.get_item(TableName=table_name, Key={"pk": {"S": "pk"}})["Item"]
    assert item["z"] == {"S": "updated"}


@mock_aws
def test_condition_check_failure_exception_is_raised_when_values_are_returned_for_an_item_with_a_top_level_list():
    # This explicitly tests for a failure in handling JSONification of DynamoType
    # when lists are at the top level of an item.
    # This exception should not be raised:
    # TypeError: Object of type DynamoType is not JSON serializable

    dynamodb_client = boto3.client("dynamodb", region_name="us-east-1")
    table_name = f"T{uuid4()}"
    dynamodb_client.create_table(
        TableName=table_name,
        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
        AttributeDefinitions=[
            {"AttributeName": "id", "AttributeType": "S"},
        ],
        BillingMode="PAY_PER_REQUEST",
    )
    record = {
        "id": {"S": "example_id"},
        "some_list": {"L": [{"M": {"hello": {"S": "h"}}}]},
    }
    dynamodb_client.put_item(
        TableName=table_name,
        Item=record,
    )

    with pytest.raises(ClientError) as error:
        dynamodb_client.update_item(
            TableName=table_name,
            Key={"id": {"S": "example_id"}},
            UpdateExpression="set some_list=list_append(some_list, :w)",
            ExpressionAttributeValues={
                ":w": {"L": [{"M": {"world": {"S": "w"}}}]},
                ":id": {"S": "incorrect id"},
            },
            ConditionExpression="id = :id",
            ReturnValuesOnConditionCheckFailure="ALL_OLD",
        )

    assert error.type.__name__ == "ConditionalCheckFailedException"
    assert error.value.response["Error"] == {
        "Message": "The conditional request failed",
        "Code": "ConditionalCheckFailedException",
    }
    assert error.value.response["Item"] == {
        "id": {"S": "example_id"},
        "some_list": {"L": [{"M": {"hello": {"S": "h"}}}]},
    }


@mock_aws
def test_condition_check_failure_exception_is_raised_when_values_are_returned_for_an_item_with_a_top_level_string_set():
    # This explicitly tests for a failure in handling JSONification of DynamoType
    # when string sets are at the top level of an item.
    # These exception should not be raised:
    # TypeError: Object of type DynamoType is not JSON serializable
    # AttributeError: 'str' object has no attribute 'to_regular_json'

    dynamodb_client = boto3.client("dynamodb", region_name="us-east-1")
    table_name = f"T{uuid4()}"
    dynamodb_client.create_table(
        TableName=table_name,
        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
        AttributeDefinitions=[
            {"AttributeName": "id", "AttributeType": "S"},
        ],
        BillingMode="PAY_PER_REQUEST",
    )
    record = {
        "id": {"S": "example_id"},
        "some_list": {"SS": ["hello"]},
    }
    dynamodb_client.put_item(
        TableName=table_name,
        Item=record,
    )

    with pytest.raises(ClientError) as error:
        dynamodb_client.update_item(
            TableName=table_name,
            Key={"id": {"S": "example_id"}},
            UpdateExpression="set some_list=list_append(some_list, :w)",
            ExpressionAttributeValues={
                ":w": {"SS": ["world"]},
                ":id": {"S": "incorrect id"},
            },
            ConditionExpression="id = :id",
            ReturnValuesOnConditionCheckFailure="ALL_OLD",
        )

    assert error.type.__name__ == "ConditionalCheckFailedException"
    assert error.value.response["Error"] == {
        "Message": "The conditional request failed",
        "Code": "ConditionalCheckFailedException",
    }
    assert error.value.response["Item"] == {
        "id": {"S": "example_id"},
        "some_list": {"SS": ["hello"]},
    }


@mock_aws
def test_condition_check_failure_exception_is_raised_when_values_are_returned_for_an_item_with_a_list_in_a_map():
    # This explicitly tests for a failure in handling JSONification of DynamoType
    # when lists are inside a map

    dynamodb_client = boto3.client("dynamodb", region_name="us-east-1")
    table_name = f"T{uuid4()}"
    dynamodb_client.create_table(
        TableName=table_name,
        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
        AttributeDefinitions=[
            {"AttributeName": "id", "AttributeType": "S"},
        ],
        BillingMode="PAY_PER_REQUEST",
    )
    record = {
        "id": {"S": "example_id"},
        "some_list_in_a_map": {
            "M": {"some_list": {"L": [{"M": {"hello": {"S": "h"}}}]}}
        },
    }
    dynamodb_client.put_item(TableName=table_name, Item=record)

    with pytest.raises(ClientError) as error:
        dynamodb_client.update_item(
            TableName=table_name,
            Key={"id": {"S": "example_id"}},
            UpdateExpression="set some_list_in_a_map.some_list=list_append(some_list_in_a_map.some_list, :w)",
            ExpressionAttributeValues={
                ":w": {"L": [{"M": {"world": {"S": "w"}}}]},
                ":id": {"S": "incorrect id"},
            },
            ConditionExpression="id = :id",
            ReturnValuesOnConditionCheckFailure="ALL_OLD",
        )

    assert error.type.__name__ == "ConditionalCheckFailedException"
    assert error.value.response["Error"] == {
        "Message": "The conditional request failed",
        "Code": "ConditionalCheckFailedException",
    }
    assert error.value.response["Item"] == {
        "id": {"S": "example_id"},
        "some_list_in_a_map": {
            "M": {"some_list": {"L": [{"M": {"hello": {"S": "h"}}}]}}
        },
    }


@mock_aws
def test_conditional_check_failed_bytes():
    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
    dynamodb.create_table(
        TableName="test_table_bytes",
        KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}],
        AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}],
        BillingMode="PAY_PER_REQUEST",
    )

    dynamodb.put_item(
        TableName="test_table_bytes",
        Item={
            "pk": {"S": "test"},
            "my_bytes": {"B": b"somebytes"},
            "my_bytes_set": {"BS": [b"byte1", b"byte2"]},
        },
    )

    with pytest.raises(ClientError) as exc:
        dynamodb.update_item(
            TableName="test_table_bytes",
            Key={"pk": {"S": "test"}},
            UpdateExpression="SET my_str = :s",
            ConditionExpression="attribute_not_exists(pk)",
            ExpressionAttributeValues={":s": {"S": "newstr"}},
            ReturnValuesOnConditionCheckFailure="ALL_OLD",
        )

    assert exc.value.response["Error"]["Code"] == "ConditionalCheckFailedException"
    assert "Item" in exc.value.response
    assert exc.value.response["Item"]["my_bytes"]["B"] == b"somebytes"
    assert exc.value.response["Item"]["my_bytes_set"]["BS"] == [b"byte1", b"byte2"]


@mock_aws
def test_between_condition_includes_zero():
    """
    A numeric attribute value of exactly 0 must satisfy a BETWEEN range that
    includes it; previously FuncBetween tested the attribute with bare
    truthiness, so Decimal("0") was treated as missing and excluded.
    """
    dynamodb = boto3.resource("dynamodb", region_name="us-east-2")
    table_name = f"T{uuid4()}"
    dynamodb.create_table(
        TableName=table_name,
        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
        AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
        BillingMode="PAY_PER_REQUEST",
    )
    table = dynamodb.Table(table_name)

    table.put_item(Item={"id": "zero", "price": 0})
    table.put_item(Item={"id": "five", "price": 5})

    result = table.scan(
        FilterExpression="price BETWEEN :lo AND :hi",
        ExpressionAttributeValues={":lo": 0, ":hi": 100},
    )
    assert {item["id"] for item in result["Items"]} == {"zero", "five"}

    result = table.scan(
        FilterExpression="price BETWEEN :lo AND :hi",
        ExpressionAttributeValues={":lo": -10, ":hi": 0},
    )
    assert {item["id"] for item in result["Items"]} == {"zero"}