1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
|
# The MIT License (MIT)
# Copyright (c) Microsoft Corporation. All rights reserved.
import unittest
import uuid
import pytest
import azure.cosmos.exceptions as exceptions
import test_config
from azure.cosmos import CosmosClient, cosmos_client
from azure.cosmos import ThroughputProperties, PartitionKey
@pytest.mark.cosmosLong
class TestAutoScale(unittest.TestCase):
client: CosmosClient = None
host = test_config.TestConfig.host
masterKey = test_config.TestConfig.masterKey
connectionPolicy = test_config.TestConfig.connectionPolicy
@classmethod
def setUpClass(cls):
if (cls.masterKey == '[YOUR_KEY_HERE]' or
cls.host == '[YOUR_ENDPOINT_HERE]'):
raise Exception(
"You must specify your Azure Cosmos account values for "
"'masterKey' and 'host' at the top of this class to run the "
"tests.")
cls.client = cosmos_client.CosmosClient(cls.host, cls.masterKey)
cls.created_database = cls.client.get_database_client(test_config.TestConfig.TEST_DATABASE_ID)
def test_autoscale_create_container(self):
container_id = None
try:
container_id = 'auto_scale'
created_container = self.created_database.create_container(
id=container_id,
partition_key=PartitionKey(path="/id"),
offer_throughput=ThroughputProperties(auto_scale_max_throughput=7000, auto_scale_increment_percent=0)
)
created_container_properties = created_container.get_throughput()
# Testing the input value of the max_throughput
assert created_container_properties.auto_scale_max_throughput == 7000
assert created_container_properties.auto_scale_increment_percent == 0
assert created_container_properties.offer_throughput is None
self.created_database.delete_container(created_container)
# Testing the incorrect passing of an input value of the max_throughput to verify negative behavior
with self.assertRaises(exceptions.CosmosHttpResponseError) as e:
self.created_database.create_container(
id='container_with_wrong_auto_scale_settings',
partition_key=PartitionKey(path="/id"),
offer_throughput=ThroughputProperties(auto_scale_max_throughput=-200, auto_scale_increment_percent=0))
assert "Requested throughput -200 is less than required minimum throughput 1000" in str(e.exception)
# Testing auto_scale_settings for the create_container_if_not_exists method
container_id = 'auto_scale_2'
created_container = self.created_database.create_container_if_not_exists(
id=container_id,
partition_key=PartitionKey(path="/id"),
offer_throughput=ThroughputProperties(auto_scale_max_throughput=1000, auto_scale_increment_percent=3)
)
created_container_properties = created_container.get_throughput()
# Testing the input value of the max_throughput
assert created_container_properties.auto_scale_max_throughput == 1000
# Testing the input value of the increment_percentage
assert created_container_properties.auto_scale_increment_percent == 3
finally:
self.created_database.delete_container(container_id)
def test_autoscale_create_database(self):
database_id = "db_auto_scale_" + str(uuid.uuid4())
try:
# Testing auto_scale_settings for the create_database method
created_database = self.client.create_database(database_id, offer_throughput=ThroughputProperties(
auto_scale_max_throughput=5000,
auto_scale_increment_percent=2))
created_db_properties = created_database.get_throughput()
# Testing the input value of the max_throughput
assert created_db_properties.auto_scale_max_throughput == 5000
# Testing the input value of the increment_percentage
assert created_db_properties.auto_scale_increment_percent == 2
self.client.delete_database(created_database.id)
# Testing auto_scale_settings for the create_database_if_not_exists method
database_id = "db_auto_scale_2_" + str(uuid.uuid4())
created_database = self.client.create_database_if_not_exists(database_id,
offer_throughput=ThroughputProperties(
auto_scale_max_throughput=9000,
auto_scale_increment_percent=11))
created_db_properties = created_database.get_throughput()
# Testing the input value of the max_throughput
assert created_db_properties.auto_scale_max_throughput == 9000
# Testing the input value of the increment_percentage
assert created_db_properties.auto_scale_increment_percent == 11
finally:
self.client.delete_database(database_id)
def test_autoscale_replace_throughput(self):
database_id = "replace_db" + str(uuid.uuid4())
container_id = None
try:
created_database = self.client.create_database(database_id, offer_throughput=ThroughputProperties(
auto_scale_max_throughput=5000,
auto_scale_increment_percent=2))
created_database.replace_throughput(
throughput=ThroughputProperties(auto_scale_max_throughput=7000, auto_scale_increment_percent=20))
created_db_properties = created_database.get_throughput()
# Testing the input value of the max_throughput
assert created_db_properties.auto_scale_max_throughput == 7000
# Testing the input value of the increment_percentage
assert created_db_properties.auto_scale_increment_percent == 20
self.client.delete_database(database_id)
container_id = "container_with_auto_scale_settings" + str(uuid.uuid4())
created_container = self.created_database.create_container(
id=container_id,
partition_key=PartitionKey(path="/id"),
offer_throughput=ThroughputProperties(auto_scale_max_throughput=5000, auto_scale_increment_percent=0))
created_container.replace_throughput(
throughput=ThroughputProperties(auto_scale_max_throughput=7000, auto_scale_increment_percent=20))
created_container_properties = created_container.get_throughput()
# Testing the input value of the replaced auto_scale settings
assert created_container_properties.auto_scale_max_throughput == 7000
assert created_container_properties.auto_scale_increment_percent == 20
finally:
self.created_database.delete_container(container_id)
if __name__ == '__main__':
unittest.main()
|
