Live tests for Cloud Shell managed identity

sdk/identity/azure-identity/dev_requirements.txt

@@ -1,3 +1,5 @@
 ../../core/azure-core
-aiohttp;python_full_version>="3.5.2"
+aiohttp;python_full_version>="3.5.3"
+pytest
+pytest-asyncio;python_full_version>="3.5.3"
 typing_extensions>=3.7.2

sdk/identity/azure-identity/tests/managed-identity-live/cloudshell.md

@@ -0,0 +1,92 @@
+# Testing azure-identity in Azure Cloud Shell
+
+# Open Azure Cloud Shell
+https://shell.azure.com/
+
+# Create an Azure Key Vault
+
+## set environment variables to simplify copy-pasting
+- RESOURCE_GROUP
+  - name of an Azure resource group
+  - must be unique in the Azure subscription
+  - e.g. 'cloudshell-identity-test'
+- KEY_VAULT_NAME
+  - 3-24 alphanumeric characters
+  - must begin with a letter
+  - must be globally unique
+
+## create a resource group
+```sh
+az group create -n $RESOURCE_GROUP --location westus2
+```
+
+## create the Key Vault
+```sh
+az keyvault create -g $RESOURCE_GROUP -n $KEY_VAULT_NAME --sku standard
+```
+
+The tests expect the vault's URI in an environment variable:
+```sh
+export AZURE_IDENTITY_TEST_VAULT_URL=$(az keyvault show -g $RESOURCE_GROUP -n $KEY_VAULT_NAME --query properties.vaultUri | tr -d '"')
+```
+
+# Run the tests
+
+## Acquire the latest code
+This may take several minutes:
+```sh
+git clone https://github.com/azure/azure-sdk-for-python --single-branch --branch master --depth 1
+```
+
+## Change working directory
+```sh
+cd azure-sdk-for-python/sdk/identity/azure-identity
+```
+
+
+## Create virtual environments
+The Azure SDK supports Python 2.7 and 3.5.3+. Python 2 and 3 should be installed
+in your Cloud Shell. However, the Python 3 version may be less than 3.5.3. Check
+this with `python3 -V`. If the version is less than 3.5.3, run tests with Python
+2 only.
+
+### Python 2.7
+```sh
+virtualenv -p python2 ~/venv2
+```
+
+### Python 3
+If your shell has at least Python 3.5.3 available, create a virtual environment
+for it:
+```sh
+virtualenv -p python3 ~/venv3
+```
+
+## For each virtual environment:
+
+### Activate
+For example:
+```sh
+source ~/venv2/bin/activate
+```
+
+### Install packages
+```sh
+pip install -r dev_requirements.txt .
+```
+
+### Run tests
+```sh
+pytest ./tests -vrs -m cloudshell
+```
+
+### Deactivate
+```sh
+deactivate
+```
+
+# Delete Azure resources
+After running tests, delete the resources provisioned earlier:
+```sh
+az group delete -n $RESOURCE_GROUP -y --no-wait
+```

sdk/identity/azure-identity/tests/managed-identity-live/conftest.py

@@ -5,6 +5,7 @@
 import os
 import sys
 
+from azure.identity._constants import EnvironmentVariables
 import pytest
 
 if sys.version_info < (3, 5, 3):
@@ -14,6 +15,10 @@
 AZURE_IDENTITY_TEST_MANAGED_IDENTITY_CLIENT_ID = "AZURE_IDENTITY_TEST_MANAGED_IDENTITY_CLIENT_ID"
 
 
+def pytest_configure(config):
+    config.addinivalue_line("markers", "cloudshell: test requires a Cloud Shell environment")
+
+
 @pytest.fixture()
 def live_managed_identity_config():
     """Live managed identity tests interact with a service to verify the credential acquires a valid access token.
@@ -34,3 +39,17 @@ def live_managed_identity_config():
         pytest.skip("this test requires azure-keyvault-secrets")
     except KeyError:
         pytest.skip("this test requires a Key Vault URL in $" + AZURE_IDENTITY_TEST_VAULT_URL)
+
+
+@pytest.fixture()
+def cloud_shell():
+    """Cloud Shell MSI is distinguished by a value for MSI_ENDPOINT but not MSI_SECRET."""
+
+    if EnvironmentVariables.MSI_ENDPOINT not in os.environ or EnvironmentVariables.MSI_SECRET in os.environ:
+        pytest.skip("Cloud Shell MSI unavailable")
+        return
+
+    try:
+        return {"vault_url": os.environ[AZURE_IDENTITY_TEST_VAULT_URL]}
+    except KeyError:
+        pytest.skip("this test requires a Key Vault URL in $" + AZURE_IDENTITY_TEST_VAULT_URL)

sdk/identity/azure-identity/tests/managed-identity-live/test_cloud_shell.py

@@ -0,0 +1,25 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+import pytest
+
+from azure.core import PipelineClient
+from azure.core.pipeline.policies import ContentDecodePolicy, HttpLoggingPolicy, RedirectPolicy, RetryPolicy
+from azure.identity import ManagedIdentityCredential
+
+
+@pytest.mark.cloudshell
+def test_cloud_shell_live(cloud_shell):
+    credential = ManagedIdentityCredential()
+    token = credential.get_token("https://vault.azure.net")
+
+    # Validate the token by sending a request to the Key Vault. The request is manual because azure-keyvault-secrets
+    # can't authenticate in Cloud Shell; the MSI endpoint there doesn't support AADv2 scopes.
+    policies = [ContentDecodePolicy(), RedirectPolicy(), RetryPolicy(), HttpLoggingPolicy()]
+    client = PipelineClient(cloud_shell["vault_url"], policies=policies)
+    list_secrets = client.get(
+        "secrets", headers={"Authorization": "Bearer " + token.token}, params={"api-version": "7.0"}
+    )
+    with client:
+        client._pipeline.run(list_secrets)

sdk/identity/azure-identity/tests/managed-identity-live/test_cloud_shell_async.py

@@ -0,0 +1,26 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+import pytest
+
+from azure.core import AsyncPipelineClient
+from azure.core.pipeline.policies import ContentDecodePolicy, HttpLoggingPolicy, AsyncRedirectPolicy, AsyncRetryPolicy
+from azure.identity.aio import ManagedIdentityCredential
+
+
+@pytest.mark.cloudshell
+@pytest.mark.asyncio
+async def test_cloud_shell_live(cloud_shell):
+    credential = ManagedIdentityCredential()
+    token = credential.get_token("https://vault.azure.net")
+
+    # Validate the token by sending a request to the Key Vault. The request is manual because azure-keyvault-secrets
+    # can't authenticate in Cloud Shell; the MSI endpoint there doesn't support AADv2 scopes.
+    policies = [ContentDecodePolicy(), AsyncRedirectPolicy(), AsyncRetryPolicy(), HttpLoggingPolicy()]
+    client = AsyncPipelineClient(cloud_shell["vault_url"], policies=policies)
+    list_secrets = client.get(
+        "secrets", headers={"Authorization": "Bearer " + token.token}, params={"api-version": "7.0"}
+    )
+    async with client:
+        await client._pipeline.run(list_secrets)