catalin/moto
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Merge pull request #1289 from terrycain/dynamodb_filter_expression_v2

Browse source 
DynamoDB FilterExpression NOT logic
This commit is contained in:
Jack Danger 2017-10-23 09:46:02 -07:00 • committed by GitHub
commit 0ca292b0ca
2 changed files with 91 additions and 30 deletions
Download patch file Download diff file Expand all files Collapse all files

38
tests/test_dynamodb2/test_dynamodb.py
View file

@ -576,10 +576,17 @@ def test_get_item_returns_consumed_capacity():
def test_filter_expression():
# TODO NOT not yet supported
row1 = moto.dynamodb2.models.Item(None, None, None, None, {'Id': {'N': '8'}, 'Subs': {'N': '5'}, 'Desc': {'S': 'Some description'}, 'KV': {'SS': ['test1', 'test2']}})
row2 = moto.dynamodb2.models.Item(None, None, None, None, {'Id': {'N': '8'}, 'Subs': {'N': '10'}, 'Desc': {'S': 'A description'}, 'KV': {'SS': ['test3', 'test4']}})
# NOT test 1
filter_expr = moto.dynamodb2.comparisons.get_filter_expression('NOT attribute_not_exists(Id)', {}, {})
filter_expr.expr(row1).should.be(True)
# NOT test 2
filter_expr = moto.dynamodb2.comparisons.get_filter_expression('NOT (Id = :v0)', {}, {':v0': {'N': 8}})
filter_expr.expr(row1).should.be(False) # Id = 8 so should be false
# AND test
filter_expr = moto.dynamodb2.comparisons.get_filter_expression('Id > :v0 AND Subs < :v1', {}, {':v0': {'N': 5}, ':v1': {'N': 7}})
filter_expr.expr(row1).should.be(True)
@ -622,6 +629,14 @@ def test_filter_expression():
filter_expr = moto.dynamodb2.comparisons.get_filter_expression('size(Desc) > size(KV)', {}, {})
filter_expr.expr(row1).should.be(True)
# Expression from @batkuip
filter_expr = moto.dynamodb2.comparisons.get_filter_expression(
'(#n0 < :v0 AND attribute_not_exists(#n1))',
{'#n0': 'Subs', '#n1': 'fanout_ts'},
{':v0': {'N': '7'}}
)
filter_expr.expr(row1).should.be(True)
@mock_dynamodb2
def test_scan_filter():
@ -712,6 +727,27 @@ def test_scan_filter3():
assert response['Count'] == 1
@mock_dynamodb2
def test_scan_filter4():
client = boto3.client('dynamodb', region_name='us-east-1')
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
# Create the DynamoDB table.
client.create_table(
TableName='test1',
AttributeDefinitions=[{'AttributeName': 'client', 'AttributeType': 'S'}, {'AttributeName': 'app', 'AttributeType': 'N'}],
KeySchema=[{'AttributeName': 'client', 'KeyType': 'HASH'}, {'AttributeName': 'app', 'KeyType': 'RANGE'}],
ProvisionedThroughput={'ReadCapacityUnits': 123, 'WriteCapacityUnits': 123}
)
table = dynamodb.Table('test1')
response = table.scan(
FilterExpression=Attr('epoch_ts').lt(7) & Attr('fanout_ts').not_exists()
)
# Just testing
assert response['Count'] == 0
@mock_dynamodb2
def test_bad_scan_filter():
client = boto3.client('dynamodb', region_name='us-east-1')