Migrate ChangeFeedTest to rx.TestSuiteBase and add internal API methods

Co-authored-by: xinlian12 <64233642+xinlian12@users.noreply.github.com>

Author: Copilot

sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/rx/ChangeFeedTest.java

@@ -4,6 +4,7 @@
 
 import com.azure.cosmos.BridgeInternal;
 import com.azure.cosmos.ConnectionMode;
+import com.azure.cosmos.ConsistencyLevel;
 import com.azure.cosmos.CosmosDiagnostics;
 import com.azure.cosmos.implementation.AsyncDocumentClient;
 import com.azure.cosmos.implementation.ClientSideRequestStatistics;
@@ -15,7 +16,6 @@
 import com.azure.cosmos.implementation.RequestOptions;
 import com.azure.cosmos.implementation.Resource;
 import com.azure.cosmos.implementation.ResourceResponse;
-import com.azure.cosmos.implementation.TestSuiteBase;
 import com.azure.cosmos.implementation.TestUtils;
 import com.azure.cosmos.implementation.Utils;
 import com.azure.cosmos.implementation.feedranges.FeedRangeEpkImpl;
@@ -58,7 +58,6 @@
 import static java.lang.annotation.ElementType.METHOD;
 import static org.assertj.core.api.Assertions.assertThat;
 
-//TODO: change to use external TestSuiteBase
 public class ChangeFeedTest extends TestSuiteBase {
 
     private static final int SETUP_TIMEOUT = 40000;
@@ -93,7 +92,7 @@ static protected DocumentCollection getCollectionDefinition(boolean enableFullFi
     }
 
     public ChangeFeedTest() {
-        super(createGatewayRxDocumentClient());
+        super(createInternalGatewayRxDocumentClient(ConsistencyLevel.SESSION, false, null, true));
         subscriberValidationTimeout = TIMEOUT;
     }
 
@@ -575,7 +574,7 @@ void populateDocumentsInternal(Method method, boolean enableFullFidelity) {
     public void before_ChangeFeedTest() {
         // set up the client
         client = clientBuilder().build();
-        createdDatabase = SHARED_DATABASE;
+        createdDatabase = SHARED_DATABASE_FOR_INTERNAL_CLIENT;
     }
 
     @AfterClass(groups = { "query", "emulator" }, timeOut = SHUTDOWN_TIMEOUT, alwaysRun = true)

sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/rx/TestSuiteBase.java

@@ -2080,4 +2080,96 @@ protected static void truncateCollection(DocumentCollection collection) {
         }
     }
 
+    /**
+     * Deletes a document using AsyncDocumentClient.
+     * @param client the AsyncDocumentClient
+     * @param documentLink the document link
+     * @param partitionKey the partition key
+     * @param etag the etag (can be null)
+     */
+    public static void deleteDocument(AsyncDocumentClient client, String documentLink, PartitionKey partitionKey, String etag) {
+        RequestOptions options = new RequestOptions();
+        options.setPartitionKey(partitionKey);
+        if (etag != null) {
+            options.setIfMatchETag(etag);
+        }
+        try {
+            client.deleteDocument(documentLink, options).block();
+        } catch (Exception e) {
+            // Ignore ResourceNotFoundException if document was already deleted, or other errors during cleanup
+        }
+    }
+
+    /**
+     * Bulk inserts documents using AsyncDocumentClient.
+     * @param client the AsyncDocumentClient
+     * @param collectionLink the collection link
+     * @param documents the documents to insert
+     * @param <T> the document type
+     */
+    public static <T extends Resource> void bulkInsert(AsyncDocumentClient client, String collectionLink, List<T> documents) {
+        bulkInsert(client, collectionLink, documents, DEFAULT_BULK_INSERT_CONCURRENCY_LEVEL);
+    }
+
+    /**
+     * Bulk inserts documents using AsyncDocumentClient with concurrency level.
+     * @param client the AsyncDocumentClient
+     * @param collectionLink the collection link
+     * @param documents the documents to insert
+     * @param concurrencyLevel the concurrency level
+     * @param <T> the document type
+     */
+    public static <T extends Resource> void bulkInsert(AsyncDocumentClient client, String collectionLink, List<T> documents, int concurrencyLevel) {
+        Flux.fromIterable(documents)
+            .flatMap(doc -> client.createDocument(collectionLink, doc, new RequestOptions(), false), concurrencyLevel)
+            .collectList()
+            .block();
+    }
+
+    /**
+     * Truncates a DocumentCollection using AsyncDocumentClient.
+     * @param client the AsyncDocumentClient
+     * @param collection the DocumentCollection
+     */
+    protected static void truncateCollection(AsyncDocumentClient client, DocumentCollection collection) {
+        logger.info("Truncating DocumentCollection {} ...", collection.getId());
+
+        String collectionLink = collection.getSelfLink();
+        if (collectionLink == null) {
+            collectionLink = TestUtils.getCollectionNameLink(
+                collection.getAltLink().split("/")[1], // dbs/{dbId}/colls/{collId} -> dbId
+                collection.getId());
+        }
+
+        CosmosQueryRequestOptions options = new CosmosQueryRequestOptions();
+        options.setMaxDegreeOfParallelism(-1);
+
+        QueryFeedOperationState dummyState = TestUtils.createDummyQueryFeedOperationState(
+            ResourceType.Document,
+            OperationType.Query,
+            options,
+            client);
+
+        try {
+            String query = "SELECT * FROM c";
+            final String finalCollectionLink = collectionLink;
+            List<Document> docs = client.queryDocuments(finalCollectionLink, query, dummyState, Document.class)
+                .flatMap(page -> Flux.fromIterable(page.getResults()))
+                .collectList()
+                .block();
+
+            if (docs != null) {
+                for (Document doc : docs) {
+                    try {
+                        client.deleteDocument(doc.getSelfLink(), new RequestOptions()).block();
+                    } catch (Exception e) {
+                        // Ignore ResourceNotFoundException if document was already deleted, or other errors during cleanup
+                    }
+                }
+            }
+        } finally {
+            safeClose(dummyState);
+        }
+    }
+
 }