test: add Cosmos DB extension tests (#1898)

* test: add Cosmos DB extension tests

* Add CosmosDB extension tests

Author: hallvictoria

workers/pyproject.toml

@@ -85,6 +85,7 @@ test-http-v2 = [
 ]
 test-deferred-bindings = [
     "azurefunctions-extensions-bindings-blob==1.1.2",
+    "azurefunctions-extensions-bindings-cosmosdb==1.0.0b1",
     "azurefunctions-extensions-bindings-eventhub==1.0.0b1; python_version < '3.14'"
 ]
 

workers/tests/emulator_tests/cosmosdb_functions/cosmosdb_functions_sdk/function_app.py

@@ -0,0 +1,55 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+import logging
+
+import azure.functions as func
+import azurefunctions.extensions.bindings.cosmosdb as cosmos
+
+app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)
+
+
+@app.route(route="cosmos")
+@app.cosmos_db_input(
+    arg_name="client",
+    connection="AzureWebJobsCosmosDBConnectionString",
+    database_name=None,
+    container_name=None)
+def cosmos_client_input(req: func.HttpRequest,
+                        client: cosmos.CosmosClient) -> str:
+    databases = client.list_databases()
+    for database in databases:
+        logging.info("Found database with ID: %s", database.get('id'))
+
+    return 'ok'
+
+
+@app.route(route="container")
+@app.cosmos_db_input(
+    arg_name="container",
+    connection="AzureWebJobsCosmosDBConnectionString",
+    database_name="test",
+    container_name="items")
+def container_proxy_input(req: func.HttpRequest,
+                          container: cosmos.ContainerProxy) -> str:
+    documents = container.query_items(
+        query="SELECT * FROM c",
+        enable_cross_partition_query=True)
+    for document in documents:
+        logging.info("Found document: %s", document)
+
+    return 'ok'
+
+
+@app.route(route="database")
+@app.cosmos_db_input(
+    arg_name="database",
+    connection="AzureWebJobsCosmosDBConnectionString",
+    database_name="test",
+    container_name=None)
+def database_proxy_input(req: func.HttpRequest,
+                         database: cosmos.DatabaseProxy) -> str:
+    containers = database.list_containers()
+    for container in containers:
+        logging.info("Found container with ID: %s", container.get('id'))
+
+    return 'ok'

workers/tests/emulator_tests/test_cosmosdb_functions.py

@@ -108,3 +108,30 @@ class TestCosmosDBFunctionsSteinGeneric(TestCosmosDBFunctions):
     def get_script_dir(cls):
         return testutils.EMULATOR_TESTS_FOLDER / 'cosmosdb_functions' / \
             'cosmosdb_functions_stein' / 'generic'
+
+
+class TestCosmosDBSDKFunctions(testutils.WebHostTestCase):
+
+    @classmethod
+    def get_script_dir(cls):
+        return testutils.EMULATOR_TESTS_FOLDER / 'cosmosdb_functions' / \
+            'cosmosdb_functions_sdk'
+
+    @classmethod
+    def get_libraries_to_install(cls):
+        return ['azurefunctions-extensions-bindings-cosmosdb']
+
+    def test_cosmos_client_input(self):
+        r = self.webhost.request('GET', 'cosmos')
+        self.assertEqual(r.status_code, 200)
+        self.assertEqual(r.text, 'ok')
+
+    def test_container_proxy_input(self):
+        r = self.webhost.request('GET', 'container')
+        self.assertEqual(r.status_code, 200)
+        self.assertEqual(r.text, 'ok')
+
+    def test_database_proxy_input(self):
+        r = self.webhost.request('GET', 'database')
+        self.assertEqual(r.status_code, 200)
+        self.assertEqual(r.text, 'ok')