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
|
# The MIT License (MIT)
# Copyright (c) Microsoft Corporation. All rights reserved.
import unittest
import uuid
import pytest
import test_config
from azure.cosmos import CosmosClient, PartitionKey, DatabaseProxy
from azure.cosmos.http_constants import HttpHeaders
# TODO: add more tests in this file once we have response headers in control plane operations
# TODO: add query tests once those changes are available
@pytest.mark.cosmosEmulator
class TestCosmosResponses(unittest.TestCase):
"""Python Cosmos Responses Tests.
"""
configs = test_config.TestConfig
host = configs.host
masterKey = configs.masterKey
client: CosmosClient = None
test_database: DatabaseProxy = None
TEST_DATABASE_ID = configs.TEST_DATABASE_ID
@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 = CosmosClient(cls.host, cls.masterKey)
cls.test_database = cls.client.get_database_client(cls.TEST_DATABASE_ID)
def test_point_operation_headers(self):
container = self.test_database.create_container(id="responses_test" + str(uuid.uuid4()),
partition_key=PartitionKey(path="/company"))
first_response = container.upsert_item({"id": str(uuid.uuid4()), "company": "Microsoft"})
lsn = first_response.get_response_headers()['lsn']
create_response = container.create_item({"id": str(uuid.uuid4()), "company": "Microsoft"})
assert len(create_response.get_response_headers()) > 0
assert int(lsn) + 1 == int(create_response.get_response_headers()['lsn'])
lsn = create_response.get_response_headers()['lsn']
read_response = container.read_item(create_response['id'], create_response['company'])
assert len(read_response.get_response_headers()) > 0
upsert_response = container.upsert_item({"id": str(uuid.uuid4()), "company": "Microsoft"})
assert len(upsert_response.get_response_headers()) > 0
assert int(lsn) + 1 == int(upsert_response.get_response_headers()['lsn'])
lsn = upsert_response.get_response_headers()['lsn']
upsert_response['replace'] = True
replace_response = container.replace_item(upsert_response['id'], upsert_response)
assert replace_response['replace'] is not None
assert len(replace_response.get_response_headers()) > 0
assert int(lsn) + 1 == int(replace_response.get_response_headers()['lsn'])
batch = []
for i in range(50):
batch.append(("create", ({"id": "item" + str(i), "company": "Microsoft"},)))
batch_response = container.execute_item_batch(batch_operations=batch, partition_key="Microsoft")
assert len(batch_response.get_response_headers()) > 0
assert int(lsn) + 1 < int(batch_response.get_response_headers()['lsn'])
if __name__ == '__main__':
unittest.main()
|
