Cleaned up code
This commit is contained in:
parent
a5895db4f8
commit
77fcafca18
3 changed files with 258 additions and 74 deletions
|
|
@ -576,29 +576,51 @@ 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']}})
|
||||
|
||||
filter1 = moto.dynamodb2.comparisons.get_filter_expression('Id > 5 AND Subs < 7', {}, {})
|
||||
filter1.expr(row1).should.be(True)
|
||||
filter1.expr(row2).should.be(False)
|
||||
# AND test
|
||||
filter_expr = moto.dynamodb2.comparisons.get_filter_expression('Id > 5 AND Subs < 7', {}, {})
|
||||
filter_expr.expr(row1).should.be(True)
|
||||
filter_expr.expr(row2).should.be(False)
|
||||
|
||||
filter2 = moto.dynamodb2.comparisons.get_filter_expression('attribute_exists(Id) AND attribute_not_exists(User)', {}, {})
|
||||
filter2.expr(row1).should.be(True)
|
||||
# OR test
|
||||
filter_expr = moto.dynamodb2.comparisons.get_filter_expression('Id = 5 OR Id=8', {}, {})
|
||||
filter_expr.expr(row1).should.be(True)
|
||||
|
||||
filter3 = moto.dynamodb2.comparisons.get_filter_expression('attribute_type(Id, N)', {}, {})
|
||||
filter3.expr(row1).should.be(True)
|
||||
# BETWEEN test
|
||||
filter_expr = moto.dynamodb2.comparisons.get_filter_expression('Id BETWEEN 5 AND 10', {}, {})
|
||||
filter_expr.expr(row1).should.be(True)
|
||||
|
||||
filter4 = moto.dynamodb2.comparisons.get_filter_expression('begins_with(Desc, Some)', {}, {})
|
||||
filter4.expr(row1).should.be(True)
|
||||
filter4.expr(row2).should.be(False)
|
||||
# PAREN test
|
||||
filter_expr = moto.dynamodb2.comparisons.get_filter_expression('Id = 8 AND (Subs = 8 OR Subs = 5)', {}, {})
|
||||
filter_expr.expr(row1).should.be(True)
|
||||
|
||||
filter5 = moto.dynamodb2.comparisons.get_filter_expression('contains(KV, test1)', {}, {})
|
||||
filter5.expr(row1).should.be(True)
|
||||
filter5.expr(row2).should.be(False)
|
||||
# IN test
|
||||
filter_expr = moto.dynamodb2.comparisons.get_filter_expression('Id IN (7,8, 9)', {}, {})
|
||||
filter_expr.expr(row1).should.be(True)
|
||||
|
||||
filter6 = moto.dynamodb2.comparisons.get_filter_expression('size(Desc) > size(KV)', {}, {})
|
||||
filter6.expr(row1).should.be(True)
|
||||
# attribute function tests
|
||||
filter_expr = moto.dynamodb2.comparisons.get_filter_expression('attribute_exists(Id) AND attribute_not_exists(User)', {}, {})
|
||||
filter_expr.expr(row1).should.be(True)
|
||||
|
||||
filter_expr = moto.dynamodb2.comparisons.get_filter_expression('attribute_type(Id, N)', {}, {})
|
||||
filter_expr.expr(row1).should.be(True)
|
||||
|
||||
# beginswith function test
|
||||
filter_expr = moto.dynamodb2.comparisons.get_filter_expression('begins_with(Desc, Some)', {}, {})
|
||||
filter_expr.expr(row1).should.be(True)
|
||||
filter_expr.expr(row2).should.be(False)
|
||||
|
||||
# contains function test
|
||||
filter_expr = moto.dynamodb2.comparisons.get_filter_expression('contains(KV, test1)', {}, {})
|
||||
filter_expr.expr(row1).should.be(True)
|
||||
filter_expr.expr(row2).should.be(False)
|
||||
|
||||
# size function test
|
||||
filter_expr = moto.dynamodb2.comparisons.get_filter_expression('size(Desc) > size(KV)', {}, {})
|
||||
filter_expr.expr(row1).should.be(True)
|
||||
|
||||
|
||||
@mock_dynamodb2
|
||||
|
|
@ -631,3 +653,126 @@ def test_scan_filter():
|
|||
FilterExpression=Attr('app').eq('app1')
|
||||
)
|
||||
assert response['Count'] == 1
|
||||
|
||||
|
||||
@mock_dynamodb2
|
||||
def test_bad_scan_filter():
|
||||
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': 'S'}],
|
||||
KeySchema=[{'AttributeName': 'client', 'KeyType': 'HASH'}, {'AttributeName': 'app', 'KeyType': 'RANGE'}],
|
||||
ProvisionedThroughput={'ReadCapacityUnits': 123, 'WriteCapacityUnits': 123}
|
||||
)
|
||||
table = dynamodb.Table('test1')
|
||||
|
||||
# Bad expression
|
||||
try:
|
||||
table.scan(
|
||||
FilterExpression='client test'
|
||||
)
|
||||
except ClientError as err:
|
||||
err.response['Error']['Code'].should.equal('ValidationError')
|
||||
else:
|
||||
raise RuntimeError('Should of raised ResourceInUseException')
|
||||
|
||||
|
||||
|
||||
@mock_dynamodb2
|
||||
def test_duplicate_create():
|
||||
client = boto3.client('dynamodb', region_name='us-east-1')
|
||||
|
||||
# Create the DynamoDB table.
|
||||
client.create_table(
|
||||
TableName='test1',
|
||||
AttributeDefinitions=[{'AttributeName': 'client', 'AttributeType': 'S'}, {'AttributeName': 'app', 'AttributeType': 'S'}],
|
||||
KeySchema=[{'AttributeName': 'client', 'KeyType': 'HASH'}, {'AttributeName': 'app', 'KeyType': 'RANGE'}],
|
||||
ProvisionedThroughput={'ReadCapacityUnits': 123, 'WriteCapacityUnits': 123}
|
||||
)
|
||||
|
||||
try:
|
||||
client.create_table(
|
||||
TableName='test1',
|
||||
AttributeDefinitions=[{'AttributeName': 'client', 'AttributeType': 'S'}, {'AttributeName': 'app', 'AttributeType': 'S'}],
|
||||
KeySchema=[{'AttributeName': 'client', 'KeyType': 'HASH'}, {'AttributeName': 'app', 'KeyType': 'RANGE'}],
|
||||
ProvisionedThroughput={'ReadCapacityUnits': 123, 'WriteCapacityUnits': 123}
|
||||
)
|
||||
except ClientError as err:
|
||||
err.response['Error']['Code'].should.equal('ResourceInUseException')
|
||||
else:
|
||||
raise RuntimeError('Should of raised ResourceInUseException')
|
||||
|
||||
|
||||
@mock_dynamodb2
|
||||
def test_delete_table():
|
||||
client = boto3.client('dynamodb', region_name='us-east-1')
|
||||
|
||||
# Create the DynamoDB table.
|
||||
client.create_table(
|
||||
TableName='test1',
|
||||
AttributeDefinitions=[{'AttributeName': 'client', 'AttributeType': 'S'}, {'AttributeName': 'app', 'AttributeType': 'S'}],
|
||||
KeySchema=[{'AttributeName': 'client', 'KeyType': 'HASH'}, {'AttributeName': 'app', 'KeyType': 'RANGE'}],
|
||||
ProvisionedThroughput={'ReadCapacityUnits': 123, 'WriteCapacityUnits': 123}
|
||||
)
|
||||
|
||||
client.delete_table(TableName='test1')
|
||||
|
||||
resp = client.list_tables()
|
||||
len(resp['TableNames']).should.equal(0)
|
||||
|
||||
try:
|
||||
client.delete_table(TableName='test1')
|
||||
except ClientError as err:
|
||||
err.response['Error']['Code'].should.equal('ResourceNotFoundException')
|
||||
else:
|
||||
raise RuntimeError('Should of raised ResourceNotFoundException')
|
||||
|
||||
|
||||
@mock_dynamodb2
|
||||
def test_delete_item():
|
||||
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': 'S'}],
|
||||
KeySchema=[{'AttributeName': 'client', 'KeyType': 'HASH'}, {'AttributeName': 'app', 'KeyType': 'RANGE'}],
|
||||
ProvisionedThroughput={'ReadCapacityUnits': 123, 'WriteCapacityUnits': 123}
|
||||
)
|
||||
client.put_item(
|
||||
TableName='test1',
|
||||
Item={
|
||||
'client': {'S': 'client1'},
|
||||
'app': {'S': 'app1'}
|
||||
}
|
||||
)
|
||||
client.put_item(
|
||||
TableName='test1',
|
||||
Item={
|
||||
'client': {'S': 'client1'},
|
||||
'app': {'S': 'app2'}
|
||||
}
|
||||
)
|
||||
|
||||
table = dynamodb.Table('test1')
|
||||
response = table.scan()
|
||||
assert response['Count'] == 2
|
||||
|
||||
# Test deletion and returning old value
|
||||
response = table.delete_item(Key={'client': 'client1', 'app': 'app1'}, ReturnValues='ALL_OLD')
|
||||
response['Attributes'].should.contain('client')
|
||||
response['Attributes'].should.contain('app')
|
||||
|
||||
response = table.scan()
|
||||
assert response['Count'] == 1
|
||||
|
||||
# Test deletion returning nothing
|
||||
response = table.delete_item(Key={'client': 'client1', 'app': 'app2'})
|
||||
len(response['Attributes']).should.equal(0)
|
||||
|
||||
response = table.scan()
|
||||
assert response['Count'] == 0
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue