diff --git a/eng/pipelines/templates/stages/cosmos-sdk-client.yml b/eng/pipelines/templates/stages/cosmos-sdk-client.yml index 44044764e9f7..290b35e46779 100644 --- a/eng/pipelines/templates/stages/cosmos-sdk-client.yml +++ b/eng/pipelines/templates/stages/cosmos-sdk-client.yml @@ -21,7 +21,7 @@ extends: BeforeTestSteps: - template: /eng/common/pipelines/templates/steps/cosmos-emulator.yml parameters: - StartParameters: '/noexplorer /noui /enablepreview /EnableSqlComputeEndpoint /SqlComputePort=9999 /disableratelimiting /partitioncount=50 /consistency=Session' + StartParameters: '/noexplorer /noui /enablepreview /EnableSqlComputeEndpoint /SqlComputePort=9999 /disableratelimiting /partitioncount=50 /consistency=Session /overrides=enablePreviousImageForDeleteInFFCF:true' MatrixConfigs: - ${{ if eq(variables['Build.Reason'], 'PullRequest') }}: - Name: Python_cosmos_emulator diff --git a/sdk/cosmos/azure-cosmos/CHANGELOG.md b/sdk/cosmos/azure-cosmos/CHANGELOG.md index a35926ac033c..c7215b4d5151 100644 --- a/sdk/cosmos/azure-cosmos/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos/CHANGELOG.md @@ -5,6 +5,8 @@ ### 4.17.2 (Unreleased) #### Features Added +* Added **provisional** support for consuming previous item images returned by `AllVersionsAndDeletes` change feed. + This **preview feature** requires service-side opt-in. See [PR 49063](https://github.com/Azure/azure-sdk-for-python/pull/49063). #### Breaking Changes diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_container.py b/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_container.py index 449642435fad..f4b78c5ef074 100644 --- a/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_container.py +++ b/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_container.py @@ -1010,6 +1010,10 @@ def query_items_change_feed( LATEST_VERSION: Query latest items from 'start_time' or 'continuation' token. ALL_VERSIONS_AND_DELETES: Query all versions and deleted items from either `start_time='Now'` or 'continuation' token. + All versions and deletes responses contain ``current`` and ``metadata`` keys. The **provisional** + ``previous`` key can contain the previous item version when previous images are enabled for the container. + To request access to this provisional feature, see `previous image opt-in + `_. :paramtype mode: Literal["LatestVersion", "AllVersionsAndDeletes"] :keyword Sequence[str] excluded_locations: Excluded locations to be skipped from preferred locations. The locations in this list are specified as the names of the azure Cosmos locations like, 'West US', 'East US' and so on. @@ -1058,6 +1062,10 @@ def query_items_change_feed( LATEST_VERSION: Query latest items from 'start_time' or 'continuation' token. ALL_VERSIONS_AND_DELETES: Query all versions and deleted items from either `start_time='Now'` or 'continuation' token. + All versions and deletes responses contain ``current`` and ``metadata`` keys. The **provisional** + ``previous`` key can contain the previous item version when previous images are enabled for the container. + To request access to this provisional feature, see `previous image opt-in + `_. :paramtype mode: Literal["LatestVersion", "AllVersionsAndDeletes"] :keyword Sequence[str] excluded_locations: Excluded locations to be skipped from preferred locations. The locations in this list are specified as the names of the azure Cosmos locations like, 'West US', 'East US' and so on. @@ -1142,6 +1150,10 @@ def query_items_change_feed( LATEST_VERSION: Query latest items from 'start_time' or 'continuation' token. ALL_VERSIONS_AND_DELETES: Query all versions and deleted items from either `start_time='Now'` or 'continuation' token. + All versions and deletes responses contain ``current`` and ``metadata`` keys. The **provisional** + ``previous`` key can contain the previous item version when previous images are enabled for the container. + To request access to this provisional feature, see `previous image opt-in + `_. :paramtype mode: Literal["LatestVersion", "AllVersionsAndDeletes"] :keyword Sequence[str] excluded_locations: Excluded locations to be skipped from preferred locations. The locations in this list are specified as the names of the azure Cosmos locations like, 'West US', 'East US' and so on. @@ -1187,6 +1199,10 @@ def query_items_change_feed( # pylint: disable=unused-argument LATEST_VERSION: Query latest items from 'start_time' or 'continuation' token. ALL_VERSIONS_AND_DELETES: Query all versions and deleted items from either `start_time='Now'` or 'continuation' token. + All versions and deletes responses contain ``current`` and ``metadata`` keys. The **provisional** + ``previous`` key can contain the previous item version when previous images are enabled for the container. + To request access to this provisional feature, see `previous image opt-in + `_. :paramtype mode: Literal["LatestVersion", "AllVersionsAndDeletes"] :keyword Sequence[str] excluded_locations: Excluded locations to be skipped from preferred locations. The locations in this list are specified as the names of the azure Cosmos locations like, 'West US', 'East US' and so on. diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/container.py b/sdk/cosmos/azure-cosmos/azure/cosmos/container.py index e292f4f17f33..34a5a7884a58 100644 --- a/sdk/cosmos/azure-cosmos/azure/cosmos/container.py +++ b/sdk/cosmos/azure-cosmos/azure/cosmos/container.py @@ -497,6 +497,10 @@ def query_items_change_feed( LATEST_VERSION: Query latest items from 'start_time' or 'continuation' token. ALL_VERSIONS_AND_DELETES: Query all versions and deleted items from either `start_time='Now'` or 'continuation' token. + All versions and deletes responses contain ``current`` and ``metadata`` keys. The **provisional** + ``previous`` key can contain the previous item version when previous images are enabled for the container. + To request access to this provisional feature, see `previous image opt-in + `_. :paramtype mode: Literal["LatestVersion", "AllVersionsAndDeletes"] :keyword Sequence[str] excluded_locations: Excluded locations to be skipped from preferred locations. The locations in this list are specified as the names of the azure Cosmos locations like, 'West US', 'East US' and so on. @@ -547,6 +551,10 @@ def query_items_change_feed( LATEST_VERSION: Query latest items from 'start_time' or 'continuation' token. ALL_VERSIONS_AND_DELETES: Query all versions and deleted items from either `start_time='Now'` or 'continuation' token. + All versions and deletes responses contain ``current`` and ``metadata`` keys. The **provisional** + ``previous`` key can contain the previous item version when previous images are enabled for the container. + To request access to this provisional feature, see `previous image opt-in + `_. :paramtype mode: Literal["LatestVersion", "AllVersionsAndDeletes"] :keyword Sequence[str] excluded_locations: Excluded locations to be skipped from preferred locations. The locations in this list are specified as the names of the azure Cosmos locations like, 'West US', 'East US' and so on. @@ -632,6 +640,10 @@ def query_items_change_feed( LATEST_VERSION: Query latest items from 'start_time' or 'continuation' token. ALL_VERSIONS_AND_DELETES: Query all versions and deleted items from either `start_time='Now'` or 'continuation' token. + All versions and deletes responses contain ``current`` and ``metadata`` keys. The **provisional** + ``previous`` key can contain the previous item version when previous images are enabled for the container. + To request access to this provisional feature, see `previous image opt-in + `_. :paramtype mode: Literal["LatestVersion", "AllVersionsAndDeletes"] :keyword Sequence[str] excluded_locations: Excluded locations to be skipped from preferred locations. The locations in this list are specified as the names of the azure Cosmos locations like, 'West US', 'East US' and so on. @@ -680,6 +692,10 @@ def query_items_change_feed( LATEST_VERSION: Query latest items from 'start_time' or 'continuation' token. ALL_VERSIONS_AND_DELETES: Query all versions and deleted items from either `start_time='Now'` or 'continuation' token. + All versions and deletes responses contain ``current`` and ``metadata`` keys. The **provisional** + ``previous`` key can contain the previous item version when previous images are enabled for the container. + To request access to this provisional feature, see `previous image opt-in + `_. :paramtype mode: Literal["LatestVersion", "AllVersionsAndDeletes"] :keyword Sequence[str] excluded_locations: Excluded locations to be skipped from preferred locations. The locations in this list are specified as the names of the azure Cosmos locations like, 'West US', 'East US' and so on. diff --git a/sdk/cosmos/azure-cosmos/samples/change_feed_management.py b/sdk/cosmos/azure-cosmos/samples/change_feed_management.py index fdefe563d0e3..364074489f42 100644 --- a/sdk/cosmos/azure-cosmos/samples/change_feed_management.py +++ b/sdk/cosmos/azure-cosmos/samples/change_feed_management.py @@ -53,6 +53,19 @@ def clean_up(container): # Deleting the current item container.delete_item(item, partition_key=item['address']['state']) + +def print_all_versions_and_deletes_change(change): + metadata = change.get('metadata', {}) + print('Operation: {}'.format(metadata.get('operationType'))) + print('Current item: {}'.format(change.get('current'))) + + # `previous` is provisional and is returned only when previous images are enabled for the container. + # Request preview access at https://aka.ms/cosmosdb-change-feed-deletes. + previous = change.get('previous') + if previous is not None: + print('Previous item: {}'.format(previous)) + + def read_change_feed(container): print('\nReading Change Feed from the beginning\n') @@ -126,7 +139,7 @@ def read_change_feed_with_all_versions_and_delete_mode(container): # This initial call was made to store a point in time in a 'continuation' token response_iterator = container.query_items_change_feed(mode="AllVersionsAndDeletes") for doc in response_iterator: - print(doc) + print_all_versions_and_deletes_change(doc) continuation_token = container.client_connection.last_response_headers['etag'] # Read all change feed with 'AllVersionsAndDeletes' mode after create items from a continuation @@ -134,13 +147,13 @@ def read_change_feed_with_all_versions_and_delete_mode(container): create_items(container, 10, 'OR') response_iterator = container.query_items_change_feed(mode="AllVersionsAndDeletes", continuation=continuation_token) for doc in response_iterator: - print(doc) + print_all_versions_and_deletes_change(doc) # Read all change feed with 'AllVersionsAndDeletes' mode after delete items from a continuation clean_up(container) response_iterator = container.query_items_change_feed(mode="AllVersionsAndDeletes", continuation=continuation_token) for doc in response_iterator: - print(doc) + print_all_versions_and_deletes_change(doc) def read_change_feed_with_all_versions_and_delete_mode_with_partition_key(container): print('\nReading Change Feed with AllVersionsAndDeletes mode from the partition key\n') @@ -149,7 +162,7 @@ def read_change_feed_with_all_versions_and_delete_mode_with_partition_key(contai # This initial call was made to store a point in time and 'partition_key' in a 'continuation' token response_iterator = container.query_items_change_feed(mode="AllVersionsAndDeletes", partition_key="CA") for doc in response_iterator: - print(doc) + print_all_versions_and_deletes_change(doc) continuation_token = container.client_connection.last_response_headers['etag'] create_items(container, 10, 'CA') @@ -158,7 +171,7 @@ def read_change_feed_with_all_versions_and_delete_mode_with_partition_key(contai # Should only print the created items with 'CA' partition key value response_iterator = container.query_items_change_feed(mode='AllVersionsAndDeletes', continuation=continuation_token) for doc in response_iterator: - print(doc) + print_all_versions_and_deletes_change(doc) continuation_token = container.client_connection.last_response_headers['etag'] clean_up(container) @@ -166,7 +179,7 @@ def read_change_feed_with_all_versions_and_delete_mode_with_partition_key(contai # Should only print the deleted items with 'CA' partition key value response_iterator = container.query_items_change_feed(mode='AllVersionsAndDeletes', continuation=continuation_token) for doc in response_iterator: - print(doc) + print_all_versions_and_deletes_change(doc) def run_sample(): client = cosmos_client.CosmosClient(HOST, {'masterKey': MASTER_KEY}) diff --git a/sdk/cosmos/azure-cosmos/samples/change_feed_management_async.py b/sdk/cosmos/azure-cosmos/samples/change_feed_management_async.py index 2efbbc3e6e49..788385f9d6ff 100644 --- a/sdk/cosmos/azure-cosmos/samples/change_feed_management_async.py +++ b/sdk/cosmos/azure-cosmos/samples/change_feed_management_async.py @@ -54,6 +54,19 @@ async def clean_up(container): # Deleting the current item await container.delete_item(item, partition_key=item['address']['state']) + +def print_all_versions_and_deletes_change(change): + metadata = change.get('metadata', {}) + print('Operation: {}'.format(metadata.get('operationType'))) + print('Current item: {}'.format(change.get('current'))) + + # `previous` is provisional and is returned only when previous images are enabled for the container. + # Request preview access at https://aka.ms/cosmosdb-change-feed-deletes. + previous = change.get('previous') + if previous is not None: + print('Previous item: {}'.format(previous)) + + async def read_change_feed(container): print('\nReading Change Feed from the beginning\n') @@ -134,7 +147,7 @@ async def read_change_feed_with_all_versions_and_delete_mode(container): # This initial call was made to store a point in time in a 'continuation' token response_iterator = container.query_items_change_feed(mode="AllVersionsAndDeletes") async for doc in response_iterator: - print(doc) + print_all_versions_and_deletes_change(doc) continuation_token = container.client_connection.last_response_headers['etag'] # Read all change feed with 'AllVersionsAndDeletes' mode after create items from a continuation @@ -142,13 +155,13 @@ async def read_change_feed_with_all_versions_and_delete_mode(container): await create_items(container, 10, 'OR') response_iterator = container.query_items_change_feed(mode="AllVersionsAndDeletes", continuation=continuation_token) async for doc in response_iterator: - print(doc) + print_all_versions_and_deletes_change(doc) # Read all change feed with 'AllVersionsAndDeletes' mode after delete items from a continuation await clean_up(container) response_iterator = container.query_items_change_feed(mode="AllVersionsAndDeletes", continuation=continuation_token) async for doc in response_iterator: - print(doc) + print_all_versions_and_deletes_change(doc) async def read_change_feed_with_all_versions_and_delete_mode_with_partition_key(container): print('\nReading Change Feed with AllVersionsAndDeletes mode from the partition key\n') @@ -157,7 +170,7 @@ async def read_change_feed_with_all_versions_and_delete_mode_with_partition_key( # This initial call was made to store a point in time and 'partition_key' in a 'continuation' token response_iterator = container.query_items_change_feed(mode="AllVersionsAndDeletes", partition_key="CA") async for doc in response_iterator: - print(doc) + print_all_versions_and_deletes_change(doc) continuation_token = container.client_connection.last_response_headers['etag'] await create_items(container, 10, 'CA') @@ -166,7 +179,7 @@ async def read_change_feed_with_all_versions_and_delete_mode_with_partition_key( # Should only print the created items with 'CA' partition key value response_iterator = container.query_items_change_feed(mode='AllVersionsAndDeletes', continuation=continuation_token) async for doc in response_iterator: - print(doc) + print_all_versions_and_deletes_change(doc) continuation_token = container.client_connection.last_response_headers['etag'] await clean_up(container) @@ -174,7 +187,7 @@ async def read_change_feed_with_all_versions_and_delete_mode_with_partition_key( # Should only print the deleted items with 'CA' partition key value response_iterator = container.query_items_change_feed(mode='AllVersionsAndDeletes', continuation=continuation_token) async for doc in response_iterator: - print(doc) + print_all_versions_and_deletes_change(doc) async def run_sample(): async with CosmosClient(HOST, MASTER_KEY) as client: diff --git a/sdk/cosmos/azure-cosmos/tests/test_change_feed_all_versions.py b/sdk/cosmos/azure-cosmos/tests/test_change_feed_all_versions.py index 1a047380923e..b791ca92a38e 100644 --- a/sdk/cosmos/azure-cosmos/tests/test_change_feed_all_versions.py +++ b/sdk/cosmos/azure-cosmos/tests/test_change_feed_all_versions.py @@ -1,14 +1,17 @@ # The MIT License (MIT) # Copyright (c) Microsoft Corporation. All rights reserved. +import time import unittest import uuid from datetime import datetime, timedelta, timezone +from types import SimpleNamespace import pytest import azure.cosmos.cosmos_client as cosmos_client import test_config +from azure.cosmos._change_feed.change_feed_iterable import ChangeFeedIterable from azure.cosmos.partition_key import PartitionKey from azure.cosmos._change_feed.change_feed_state import ChangeFeedStateV2 @@ -18,8 +21,12 @@ METADATA = 'metadata' OPERATION_TYPE = 'operationType' CREATE = 'create' +REPLACE = 'replace' DELETE = 'delete' E_TAG = 'etag' +VERSION = 'version' +TTL_SECONDS = 5 +TTL_TEST_TIMEOUT_SECONDS = 250 @pytest.fixture(scope="class") def setup(): @@ -45,18 +52,19 @@ def round_time(): return utc_now - timedelta(microseconds=utc_now.microsecond) def assert_change_feed(expected, actual): - if len(actual) == 0: - assert len(expected) == len(actual) - return - - #TODO: remove this if we can add flag to get 'previous' always - for item in actual: - if METADATA in item and item[METADATA][OPERATION_TYPE] == DELETE: - if ID in item[METADATA]: - item[PREVIOUS] = {ID: item[METADATA][ID]} + assert len(expected) == len(actual) + + # Delete entries don't have a current item, and previous is optional. + def sort_key(change): + return ( + change[METADATA][OPERATION_TYPE], + change.get(CURRENT, {}).get(ID) + or change.get(PREVIOUS, {}).get(ID) + or change[METADATA].get(ID), + ) - # Sort actual by operation_type and id - actual = sorted(actual, key=lambda k: (k[METADATA][OPERATION_TYPE], k[CURRENT][ID]) if k[METADATA][OPERATION_TYPE] == CREATE else (k[METADATA][OPERATION_TYPE], k[PREVIOUS][ID])) + expected = sorted(expected, key=sort_key) + actual = sorted(actual, key=sort_key) for expected_change_feed, actual_change_feed in zip(expected, actual): for expected_type, expected_data in expected_change_feed.items(): @@ -78,6 +86,43 @@ def _is_all_versions_and_deletes_not_enabled(error: Exception) -> bool: and "must be enabled" in message ) + +@pytest.mark.cosmosEmulator +@pytest.mark.unittest +class TestAllVersionsChangeFeedResponse: + def test_previous_images_are_preserved(self): + changes = [ + { + CURRENT: {ID: "item", "pk": "pk", VERSION: 1}, + METADATA: {OPERATION_TYPE: CREATE}, + }, + { + CURRENT: {ID: "item", "pk": "pk", VERSION: 2}, + PREVIOUS: {ID: "item", "pk": "pk", VERSION: 1, "_etag": "old-etag"}, + METADATA: {OPERATION_TYPE: REPLACE, "previousImageLSN": 1}, + }, + { + CURRENT: {}, + PREVIOUS: {ID: "item", "pk": "pk", VERSION: 2, "_etag": "new-etag"}, + METADATA: {OPERATION_TYPE: DELETE, ID: "item", "timeToLiveExpired": False}, + }, + { + CURRENT: {}, + METADATA: {OPERATION_TYPE: DELETE, ID: "expired", "timeToLiveExpired": True}, + }, + ] + iterable = object.__new__(ChangeFeedIterable) + iterable._client = SimpleNamespace(last_response_headers={E_TAG: '"continuation"'}) + + continuation, unpacked = iterable._unpack(changes) + + assert continuation == '"continuation"' + assert unpacked is changes + assert unpacked[1][PREVIOUS][VERSION] == 1 + assert unpacked[2][PREVIOUS][VERSION] == 2 + assert PREVIOUS not in unpacked[3] + + @pytest.mark.cosmosEmulator @pytest.mark.cosmosAADLong @pytest.mark.unittest @@ -127,26 +172,58 @@ def test_query_change_feed_all_versions_and_deletes(self, setup): cont_token2 = created_collection.client_connection.last_response_headers['etag'] assert_change_feed(expected_change_feeds, actual_change_feeds) - ## Test change_feed for deleted items + ## Test change_feed for replaced items + replaced_items = [] for item in created_items: - created_collection.delete_item(item=item, partition_key=item['pk']) + replacement = {partition_key: item[partition_key], ID: item[ID], VERSION: 2} + replaced_items.append(created_collection.replace_item(item=item[ID], body=replacement)) query_iterable = created_collection.query_items_change_feed( continuation=cont_token2, mode=mode, ) - expected_change_feeds = [{CURRENT: {}, PREVIOUS: {ID: f'doc{i}'}, METADATA: {OPERATION_TYPE: DELETE}} for i in range(4)] + expected_change_feeds = [ + {CURRENT: {ID: f'doc{i}', VERSION: 2}, METADATA: {OPERATION_TYPE: REPLACE}} + for i in range(4) + ] + actual_change_feeds = list(query_iterable) + cont_token3 = created_collection.client_connection.last_response_headers[E_TAG] + assert_change_feed(expected_change_feeds, actual_change_feeds) + + ## Test change_feed for deleted items + for item in replaced_items: + created_collection.delete_item(item=item, partition_key=item['pk']) + query_iterable = created_collection.query_items_change_feed( + continuation=cont_token3, + mode=mode, + ) + + expected_change_feeds = [] + for i in range(4): + expected_change = { + CURRENT: {}, + METADATA: {OPERATION_TYPE: DELETE, ID: f'doc{i}'}, + } + if setup["is_emulator"]: + expected_change[PREVIOUS] = { + ID: f'doc{i}', + partition_key: f'pk{i}', + VERSION: 2, + } + expected_change_feeds.append(expected_change) actual_change_feeds = list(query_iterable) assert_change_feed(expected_change_feeds, actual_change_feeds) - ## Test change_feed for created/deleted items + ## Test change_feed for created/replaced/deleted items query_iterable = created_collection.query_items_change_feed( continuation=cont_token1, mode = mode ) expected_change_feeds = [{CURRENT: {ID: f'doc{i}'}, METADATA: {OPERATION_TYPE: CREATE}} for i in range(4)]\ - + [{CURRENT: {}, PREVIOUS: {ID: f'doc{i}'}, METADATA: {OPERATION_TYPE: DELETE}} for i in range(4)] + + [{CURRENT: {ID: f'doc{i}', VERSION: 2}, + METADATA: {OPERATION_TYPE: REPLACE}} for i in range(4)]\ + + expected_change_feeds actual_change_feeds = list(query_iterable) assert_change_feed(expected_change_feeds, actual_change_feeds) @@ -169,6 +246,49 @@ def test_query_change_feed_all_versions_and_deletes(self, setup): actual_change_feeds = list(query_iterable) assert_change_feed(expected_change_feeds, actual_change_feeds) + @pytest.mark.timeout(TTL_TEST_TIMEOUT_SECONDS + 50) + def test_query_change_feed_ttl_delete(self, setup): + if not setup["is_emulator"]: + pytest.skip("TTL expiration timing is validated only against the emulator.") + + partition_key = 'pk' + cid = "change_feed_ttl_test_" + str(uuid.uuid4()) + setup["key_db"].create_container( + cid, + PartitionKey(path=f"/{partition_key}"), + default_ttl=-1, + change_feed_policy={"retentionDuration": 10}, + ) + created_collection = setup["created_db"].get_container_client(cid) + + list(created_collection.query_items_change_feed(mode='AllVersionsAndDeletes')) + continuation = created_collection.client_connection.last_response_headers[E_TAG] + created_collection.create_item( + body={ID: 'ttl-item', partition_key: 'ttl-pk', 'ttl': TTL_SECONDS} + ) + + ttl_delete = None + deadline = time.monotonic() + TTL_TEST_TIMEOUT_SECONDS + while time.monotonic() < deadline: + changes = list(created_collection.query_items_change_feed(continuation=continuation)) + continuation = created_collection.client_connection.last_response_headers[E_TAG] + ttl_delete = next( + ( + change for change in changes + if change[METADATA][OPERATION_TYPE] == DELETE + and change[METADATA].get("timeToLiveExpired") is True + ), + None, + ) + if ttl_delete is not None: + break + time.sleep(1) + + assert ttl_delete is not None, "Timed out waiting for the TTL delete change." + assert ttl_delete[METADATA][ID] == 'ttl-item' + assert ttl_delete[METADATA]["partitionKey"] == {partition_key: 'ttl-pk'} + assert ttl_delete.get(PREVIOUS) is None + def test_query_change_feed_all_versions_and_deletes_errors(self, setup): cid = "change_feed_test_" + str(uuid.uuid4()) # Container creation is control-plane and uses key-auth key_db. diff --git a/sdk/cosmos/azure-cosmos/tests/test_change_feed_all_versions_async.py b/sdk/cosmos/azure-cosmos/tests/test_change_feed_all_versions_async.py index 5c996327e039..0794e2553595 100644 --- a/sdk/cosmos/azure-cosmos/tests/test_change_feed_all_versions_async.py +++ b/sdk/cosmos/azure-cosmos/tests/test_change_feed_all_versions_async.py @@ -1,15 +1,19 @@ # The MIT License (MIT) # Copyright (c) Microsoft Corporation. All rights reserved. +import asyncio +import time import unittest import uuid from datetime import datetime, timedelta, timezone +from types import SimpleNamespace import pytest import pytest_asyncio import test_config from azure.cosmos.aio import CosmosClient +from azure.cosmos._change_feed.aio.change_feed_iterable import ChangeFeedIterable from azure.cosmos.partition_key import PartitionKey ID = 'id' @@ -18,8 +22,12 @@ METADATA = 'metadata' OPERATION_TYPE = 'operationType' CREATE = 'create' +REPLACE = 'replace' DELETE = 'delete' E_TAG = 'etag' +VERSION = 'version' +TTL_SECONDS = 5 +TTL_TEST_TIMEOUT_SECONDS = 250 @pytest_asyncio.fixture() async def setup(): @@ -63,18 +71,19 @@ def _is_all_versions_and_deletes_not_enabled(error: Exception) -> bool: ) async def assert_change_feed(expected, actual): - if len(actual) == 0: - assert len(expected) == len(actual) - return - - #TODO: remove this if we can add flag to get 'previous' always - for item in actual: - if METADATA in item and item[METADATA][OPERATION_TYPE] == DELETE: - if ID in item[METADATA]: - item[PREVIOUS] = {ID: item[METADATA][ID]} + assert len(expected) == len(actual) + + # Delete entries don't have a current item, and previous is optional. + def sort_key(change): + return ( + change[METADATA][OPERATION_TYPE], + change.get(CURRENT, {}).get(ID) + or change.get(PREVIOUS, {}).get(ID) + or change[METADATA].get(ID), + ) - # Sort actual by operation_type and id - actual = sorted(actual, key=lambda k: (k[METADATA][OPERATION_TYPE], k[CURRENT][ID]) if k[METADATA][OPERATION_TYPE] == CREATE else (k[METADATA][OPERATION_TYPE], k[PREVIOUS][ID])) + expected = sorted(expected, key=sort_key) + actual = sorted(actual, key=sort_key) for expected_change_feed, actual_change_feed in zip(expected, actual): for expected_type, expected_data in expected_change_feed.items(): @@ -84,6 +93,44 @@ async def assert_change_feed(expected, actual): assert key in actual_data assert expected_data[key] == actual_data[key] + +@pytest.mark.cosmosEmulator +@pytest.mark.unittest +@pytest.mark.asyncio +class TestAllVersionsChangeFeedResponseAsync: + async def test_previous_images_are_preserved_async(self): + changes = [ + { + CURRENT: {ID: "item", "pk": "pk", VERSION: 1}, + METADATA: {OPERATION_TYPE: CREATE}, + }, + { + CURRENT: {ID: "item", "pk": "pk", VERSION: 2}, + PREVIOUS: {ID: "item", "pk": "pk", VERSION: 1, "_etag": "old-etag"}, + METADATA: {OPERATION_TYPE: REPLACE, "previousImageLSN": 1}, + }, + { + CURRENT: {}, + PREVIOUS: {ID: "item", "pk": "pk", VERSION: 2, "_etag": "new-etag"}, + METADATA: {OPERATION_TYPE: DELETE, ID: "item", "timeToLiveExpired": False}, + }, + { + CURRENT: {}, + METADATA: {OPERATION_TYPE: DELETE, ID: "expired", "timeToLiveExpired": True}, + }, + ] + iterable = object.__new__(ChangeFeedIterable) + iterable._client = SimpleNamespace(last_response_headers={E_TAG: '"continuation"'}) + + continuation, unpacked = await iterable._unpack(changes) + + assert continuation == '"continuation"' + assert unpacked is changes + assert unpacked[1][PREVIOUS][VERSION] == 1 + assert unpacked[2][PREVIOUS][VERSION] == 2 + assert PREVIOUS not in unpacked[3] + + @pytest.mark.cosmosEmulator @pytest.mark.cosmosAADLong @pytest.mark.asyncio @@ -134,28 +181,58 @@ async def test_query_change_feed_all_versions_and_deletes_async(self, setup): cont_token2 = created_collection.client_connection.last_response_headers['etag'] await assert_change_feed(expected_change_feeds, actual_change_feeds) - ## Test change_feed for deleted items + ## Test change_feed for replaced items + replaced_items = [] for item in created_items: - await created_collection.delete_item(item=item, partition_key=item['pk']) + replacement = {partition_key: item[partition_key], ID: item[ID], VERSION: 2} + replaced_items.append(await created_collection.replace_item(item=item[ID], body=replacement)) query_iterable = created_collection.query_items_change_feed( continuation=cont_token2, mode=mode, ) - expected_change_feeds = [{CURRENT: {}, PREVIOUS: {ID: f'doc{i}'}, METADATA: {OPERATION_TYPE: DELETE}} for i in - range(4)] + expected_change_feeds = [ + {CURRENT: {ID: f'doc{i}', VERSION: 2}, METADATA: {OPERATION_TYPE: REPLACE}} + for i in range(4) + ] actual_change_feeds = [item async for item in query_iterable] + cont_token3 = created_collection.client_connection.last_response_headers[E_TAG] await assert_change_feed(expected_change_feeds, actual_change_feeds) - ## Test change_feed for created/deleted items + ## Test change_feed for deleted items + for item in replaced_items: + await created_collection.delete_item(item=item, partition_key=item['pk']) + query_iterable = created_collection.query_items_change_feed( + continuation=cont_token3, + mode=mode, + ) + + expected_change_feeds = [] + for i in range(4): + expected_change = { + CURRENT: {}, + METADATA: {OPERATION_TYPE: DELETE, ID: f'doc{i}'}, + } + if setup["is_emulator"]: + expected_change[PREVIOUS] = { + ID: f'doc{i}', + partition_key: f'pk{i}', + VERSION: 2, + } + expected_change_feeds.append(expected_change) + actual_change_feeds = [item async for item in query_iterable] + await assert_change_feed(expected_change_feeds, actual_change_feeds) + + ## Test change_feed for created/replaced/deleted items query_iterable = created_collection.query_items_change_feed( continuation=cont_token1, mode=mode ) expected_change_feeds = [{CURRENT: {ID: f'doc{i}'}, METADATA: {OPERATION_TYPE: CREATE}} for i in range(4)] \ - + [{CURRENT: {}, PREVIOUS: {ID: f'doc{i}'}, METADATA: {OPERATION_TYPE: DELETE}} for i in - range(4)] + + [{CURRENT: {ID: f'doc{i}', VERSION: 2}, + METADATA: {OPERATION_TYPE: REPLACE}} for i in range(4)] \ + + expected_change_feeds actual_change_feeds = [item async for item in query_iterable] await assert_change_feed(expected_change_feeds, actual_change_feeds) @@ -178,6 +255,52 @@ async def test_query_change_feed_all_versions_and_deletes_async(self, setup): actual_change_feeds = [item async for item in query_iterable] await assert_change_feed(expected_change_feeds, actual_change_feeds) + @pytest.mark.timeout(TTL_TEST_TIMEOUT_SECONDS + 50) + async def test_query_change_feed_ttl_delete_async(self, setup): + if not setup["is_emulator"]: + pytest.skip("TTL expiration timing is validated only against the emulator.") + + partition_key = 'pk' + cid = "change_feed_ttl_test_" + str(uuid.uuid4()) + await setup["key_db"].create_container( + cid, + PartitionKey(path=f"/{partition_key}"), + default_ttl=-1, + change_feed_policy={"retentionDuration": 10}, + ) + created_collection = setup["created_db"].get_container_client(cid) + + initial_feed = created_collection.query_items_change_feed(mode='AllVersionsAndDeletes') + _ = [item async for item in initial_feed] + continuation = created_collection.client_connection.last_response_headers[E_TAG] + await created_collection.create_item( + body={ID: 'ttl-item', partition_key: 'ttl-pk', 'ttl': TTL_SECONDS} + ) + + ttl_delete = None + deadline = time.monotonic() + TTL_TEST_TIMEOUT_SECONDS + while time.monotonic() < deadline: + changes = [ + item async for item in created_collection.query_items_change_feed(continuation=continuation) + ] + continuation = created_collection.client_connection.last_response_headers[E_TAG] + ttl_delete = next( + ( + change for change in changes + if change[METADATA][OPERATION_TYPE] == DELETE + and change[METADATA].get("timeToLiveExpired") is True + ), + None, + ) + if ttl_delete is not None: + break + await asyncio.sleep(1) + + assert ttl_delete is not None, "Timed out waiting for the TTL delete change." + assert ttl_delete[METADATA][ID] == 'ttl-item' + assert ttl_delete[METADATA]["partitionKey"] == {partition_key: 'ttl-pk'} + assert ttl_delete.get(PREVIOUS) is None + async def test_query_change_feed_all_versions_and_deletes_errors_async(self, setup): cid = "change_feed_test_" + str(uuid.uuid4()) # Container creation is control-plane and uses key-auth key_db.