81 lines
2.8 KiB
Python
81 lines
2.8 KiB
Python
import boto3
|
|
import sure # noqa # pylint: disable=unused-import
|
|
|
|
from moto import mock_dynamodb
|
|
|
|
|
|
@mock_dynamodb
|
|
def test_update_table__billing_mode():
|
|
client = boto3.client("dynamodb", region_name="us-east-1")
|
|
client.create_table(
|
|
TableName="test",
|
|
AttributeDefinitions=[
|
|
{"AttributeName": "client", "AttributeType": "S"},
|
|
{"AttributeName": "app", "AttributeType": "S"},
|
|
],
|
|
KeySchema=[
|
|
{"AttributeName": "client", "KeyType": "HASH"},
|
|
{"AttributeName": "app", "KeyType": "RANGE"},
|
|
],
|
|
BillingMode="PAY_PER_REQUEST",
|
|
)
|
|
|
|
client.update_table(
|
|
TableName="test",
|
|
BillingMode="PROVISIONED",
|
|
ProvisionedThroughput={"ReadCapacityUnits": 1, "WriteCapacityUnits": 1},
|
|
)
|
|
|
|
actual = client.describe_table(TableName="test")["Table"]
|
|
actual.should.have.key("BillingModeSummary").equals({"BillingMode": "PROVISIONED"})
|
|
actual.should.have.key("ProvisionedThroughput").equals(
|
|
{"ReadCapacityUnits": 1, "WriteCapacityUnits": 1}
|
|
)
|
|
|
|
|
|
@mock_dynamodb
|
|
def test_update_table_throughput():
|
|
conn = boto3.resource("dynamodb", region_name="us-west-2")
|
|
table = conn.create_table(
|
|
TableName="messages",
|
|
KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
|
|
AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
|
|
ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
|
|
)
|
|
table.provisioned_throughput["ReadCapacityUnits"].should.equal(5)
|
|
table.provisioned_throughput["WriteCapacityUnits"].should.equal(5)
|
|
|
|
table.update(
|
|
ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 6}
|
|
)
|
|
|
|
table.provisioned_throughput["ReadCapacityUnits"].should.equal(5)
|
|
table.provisioned_throughput["WriteCapacityUnits"].should.equal(6)
|
|
|
|
|
|
@mock_dynamodb
|
|
def test_update_table__enable_stream():
|
|
conn = boto3.client("dynamodb", region_name="us-east-1")
|
|
|
|
resp = conn.create_table(
|
|
TableName="test-stream-update",
|
|
KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
|
|
AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
|
|
ProvisionedThroughput={"ReadCapacityUnits": 1, "WriteCapacityUnits": 1},
|
|
)
|
|
|
|
assert "StreamSpecification" not in resp["TableDescription"]
|
|
|
|
resp = conn.update_table(
|
|
TableName="test-stream-update",
|
|
StreamSpecification={"StreamEnabled": True, "StreamViewType": "NEW_IMAGE"},
|
|
)
|
|
|
|
assert "StreamSpecification" in resp["TableDescription"]
|
|
assert resp["TableDescription"]["StreamSpecification"] == {
|
|
"StreamEnabled": True,
|
|
"StreamViewType": "NEW_IMAGE",
|
|
}
|
|
assert "LatestStreamLabel" in resp["TableDescription"]
|
|
assert "LatestStreamArn" in resp["TableDescription"]
|